Files
mini-openclaw-v2/backend/kaiju_history_data.py
T
qiaoqichaoandClaude Code d56c699ce6 chore(v2): 从兴邦预审v2白名单抽取独立可运行基座
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单,
从 55e2cdb 抽取 224 个文件(2.48 MB)到独立目录 F:/mini-OpenClaw-v2。

- 排除:凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等
  4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/
  ori_file)、Office 临时文件
- 目标目录独立 git 仓库,初始恢复点;Gitea 远程与分支待 OQ-06 Q2 定案
- 未部署、未启动服务、未调用模型

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-29 12:58:26 +08:00

318 lines
13 KiB
Python

"""兴邦金租历史评审数据源(虚拟示例数据,2024-2026年)
用于智能分配算法的数据驱动决策:
- 主审员产能画像(平均评审天数、行业专长、质量分)
- 历史项目追溯(按主审员/行业查询)
- 多维度分配推荐依据
"""
import random
from datetime import datetime, timedelta
from typing import List, Dict, Optional
# ============ 历史评审项目数据结构 ============
class HistoricalProject:
"""历史评审项目记录"""
def __init__(
self,
project_id: str,
project_name: str,
industry: str,
credit_amount: float, # 万元
complexity: int, # 1-5,难度
planned_days: int, # 计划评审天数
actual_days: int, # 实际评审天数
reviewer: str, # 主审员
outcome: str, # 通过/否决/撤并
quality_score: int, # 0-100,历史质量分
review_date: str, # YYYY-MM-DD
):
self.project_id = project_id
self.project_name = project_name
self.industry = industry
self.credit_amount = credit_amount
self.complexity = complexity
self.planned_days = planned_days
self.actual_days = actual_days
self.reviewer = reviewer
self.outcome = outcome
self.quality_score = quality_score
self.review_date = review_date
def to_dict(self) -> dict:
return {
"project_id": self.project_id,
"project_name": self.project_name,
"industry": self.industry,
"credit_amount": self.credit_amount,
"complexity": self.complexity,
"planned_days": self.planned_days,
"actual_days": self.actual_days,
"reviewer": self.reviewer,
"outcome": self.outcome,
"quality_score": self.quality_score,
"review_date": self.review_date,
}
# ============ 虚拟数据生成(体现真实规律)============
def _generate_historical_data() -> List[HistoricalProject]:
"""生成2024-2026年虚拟历史评审数据
数据规律设计:
- 李明:制造业专长,效率高(实际天数<计划),质量分85-95
- 王芳:新能源专长,质量高(90-98),但项目复杂度高时需时较长
- 张强:交通运输专长,负荷较重(项目多),质量稳定80-90
- 刘洋:医疗健康专长,细致但速度慢(实际天数略超计划),质量92-98
- 陈晨:设备租赁专长,全行业覆盖,效率中等,质量78-88
- 赵磊:环保行业专长,项目少但质量高,88-96
"""
projects = []
project_counter = 1
# 主审员配置:(姓名, 主行业, 次要行业, 项目数, 效率倍数, 质量基数, 质量波动)
reviewers_config = [
("李明", "制造业", ["装备制造", "机械设备", "汽车零部件"], 22, 0.85, 88, 7),
("王芳", "新能源", ["光伏设备", "风电设备", "新能源汽车"], 20, 1.05, 92, 6),
("张强", "交通运输", ["物流运输", "港口码头", "公路客运"], 25, 0.95, 84, 8),
("刘洋", "医疗健康", ["医疗器械", "医药制造", "医院设备"], 18, 1.15, 94, 4),
("陈晨", "设备租赁", ["工程机械", "办公设备", "生产设备"], 20, 1.0, 82, 8),
("赵磊", "环保", ["环保设备", "污水处理", "固废处理"], 16, 1.08, 90, 6),
]
# 行业项目名称模板
industry_templates = {
"制造业": ["精密机械加工设备", "自动化生产线", "数控机床", "模具制造设备"],
"装备制造": ["重型装备", "专用设备", "成套装备"],
"机械设备": ["工业机器人", "包装机械", "印刷设备"],
"汽车零部件": ["汽车配件生产线", "发动机零部件", "车身冲压设备"],
"新能源": ["分布式光伏电站", "储能设备", "充电桩建设"],
"光伏设备": ["光伏组件生产线", "硅片切割设备", "光伏发电项目"],
"风电设备": ["风力发电机组", "风电场建设", "海上风电项目"],
"新能源汽车": ["新能源客车", "电动物流车", "充电设施"],
"交通运输": ["货运车队", "冷链物流车辆", "运输设备更新"],
"物流运输": ["智能仓储设备", "配送车辆", "物流信息系统"],
"港口码头": ["港口装卸设备", "集装箱堆场", "码头机械"],
"公路客运": ["客运班车", "旅游大巴", "公交车辆"],
"医疗健康": ["医学影像设备", "手术器械", "检验检测设备"],
"医疗器械": ["CT设备", "超声设备", "监护仪"],
"医药制造": ["制药生产线", "GMP车间改造", "原料药设备"],
"医院设备": ["病床护理设备", "消毒灭菌设备", "医用气体系统"],
"设备租赁": ["工程机械租赁", "办公设备租赁", "生产设备租赁"],
"工程机械": ["挖掘机", "起重机", "推土机"],
"办公设备": ["复印打印设备", "办公家具", "会议系统"],
"生产设备": ["注塑机", "冲压设备", "焊接设备"],
"环保": ["环保治理设备", "节能改造项目", "清洁能源设备"],
"环保设备": ["废气处理设备", "除尘设备", "脱硫脱硝装置"],
"污水处理": ["污水处理站", "中水回用系统", "污泥处理设备"],
"固废处理": ["垃圾焚烧设备", "危废处理设施", "资源回收系统"],
}
# 生成每位主审员的历史项目
start_date = datetime(2024, 1, 1)
end_date = datetime(2026, 8, 31)
for reviewer_name, main_industry, sub_industries, num_projects, efficiency, quality_base, quality_range in reviewers_config:
# 该主审员的行业池(主行业占70%,次要行业占30%)
industries_pool = [main_industry] * 7 + sub_industries * 3
for i in range(num_projects):
# 随机日期(制造业、新能源集中在2-6月、9-11月)
if main_industry in ["制造业", "新能源", "装备制造"]:
# 集中在业务旺季
month_choices = [2, 3, 4, 5, 6, 9, 10, 11]
month = random.choice(month_choices)
else:
month = random.randint(1, 12)
year = random.choice([2024, 2025, 2026])
if year == 2026 and month > 8:
month = random.randint(1, 8)
day = random.randint(1, 28)
review_date = datetime(year, month, day)
# 行业选择(主行业为主)
industry = random.choice(industries_pool)
# 项目名称
company_prefix = random.choice([
"阳光", "华润", "中联", "鑫盛", "宏泰", "瑞丰", "恒通", "盛达",
"金辉", "博瑞", "创新", "天成", "嘉诚", "永安", "兴旺", "腾飞"
])
equipment_name = random.choice(industry_templates.get(industry, ["设备租赁项目"]))
project_name = f"{company_prefix}{equipment_name}融资租赁项目"
# 授信金额(万元):复杂度越高金额越大
complexity = random.choices([1, 2, 3, 4, 5], weights=[5, 15, 40, 30, 10])[0]
if complexity == 1:
credit_amount = random.randint(500, 1500)
elif complexity == 2:
credit_amount = random.randint(1200, 3000)
elif complexity == 3:
credit_amount = random.randint(2500, 6000)
elif complexity == 4:
credit_amount = random.randint(5000, 10000)
else: # 5
credit_amount = random.randint(8000, 20000)
# 计划评审天数(复杂度决定)
planned_days = {1: 5, 2: 7, 3: 10, 4: 12, 5: 15}[complexity]
# 实际评审天数(受主审员效率影响)
actual_days = int(planned_days * efficiency * random.uniform(0.9, 1.1))
actual_days = max(3, actual_days) # 最少3天
# 质量分(受主审员能力+复杂度影响)
# 复杂度高的项目,质量分略有波动
quality_penalty = (complexity - 3) * 2 if complexity >= 4 else 0
quality_score = quality_base + random.randint(-quality_range, quality_range) - quality_penalty
quality_score = max(60, min(100, quality_score))
# 评审结果(质量分高=通过率高)
if quality_score >= 85:
outcome = random.choices(["通过", "否决", "撤并"], weights=[90, 5, 5])[0]
elif quality_score >= 75:
outcome = random.choices(["通过", "否决", "撤并"], weights=[80, 12, 8])[0]
else:
outcome = random.choices(["通过", "否决", "撤并"], weights=[60, 25, 15])[0]
# 如果返工(实际天数超出计划较多),质量分降低
if actual_days > planned_days * 1.3:
quality_score = max(quality_score - 10, 60)
project = HistoricalProject(
project_id=f"PROJ-{year}-{project_counter:04d}",
project_name=project_name,
industry=industry,
credit_amount=credit_amount,
complexity=complexity,
planned_days=planned_days,
actual_days=actual_days,
reviewer=reviewer_name,
outcome=outcome,
quality_score=quality_score,
review_date=review_date.strftime("%Y-%m-%d"),
)
projects.append(project)
project_counter += 1
# 按日期排序
projects.sort(key=lambda p: p.review_date)
return projects
# ============ 全局数据缓存 ============
_HISTORICAL_PROJECTS: List[HistoricalProject] = _generate_historical_data()
# ============ 查询接口 ============
def get_all_projects() -> List[Dict]:
"""获取所有历史项目"""
return [p.to_dict() for p in _HISTORICAL_PROJECTS]
def get_projects_by_reviewer(reviewer: str) -> List[Dict]:
"""按主审员查询历史项目"""
filtered = [p for p in _HISTORICAL_PROJECTS if p.reviewer == reviewer]
return [p.to_dict() for p in filtered]
def get_projects_by_industry(industry: str) -> List[Dict]:
"""按行业查询历史项目(模糊匹配)"""
filtered = [p for p in _HISTORICAL_PROJECTS if industry in p.industry]
return [p.to_dict() for p in filtered]
def get_projects_filtered(reviewer: Optional[str] = None, industry: Optional[str] = None) -> List[Dict]:
"""组合过滤查询"""
filtered = _HISTORICAL_PROJECTS
if reviewer:
filtered = [p for p in filtered if p.reviewer == reviewer]
if industry:
filtered = [p for p in filtered if industry in p.industry]
return [p.to_dict() for p in filtered]
def get_reviewer_stats() -> Dict[str, Dict]:
"""获取所有主审员的统计画像
返回格式:
{
"李明": {
"total_projects": 22,
"avg_actual_days": 8.5,
"avg_quality_score": 88.2,
"industry_expertise": {
"制造业": {"count": 15, "avg_quality": 90.1},
"装备制造": {"count": 7, "avg_quality": 86.3}
},
"pass_rate": 0.91
},
...
}
"""
stats = {}
reviewers = list(set(p.reviewer for p in _HISTORICAL_PROJECTS))
for reviewer in reviewers:
projects = [p for p in _HISTORICAL_PROJECTS if p.reviewer == reviewer]
# 基础统计
total_projects = len(projects)
avg_actual_days = sum(p.actual_days for p in projects) / total_projects if total_projects > 0 else 0
avg_quality = sum(p.quality_score for p in projects) / total_projects if total_projects > 0 else 0
pass_count = len([p for p in projects if p.outcome == "通过"])
pass_rate = pass_count / total_projects if total_projects > 0 else 0
# 行业专长统计
industry_expertise = {}
for project in projects:
industry = project.industry
if industry not in industry_expertise:
industry_expertise[industry] = {"count": 0, "quality_sum": 0}
industry_expertise[industry]["count"] += 1
industry_expertise[industry]["quality_sum"] += project.quality_score
# 计算每个行业的平均质量分
for industry in industry_expertise:
count = industry_expertise[industry]["count"]
quality_sum = industry_expertise[industry]["quality_sum"]
industry_expertise[industry] = {
"count": count,
"avg_quality": quality_sum / count if count > 0 else 0
}
stats[reviewer] = {
"total_projects": total_projects,
"avg_actual_days": round(avg_actual_days, 1),
"avg_quality_score": round(avg_quality, 1),
"industry_expertise": industry_expertise,
"pass_rate": round(pass_rate, 2)
}
return stats
# ============ 初始化时输出统计信息 ============
if __name__ == "__main__":
print(f"✅ 已生成 {len(_HISTORICAL_PROJECTS)} 条历史评审记录")
print("\n主审员统计:")
stats = get_reviewer_stats()
for reviewer, data in stats.items():
print(f" {reviewer}: {data['total_projects']}个项目, 平均{data['avg_actual_days']}天, 质量{data['avg_quality_score']}分")