feat: 增长质量诊断模块 — 后端评分API+前端驾驶舱卡片+雷达图+明细下钻+双实体对比

- 新增 POST /api/cma/predict/growth-quality 评分API - 五维度评分引擎（营收结构/利润结构/现金资产/增长驱动/组织效率） - 自动诊断结论生成 - 明细指标+改善建议 - 双实体对比模式 - 前端 GrowthQuality.vue 组件 - 综合评分大字展示 - 五维度评分条 + 状态颜色 - ECharts雷达图 - 明细下钻弹窗（指标+建议） - 双实体并列对比（VS模式） - 集成到 Dashboard.vue CEO驾驶舱视图 - 酣客测试验证：综合1.4分（越增长越重） - 博海对比验证：综合3.6分（增长质量中等）
2026-07-19 10:46:43 +08:00
parent 2d471aa07b
commit f85c3d2c4e
5 changed files with 753 additions and 0 deletions
@@ -331,3 +331,275 @@ def api_real_option(data: dict):
        raise
    except Exception as e:
        raise HTTPException(400, f"实物期权计算失败: {str(e)}")
+
+
+# ── 增长质量诊断 ─────────────────────────────────────────────
+
+def _score_revenue_structure(entity: dict) -> int:
+    """营收结构评分：渠补率越低越好"""
+    rebate_rate = float(entity.get("rebateRate", entity.get("rebate_rate", 0)))
+    if rebate_rate > 80: return 1
+    if rebate_rate > 60: return 2
+    if rebate_rate > 40: return 3
+    if rebate_rate > 20: return 4
+    return 5
+
+
+def _score_profit_structure(entity: dict) -> int:
+    """利润结构评分：真实毛利率越高越好"""
+    gross_margin = float(entity.get("trueGrossMargin", entity.get("true_gross_margin", 0)))
+    if gross_margin < 0: return 1
+    if gross_margin < 10: return 2
+    if gross_margin < 20: return 3
+    if gross_margin < 30: return 4
+    return 5
+
+
+def _score_cash_assets(entity: dict) -> int:
+    """现金资产评分：现金比率越高越好"""
+    cash_ratio = float(entity.get("cashRatio", entity.get("cash_ratio", 0)))
+    if cash_ratio < 10: return 1
+    if cash_ratio < 30: return 2
+    if cash_ratio < 50: return 3
+    if cash_ratio < 100: return 4
+    return 5
+
+
+def _score_growth_driver(entity: dict) -> int:
+    """增长驱动评分：费用增速相对收入增速越低越好"""
+    expense_growth = float(entity.get("expenseGrowthRate", entity.get("expense_growth_rate", 0)))
+    revenue_growth = float(entity.get("revenueGrowthRate", entity.get("revenue_growth_rate", 1)))
+    if revenue_growth <= 0: revenue_growth = 1  # prevent div by zero
+    ratio = expense_growth / revenue_growth
+    if ratio > 1.5: return 1
+    if ratio > 1.2: return 2
+    if ratio > 1.0: return 3
+    if ratio > 0.8: return 4
+    return 5
+
+
+def _score_org_efficiency(entity: dict) -> int:
+    """组织效率评分：管理费/净收入越低越好"""
+    mgmt_ratio = float(entity.get("mgmtRatio", entity.get("mgmt_ratio", 0)))
+    if mgmt_ratio > 300: return 1
+    if mgmt_ratio > 200: return 2
+    if mgmt_ratio > 100: return 3
+    if mgmt_ratio > 50: return 4
+    return 5
+
+
+def _diagnosis_text(overall: float, dimensions: dict, entity_name: str) -> str:
+    """根据评分生成诊断结论"""
+    lines = []
+    low_dims = {k: v for k, v in dimensions.items() if v["score"] <= 2}
+    mid_dims = {k: v for k, v in dimensions.items() if 2 < v["score"] < 4}
+
+    dim_labels = {
+        "revenueStructure": "营收结构",
+        "profitStructure": "利润结构",
+        "cashAssets": "现金资产",
+        "growthDriver": "增长驱动",
+        "orgEfficiency": "组织效率",
+    }
+
+    if overall < 2:
+        lines.append(f"{entity_name}的增长质量评分仅{overall}分，属于「越增长越重」类型。")
+        lines.append("增长主要依赖资源投入而非核心能力积累，可持续性堪忧。")
+    elif overall < 3:
+        lines.append(f"{entity_name}的增长质量评分{overall}分，需重点关注。")
+        lines.append("部分维度存在风险，增长质量有待改善。")
+    elif overall < 4:
+        lines.append(f"{entity_name}的增长质量评分{overall}分，处于中等水平。")
+        lines.append("多数维度表现尚可，仍有优化空间。")
+    else:
+        lines.append(f"{entity_name}的增长质量评分{overall}分，「越增长越轻」。")
+        lines.append("增长模式健康，具备持续增长能力。")
+
+    if low_dims:
+        low_names = [dim_labels.get(k, k) for k in low_dims]
+        lines.append(f"⚠️ 需重点关注：{'、'.join(low_names)}评分偏低（≤2分）。")
+
+    if mid_dims:
+        mid_names = [dim_labels.get(k, k) for k in mid_dims]
+        lines.append(f"💡 可优化：{'、'.join(mid_names)}有提升空间。")
+
+    # 具体建议（硬编码的关键诊断）
+    if dimensions.get("revenueStructure", {}).get("score", 5) <= 2:
+        lines.append("• 营收依赖渠道返利，建议降低渠补率、拓展直销渠道。")
+    if dimensions.get("orgEfficiency", {}).get("score", 5) <= 2:
+        lines.append("• 管理费率高企，建议精简费用结构、优化运营效率。")
+    if dimensions.get("cashAssets", {}).get("score", 5) <= 2:
+        lines.append("• 现金比率极低，存在断流风险，建议加强现金流管理。")
+    if dimensions.get("growthDriver", {}).get("score", 5) <= 2:
+        lines.append("• 费用增速远超收入增速，增长不可持续，需控制费用膨胀。")
+
+    return "\n".join(lines)
+
+
+def _generate_improvement_suggestions(dimension: str, score: int, entity: dict) -> list:
+    """为指定维度生成改善建议"""
+    suggestions = []
+    if dimension == "revenueStructure":
+        rebate = float(entity.get("rebateRate", entity.get("rebate_rate", 0)))
+        if score <= 2:
+            target_rebate = max(rebate - 10, 0)
+            savings = f"释放现金{round(rebate - target_rebate, 1)}%/月"
+            suggestions.append(f"渠补谈判：{rebate}%→{target_rebate}%（{savings}）")
+            suggestions.append("客户分散：拓展直销渠道，降低渠道依赖")
+            suggestions.append("渠补制度：分级管理，差异化返利")
+        else:
+            suggestions.append("维持现有渠补政策")
+    elif dimension == "profitStructure":
+        gm = float(entity.get("trueGrossMargin", entity.get("true_gross_margin", 0)))
+        if score <= 2:
+            suggestions.append(f"成本优化：毛利率仅{gm}%，需分析成本构成")
+            suggestions.append("产品结构：提高高毛利产品占比")
+            suggestions.append("定价策略：评估提价空间")
+        else:
+            suggestions.append("维持毛利率水平")
+    elif dimension == "cashAssets":
+        cr = float(entity.get("cashRatio", entity.get("cash_ratio", 0)))
+        if score <= 2:
+            suggestions.append(f"现金管理：现金比率仅{cr}%，存在断流风险")
+            suggestions.append("应收账款：加快回款周期")
+            suggestions.append("融资安排：准备短期授信额度")
+        else:
+            suggestions.append("维持现金流健康")
+    elif dimension == "growthDriver":
+        eg = float(entity.get("expenseGrowthRate", entity.get("expense_growth_rate", 0)))
+        if score <= 2:
+            suggestions.append(f"费用管控：费用增速{eg}倍于收入，需严控费用")
+            suggestions.append("预算管理：建立费用增长红线机制")
+            suggestions.append("投资回报：评估每项投入的ROI")
+        else:
+            suggestions.append("维持费用增长与收入增长匹配")
+    elif dimension == "orgEfficiency":
+        mr = float(entity.get("mgmtRatio", entity.get("mgmt_ratio", 0)))
+        if score <= 2:
+            suggestions.append(f"管理效率：管理费/净收入{mr}%，急需降本增效")
+            suggestions.append("组织精简：评估管理层级压缩空间")
+            suggestions.append("流程优化：推进数字化降本")
+        else:
+            suggestions.append("维持管理效率水平")
+    return suggestions
+
+
+def _get_dim_detail_indicators(dimension: str, entity: dict) -> list:
+    """获取维度的明细诊断指标"""
+    indicators = []
+    if dimension == "revenueStructure":
+        rebate = float(entity.get("rebateRate", entity.get("rebate_rate", 0)))
+        net_ratio = round(100 - rebate, 1)
+        indicators.append({"label": "渠补率", "value": f"{rebate}%", 
+                          "verdict": "收入依赖渠道返利" if rebate > 50 else "渠道依赖程度中等",
+                          "status": "danger" if rebate > 50 else "warning" if rebate > 20 else "success"})
+        indicators.append({"label": "净收入占比", "value": f"{net_ratio}%",
+                          "verdict": f"仅{net_ratio}%归公司" if net_ratio < 30 else "净收入占比合理",
+                          "status": "danger" if net_ratio < 30 else "success"})
+    elif dimension == "profitStructure":
+        gm = float(entity.get("trueGrossMargin", entity.get("true_gross_margin", 0)))
+        indicators.append({"label": "真实毛利率", "value": f"{gm}%",
+                          "verdict": "毛利偏低" if gm < 15 else "毛利正常",
+                          "status": "danger" if gm < 10 else "warning" if gm < 20 else "success"})
+    elif dimension == "cashAssets":
+        cr = float(entity.get("cashRatio", entity.get("cash_ratio", 0)))
+        indicators.append({"label": "现金比率", "value": f"{cr}%",
+                          "verdict": "断流风险" if cr < 5 else "现金紧张" if cr < 30 else "现金充足",
+                          "status": "danger" if cr < 5 else "warning" if cr < 30 else "success"})
+    elif dimension == "growthDriver":
+        eg = float(entity.get("expenseGrowthRate", entity.get("expense_growth_rate", 0)))
+        rg = float(entity.get("revenueGrowthRate", entity.get("revenue_growth_rate", 1)))
+        ratio = eg / rg if rg > 0 else 99
+        indicators.append({"label": "费用增速/收入增速", "value": f"{ratio:.1f}倍",
+                          "verdict": "费用增速过快" if ratio > 1.5 else "费用可控" if ratio > 1 else "增长健康",
+                          "status": "danger" if ratio > 1.5 else "warning" if ratio > 1 else "success"})
+    elif dimension == "orgEfficiency":
+        mr = float(entity.get("mgmtRatio", entity.get("mgmt_ratio", 0)))
+        indicators.append({"label": "管理费/净收入", "value": f"{mr}%",
+                          "verdict": "管理费极高" if mr > 200 else "管理费偏高" if mr > 100 else "管理费正常",
+                          "status": "danger" if mr > 200 else "warning" if mr > 100 else "success"})
+    return indicators
+
+
+@router.post("/growth-quality")
+def api_growth_quality(data: dict):
+    """增长质量诊断 — 五维度评分+综合评分+诊断结论"""
+    try:
+        entity = data.get("entity", data)
+        entity_name = entity.get("entity", entity.get("name", "该企业"))
+        period = entity.get("period", data.get("period", "当前"))
+
+        # 五维度评分
+        dim_scores = {
+            "revenueStructure": _score_revenue_structure(entity),
+            "profitStructure": _score_profit_structure(entity),
+            "cashAssets": _score_cash_assets(entity),
+            "growthDriver": _score_growth_driver(entity),
+            "orgEfficiency": _score_org_efficiency(entity),
+        }
+
+        overall = round(sum(dim_scores.values()) / 5, 1)
+
+        # 综合等级
+        if overall >= 4:
+            level = "🟢 越增长越轻"
+            level_type = "excellent"
+        elif overall >= 3:
+            level = "🟡 增长质量中等"
+            level_type = "medium"
+        elif overall >= 2:
+            level = "🟠 需关注"
+            level_type = "warning"
+        else:
+            level = "🔴 越增长越重"
+            level_type = "danger"
+
+        # 诊断结论
+        dimensions_payload = {}
+        detail_payload = {}
+        for dim, score in dim_scores.items():
+            dimensions_payload[dim] = {"score": score, "weight": 20}
+            detail_payload[dim] = {
+                "score": score,
+                "indicators": _get_dim_detail_indicators(dim, entity),
+                "suggestions": _generate_improvement_suggestions(dim, score, entity),
+            }
+
+        diagnosis = _diagnosis_text(overall, dimensions_payload, entity_name)
+
+        # 对比数据（如果请求中包含多个实体）
+        compare = data.get("compare", None)
+        compare_result = None
+        if compare:
+            compare_entity = compare
+            compare_name = compare_entity.get("entity", compare_entity.get("name", "对比企业"))
+            cdims = {
+                "revenueStructure": _score_revenue_structure(compare_entity),
+                "profitStructure": _score_profit_structure(compare_entity),
+                "cashAssets": _score_cash_assets(compare_entity),
+                "growthDriver": _score_growth_driver(compare_entity),
+                "orgEfficiency": _score_org_efficiency(compare_entity),
+            }
+            compare_overall = round(sum(cdims.values()) / 5, 1)
+            compare_result = {
+                "entity_name": compare_name,
+                "overall": compare_overall,
+                "dimensions": {k: {"score": v, "weight": 20} for k, v in cdims.items()},
+                "level": ("🟢 越增长越轻" if compare_overall >= 4 else
+                          "🟡 增长质量中等" if compare_overall >= 3 else
+                          "🟠 需关注" if compare_overall >= 2 else "🔴 越增长越重"),
+            }
+
+        return {
+            "entity_name": entity_name,
+            "period": period,
+            "overall": overall,
+            "level": level,
+            "level_type": level_type,
+            "dimensions": dimensions_payload,
+            "detail": detail_payload,
+            "diagnosis": diagnosis,
+            "compare": compare_result,
+        }
+    except Exception as e:
+        raise HTTPException(400, f"增长质量诊断失败: {str(e)}")