selfrelease
7ec4fb0747
- LLM 客户端：全部 SSE 流式输出，兼容 OpenAI 接口
- AI 月报解析：SSE 流式端点 POST /reports/{id}/parse
- 健康度计算引擎：四维评分（财务/经营/AI商业化/AI成本）
- 风险自动检测引擎：6 条规则自动检测指标越界
- AI Copilot：SSE 流式对话 POST /copilot/chat
- 权限中间件：角色级 + 字段级权限控制
- 测试：21 个新测试（健康度 8 + 风险检测 8 + 权限 5），总计 64 passed
161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
"""风险自动检测引擎。
|
|
|
|
基于月报结构化数据，检测指标越界并自动生成风险事件。
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 风险检测规则
|
|
RULES = [
|
|
{
|
|
"name": "现金跑道不足",
|
|
"type": "financial",
|
|
"severity": "critical",
|
|
"condition": lambda d: _get_runway(d) < 3 and _get_runway(d) > 0,
|
|
"title": "现金跑道不足 3 个月",
|
|
"description": "当前现金跑道仅 {runway} 个月，需紧急融资",
|
|
"suggested_action": "立即启动融资对话，评估 bridge loan 可能性",
|
|
},
|
|
{
|
|
"name": "现金跑道预警",
|
|
"type": "financial",
|
|
"severity": "high",
|
|
"condition": lambda d: 3 <= _get_runway(d) < 6,
|
|
"title": "现金跑道低于 6 个月",
|
|
"description": "当前现金跑道 {runway} 个月，需加快融资进度",
|
|
"suggested_action": "与创始人沟通融资时间表，准备备选方案",
|
|
},
|
|
{
|
|
"name": "烧钱率上升",
|
|
"type": "financial",
|
|
"severity": "medium",
|
|
"condition": lambda d: _get_burn_trend(d) == "up",
|
|
"title": "烧钱率持续上升",
|
|
"description": "月度烧钱率呈上升趋势，需关注成本控制",
|
|
"suggested_action": "审查主要支出项，制定成本优化计划",
|
|
},
|
|
{
|
|
"name": "营收下滑",
|
|
"type": "financial",
|
|
"severity": "high",
|
|
"condition": lambda d: _get_yoy(revenue=d) < 0,
|
|
"title": "营收同比下滑",
|
|
"description": "营收同比下降 {yoy}%，需关注业务增长",
|
|
"suggested_action": "分析营收下滑原因，调整商业策略",
|
|
},
|
|
{
|
|
"name": "高人员流失",
|
|
"type": "org",
|
|
"severity": "medium",
|
|
"condition": lambda d: _get_departures(d) > 5,
|
|
"title": "人员流失率较高",
|
|
"description": "本月离职 {departures} 人，需关注团队稳定性",
|
|
"suggested_action": "了解离职原因，评估核心岗位风险",
|
|
},
|
|
{
|
|
"name": "团队净缩减",
|
|
"type": "org",
|
|
"severity": "high",
|
|
"condition": lambda d: _get_net_headcount(d) < -3,
|
|
"title": "团队规模显著缩减",
|
|
"description": "本月团队净减少 {net} 人，需关注组织健康",
|
|
"suggested_action": "与创始人沟通团队规划，评估关键岗位覆盖",
|
|
},
|
|
]
|
|
|
|
|
|
def _get_runway(data: dict[str, Any]) -> float:
|
|
"""获取现金跑道月数。"""
|
|
cash = data.get("cash_balance", {})
|
|
try:
|
|
return float(cash.get("runway_months", 0) or 0)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
|
|
def _get_burn_trend(data: dict[str, Any]) -> str:
|
|
"""获取烧钱率趋势。"""
|
|
burn = data.get("burn_rate", {})
|
|
return burn.get("trend", "")
|
|
|
|
|
|
def _get_yoy(revenue: dict[str, Any], d: dict[str, Any] = None) -> float:
|
|
"""获取营收同比增长率。"""
|
|
if d is None:
|
|
d = revenue
|
|
revenue = d.get("revenue", {})
|
|
yoy = revenue.get("yoy_change", "")
|
|
try:
|
|
return float(str(yoy).replace("%", "").replace("+", "") or 0)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
|
|
def _get_departures(data: dict[str, Any]) -> float:
|
|
"""获取离职人数。"""
|
|
hc = data.get("headcount", {})
|
|
try:
|
|
return float(hc.get("departures", 0) or 0)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
|
|
def _get_net_headcount(data: dict[str, Any]) -> float:
|
|
"""获取团队净变化。"""
|
|
hc = data.get("headcount", {})
|
|
try:
|
|
new = float(hc.get("new_hires", 0) or 0)
|
|
dep = float(hc.get("departures", 0) or 0)
|
|
return new - dep
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
|
|
def detect_risks(
|
|
structured_data: dict[str, Any],
|
|
company_id: str,
|
|
) -> list[dict[str, Any]]:
|
|
"""从月报结构化数据中检测风险事件。
|
|
|
|
Args:
|
|
structured_data: 月报 AI 解析后的结构化数据
|
|
company_id: 企业 ID
|
|
|
|
Returns:
|
|
风险事件列表，每项包含 type/severity/title/description/suggested_action
|
|
"""
|
|
if not structured_data:
|
|
return []
|
|
|
|
risks: list[dict[str, Any]] = []
|
|
runway = _get_runway(structured_data)
|
|
yoy = _get_yoy(structured_data)
|
|
departures = _get_departures(structured_data)
|
|
net_hc = _get_net_headcount(structured_data)
|
|
|
|
for rule in RULES:
|
|
try:
|
|
if rule["condition"](structured_data):
|
|
risk = {
|
|
"company_id": company_id,
|
|
"type": rule["type"],
|
|
"severity": rule["severity"],
|
|
"title": rule["title"],
|
|
"description": rule["description"].format(
|
|
runway=runway,
|
|
yoy=abs(yoy),
|
|
departures=departures,
|
|
net=abs(net_hc),
|
|
),
|
|
"suggested_action": rule["suggested_action"],
|
|
}
|
|
risks.append(risk)
|
|
logger.info("检测到风险: %s — %s", rule["name"], risk["title"])
|
|
except Exception as e:
|
|
logger.warning("风险检测规则 '%s' 执行异常: %s", rule["name"], e)
|
|
|
|
return risks
|