Files
cma-management/backend/app/utils/kpi_forecast_engine.py
T
Hermes CI Fix 6b6043536a feat: 预测性成本智能升级 — 历史回归弹性校准 + 预测偏差告警
升级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)
2026-08-25 00:55:50 +08:00

530 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""KPI预测引擎 — 基于历史KPI值做趋势预测(预测性成本智能 MVP)
模型(MVP原则:简单可用,不上深度学习):
- linear 线性回归(最小二乘 y = a + b·x),输出95%预测区间
- moving_average 简单移动平均(默认窗口3期),输出均值±波动区间
置信度诚实标注:基于历史数据量 + 拟合优度(R² / 波动率CV)综合打分,
数据不足时明确给出 low,不做虚假高置信。
复用 cash_forecast_engine.get_entity_kpi_history 取历史数据(不重复写查询)。
"""
import logging
import math
from typing import Optional
from sqlalchemy.orm import Session
from app.utils.cash_forecast_engine import get_entity_kpi_history, find_kpi
logger = logging.getLogger("cma.kpi_forecast")
MODELS = ("linear", "moving_average")
DEFAULT_PERIODS = 3
TREND_THRESHOLD_PCT = 3.0 # |趋势百分比| ≥ 3% 判定为有明确趋势方向
# 置信度档位
CONF_LEVELS = {3: "high", 2: "medium", 1: "low"}
# 中文映射(供 summary 使用)
TREND_CN = {"up": "上升", "down": "下降", "flat": "基本平稳"}
CONF_CN = {"high": "高", "medium": "中", "low": "低"}
def next_period(period: str, steps: int = 1) -> str:
"""期数递增:"2026-05" + 1 → "2026-06";解析失败时退化为 period+N"""
try:
y, m = str(period).split("-")
total = int(y) * 12 + (int(m) - 1) + steps
return f"{total // 12:04d}-{total % 12 + 1:02d}"
except Exception:
return f"{period}+{steps}"
def _t_crit(n: int) -> float:
"""95%双尾学生t临界值近似(小样本查表取保守值,大样本趋近1.96"""
table = {
2: 12.71, 3: 4.30, 4: 3.18, 5: 2.78, 6: 2.57, 7: 2.45,
8: 2.31, 9: 2.26, 10: 2.23, 12: 2.18, 15: 2.13,
20: 2.09, 30: 2.04, 60: 2.00,
}
for k in sorted(table):
if n <= k:
return table[k]
return 1.96
def _std(values: list) -> float:
"""样本标准差(n>=2),n==1 返回0"""
n = len(values)
if n < 2:
return 0.0
mean = sum(values) / n
return math.sqrt(sum((v - mean) ** 2 for v in values) / (n - 1))
def _rel_trend_pct(values: list) -> float:
"""趋势百分比 = 线性回归斜率 / |均值| × 100(与 cash_forecast_engine.calc_trend 同口径)"""
n = len(values)
if n < 2:
return 0.0
xbar = (n - 1) / 2.0
ybar = sum(values) / n
sxx = sum((i - xbar) ** 2 for i in range(n))
if sxx == 0:
return 0.0
slope = sum((i - xbar) * (values[i] - ybar) for i in range(n)) / sxx
return slope / max(abs(ybar), 1.0) * 100
def judge_trend(trend_pct: float, threshold: float = TREND_THRESHOLD_PCT) -> str:
"""趋势方向判定:up / down / flat"""
if trend_pct > threshold:
return "up"
if trend_pct < -threshold:
return "down"
return "flat"
def _compute_r2(values: list, pred_fn) -> float:
"""拟合优度 R²(0~1),数据无波动时视为完全拟合"""
ybar = sum(values) / len(values)
ss_tot = sum((v - ybar) ** 2 for v in values)
if ss_tot == 0:
return 1.0
ss_res = sum((v - pred_fn(i)) ** 2 for i, v in enumerate(values))
return max(0.0, 1.0 - ss_res / ss_tot)
def compute_confidence(n: int, model: str, r2: Optional[float] = None,
cv: Optional[float] = None) -> str:
"""置信度诚实标注:数据量基数 + 拟合优度修正
- 数据量:n>=12 → 3分;n>=6 → 2分;否则 1分
- linearR²>=0.7 +1R²<0.3 -1
- moving_averageCV<0.3 +1(低波动更可信);CV>0.6 -1
"""
score = 3 if n >= 12 else (2 if n >= 6 else 1)
# 拟合度修正仅在样本量足够时生效:
# n<4 时 R² 无统计意义(2点直线必然R²=1.0),CV 也噪声大,不做上调,避免虚假高置信
if n >= 4:
if model == "linear" and r2 is not None:
if r2 >= 0.7:
score += 1
elif r2 < 0.3:
score -= 1
elif model == "moving_average" and cv is not None:
if cv < 0.3:
score += 1
elif cv > 0.6:
score -= 1
score = max(1, min(3, score))
return CONF_LEVELS[score]
def linear_forecast(values: list, periods: int = 3) -> dict:
"""线性回归预测 — 返回未来periods期预测值 + 95%预测区间 + 拟合统计量"""
n = len(values)
x = list(range(n))
xbar = (n - 1) / 2.0
ybar = sum(values) / n
sxx = sum((i - xbar) ** 2 for i in x)
slope = sum((i - xbar) * (values[i] - ybar) for i in x) / sxx if sxx else 0.0
intercept = ybar - slope * xbar
def pred(i: int) -> float:
return intercept + slope * i
# 残差标准误(n>=3 用 n-2 自由度;n==2 用样本标准差近似)
if n >= 3:
resid = [values[i] - pred(i) for i in x]
se = math.sqrt(sum(r * r for r in resid) / (n - 2))
else:
se = _std(values)
if se == 0:
se = max(abs(ybar) * 0.05, 1e-9) # 完全拟合时给最小带,避免零宽区间
t_crit = _t_crit(n)
forecast = []
for k in range(periods):
x0 = n + k
predicted = pred(x0)
se_pred = se * math.sqrt(1.0 + 1.0 / n + (x0 - xbar) ** 2 / max(sxx, 1e-9)) * t_crit
band = max(se_pred, abs(predicted) * 0.02)
forecast.append({
"predicted": round(predicted, 2),
"lower": round(predicted - band, 2),
"upper": round(predicted + band, 2),
})
r2 = _compute_r2(values, pred)
trend_pct = slope / max(abs(ybar), 1.0) * 100
return {
"forecast": forecast,
"slope": slope,
"intercept": intercept,
"r2": round(r2, 3),
"trend_pct": round(trend_pct, 2),
"se": round(se, 4),
}
def moving_average_forecast(values: list, periods: int = 3, window: int = 3) -> dict:
"""简单移动平均预测 — 未来各期预测值 = 最近window期均值;区间=均值±1.96×波动"""
n = len(values)
w = max(1, min(window, n))
base = sum(values[-w:]) / w
std = _std(values)
if std == 0:
std = max(abs(base) * 0.05, 1e-9)
band = max(1.96 * std, abs(base) * 0.02)
forecast = [{
"predicted": round(base, 2),
"lower": round(base - band, 2),
"upper": round(base + band, 2),
} for _ in range(periods)]
cv = std / abs(base) if base else 0.0
trend_pct = _rel_trend_pct(values)
return {
"forecast": forecast,
"window": w,
"mean": round(base, 2),
"std": round(std, 4),
"cv": round(cv, 3),
"trend_pct": round(trend_pct, 2),
}
def build_summary(kpi_name: str, unit: str, trend: str, next_target: Optional[float],
periods: int, n_history: int, confidence: str, model: str) -> str:
"""中文一句话解读"""
trend_cn = TREND_CN.get(trend, trend)
conf_cn = CONF_CN.get(confidence, confidence)
unit_txt = unit or ""
if periods <= 0:
return f"基于{n_history}期历史数据,{kpi_name}当前趋势{trend_cn}(模型:{model},置信度:{conf_cn}),未请求未来期数预测"
target_txt = f"{next_target:,.2f}{unit_txt}" if next_target is not None else "—"
return (
f"基于{n_history}期历史数据,{kpi_name}未来{periods}期预计{trend_cn}"
f"下一期预测值约{target_txt}(模型:{model},置信度:{conf_cn}"
)
def forecast_kpi(entity_id: int, kpi_code: str, db: Session,
periods: int = DEFAULT_PERIODS, model: str = "linear") -> Optional[dict]:
"""单个KPI预测(多租户隔离:历史数据通过 entity_id 维度查询)
返回 None 表示 KPI 不存在或历史数据不足(<2条)。
"""
if model not in MODELS:
model = "linear"
history = get_entity_kpi_history(entity_id, kpi_code, db, limit_months=120)
if not history:
return None
hist_asc = list(reversed(history)) # 按 period 升序
values = [float(v.actual_value) for v in hist_asc if v.actual_value is not None]
if len(values) < 2:
return None
kpi_def = find_kpi(db, entity_id, [kpi_code])
kpi_name = str(kpi_def.kpi_name) if kpi_def else kpi_code
unit = str(kpi_def.unit or "") if kpi_def else ""
if model == "moving_average":
res = moving_average_forecast(values, periods)
confidence = compute_confidence(len(values), model, cv=res["cv"])
else:
res = linear_forecast(values, periods)
confidence = compute_confidence(len(values), model, r2=res["r2"])
trend = judge_trend(res["trend_pct"])
# 未来期数(基于最近一期 period 递增)
last_period = hist_asc[-1].period
forecast = []
for k in range(periods):
fp = res["forecast"][k]
forecast.append({
"period": next_period(last_period, k + 1),
"predicted": fp["predicted"],
"lower": fp["lower"],
"upper": fp["upper"],
})
next_target = forecast[0]["predicted"] if forecast else None
summary = build_summary(kpi_name, unit, trend, next_target, periods,
len(values), confidence, model)
return {
"entity_id": entity_id,
"kpi": {"code": kpi_code, "name": kpi_name, "unit": unit},
"model": model,
"periods": periods,
"trend": trend,
"trend_pct": res["trend_pct"],
"confidence": confidence,
"history_count": len(values),
"history": [{"period": v.period, "value": round(float(v.actual_value), 2)} for v in hist_asc],
"forecast": forecast,
"next_target": next_target,
"summary": summary,
}
def forecast_finance_kpis(entity_id: int, db: Session,
periods: int = DEFAULT_PERIODS, model: str = "linear",
min_history: int = 3) -> list:
"""批量预测该企业全部财务维度KPI(历史≥min_history条),按可预测性排序"""
from app.models import KPIDefinition
kpis = db.query(KPIDefinition).filter(
KPIDefinition.entity_id == entity_id,
KPIDefinition.dimension == "finance",
KPIDefinition.status == "active",
).all()
results = []
for kpi in kpis:
r = forecast_kpi(entity_id, str(kpi.kpi_code), db, periods=periods, model=model)
if r and r["history_count"] >= min_history:
results.append(r)
# 可预测性排序:置信度(high=3/medium=2/low=1) 优先,其次历史数据量
score = {"high": 3, "medium": 2, "low": 1}
results.sort(key=lambda r: (score.get(r["confidence"], 0), r["history_count"]), reverse=True)
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 = [
{"key": "oil", "name": "原油价格", "unit": "美元/桶",
"desc": "油价↑ → 运输/能源成本↑ → 成本类KPI↑、利润类KPI↓"},
{"key": "usd", "name": "美元汇率", "unit": "USD/CNY",
"desc": "美元↑ → 进口成本↑(成本类↑)、出口收入↑(营收类↑)"},
{"key": "cpi", "name": "CPI通胀率", "unit": "%",
"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": { # 成本/费用类: 宏观涨 → 成本涨
"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,
"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% → 调整后预测值: 方向+ 因素涨预测涨; 方向- 因素涨预测跌
负值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)