"""增长质量诊断 API — 五维评分 + 诊断结论 + 跨期对比 数据来源：KPI字典 (kpi_definitions) + KPI实际值 (kpi_values) 五维度：营收增长 / 利润质量 / 现金质量 / 增长效率 / 组织健康 评分区间：0-100（>=80 好 / 60-79 中 / <60 差） """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from typing import Optional, List from datetime import datetime import re from app.database import get_db from app.deps import get_entity_id from app.auth_middleware import require_auth, require_role from app.models import KPIDefinition, KPIValue, Entity router = APIRouter(prefix="/api/cma/growth-quality", tags=["增长质量诊断"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], ) # 五维度定义 DIMENSIONS = [ {"key": "revenueGrowth", "name": "营收增长", "weight": 20, "icon": "📈", "desc": "营收增速、增长持续性"}, {"key": "profitQuality", "name": "利润质量", "weight": 20, "icon": "💰", "desc": "毛利率/净利率趋势、利润与收入匹配"}, {"key": "cashQuality", "name": "现金质量", "weight": 20, "icon": "🏦", "desc": "经营现金流与净利润匹配度（含金量）"}, {"key": "growthEfficiency", "name": "增长效率", "weight": 20, "icon": "🚀", "desc": "获客成本、单位增长投入产出"}, {"key": "orgHealth", "name": "组织健康", "weight": 20, "icon": "⚡", "desc": "人效、费用结构"}, ] # 关键KPI编码 → 维度用途 KPI_CODES = { "F_REVENUE": "营业收入(万元)", "F_NET_PROFIT": "净利润(万元)", "F_OP_CFLOW": "经营性现金流(万元)", "F_GROSS_MARGIN": "毛利率(%)", "F_OP_PROFIT_MARGIN": "经营利润率(%)", "F_COST_RATIO": "费用率(%)", "F_REVENUE_GROWTH": "收入增长率(%)", "F_FCF": "自由现金流(万元)", "C_NEW_CLIENTS": "新客户数", "C_REBATE_RATE": "渠补率(%)", "C_SATISFACTION": "客户满意度", "L_TRAINING": "培训完成率", "P_DELIVERY": "交付及时率", "F_AR_DAYS": "应收账款周转天数", } MONTH_RE = re.compile(r"^\d{4}-\d{2}$") def _fetch_kpi_values(db: Session, entity_id: int, period: str) -> dict: """拉取某实体某期间的全部KPI值 {kpi_code: actual_value}""" rows = (db.query(KPIDefinition.kpi_code, KPIValue.actual_value) .join(KPIValue, KPIValue.kpi_id == KPIDefinition.id) .filter(KPIDefinition.entity_id == entity_id, KPIValue.period == period, KPIValue.actual_value.isnot(None)) .all()) return {code: value for code, value in rows} def _fetch_history(db: Session, entity_id: int, limit: int = 12) -> List[dict]: """拉取最近 N 个期间（按月，含数据）的 KPI 值，供趋势/持续性分析""" periods = (db.query(KPIValue.period) .join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id) .filter(KPIDefinition.entity_id == entity_id, KPIDefinition.kpi_code.in_(["F_REVENUE", "F_NET_PROFIT", "F_OP_CFLOW"])) .distinct().all()) plist = sorted({p[0] for p in periods}, reverse=True) # 只保留 YYYY-MM 格式，按时间排序（旧→新） months = sorted([p for p in plist if MONTH_RE.match(p)]) hist = [] for p in months[-limit:]: hist.append({"period": p, **{k: None for k in KPI_CODES}}) if not hist: return [] # 批量取数 rows = (db.query(KPIDefinition.kpi_code, KPIValue.period, KPIValue.actual_value) .join(KPIValue, KPIValue.kpi_id == KPIDefinition.id) .filter(KPIDefinition.entity_id == entity_id, KPIValue.period.in_([h["period"] for h in hist])) .all()) idx = {h["period"]: h for h in hist} for code, period, val in rows: if period in idx and code in idx[period]: idx[period][code] = val return hist def _prev_period(period: str) -> Optional[str]: """计算上期（YYYY-MM → 上一月；其他格式 → None）""" m = MONTH_RE.match(period) if not m: return None y, mo = int(period[:4]), int(period[5:7]) if mo == 1: return f"{y-1:04d}-12" return f"{y:04d}-{mo-1:02d}" def _yoy_period(period: str) -> Optional[str]: """计算去年同期（YYYY-MM → 去年同月）""" m = MONTH_RE.match(period) if not m: return None return f"{int(period[:4])-1:04d}-{period[5:7]}" # ═══════════════════════════════════════════════ # 五维度评分引擎（0-100） # ═══════════════════════════════════════════════ def _clamp(v: float, lo: float = 0.0, hi: float = 100.0) -> float: return max(lo, min(hi, v)) def _linear(value, points: List[tuple]): """分段线性插值评分: points = [(x, score), ...] 按 x 升序""" if value is None: return 50.0 if value <= points[0][0]: return points[0][1] if value >= points[-1][0]: return points[-1][1] for (x1, s1), (x2, s2) in zip(points, points[1:]): if x1 <= value <= x2: if x2 == x1: return s1 return s1 + (s2 - s1) * (value - x1) / (x2 - x1) return 50.0 # ── 1. 营收增长：增速 + 持续性 ── def _score_revenue_growth(cur: dict, prev: dict, history: List[dict]) -> float: rev_cur = cur.get("F_REVENUE") rev_prev = prev.get("F_REVENUE") if prev else None # 优先用 KPI 直接给的收入增长率 kpi_growth = cur.get("F_REVENUE_GROWTH") growth = None if kpi_growth is not None: growth = float(kpi_growth) elif rev_cur is not None and rev_prev: growth = (rev_cur - rev_prev) / rev_prev * 100 if rev_prev else None score = _linear(growth, [ (-30, 5), (-20, 15), (-10, 30), (0, 45), (5, 60), (10, 70), (20, 82), (30, 90), (50, 96), ]) # 增长持续性：近6个月中收入增长月占比 if len(history) >= 2: revs = [h.get("F_REVENUE") for h in history if h.get("F_REVENUE") is not None] ups = 0 for i in range(1, len(revs)): if revs[i] > revs[i - 1]: ups += 1 persist = ups / (len(revs) - 1) if len(revs) > 1 else 0.5 score = score * 0.7 + persist * 100 * 0.3 return round(_clamp(score), 1) # ── 2. 利润质量：毛利率/净利率水平 + 趋势 + 收入匹配 ── def _score_profit_quality(cur: dict, prev: dict) -> float: gm = cur.get("F_GROSS_MARGIN") np_ = cur.get("F_NET_PROFIT") rev = cur.get("F_REVENUE") net_margin = (np_ / rev * 100) if (np_ is not None and rev) else None opm = cur.get("F_OP_PROFIT_MARGIN") gm_score = _linear(gm, [(-10, 5), (0, 10), (10, 25), (20, 45), (30, 62), (40, 75), (55, 88), (70, 95)]) nm_score = _linear(net_margin, [(-50, 0), (-20, 10), (-10, 20), (0, 35), (10, 60), (20, 78), (30, 90)]) opm_score = _linear(opm, [(-20, 10), (0, 30), (10, 55), (20, 75), (35, 90)]) # 毛利率 40% + 净利率 40% + 经营利润率 20% base = gm_score * 0.4 + nm_score * 0.4 + opm_score * 0.2 # 利润与收入匹配：收入升但利润降 → 扣分 if prev and rev is not None and np_ is not None: prev_rev = prev.get("F_REVENUE") prev_np = prev.get("F_NET_PROFIT") if prev_rev and prev_np is not None: rev_up = rev > prev_rev np_down = np_ < prev_np if rev_up and np_down: base -= 10 elif np_down: base -= 5 return round(_clamp(base), 1) # ── 3. 现金质量：含金量(OCF/净利润) + 现金流强度 ── def _score_cash_quality(cur: dict) -> float: ocf = cur.get("F_OP_CFLOW") np_ = cur.get("F_NET_PROFIT") rev = cur.get("F_REVENUE") fcf = cur.get("F_FCF") # 含金量 = OCF / 净利润（净利润>0时） gold = None if ocf is not None and np_ is not None and np_ > 0: gold = ocf / np_ # 净利润<=0：利润为负，含金量指标失效 → 低分（除非现金流强） gold_score = _linear(gold, [(0, 10), (0.5, 35), (0.8, 55), (1.0, 70), (1.2, 85), (1.5, 95)]) if np_ is not None and np_ <= 0: gold_score = 15 if (ocf is None or ocf <= 0) else 35 # 现金流强度 = OCF / 收入 ocf_ratio = (ocf / rev * 100) if (ocf is not None and rev) else None ocf_score = _linear(ocf_ratio, [(-20, 5), (0, 20), (10, 50), (20, 75), (30, 90), (50, 98)]) fcf_score = _linear(fcf, [(-100, 10), (-20, 30), (0, 50), (20, 70), (100, 90)]) if fcf is not None else 50.0 score = gold_score * 0.5 + ocf_score * 0.35 + fcf_score * 0.15 return round(_clamp(score), 1) # ── 4. 增长效率：费用率水平 + 单位增长投入产出 + 获客成本 ── def _score_growth_efficiency(cur: dict, prev: dict) -> float: cost_ratio = cur.get("F_COST_RATIO") cost_score = _linear(cost_ratio, [(10, 95), (20, 82), (30, 68), (40, 55), (55, 40), (70, 25), (90, 10)]) # 费用增速 vs 收入增速（用费用率变化近似） eff_score = 60.0 if prev is not None and cost_ratio is not None: prev_cr = prev.get("F_COST_RATIO") if prev_cr: cr_change = cost_ratio - prev_cr eff_score = _linear(cr_change, [(-15, 95), (-5, 80), (0, 65), (5, 45), (15, 25), (30, 10)]) # 获客成本代理：收入/新客户数（越高越高效） rev = cur.get("F_REVENUE") new_clients = cur.get("C_NEW_CLIENTS") cac_score = 60.0 if rev is not None and new_clients: per_client = rev / new_clients cac_score = _linear(per_client, [(0, 40), (50, 50), (200, 65), (500, 78), (1000, 88)]) score = cost_score * 0.45 + eff_score * 0.35 + cac_score * 0.2 return round(_clamp(score), 1) # ── 5. 组织健康：费用结构 + 人效/运营质量 ── def _score_org_health(cur: dict) -> float: cost_ratio = cur.get("F_COST_RATIO") # 费用结构（费用率越低越健康） cost_score = _linear(cost_ratio, [(10, 95), (20, 82), (30, 68), (40, 55), (55, 40), (70, 25), (90, 10)]) # 运营/人效质量代理：满意度、培训、交付、应收 sat = cur.get("C_SATISFACTION") train = cur.get("L_TRAINING") deliver = cur.get("P_DELIVERY") ar_days = cur.get("F_AR_DAYS") op_vals = [v for v in [sat, train, deliver] if v is not None] op_score = (sum(op_vals) / len(op_vals)) if op_vals else 55.0 ar_score = _linear(ar_days, [(15, 95), (30, 80), (45, 65), (60, 50), (90, 30), (120, 15)]) if ar_days is not None else 55.0 score = cost_score * 0.4 + op_score * 0.35 + ar_score * 0.25 return round(_clamp(score), 1) _SCORERS = { "revenueGrowth": _score_revenue_growth, "profitQuality": _score_profit_quality, "cashQuality": _score_cash_quality, "growthEfficiency": _score_growth_efficiency, "orgHealth": _score_org_health, } # ═══════════════════════════════════════════════ # 明细指标 + 改善建议 # ═══════════════════════════════════════════════ def _fmt(v, unit=""): if v is None: return "—" if isinstance(v, float) and v == int(v): return f"{int(v)}{unit}" return f"{round(v, 2)}{unit}" def _dim_indicators(dim_key: str, cur: dict, prev: dict) -> List[dict]: """维度明细指标（label/value/verdict/status）""" inds = [] def add(label, value, verdict, status): inds.append({"label": label, "value": value, "verdict": verdict, "status": status}) if dim_key == "revenueGrowth": rev, prev_rev = cur.get("F_REVENUE"), (prev or {}).get("F_REVENUE") growth = None if rev is not None and prev_rev: growth = (rev - prev_rev) / prev_rev * 100 add("营业收入", _fmt(rev, "万"), "环比" + (_fmt(growth, "%") if growth is not None else "无上期数据"), "success" if (growth or 0) >= 0 else "danger") add("收入增长率(KPI)", _fmt(cur.get("F_REVENUE_GROWTH"), "%"), "KPI直接值" if cur.get("F_REVENUE_GROWTH") is not None else "未录入", "success" if (cur.get("F_REVENUE_GROWTH") or 0) >= 10 else "warning") elif dim_key == "profitQuality": rev, np_ = cur.get("F_REVENUE"), cur.get("F_NET_PROFIT") nm = (np_ / rev * 100) if (np_ is not None and rev) else None add("毛利率", _fmt(cur.get("F_GROSS_MARGIN"), "%"), "毛利健康" if (cur.get("F_GROSS_MARGIN") or 0) >= 30 else "毛利偏低", "success" if (cur.get("F_GROSS_MARGIN") or 0) >= 30 else "danger") add("净利率", _fmt(nm, "%"), "盈利" if (nm or 0) > 0 else "亏损", "success" if (nm or 0) > 10 else "danger") add("经营利润率", _fmt(cur.get("F_OP_PROFIT_MARGIN"), "%"), "正常" if (cur.get("F_OP_PROFIT_MARGIN") or 0) >= 15 else "偏低", "success" if (cur.get("F_OP_PROFIT_MARGIN") or 0) >= 15 else "warning") elif dim_key == "cashQuality": ocf, np_ = cur.get("F_OP_CFLOW"), cur.get("F_NET_PROFIT") gold = (ocf / np_) if (ocf is not None and np_ and np_ > 0) else None add("经营现金流", _fmt(ocf, "万"), "现金流入" if (ocf or 0) > 0 else "现金流出", "success" if (ocf or 0) > 0 else "danger") add("含金量(OCF/净利润)", _fmt(gold, "倍"), "含金量高" if (gold or 0) >= 1 else ("利润为负" if (np_ or 0) <= 0 else "含金量低"), "success" if (gold or 0) >= 1 else "danger") add("自由现金流", _fmt(cur.get("F_FCF"), "万"), "正常" if (cur.get("F_FCF") or 0) > 0 else "为负", "success" if (cur.get("F_FCF") or 0) > 0 else "warning") elif dim_key == "growthEfficiency": rev, nc = cur.get("F_REVENUE"), cur.get("C_NEW_CLIENTS") per = (rev / nc) if (rev is not None and nc) else None add("费用率", _fmt(cur.get("F_COST_RATIO"), "%"), "费用可控" if (cur.get("F_COST_RATIO") or 0) <= 30 else "费用偏高", "success" if (cur.get("F_COST_RATIO") or 0) <= 30 else "warning") add("单位客户营收(万/户)", _fmt(per), "获客效率高" if (per or 0) >= 200 else "获客效率一般", "success" if (per or 0) >= 500 else "warning") add("渠补率", _fmt(cur.get("C_REBATE_RATE"), "%"), "渠道依赖" if (cur.get("C_REBATE_RATE") or 0) > 50 else "渠道健康", "danger" if (cur.get("C_REBATE_RATE") or 0) > 50 else "success") elif dim_key == "orgHealth": add("费用率(结构)", _fmt(cur.get("F_COST_RATIO"), "%"), "结构健康" if (cur.get("F_COST_RATIO") or 0) <= 30 else "结构偏重", "success" if (cur.get("F_COST_RATIO") or 0) <= 30 else "warning") add("应收周转天数", _fmt(cur.get("F_AR_DAYS"), "天"), "回款快" if (cur.get("F_AR_DAYS") or 0) <= 45 else "回款偏慢", "success" if (cur.get("F_AR_DAYS") or 0) <= 45 else "warning") add("客户满意度", _fmt(cur.get("C_SATISFACTION")), "满意" if (cur.get("C_SATISFACTION") or 0) >= 80 else "待提升", "success" if (cur.get("C_SATISFACTION") or 0) >= 80 else "warning") add("培训完成率", _fmt(cur.get("L_TRAINING"), "%"), "学习投入足" if (cur.get("L_TRAINING") or 0) >= 80 else "学习投入不足", "success" if (cur.get("L_TRAINING") or 0) >= 80 else "warning") return inds def _dim_suggestions(dim_key: str, score: float, cur: dict) -> List[str]: """按维度评分生成改善建议""" if score >= 80: return ["该维度表现良好，建议保持并固化为标准流程"] sug = [] if dim_key == "revenueGrowth": sug = ["挖掘存量客户复购，稳定收入基本盘", "拓展新渠道/新产品线，提升营收增速", "跟踪F_REVENUE_GROWTH KPI按月更新，建立增长预警线"] elif dim_key == "profitQuality": sug = ["排查毛利率下滑原因（成本/价格/渠补），优先止血", "控制费用增速不超过收入增速，改善净利率", "对亏损产品线做盈亏平衡分析，必要时收缩"] elif dim_key == "cashQuality": sug = ["加强应收账款催收，缩短回款周期", "压缩非必要开支，提升经营现金流净额", "建立现金流月度滚动预测，防范断流风险"] elif dim_key == "growthEfficiency": sug = ["优化费用结构，降低费用率至30%以下", "评估渠道返利政策，降低渠补率与渠道依赖", "提高获客转化率，降低单位获客成本"] elif dim_key == "orgHealth": sug = ["精简组织与费用结构，提升人效", "强化培训与人才梯队建设（盯L_TRAINING）", "优化应收管理，缩短周转天数"] return sug def _level_of(overall: float) -> dict: if overall >= 80: return {"level": "好", "level_type": "success", "desc": "增长质量优秀，增长可持续"} if overall >= 60: return {"level": "中", "level_type": "warning", "desc": "增长质量中等，存在优化空间"} return {"level": "差", "level_type": "danger", "desc": "增长质量堪忧，需立即干预"} # ═══════════════════════════════════════════════ # 诊断主流程 # ═══════════════════════════════════════════════ def _diagnose(db: Session, entity_id: int, period: str, history: List[dict]): """对单个期间执行五维诊断，返回完整诊断对象""" cur = _fetch_kpi_values(db, entity_id, period) prev_period = _prev_period(period) prev = _fetch_kpi_values(db, entity_id, prev_period) if prev_period else {} scores = {} dims_payload = {} for dim in DIMENSIONS: key = dim["key"] scorer = _SCORERS[key] if key == "revenueGrowth": s = scorer(cur, prev, history) elif key in ("cashQuality", "orgHealth"): s = scorer(cur) else: s = scorer(cur, prev) scores[key] = s dims_payload[key] = { "key": key, "name": dim["name"], "icon": dim["icon"], "desc": dim["desc"], "weight": dim["weight"], "score": s, "indicators": _dim_indicators(key, cur, prev), "suggestions": _dim_suggestions(key, s, cur), } overall = round(sum(scores.values()) / len(scores), 1) level = _level_of(overall) # 诊断结论文本 low_dims = [d for d in DIMENSIONS if scores[d["key"]] < 60] mid_dims = [d for d in DIMENSIONS if 60 <= scores[d["key"]] < 80] lines = [f"{period} 综合增长质量评分 {overall} 分（{level['level']}）：{level['desc']}。"] if low_dims: lines.append("需重点关注：" + "、".join(f"{d['name']}({scores[d['key']]}分)" for d in low_dims) + "。") if mid_dims: lines.append("可优化：" + "、".join(f"{d['name']}({scores[d['key']]}分)" for d in mid_dims) + "。") if not low_dims: lines.append("各维度均处于健康区间，增长质量扎实。") diagnosis = "".join(lines) return { "entity_id": entity_id, "period": period, "overall": overall, "level": level["level"], "level_type": level["level_type"], "diagnosis": diagnosis, "dimensions": dims_payload, "kpi_available": {k: cur.get(k) is not None for k in KPI_CODES}, } @router.get("/periods") def list_periods(entity_id: int = Depends(get_entity_id), db: Session = Depends(get_db)): """列出某实体有KPI数据的期间（按月，含数据覆盖度，用于前端默认期间选择）""" rows = (db.query(KPIValue.period, KPIValue.kpi_id) .join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id) .filter(KPIDefinition.entity_id == entity_id) .all()) counts: dict = {} for period, kpi_id in rows: if MONTH_RE.match(period or ""): counts[period] = counts.get(period, 0) + 1 periods = sorted(counts.keys(), reverse=True) return {"entity_id": entity_id, "periods": periods, "coverage": {p: counts[p] for p in periods}} @router.get("/diagnosis") def growth_quality_diagnosis( entity_id: int = Depends(get_entity_id), period: Optional[str] = Query(None, description="期间 YYYY-MM，默认最近有数据期间"), db: Session = Depends(get_db), ): """增长质量诊断 — 五维评分(0-100) + 诊断结论 + 跨期对比（本期/上期/去年同期）""" ent = db.query(Entity).filter(Entity.id == entity_id).first() if not ent: raise HTTPException(404, f"实体 {entity_id} 不存在") history = _fetch_history(db, entity_id, limit=12) if not history: raise HTTPException(400, "该实体暂无月度KPI数据，请先录入KPI实际值") # 默认取最近且有足够数据覆盖的期间（>=5个KPI值，退化为最近一个） if not period: cov_rows = (db.query(KPIValue.period) .join(KPIDefinition, KPIDefinition.id == KPIValue.kpi_id) .filter(KPIDefinition.entity_id == entity_id) .all()) cov: dict = {} for (p,) in cov_rows: if MONTH_RE.match(p or ""): cov[p] = cov.get(p, 0) + 1 candidates = sorted([p for p in cov if cov[p] >= 5], reverse=True) period = candidates[0] if candidates else history[-1]["period"] current = _diagnose(db, entity_id, period, history) # 跨期对比：上期 + 去年同期 prev_p = _prev_period(period) yoy_p = _yoy_period(period) prev_data = _fetch_kpi_values(db, entity_id, prev_p) if prev_p else {} yoy_data = _fetch_kpi_values(db, entity_id, yoy_p) if yoy_p else {} comparison = { "current": { "period": period, "overall": current["overall"], "level": current["level"], "level_type": current["level_type"], "dimensions": {k: v["score"] for k, v in current["dimensions"].items()}, }, } if prev_p and prev_data: pdiag = _diagnose(db, entity_id, prev_p, history) comparison["previous"] = { "period": prev_p, "overall": pdiag["overall"], "level": pdiag["level"], "level_type": pdiag["level_type"], "dimensions": {k: v["score"] for k, v in pdiag["dimensions"].items()}, } if yoy_p and yoy_data: ydiag = _diagnose(db, entity_id, yoy_p, history) comparison["yoy"] = { "period": yoy_p, "overall": ydiag["overall"], "level": ydiag["level"], "level_type": ydiag["level_type"], "dimensions": {k: v["score"] for k, v in ydiag["dimensions"].items()}, } # 趋势：近12个月综合评分 trend = [] for h in history: try: d = _diagnose(db, entity_id, h["period"], history) trend.append({"period": h["period"], "overall": d["overall"]}) except Exception: continue return { "entity": {"id": ent.id, "name": ent.name, "short_name": ent.short_name}, "period": period, "overall": current["overall"], "level": current["level"], "level_type": current["level_type"], "diagnosis": current["diagnosis"], "dimensions": current["dimensions"], "comparison": comparison, "trend": trend, "kpi_available": current["kpi_available"], }