selfrelease
fad458b2a7
- UIUX 文档：填充 19 个缺口（多主体画像/健康度/AI+看板/增长域/洞察域/创始人端/OODA/助推/商密） - UIUX 文档：插入 6 个新章节（十四~十九），旧章节重编号为二十~三十一，更新目录和交叉引用 - 作业指导书 x5：导航改为 6 域分组，新增 Context Bar/工作模式/Insight Rail/决策线程/多工作区等 UI 概念 - 新建 docs/2-task-uiux.md：50 个代码落地开发任务，按 P0-P6 分优先级 + 8 Sprint 规划 - 后端/前端：大量新增模型、路由、组件（来自之前 Phase 开发）
434 lines
13 KiB
Python
434 lines
13 KiB
Python
"""健康度评分计算引擎。
|
||
|
||
基于月报结构化数据，计算多维度评分：
|
||
- 基础 4 维度：财务、经营、AI 商业化、AI 成本
|
||
- T2.9 扩展 5 维度：组织人才、产品技术、市场竞争、治理合规、融资资本
|
||
- T3.11 扩展 5 维度：协同赋能、AI 模型产品、数据合规、团队技术、客户成功
|
||
|
||
总分 = 加权平均，输出 0-100 分。
|
||
"""
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 14 维度权重配置
|
||
WEIGHTS = {
|
||
"financial": 0.15,
|
||
"operational": 0.10,
|
||
"ai_commercial": 0.10,
|
||
"ai_cost": 0.05,
|
||
"org_talent": 0.10,
|
||
"product_tech": 0.10,
|
||
"market_compete": 0.10,
|
||
"governance": 0.05,
|
||
"financing": 0.05,
|
||
"synergy": 0.05,
|
||
"ai_model_product": 0.05,
|
||
"data_compliance": 0.05,
|
||
"team_tech": 0.03,
|
||
"customer_success": 0.07,
|
||
}
|
||
|
||
|
||
def _safe_float(value: Any, default: float = 0.0) -> float:
|
||
"""安全转换为 float。"""
|
||
if value is None or value == "":
|
||
return default
|
||
try:
|
||
return float(value)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
|
||
|
||
def _calc_financial_score(data: dict[str, Any]) -> float:
|
||
"""计算财务健康度。
|
||
|
||
指标：
|
||
- 现金跑道（runway_months）：>12 月=90+，6-12=60-90，<6=<60
|
||
- 营收同比增长（yoy_change）：正=加分，负=减分
|
||
- 烧钱率趋势（burn_rate trend）：down=加分，up=减分
|
||
"""
|
||
score = 50.0 # 基础分
|
||
|
||
cash = data.get("cash_balance", {})
|
||
runway = _safe_float(cash.get("runway_months"))
|
||
if runway > 0:
|
||
if runway >= 12:
|
||
score += 30
|
||
elif runway >= 6:
|
||
score += 15
|
||
elif runway >= 3:
|
||
score -= 10
|
||
else:
|
||
score -= 30
|
||
|
||
revenue = data.get("revenue", {})
|
||
yoy = revenue.get("yoy_change", "")
|
||
if yoy:
|
||
yoy_val = _safe_float(str(yoy).replace("%", "").replace("+", ""))
|
||
if yoy_val > 0:
|
||
score += 15
|
||
elif yoy_val < 0:
|
||
score -= 15
|
||
|
||
burn = data.get("burn_rate", {})
|
||
trend = burn.get("trend", "")
|
||
if trend == "down":
|
||
score += 10
|
||
elif trend == "up":
|
||
score -= 10
|
||
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_operational_score(data: dict[str, Any]) -> float:
|
||
"""计算经营健康度。
|
||
|
||
指标：
|
||
- 团队规模变化（headcount）：净增长=加分
|
||
- 关键指标达成情况
|
||
"""
|
||
score = 60.0
|
||
|
||
headcount = data.get("headcount", {})
|
||
new_hires = _safe_float(headcount.get("new_hires"))
|
||
departures = _safe_float(headcount.get("departures"))
|
||
net_change = new_hires - departures
|
||
if net_change > 0:
|
||
score += 15
|
||
elif net_change < 0:
|
||
score -= 10
|
||
if departures > 5:
|
||
score -= 10 # 高流失率
|
||
|
||
key_metrics = data.get("key_metrics", [])
|
||
if key_metrics:
|
||
positive_count = sum(
|
||
1 for m in key_metrics
|
||
if _safe_float(str(m.get("change", "")).replace("%", "").replace("+", "")) > 0
|
||
)
|
||
score += (positive_count / len(key_metrics)) * 20
|
||
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_ai_commercial_score(data: dict[str, Any]) -> float:
|
||
"""计算 AI 商业化度。
|
||
|
||
基于 key_metrics 中 AI 相关指标的达成情况。
|
||
"""
|
||
score = 50.0
|
||
|
||
key_metrics = data.get("key_metrics", [])
|
||
ai_metrics = [
|
||
m for m in key_metrics
|
||
if "ai" in str(m.get("name", "")).lower()
|
||
or "模型" in str(m.get("name", ""))
|
||
or "推理" in str(m.get("name", ""))
|
||
]
|
||
|
||
if ai_metrics:
|
||
for m in ai_metrics:
|
||
change = str(m.get("change", ""))
|
||
val = _safe_float(change.replace("%", "").replace("+", ""))
|
||
if val > 0:
|
||
score += 15
|
||
elif val < 0:
|
||
score -= 10
|
||
else:
|
||
# 无 AI 相关指标，给中等偏下分数
|
||
score = 40.0
|
||
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_ai_cost_score(data: dict[str, Any]) -> float:
|
||
"""计算 AI 成本效率。
|
||
|
||
基于 burn_rate 和 AI 相关支出估算。
|
||
"""
|
||
score = 55.0
|
||
|
||
burn = data.get("burn_rate", {})
|
||
trend = burn.get("trend", "")
|
||
if trend == "down":
|
||
score += 20
|
||
elif trend == "up":
|
||
score -= 15
|
||
|
||
# 如果有 key_metrics 中的成本相关指标
|
||
key_metrics = data.get("key_metrics", [])
|
||
cost_metrics = [
|
||
m for m in key_metrics
|
||
if "成本" in str(m.get("name", "")) or "cost" in str(m.get("name", "")).lower()
|
||
]
|
||
for m in cost_metrics:
|
||
change = str(m.get("change", ""))
|
||
val = _safe_float(change.replace("%", "").replace("+", ""))
|
||
if val < 0: # 成本下降是好事
|
||
score += 10
|
||
elif val > 0:
|
||
score -= 10
|
||
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_org_talent_score(data: dict[str, Any]) -> float:
|
||
"""计算组织人才健康度。
|
||
|
||
指标：团队规模变化、流失率、关键岗位填补。
|
||
"""
|
||
score = 60.0
|
||
headcount = data.get("headcount", {})
|
||
new_hires = _safe_float(headcount.get("new_hires"))
|
||
departures = _safe_float(headcount.get("departures"))
|
||
total = _safe_float(headcount.get("total"), 1)
|
||
if total > 0:
|
||
turnover_rate = departures / total
|
||
if turnover_rate < 0.05:
|
||
score += 20
|
||
elif turnover_rate < 0.10:
|
||
score += 10
|
||
elif turnover_rate > 0.20:
|
||
score -= 20
|
||
elif turnover_rate > 0.15:
|
||
score -= 10
|
||
if new_hires > 0:
|
||
score += 10
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_product_tech_score(data: dict[str, Any]) -> float:
|
||
"""计算产品技术健康度。
|
||
|
||
指标：产品迭代频率、技术指标达成。
|
||
"""
|
||
score = 55.0
|
||
key_metrics = data.get("key_metrics", [])
|
||
tech_metrics = [
|
||
m for m in key_metrics
|
||
if any(k in str(m.get("name", "")).lower()
|
||
for k in ["产品", "product", "迭代", "release", "技术", "tech"])
|
||
]
|
||
if tech_metrics:
|
||
for m in tech_metrics:
|
||
change = str(m.get("change", ""))
|
||
val = _safe_float(change.replace("%", "").replace("+", ""))
|
||
if val > 0:
|
||
score += 12
|
||
elif val < 0:
|
||
score -= 8
|
||
else:
|
||
score = 50.0
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_market_compete_score(data: dict[str, Any]) -> float:
|
||
"""计算市场竞争健康度。
|
||
|
||
指标：市场份额变化、竞品动态、客户增长。
|
||
"""
|
||
score = 55.0
|
||
key_metrics = data.get("key_metrics", [])
|
||
market_metrics = [
|
||
m for m in key_metrics
|
||
if any(k in str(m.get("name", ""))
|
||
for k in ["市场", "份额", "客户", "竞品", "MAU", "DAU", "GMV"])
|
||
]
|
||
if market_metrics:
|
||
for m in market_metrics:
|
||
change = str(m.get("change", ""))
|
||
val = _safe_float(change.replace("%", "").replace("+", ""))
|
||
if val > 0:
|
||
score += 12
|
||
elif val < 0:
|
||
score -= 8
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_governance_score(data: dict[str, Any]) -> float:
|
||
"""计算治理合规健康度。
|
||
|
||
指标：董事会召开频率、合规事件。
|
||
"""
|
||
score = 70.0
|
||
governance = data.get("governance", {})
|
||
if governance.get("board_meeting_held"):
|
||
score += 10
|
||
if governance.get("compliance_issues"):
|
||
score -= 20
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_financing_score(data: dict[str, Any]) -> float:
|
||
"""计算融资资本健康度。
|
||
|
||
指标：现金跑道、融资进度。
|
||
"""
|
||
score = 55.0
|
||
cash = data.get("cash_balance", {})
|
||
runway = _safe_float(cash.get("runway_months"))
|
||
if runway >= 18:
|
||
score += 25
|
||
elif runway >= 12:
|
||
score += 15
|
||
elif runway >= 6:
|
||
score += 5
|
||
elif runway < 3:
|
||
score -= 25
|
||
financing = data.get("financing", {})
|
||
if financing.get("in_progress"):
|
||
score += 10
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_synergy_score(data: dict[str, Any]) -> float:
|
||
"""计算协同赋能健康度（T3.11）。"""
|
||
score = 55.0
|
||
synergy = data.get("synergy", {})
|
||
if synergy.get("active_count", 0) > 0:
|
||
score += min(20, synergy.get("active_count", 0) * 5)
|
||
if synergy.get("completed_count", 0) > 0:
|
||
score += 10
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_ai_model_product_score(data: dict[str, Any]) -> float:
|
||
"""计算 AI 模型产品健康度（T3.11）。"""
|
||
score = 50.0
|
||
ai_data = data.get("ai_metrics", {})
|
||
if ai_data.get("model_accuracy"):
|
||
score += 15
|
||
if ai_data.get("inference_cost_trend") == "down":
|
||
score += 10
|
||
if ai_data.get("data_quality_score"):
|
||
score += min(15, _safe_float(ai_data.get("data_quality_score")) * 0.15)
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_data_compliance_score(data: dict[str, Any]) -> float:
|
||
"""计算数据合规健康度（T3.11）。"""
|
||
score = 70.0
|
||
compliance = data.get("data_compliance", {})
|
||
if compliance.get("issues_count", 0) > 0:
|
||
score -= min(30, compliance.get("issues_count", 0) * 10)
|
||
if compliance.get("audit_passed"):
|
||
score += 15
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_team_tech_score(data: dict[str, Any]) -> float:
|
||
"""计算团队技术健康度（T3.11）。"""
|
||
score = 55.0
|
||
team = data.get("team_tech", {})
|
||
if team.get("tech_lead_count", 0) > 0:
|
||
score += 15
|
||
if team.get("patent_count", 0) > 0:
|
||
score += min(15, team.get("patent_count", 0) * 3)
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _calc_customer_success_score(data: dict[str, Any]) -> float:
|
||
"""计算客户成功健康度（T3.11）。"""
|
||
score = 55.0
|
||
cs = data.get("customer_success", {})
|
||
retention = _safe_float(cs.get("retention_rate"), -1)
|
||
if retention >= 0:
|
||
if retention >= 0.90:
|
||
score += 25
|
||
elif retention >= 0.80:
|
||
score += 15
|
||
elif retention < 0.70:
|
||
score -= 15
|
||
nps = _safe_float(cs.get("nps"))
|
||
if nps > 0:
|
||
score += min(15, nps * 0.15)
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def calculate_health_score(structured_data: dict[str, Any]) -> dict[str, float]:
|
||
"""计算 14 维度健康度评分。
|
||
|
||
Args:
|
||
structured_data: 月报 AI 解析后的结构化数据
|
||
|
||
Returns:
|
||
包含 total_score 和 14 个维度分数的字典
|
||
"""
|
||
if not structured_data:
|
||
return {
|
||
"total_score": 0.0,
|
||
"financial_score": 0.0,
|
||
"operational_score": 0.0,
|
||
"ai_commercial_score": 0.0,
|
||
"ai_cost_score": 0.0,
|
||
"org_talent_score": 0.0,
|
||
"product_tech_score": 0.0,
|
||
"market_compete_score": 0.0,
|
||
"governance_score": 0.0,
|
||
"financing_score": 0.0,
|
||
"synergy_score": 0.0,
|
||
"ai_model_product_score": 0.0,
|
||
"data_compliance_score": 0.0,
|
||
"team_tech_score": 0.0,
|
||
"customer_success_score": 0.0,
|
||
}
|
||
|
||
scores = {
|
||
"financial_score": _calc_financial_score(structured_data),
|
||
"operational_score": _calc_operational_score(structured_data),
|
||
"ai_commercial_score": _calc_ai_commercial_score(structured_data),
|
||
"ai_cost_score": _calc_ai_cost_score(structured_data),
|
||
"org_talent_score": _calc_org_talent_score(structured_data),
|
||
"product_tech_score": _calc_product_tech_score(structured_data),
|
||
"market_compete_score": _calc_market_compete_score(structured_data),
|
||
"governance_score": _calc_governance_score(structured_data),
|
||
"financing_score": _calc_financing_score(structured_data),
|
||
"synergy_score": _calc_synergy_score(structured_data),
|
||
"ai_model_product_score": _calc_ai_model_product_score(structured_data),
|
||
"data_compliance_score": _calc_data_compliance_score(structured_data),
|
||
"team_tech_score": _calc_team_tech_score(structured_data),
|
||
"customer_success_score": _calc_customer_success_score(structured_data),
|
||
}
|
||
|
||
weight_keys = [
|
||
"financial", "operational", "ai_commercial", "ai_cost",
|
||
"org_talent", "product_tech", "market_compete", "governance", "financing",
|
||
"synergy", "ai_model_product", "data_compliance", "team_tech", "customer_success",
|
||
]
|
||
score_keys = [
|
||
"financial_score", "operational_score", "ai_commercial_score", "ai_cost_score",
|
||
"org_talent_score", "product_tech_score", "market_compete_score", "governance_score",
|
||
"financing_score", "synergy_score", "ai_model_product_score", "data_compliance_score",
|
||
"team_tech_score", "customer_success_score",
|
||
]
|
||
|
||
total = sum(scores[sk] * WEIGHTS[wk] for sk, wk in zip(score_keys, weight_keys))
|
||
scores["total_score"] = round(total, 1)
|
||
|
||
result = {k: round(v, 1) for k, v in scores.items()}
|
||
logger.info("健康度评分计算完成（14 维度）: %s", result)
|
||
return result
|
||
|
||
|
||
def determine_trend(current_score: float, previous_score: float | None) -> str:
|
||
"""判断评分趋势。
|
||
|
||
Args:
|
||
current_score: 当前评分
|
||
previous_score: 上期评分（如有）
|
||
|
||
Returns:
|
||
"up" / "down" / "stable"
|
||
"""
|
||
if previous_score is None:
|
||
return "stable"
|
||
diff = current_score - previous_score
|
||
if diff > 5:
|
||
return "up"
|
||
elif diff < -5:
|
||
return "down"
|
||
return "stable"
|