From 6b6043536a5aa64a8b3f349ee021b04693dc1ac9 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Tue, 25 Aug 2026 00:55:50 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=A2=84=E6=B5=8B=E6=80=A7=E6=88=90?= =?UTF-8?q?=E6=9C=AC=E6=99=BA=E8=83=BD=E5=8D=87=E7=BA=A7=20=E2=80=94=20?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E5=9B=9E=E5=BD=92=E5=BC=B9=E6=80=A7=E6=A0=A1?= =?UTF-8?q?=E5=87=86=20+=20=E9=A2=84=E6=B5=8B=E5=81=8F=E5=B7=AE=E5=91=8A?= =?UTF-8?q?=E8=AD=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 升级1: 宏观敏感性弹性历史校准 - 内置宏观历史数据(oil/usd/cpi 2026-01~07月度) - 变化率弹性: 同period匹配KPI历史vs因素历史算弹性 - 合理性校验: |弹性|超出[0.01,0.5]视为噪声回退规则(诚实标注) 升级2: 预测偏差告警闭环 - 新表 kpi_forecast_log(预测历史)+模型KpiForecastLog - 预测时落库(同KPI同预测期覆盖) - alert_rules 支持 rule_type=forecast_deviation(threshold_pct) - POST /alert-rules/run-forecast-deviation: 预测vs实际偏差>阈值生成预警(去重, 超2倍阈值红色) - 端到端验证: 模拟实际500vs预测399.55→偏差20.1%>5%→红色预警生成 回归: pytest 40 passed(predict+alerts) --- backend/app/api/alert_rules.py | 77 +++++++++++- backend/app/api/predict.py | 14 ++- backend/app/models/__init__.py | 15 +++ backend/app/utils/kpi_forecast_engine.py | 147 ++++++++++++++++++++++- 4 files changed, 249 insertions(+), 4 deletions(-) diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index 18266c11..e664261b 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -107,7 +107,7 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not kpi: raise HTTPException(404, "KPI不存在") - if rule_type not in ("static", "dynamic", "trend_up", "trend_down"): + if rule_type not in ("static", "dynamic", "trend_up", "trend_down", "forecast_deviation"): # 升级2b: 预测偏差 raise HTTPException(400, f"不支持的规则类型: {rule_type}") rule = AlertRule( @@ -619,3 +619,78 @@ def generate_alert_suggestions(db: Session = Depends(get_db)): db.commit() return {"message": f"已为{updated}条预警生成情景建议", "updated": updated} + + +@router.post("/run-forecast-deviation") +def run_forecast_deviation_check( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """预测偏差检查（升级2b, 2026-08-25）— alert_rules type=forecast_deviation + 对每条偏差规则: 取最新预测log(kpi_forecast_log) vs 该期实际值(kpi_values)， + 偏差 > threshold_pct → 生成/更新 pending 预警（去重）""" + from app.models import KpiForecastLog + rules = db.query(AlertRule).filter( + AlertRule.entity_id == entity_id, + AlertRule.rule_type == "forecast_deviation", + AlertRule.enabled == 1, + ).all() + if not rules: + return {"message": "无预测偏差规则，可先创建 rule_type=forecast_deviation 规则", "generated": 0} + + generated = 0 + for rule in rules: + try: + params = rule.params or {} + threshold = float(params.get("threshold_pct", 15)) + # 最新预测 + log = db.query(KpiForecastLog).filter( + KpiForecastLog.entity_id == entity_id, + KpiForecastLog.kpi_id == rule.kpi_id, + ).order_by(KpiForecastLog.created_at.desc()).first() + if not log or log.forecast_value is None: + continue + # 该预测期的实际值（同period匹配；兼容 2026-H1 等半年度） + actual = db.query(KPIValue).filter( + KPIValue.kpi_id == rule.kpi_id, + KPIValue.period == log.period, + ).order_by(KPIValue.id.desc()).first() + if not actual or not actual.actual_value: + continue + base = abs(actual.actual_value) + if base < 1e-9: + continue + deviation_pct = abs(log.forecast_value - actual.actual_value) / base * 100 + if deviation_pct <= threshold: + continue + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first() + kpi_label = f"{kpi.kpi_name}({kpi.kpi_code})" if kpi else f"KPI#{rule.kpi_id}" + alert_level = "red" if deviation_pct > threshold * 2 else "yellow" + alert_message = ( + f"预测偏差 {deviation_pct:.1f}% > 阈值{threshold}%：" + f"{kpi_label} 预测{log.period}={log.forecast_value}，实际={actual.actual_value}" + ) + # 去重: 同KPI+period 已有 pending 偏差预警 + existing = db.query(KPIAlert).filter( + KPIAlert.kpi_id == rule.kpi_id, + KPIAlert.alert_message.like(f"%预测偏差%{log.period}%"), + KPIAlert.status == "pending", + ).first() + if existing: + existing.alert_message = alert_message + existing.alert_level = alert_level + else: + db.add(KPIAlert( + kpi_id=rule.kpi_id, + kpi_value_id=actual.id, + alert_level=alert_level, + alert_message=alert_message, + alert_type="forecast", + status="pending", + )) + generated += 1 + except Exception as e: + logger.error(f"预测偏差检查失败 rule_id={rule.id}: {e}") + continue + db.commit() + return {"message": f"预测偏差检查完成: {generated}条", "generated": generated} diff --git a/backend/app/api/predict.py b/backend/app/api/predict.py index f49fad6d..ce005f8e 100644 --- a/backend/app/api/predict.py +++ b/backend/app/api/predict.py @@ -719,7 +719,8 @@ 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, + MACRO_FACTORS, factor_sensitivity_for_kpi, factor_sensitivity_with_history, + adjusted_next_with_factor, save_forecast_logs, ) @@ -758,6 +759,10 @@ def api_kpi_forecast_finance( if model not in MODELS: raise HTTPException(400, f"不支持的模型: {model}，可选: {'/'.join(MODELS)}") results = forecast_finance_kpis(entity_id, db, periods=periods, model=model) + try: + save_forecast_logs(entity_id, results, db, model=model) # 升级2a: 预测落库(供偏差告警) + except Exception as e: + logger.warning(f"预测落库失败(不影响返回): {e}") return { "entity_id": entity_id, "model": model, @@ -786,7 +791,9 @@ def api_kpi_forecast_sensitivity( matrix = [] for r in results: kpi_info = r.get("kpi", {}) - sens = factor_sensitivity_for_kpi(kpi_info.get("name", ""), kpi_info.get("code", "")) + # v2: 有历史数据用变化率弹性校准，无数据回退规则推断 + sens = factor_sensitivity_with_history( + kpi_info.get("name", ""), kpi_info.get("code", ""), r.get("history", [])) next_val = r.get("next_target") factor_effects = [] for s in sens: @@ -798,6 +805,9 @@ def api_kpi_forecast_sensitivity( "factor_unit": s["factor_unit"], "direction": s["direction"], "elasticity": s["elasticity"], + "elasticity_source": s.get("elasticity_source", "rule"), + "matched_periods": s.get("matched_periods"), + "rule_direction": s.get("rule_direction"), "adj_up": up_val, "adj_down": down_val, }) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 64ba8309..e8c69e41 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -306,6 +306,21 @@ class MpmResult(Base): created_at = Column(DateTime, server_default=func.now()) +class KpiForecastLog(Base): + """KPI预测历史 — 预测偏差告警数据源 (2026-08-25 升级2a)""" + __tablename__ = "kpi_forecast_log" + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False, comment="企业ID") + kpi_id = Column(Integer, nullable=False, comment="KPI ID") + kpi_code = Column(String(50), nullable=False, comment="KPI编码") + period = Column(String(20), nullable=False, comment="预测期间") + forecast_value = Column(Float, nullable=True, comment="预测值") + model = Column(String(30), default="linear", comment="预测模型") + confidence = Column(String(10), nullable=True, comment="置信度") + trend = Column(String(10), nullable=True, comment="趋势") + created_at = Column(DateTime, server_default=func.now()) + + class BotBridgeConfig(Base): """Bot桥接鉴权配置""" __tablename__ = "bot_bridge_config" diff --git a/backend/app/utils/kpi_forecast_engine.py b/backend/app/utils/kpi_forecast_engine.py index 362fce2c..c2563794 100644 --- a/backend/app/utils/kpi_forecast_engine.py +++ b/backend/app/utils/kpi_forecast_engine.py @@ -295,10 +295,55 @@ def forecast_finance_kpis(entity_id: int, db: Session, return results +def save_forecast_logs(entity_id: int, results: list, db: Session, model: str = "linear") -> int: + """预测结果落库 kpi_forecast_log（预测偏差告警数据源, 2026-08-25 升级2a） + 存每KPI的下一期预测；同KPI同预测期覆盖（保留最新）""" + from app.models import KpiForecastLog, KPIDefinition + # 预加载 KPI id 映射（返回结果里的 kpi 无 id 字段，需从DB查） + kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id).all()} + saved = 0 + for r in results: + kpi_info = r.get("kpi", {}) + fc = r.get("forecast") or [] + if not fc: + continue + first = fc[0] + period = first.get("period", "") + val = first.get("predicted") or first.get("value") or first.get("next_value") + if not period or val is None: + continue + kpi_code = kpi_info.get("code", "") + kpi_id = kpi_info.get("id") or kpi_map.get(kpi_code) + if not kpi_id: + continue + # 同KPI同预测期覆盖 + existing = db.query(KpiForecastLog).filter( + KpiForecastLog.entity_id == entity_id, + KpiForecastLog.kpi_id == kpi_id, + KpiForecastLog.period == period, + ).first() + if existing: + existing.forecast_value = float(val) + existing.model = model + existing.confidence = r.get("confidence") + existing.trend = r.get("trend") + else: + db.add(KpiForecastLog( + entity_id=entity_id, kpi_id=kpi_id, kpi_code=kpi_code, + period=period, forecast_value=float(val), + model=model, confidence=r.get("confidence"), trend=r.get("trend"), + )) + saved += 1 + db.commit() + return saved + + # ════════════════════════════════════════════════════════════ # 宏观敏感性因素联动（IMA 2026.7 Predictive Cost Intelligence 完整版） # 内置宏观因素 → 按KPI类型推断弹性系数 → 调整预测值 # MVP：弹性系数为规则推断+可调，非历史回归（诚实标注"模型弹性"） +# v2(2026-08-25): 内置宏观历史数据 → 变化率弹性校准（有数据用回归，无数据回退规则） # ════════════════════════════════════════════════════════════ MACRO_FACTORS = [ @@ -310,6 +355,28 @@ MACRO_FACTORS = [ "desc": "CPI↑ → 成本↑、名义营收↑"}, ] +# 内置宏观因素历史数据（月度，2026-01 ~ 2026-07，供变化率弹性校准） +MACRO_FACTOR_HISTORY = { + "oil": [ + {"period": "2026-01", "value": 74.0}, {"period": "2026-02", "value": 78.0}, + {"period": "2026-03", "value": 76.0}, {"period": "2026-04", "value": 82.0}, + {"period": "2026-05", "value": 79.0}, {"period": "2026-06", "value": 85.0}, + {"period": "2026-07", "value": 88.0}, + ], + "usd": [ + {"period": "2026-01", "value": 7.05}, {"period": "2026-02", "value": 7.08}, + {"period": "2026-03", "value": 7.06}, {"period": "2026-04", "value": 7.10}, + {"period": "2026-05", "value": 7.12}, {"period": "2026-06", "value": 7.15}, + {"period": "2026-07", "value": 7.18}, + ], + "cpi": [ + {"period": "2026-01", "value": 1.8}, {"period": "2026-02", "value": 1.9}, + {"period": "2026-03", "value": 1.9}, {"period": "2026-04", "value": 2.0}, + {"period": "2026-05", "value": 2.1}, {"period": "2026-06", "value": 2.1}, + {"period": "2026-07", "value": 2.2}, + ], +} + # KPI 类别关键词 → 因素方向/弹性 (direction: +因素涨KPI涨, -因素涨KPI跌) FACTOR_RULES = { "cost": { # 成本/费用类: 宏观涨 → 成本涨 @@ -353,7 +420,7 @@ def infer_kpi_category(kpi_name: str, kpi_code: str = "") -> str: def factor_sensitivity_for_kpi(kpi_name: str, kpi_code: str = "") -> list: - """返回该KPI对3个宏观因素的敏感性（方向+弹性）""" + """返回该KPI对3个宏观因素的敏感性（方向+弹性）— 规则推断版""" cat = infer_kpi_category(kpi_name, kpi_code) rules = FACTOR_RULES.get(cat, FACTOR_RULES["profit"]) out = [] @@ -367,10 +434,88 @@ def factor_sensitivity_for_kpi(kpi_name: str, kpi_code: str = "") -> list: "direction": r["direction"], "elasticity": r["elasticity"], "category": cat, + "elasticity_source": "rule", }) return out +def _rate_of_change(series: list) -> list: + """相邻期变化率列表 [(period, pct), ...]""" + out = [] + for i in range(1, len(series)): + prev, cur = series[i - 1], series[i] + if prev and prev.get("value"): + pct = (cur["value"] - prev["value"]) / prev["value"] * 100 + out.append((cur["period"], pct)) + return out + + +def elasticity_from_history(kpi_history: list, factor_key: str, + direction: str) -> Optional[dict]: + """变化率弹性校准：KPI历史 vs 宏观因素历史（同period匹配） + 弹性 = mean(KPI变化率 / 因素变化率)（符号由实际数据决定） + 匹配期数 < 2 或无因素数据 → 返回 None（回退规则） + """ + factor_hist = MACRO_FACTOR_HISTORY.get(factor_key) + if not factor_hist or not kpi_history: + return None + kpi_by_period = {h.get("period"): h.get("value") for h in kpi_history if h.get("value") is not None} + ratios = [] + # 因素相邻期变化率 + for i in range(1, len(factor_hist)): + fp = factor_hist[i]["period"] + fv = factor_hist[i]["value"] + fv_prev = factor_hist[i - 1]["value"] + if not fv_prev: + continue + f_chg = (fv - fv_prev) / fv_prev * 100 + # KPI 同期值（以及上一期，用于算KPI变化） + k_cur = kpi_by_period.get(fp) + # KPI 在因素上一期的值（模糊匹配上一月度） + k_prev = kpi_by_period.get(factor_hist[i - 1]["period"]) + if k_cur is not None and k_prev not in (None, 0) and abs(f_chg) > 0.01: + k_chg = (k_cur - k_prev) / k_prev * 100 + ratios.append(k_chg / f_chg) + if len(ratios) < 2: + return None + import statistics + raw_elasticity = statistics.median(ratios) + # 弹性合理性校验: |弹性| 超出 [0.01, 0.5] 视为数据噪声 → 回退规则推断（诚实标注，不用失真校准） + if not (0.01 <= abs(raw_elasticity) <= 0.5): + return None + elasticity = round(raw_elasticity, 4) + # 方向由数据符号决定；数据符号与规则方向冲突时以数据为准（标注） + data_direction = "+" if elasticity >= 0 else "-" + return { + "elasticity": abs(elasticity), + "direction": data_direction, + "matched_periods": len(ratios), + "elasticity_source": "history", + "rule_direction": direction, + } + + +def factor_sensitivity_with_history(kpi_name: str, kpi_code: str = "", + kpi_history: Optional[list] = None) -> list: + """增强版敏感性：有历史数据用变化率弹性校准，无数据回退规则推断""" + base = factor_sensitivity_for_kpi(kpi_name, kpi_code) + out = [] + for s in base: + hist_el = elasticity_from_history(kpi_history or [], s["factor_key"], s["direction"]) if kpi_history else None + if hist_el: + out.append({ + **s, + "elasticity": hist_el["elasticity"], + "direction": hist_el["direction"], + "elasticity_source": hist_el["elasticity_source"], + "matched_periods": hist_el["matched_periods"], + "rule_direction": hist_el["rule_direction"], + }) + else: + out.append(s) + return out + + def adjusted_next_with_factor(next_target: Optional[float], pct: float, direction: str, elasticity: float) -> Optional[float]: """因素变动 pct% → 调整后预测值: 方向+ 因素涨预测涨; 方向- 因素涨预测跌