149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
六维风险评分器
|
|
整合所有数据源和分析结果，生成结构化六维风险画像
|
|
"""
|
|
import logging
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# 风险等级映射
|
|
RISK_LEVELS = {
|
|
(0, 30): {"level": "低", "color": "#4CAF50", "emoji": "🟢"},
|
|
(30, 50): {"level": "中低", "color": "#8BC34A", "emoji": "🟡"},
|
|
(50, 70): {"level": "中高", "color": "#FF9800", "emoji": "🟠"},
|
|
(70, 90): {"level": "高", "color": "#F44336", "emoji": "🔴"},
|
|
(90, 101): {"level": "极高", "color": "#B71C1C", "emoji": "⛔"},
|
|
}
|
|
|
|
|
|
def get_risk_level(score: int) -> dict:
|
|
"""根据分数获取风险等级详情"""
|
|
for (low, high), info in RISK_LEVELS.items():
|
|
if low <= score < high:
|
|
return info
|
|
return {"level": "未知", "color": "#9E9E9E", "emoji": "❓"}
|
|
|
|
|
|
def calculate_six_dimension_scores(company_data: dict) -> dict:
|
|
"""
|
|
基于企业数据直接计算六维风险评分（不依赖 LLM）
|
|
用于快速预览和规则引擎降级场景
|
|
"""
|
|
financials = company_data.get("financials", {})
|
|
compliance = company_data.get("compliance", {})
|
|
tech_route = company_data.get("tech_route", {})
|
|
personnel = company_data.get("core_tech_personnel", [])
|
|
supply_chain = company_data.get("supply_chain", {})
|
|
|
|
scores = {}
|
|
|
|
# 1. 技术路线颠覆风险
|
|
competing = len(tech_route.get("competing_techs", []))
|
|
moat = tech_route.get("tech_moat", "")
|
|
tech_score = min(25 + competing * 15, 85)
|
|
if "差距" in moat or "受制" in moat or "威胁" in moat:
|
|
tech_score = min(tech_score + 15, 95)
|
|
if "领先" in moat or "第一" in moat:
|
|
tech_score = max(tech_score - 10, 10)
|
|
scores["tech_disruption"] = tech_score
|
|
|
|
# 2. 核心人员流失风险
|
|
departed = [p for p in personnel if "离职" in p.get("status", "")]
|
|
high_imp = [p for p in personnel if p.get("importance") == "极高"]
|
|
if departed:
|
|
scores["talent_loss"] = 80
|
|
elif len(high_imp) <= 1 and len(personnel) <= 2:
|
|
scores["talent_loss"] = 55
|
|
else:
|
|
scores["talent_loss"] = 25
|
|
|
|
# 3. 算法/数据合规风险
|
|
algo_status = compliance.get("algo_filing_status", "")
|
|
data_risk = compliance.get("data_export_risk", "低")
|
|
algo_score = 20
|
|
if "未" in algo_status:
|
|
algo_score = 75
|
|
elif data_risk == "高":
|
|
algo_score = 65
|
|
elif data_risk == "中":
|
|
algo_score = 40
|
|
elif "已备案" in algo_status:
|
|
algo_score = 15
|
|
scores["algo_compliance"] = algo_score
|
|
|
|
# 4. 地缘政治/出口管制风险
|
|
entity_status = compliance.get("entity_list_status", "")
|
|
if "被列入" in entity_status:
|
|
scores["geopolitical"] = 92
|
|
elif supply_chain.get("supplier_concentration_risk") == "极高":
|
|
scores["geopolitical"] = 68
|
|
elif supply_chain.get("supplier_concentration_risk") == "高":
|
|
scores["geopolitical"] = 50
|
|
else:
|
|
scores["geopolitical"] = 18
|
|
|
|
# 5. 研发资本化操纵风险
|
|
cap_rate = financials.get("rd_capitalization_rate", 0)
|
|
if cap_rate >= 0.4:
|
|
scores["rd_capitalization"] = 90
|
|
elif cap_rate >= 0.3:
|
|
scores["rd_capitalization"] = 72
|
|
elif cap_rate >= 0.15:
|
|
scores["rd_capitalization"] = 48
|
|
elif cap_rate > 0:
|
|
scores["rd_capitalization"] = 25
|
|
else:
|
|
scores["rd_capitalization"] = 10
|
|
|
|
# 6. 客户/供应商集中风险
|
|
cust_ratio = financials.get("top5_customer_ratio", 0)
|
|
supp_ratio = financials.get("top5_supplier_ratio", 0)
|
|
max_conc = max(cust_ratio, supp_ratio)
|
|
if max_conc >= 0.8:
|
|
scores["concentration"] = 88
|
|
elif max_conc >= 0.6:
|
|
scores["concentration"] = 68
|
|
elif max_conc >= 0.4:
|
|
scores["concentration"] = 42
|
|
else:
|
|
scores["concentration"] = 18
|
|
|
|
# 加权综合
|
|
weights = {
|
|
"tech_disruption": 0.20,
|
|
"talent_loss": 0.15,
|
|
"algo_compliance": 0.15,
|
|
"geopolitical": 0.20,
|
|
"rd_capitalization": 0.15,
|
|
"concentration": 0.15,
|
|
}
|
|
comprehensive = int(sum(scores[k] * weights[k] for k in scores))
|
|
|
|
# 维度中文名映射
|
|
dim_names = {
|
|
"tech_disruption": "技术路线颠覆",
|
|
"talent_loss": "核心人员流失",
|
|
"algo_compliance": "算法/数据合规",
|
|
"geopolitical": "地缘政治/出口管制",
|
|
"rd_capitalization": "研发资本化操纵",
|
|
"concentration": "客户/供应商集中",
|
|
}
|
|
|
|
return {
|
|
"scores": scores,
|
|
"comprehensive_score": comprehensive,
|
|
"risk_level": get_risk_level(comprehensive),
|
|
"dimension_details": {
|
|
k: {
|
|
"name": dim_names[k],
|
|
"score": v,
|
|
"level": get_risk_level(v),
|
|
"weight": weights[k],
|
|
}
|
|
for k, v in scores.items()
|
|
},
|
|
}
|