# -*- coding: utf-8 -*- """ 系统集成测试脚本 验证所有核心模块是否正常工作 """ import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) def test_data_layer(): """测试数据层""" print("=" * 50) print("测试 1: 数据层") print("=" * 50) from collectors.financial_collector import get_all_companies, get_company_by_code companies = get_all_companies() print(f" 已加载 {len(companies)} 家企业") assert len(companies) == 10, f"期望10家企业，实际{len(companies)}" c = get_company_by_code("688256") assert c is not None, "未找到寒武纪(688256)" print(f" 寒武纪: {c['short_name']}, 行业: {c['industry']}") print(f" 制裁状态: {c['compliance']['entity_list_status']}") print(" ✅ 数据层测试通过\n") def test_entity_list(): """测试实体清单匹配""" print("=" * 50) print("测试 2: 实体清单匹配") print("=" * 50) from collectors.entity_list_collector import check_entity_list, check_supply_chain_sanctions result = check_entity_list("中芯国际") print(f" 中芯国际制裁状态: {result['is_sanctioned']}, 匹配: {result['match_type']}") assert result["is_sanctioned"] == True result2 = check_entity_list("金山办公") print(f" 金山办公制裁状态: {result2['is_sanctioned']}") assert result2["is_sanctioned"] == False # 供应链制裁穿透 suppliers = ["ASML", "东京电子", "某国内供应商"] sanctions = check_supply_chain_sanctions("中芯国际", suppliers) print(f" 供应链制裁穿透: 发现 {len(sanctions)} 个受制裁供应商") print(" ✅ 实体清单测试通过\n") def test_compliance(): """测试合规数据""" print("=" * 50) print("测试 3: 算法备案合规") print("=" * 50) from collectors.compliance_collector import check_algo_filing, assess_algo_compliance_risk from collectors.financial_collector import get_company_by_code filing = check_algo_filing("金山办公") print(f" 金山办公算法备案: {filing['has_filing']}, 备案数: {len(filing['filings'])}") assert filing["has_filing"] == True company = get_company_by_code("688787") # 海天瑞声 risk = assess_algo_compliance_risk(company) print(f" 海天瑞声合规风险: {risk['overall_risk_level']}") print(" ✅ 合规数据测试通过\n") def test_knowledge_graph(): """测试知识图谱""" print("=" * 50) print("测试 4: 知识图谱构建") print("=" * 50) from knowledge_graph.graph_builder import build_graph, get_graph_stats G = build_graph() stats = get_graph_stats(G) print(f" 节点数: {stats['total_nodes']}") print(f" 边数: {stats['total_edges']}") print(f" 受制裁节点: {stats['sanctioned_nodes']}") print(f" 节点类型: {stats['node_types']}") assert stats["total_nodes"] > 20 assert stats["total_edges"] > 15 print(" ✅ 知识图谱测试通过\n") def test_contagion(): """测试供应链传染分析""" print("=" * 50) print("测试 5: 供应链风险传染分析") print("=" * 50) from knowledge_graph.graph_builder import build_graph from knowledge_graph.contagion_analyzer import analyze_contagion G = build_graph() # 中芯国际 - 应该有大量直接风险 result = analyze_contagion(G, "中芯国际") print(f" 中芯国际供应链风险评分: {result['risk_score']}/100") print(f" 直接风险: {len(result['direct_risks'])} 个") print(f" 间接风险: {len(result['indirect_risks'])} 个") print(f" 传染路径: {len(result['contagion_paths'])} 条") # 海尔生物 - 应该风险较低 result2 = analyze_contagion(G, "海尔生物") print(f" 海尔生物供应链风险评分: {result2['risk_score']}/100") print(" ✅ 传染分析测试通过\n") def test_risk_scorer(): """测试六维风险评分""" print("=" * 50) print("测试 6: 六维风险评分") print("=" * 50) from collectors.financial_collector import get_all_companies from risk_engine.risk_scorer import calculate_six_dimension_scores companies = get_all_companies() print(f" {'企业':<10} {'综合':>6} {'技术':>6} {'人员':>6} {'合规':>6} {'地缘':>6} {'资本化':>6} {'集中':>6}") print(" " + "-" * 60) for comp in companies: result = calculate_six_dimension_scores(comp) s = result["scores"] print(f" {comp['short_name']:<10} {result['comprehensive_score']:>5} " f"{s['tech_disruption']:>5} {s['talent_loss']:>5} " f"{s['algo_compliance']:>5} {s['geopolitical']:>5} " f"{s['rd_capitalization']:>5} {s['concentration']:>5}") print(" ✅ 风险评分测试通过\n") def test_agents(): """测试多智能体（规则引擎模式）""" print("=" * 50) print("测试 7: 多智能体辩论（规则引擎降级模式）") print("=" * 50) from collectors.financial_collector import get_company_by_code from agents.law_agent import LawAgent from agents.tech_agent import TechAgent from agents.finance_agent import FinanceAgent from agents.judge_agent import JudgeAgent company = get_company_by_code("688787") # 海天瑞声 - 高风险案例 law = LawAgent() law_result = law._rule_based_evaluation(company) print(f" 法务风险: {law_result['overall_law_risk']['score']}分 ({law_result['overall_law_risk']['level']})") tech = TechAgent() tech_result = tech._rule_based_evaluation(company) print(f" 技术风险: {tech_result['overall_tech_risk']['score']}分 ({tech_result['overall_tech_risk']['level']})") fin = FinanceAgent() fin_result = fin._rule_based_evaluation(company) print(f" 财务风险: {fin_result['overall_fin_risk']['score']}分 ({fin_result['overall_fin_risk']['level']})") judge = JudgeAgent() judge_result = judge._rule_based_evaluation(company, law_result, tech_result, fin_result) print(f" 综合评分: {judge_result['comprehensive_score']}分") print(f" 核保决策: {judge_result['underwriting_decision']}") print(f" 关键风险: {judge_result['key_risks']}") print(" ✅ 多智能体测试通过\n") def test_dynamic_pricing(): """测试动态定价""" print("=" * 50) print("测试 8: 动态保险定价") print("=" * 50) from risk_engine.dynamic_pricing import calculate_all_products # 高风险企业 scores_high = { "tech_disruption": 80, "talent_loss": 75, "algo_compliance": 60, "geopolitical": 90, "rd_capitalization": 45, "concentration": 85, } results = calculate_all_products(75, scores_high, "芯片", 1_170_000_000) print(" 寒武纪（高风险）保费方案:") for r in results: print(f" {r['product_name']}: 基础¥{r['base_premium']:,.0f} -> 最终¥{r['final_premium']:,.0f} " f"(倍率{r['risk_multiplier']:.2f}) {'✅可保' if r['is_insurable'] else '❌拒保'}") # 低风险企业 scores_low = { "tech_disruption": 30, "talent_loss": 25, "algo_compliance": 15, "geopolitical": 18, "rd_capitalization": 10, "concentration": 20, } results2 = calculate_all_products(20, scores_low, "医疗器械", 2_580_000_000) print(" 南微医学（低风险）保费方案:") for r in results2: print(f" {r['product_name']}: 基础¥{r['base_premium']:,.0f} -> 最终¥{r['final_premium']:,.0f} " f"(倍率{r['risk_multiplier']:.2f})") print(" ✅ 动态定价测试通过\n") if __name__ == "__main__": print("\n🛡️ 科创企业智能风控系统 - 集成测试\n") test_data_layer() test_entity_list() test_compliance() test_knowledge_graph() test_contagion() test_risk_scorer() test_agents() test_dynamic_pricing() print("=" * 50) print("🎉 所有测试通过！系统可以正常运行。") print("=" * 50) print("\n启动可视化系统: streamlit run app.py")