"""兴邦金租历史评审数据源（虚拟示例数据，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']}分")