feat: 预测性成本智能·宏观敏感性因素联动(IMA 2026.7完整版)

- 内置3宏观因素: 原油价格/美元汇率/CPI通胀率 - 敏感性引擎: 按KPI类别推断弹性(成本类油价0.15/利润类0.12/营收类0.08), 方向+因素涨KPI涨 - 负值KPI(亏损)方向反转修复: 油价涨→净利更亏 - API: GET /predict/kpi-forecast/sensitivity?pct=10 → KPI×因素矩阵(±pct调整后预测) - 前端: 敏感性幅度选择(±5/10/20%) + 敏感性矩阵表(同向/反向+↑↓调整值) - 诚实标注: 模型弹性(规则推断,非历史回归), 后续可用宏观历史数据回归校准 - 验证: Chrome实测页面+API矩阵, pytest 46 passed
2026-08-25 00:44:45 +08:00
parent 1c8b01d682
commit 8ec846c6df
4 changed files with 217 additions and 3 deletions
@@ -1,6 +1,6 @@
 """预测模拟API — 管理会计OS"""
 import logging
-from fastapi import APIRouter, HTTPException, Depends, Request
+from fastapi import APIRouter, HTTPException, Depends, Request, Query
 from app.utils.predict_engine import (
    cvp_analysis, npv, irr,
    sensitivity_analysis, scenario_analysis,
@@ -719,6 +719,7 @@ def api_growth_quality(request: Request, data: dict):
 # ── KPI趋势预测（预测性成本智能 MVP） ────────────────────────────
 from app.utils.kpi_forecast_engine import (  # noqa: E402
    MODELS, forecast_kpi, forecast_finance_kpis,
+    MACRO_FACTORS, factor_sensitivity_for_kpi, adjusted_next_with_factor,
 )


@@ -764,3 +765,56 @@ def api_kpi_forecast_finance(
        "total": len(results),
        "data": results,
    }
+
+
+@router.get("/kpi-forecast/sensitivity")
+def api_kpi_forecast_sensitivity(
+    pct: float = Query(10, description="宏观因素变动幅度% (±)"),
+    periods: int = Query(3),
+    model: str = Query("linear"),
+    entity_id: int = Depends(get_entity_id),
+    db: Session = Depends(get_db),
+):
+    """宏观敏感性因素联动（IMA 2026.7）— 财务KPI × 宏观因素(油价/汇率/CPI)敏感性矩阵
+    输出：每个KPI的预测值 + 各因素 ±pct% 情景下的调整后预测值
+    MVP：弹性系数为规则推断（按KPI类别），诚实标注"模型弹性"非历史回归"""
+    if abs(pct) > 50:
+        raise HTTPException(400, "pct 必须在 ±50 以内")
+    if model not in MODELS:
+        raise HTTPException(400, f"不支持的模型: {model}，可选: {'/'.join(MODELS)}")
+    results = forecast_finance_kpis(entity_id, db, periods=periods, model=model)
+    matrix = []
+    for r in results:
+        kpi_info = r.get("kpi", {})
+        sens = factor_sensitivity_for_kpi(kpi_info.get("name", ""), kpi_info.get("code", ""))
+        next_val = r.get("next_target")
+        factor_effects = []
+        for s in sens:
+            up_val = adjusted_next_with_factor(next_val, pct, s["direction"], s["elasticity"])
+            down_val = adjusted_next_with_factor(next_val, -pct, s["direction"], s["elasticity"])
+            factor_effects.append({
+                "factor_key": s["factor_key"],
+                "factor_name": s["factor_name"],
+                "factor_unit": s["factor_unit"],
+                "direction": s["direction"],
+                "elasticity": s["elasticity"],
+                "adj_up": up_val,
+                "adj_down": down_val,
+            })
+        matrix.append({
+            "kpi": kpi_info,
+            "category": sens[0]["category"] if sens else "profit",
+            "next_target": next_val,
+            "confidence": r.get("confidence"),
+            "trend": r.get("trend"),
+            "factors": factor_effects,
+        })
+    return {
+        "entity_id": entity_id,
+        "model": model,
+        "periods": periods,
+        "pct": pct,
+        "factors": MACRO_FACTORS,
+        "total": len(matrix),
+        "data": matrix,
+    }
@@ -293,3 +293,92 @@ def forecast_finance_kpis(entity_id: int, db: Session,
    score = {"high": 3, "medium": 2, "low": 1}
    results.sort(key=lambda r: (score.get(r["confidence"], 0), r["history_count"]), reverse=True)
    return results
+
+
+# ════════════════════════════════════════════════════════════
+# 宏观敏感性因素联动（IMA 2026.7 Predictive Cost Intelligence 完整版）
+# 内置宏观因素 → 按KPI类型推断弹性系数 → 调整预测值
+# MVP：弹性系数为规则推断+可调，非历史回归（诚实标注"模型弹性"）
+# ════════════════════════════════════════════════════════════
+
+MACRO_FACTORS = [
+    {"key": "oil", "name": "原油价格", "unit": "美元/桶",
+     "desc": "油价↑ → 运输/能源成本↑ → 成本类KPI↑、利润类KPI↓"},
+    {"key": "usd", "name": "美元汇率", "unit": "USD/CNY",
+     "desc": "美元↑ → 进口成本↑(成本类↑)、出口收入↑(营收类↑)"},
+    {"key": "cpi", "name": "CPI通胀率", "unit": "%",
+     "desc": "CPI↑ → 成本↑、名义营收↑"},
+]
+
+# KPI 类别关键词 → 因素方向/弹性 (direction: +因素涨KPI涨, -因素涨KPI跌)
+FACTOR_RULES = {
+    "cost": {  # 成本/费用类: 宏观涨 → 成本涨
+        "oil": {"direction": "+", "elasticity": 0.15},
+        "usd": {"direction": "+", "elasticity": 0.10},
+        "cpi": {"direction": "+", "elasticity": 0.10},
+    },
+    "revenue": {  # 营收类: 通胀涨→名义营收涨
+        "oil": {"direction": "-", "elasticity": 0.05},
+        "usd": {"direction": "+", "elasticity": 0.08},
+        "cpi": {"direction": "+", "elasticity": 0.08},
+    },
+    "profit": {  # 利润类: 宏观涨 → 成本挤压利润
+        "oil": {"direction": "-", "elasticity": 0.12},
+        "usd": {"direction": "-", "elasticity": 0.08},
+        "cpi": {"direction": "-", "elasticity": 0.08},
+    },
+    "cash": {  # 现金流类
+        "oil": {"direction": "-", "elasticity": 0.06},
+        "usd": {"direction": "-", "elasticity": 0.04},
+        "cpi": {"direction": "-", "elasticity": 0.05},
+    },
+}
+
+# 类别关键词匹配（长词优先）
+CATEGORY_KEYWORDS = [
+    ("profit", ["净利润", "净利", "利润", "毛利", "ROE", "ROI", "EVA", "收益率", "报酬率"]),
+    ("revenue", ["营收", "收入", "销售额", "销售", "产值", "客单"]),
+    ("cost", ["费用率", "成本率", "费用", "成本", "费率", "应付", "返利", "渠补", "税"]),
+    ("cash", ["现金流", "现金", "回款", "FCF", "资金"]),
+]
+
+
+def infer_kpi_category(kpi_name: str, kpi_code: str = "") -> str:
+    """按KPI名称/编码推断类别: profit/revenue/cost/cash，兜底 profit（保守）"""
+    n = (kpi_name or "") + " " + (kpi_code or "")
+    for cat, kws in CATEGORY_KEYWORDS:
+        if any(kw in n for kw in kws):
+            return cat
+    return "profit"
+
+
+def factor_sensitivity_for_kpi(kpi_name: str, kpi_code: str = "") -> list:
+    """返回该KPI对3个宏观因素的敏感性（方向+弹性）"""
+    cat = infer_kpi_category(kpi_name, kpi_code)
+    rules = FACTOR_RULES.get(cat, FACTOR_RULES["profit"])
+    out = []
+    for f in MACRO_FACTORS:
+        r = rules.get(f["key"], {"direction": "-", "elasticity": 0.05})
+        out.append({
+            "factor_key": f["key"],
+            "factor_name": f["name"],
+            "factor_unit": f["unit"],
+            "factor_desc": f["desc"],
+            "direction": r["direction"],
+            "elasticity": r["elasticity"],
+            "category": cat,
+        })
+    return out
+
+
+def adjusted_next_with_factor(next_target: Optional[float], pct: float,
+                              direction: str, elasticity: float) -> Optional[float]:
+    """因素变动 pct% → 调整后预测值: 方向+ 因素涨预测涨; 方向- 因素涨预测跌
+    负值KPI(亏损)方向反转: 方向- 时因素涨 → 更亏(更负)"""
+    if next_target is None:
+        return None
+    factor_change = pct * 0.01  # ±5% → 0.05
+    sign = 1.0 if direction == "+" else -1.0
+    if next_target < 0:
+        sign = -sign  # 负值(亏损): 因素涨 → 更亏
+    return round(next_target * (1 + sign * factor_change * elasticity), 2)