206 lines
8.0 KiB
Python
206 lines
8.0 KiB
Python
# -*- 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")
|