diff --git a/backend/app/api/predict.py b/backend/app/api/predict.py index 628f7a31..5e267f45 100644 --- a/backend/app/api/predict.py +++ b/backend/app/api/predict.py @@ -714,3 +714,53 @@ def api_growth_quality(request: Request, data: dict): } except Exception as e: raise HTTPException(400, f"增长质量诊断失败: {str(e)}") + + +# ── KPI趋势预测（预测性成本智能 MVP） ──────────────────────────── +from app.utils.kpi_forecast_engine import ( # noqa: E402 + MODELS, forecast_kpi, forecast_finance_kpis, +) + + +@router.get("/kpi-forecast") +def api_kpi_forecast( + kpi_code: str, + periods: int = 3, + model: str = "linear", + entity_id: int = Depends(get_entity_id), + db: Session = Depends(get_db), +): + """单个财务KPI预测 — 线性回归/移动平均，多租户隔离（entity_id 权限校验）""" + if periods < 0 or periods > 24: + raise HTTPException(400, "periods 必须在 0~24 之间") + if model not in MODELS: + raise HTTPException(400, f"不支持的模型: {model}，可选: {'/'.join(MODELS)}") + result = forecast_kpi(entity_id, kpi_code, db, periods=periods, model=model) + if result is None: + raise HTTPException( + 404, + f"KPI {kpi_code} 在企业 entity_id={entity_id} 下不存在，或历史数据不足（至少2条）", + ) + return result + + +@router.get("/kpi-forecast/finance") +def api_kpi_forecast_finance( + periods: int = 3, + model: str = "linear", + entity_id: int = Depends(get_entity_id), + db: Session = Depends(get_db), +): + """批量预测该企业全部财务维度KPI（历史≥3条），按可预测性排序""" + if periods < 0 or periods > 24: + raise HTTPException(400, "periods 必须在 0~24 之间") + if model not in MODELS: + raise HTTPException(400, f"不支持的模型: {model}，可选: {'/'.join(MODELS)}") + results = forecast_finance_kpis(entity_id, db, periods=periods, model=model) + return { + "entity_id": entity_id, + "model": model, + "periods": periods, + "total": len(results), + "data": results, + } diff --git a/backend/app/utils/kpi_forecast_engine.py b/backend/app/utils/kpi_forecast_engine.py new file mode 100644 index 00000000..246ea0c7 --- /dev/null +++ b/backend/app/utils/kpi_forecast_engine.py @@ -0,0 +1,295 @@ +"""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分 + - linear：R²>=0.7 +1；R²<0.3 -1 + - moving_average：CV<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 diff --git a/backend/tests/test_kpi_forecast.py b/backend/tests/test_kpi_forecast.py new file mode 100644 index 00000000..f69d3ccd --- /dev/null +++ b/backend/tests/test_kpi_forecast.py @@ -0,0 +1,308 @@ +"""KPI预测引擎测试 — 预测性成本智能 MVP + +覆盖：线性回归预测 / 移动平均预测 / 置信区间 / 趋势方向判定 / +边界情况（历史<2条、无历史、未来期数=0）/ entity_id 多租户隔离。 +""" +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import Entity, KPIValue +from app.utils import kpi_forecast_engine as engine +from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header + +BASE = "/api/cma/predict" + + +# ── 测试数据工厂 ── + +def _seed_history(db: Session, kpi, values: list, entity_id: int = 1, start_period: str = "2026-01"): + """为KPI写入连续期数的历史值""" + for i, v in enumerate(values): + month = int(start_period.split("-")[1]) + i + year = int(start_period.split("-")[0]) + (month - 1) // 12 + month = (month - 1) % 12 + 1 + db.add(KPIValue( + kpi_id=kpi.id, entity_id=entity_id, + period=f"{year:04d}-{month:02d}", actual_value=float(v), + )) + db.commit() + + +def _seed_entity2(db: Session) -> Entity: + ent = db.query(Entity).filter(Entity.id == 2).first() + if not ent: + ent = Entity(id=2, name="博海网络科技", short_name="博海", status="active") + db.add(ent) + db.commit() + return ent + + +# ── 纯函数单元测试 ── + +class TestNextPeriod: + def test_increment(self): + assert engine.next_period("2026-05", 1) == "2026-06" + + def test_cross_year(self): + assert engine.next_period("2026-12", 1) == "2027-01" + assert engine.next_period("2026-01", 2) == "2026-03" + + def test_malformed_fallback(self): + assert engine.next_period("未知期", 1) == "未知期+1" + + +class TestLinearForecast: + def test_known_line(self): + """y = 2x + 1 (x=0..4) → 预测 x=5→11, x=6→13""" + values = [1.0, 3.0, 5.0, 7.0, 9.0] + res = engine.linear_forecast(values, periods=3) + assert len(res["forecast"]) == 3 + assert res["forecast"][0]["predicted"] == pytest.approx(11.0, abs=1e-6) + assert res["forecast"][1]["predicted"] == pytest.approx(13.0, abs=1e-6) + assert res["forecast"][2]["predicted"] == pytest.approx(15.0, abs=1e-6) + # 完全拟合 → R²=1 + assert res["r2"] == pytest.approx(1.0, abs=1e-6) + + def test_confidence_interval_bounds(self): + """每个预测点 lower ≤ predicted ≤ upper""" + import random + random.seed(42) + values = [random.uniform(50, 150) for _ in range(10)] + res = engine.linear_forecast(values, periods=4) + for f in res["forecast"]: + assert f["lower"] <= f["predicted"] <= f["upper"] + + +class TestMovingAverageForecast: + def test_window_mean(self): + """最近3期均值作为预测值""" + values = [10.0, 20.0, 30.0, 40.0, 50.0] + res = engine.moving_average_forecast(values, periods=3) + assert len(res["forecast"]) == 3 + assert res["mean"] == pytest.approx(40.0) # (30+40+50)/3 + for f in res["forecast"]: + assert f["predicted"] == pytest.approx(40.0) + assert f["lower"] <= f["predicted"] <= f["upper"] + + def test_window_smaller_than_history(self): + values = [5.0, 6.0] + res = engine.moving_average_forecast(values, periods=2) + assert res["window"] == 2 + assert res["forecast"][0]["predicted"] == pytest.approx(5.5) + + +class TestTrendAndConfidence: + def test_trend_up(self): + assert engine.judge_trend(8.0) == "up" + + def test_trend_down(self): + assert engine.judge_trend(-8.0) == "down" + + def test_trend_flat(self): + assert engine.judge_trend(1.0) == "flat" + assert engine.judge_trend(-1.0) == "flat" + + def test_confidence_high(self): + assert engine.compute_confidence(20, "linear", r2=0.9) == "high" + + def test_confidence_medium(self): + assert engine.compute_confidence(8, "linear", r2=0.5) == "medium" + + def test_confidence_low(self): + assert engine.compute_confidence(3, "linear", r2=0.1) == "low" + # 移动平均高波动 → low + assert engine.compute_confidence(10, "moving_average", cv=0.8) == "low" + + +# ── 引擎级测试（真实DB） ── + +class TestForecastKpi: + def test_forecast_kpi_structure(self, db: Session): + """完整返回结构""" + kpi = create_test_kpi(db, kpi_code="F_TEST_REV", kpi_name="测试收入", + dimension="finance", unit="万元") + _seed_history(db, kpi, [100, 110, 120, 130, 140]) + r = engine.forecast_kpi(1, "F_TEST_REV", db, periods=3) + assert r is not None + assert r["kpi"]["code"] == "F_TEST_REV" + assert r["kpi"]["name"] == "测试收入" + assert r["kpi"]["unit"] == "万元" + assert r["trend"] == "up" + assert r["confidence"] in ("high", "medium", "low") + assert r["history_count"] == 5 + assert len(r["history"]) == 5 + assert len(r["forecast"]) == 3 + # 历史升序 + periods = [h["period"] for h in r["history"]] + assert periods == sorted(periods) + # 预测期号在历史之后 + assert r["forecast"][0]["period"] > periods[-1] + assert r["next_target"] == r["forecast"][0]["predicted"] + assert "测试收入" in r["summary"] + + def test_forecast_kpi_moving_average_model(self, db: Session): + kpi = create_test_kpi(db, kpi_code="F_TEST_MA", kpi_name="测试费用", unit="元") + _seed_history(db, kpi, [10, 12, 11, 13, 12]) + r = engine.forecast_kpi(1, "F_TEST_MA", db, periods=2, model="moving_average") + assert r["model"] == "moving_average" + assert len(r["forecast"]) == 2 + + def test_forecast_kpi_insufficient_history(self, db: Session): + kpi = create_test_kpi(db, kpi_code="F_TEST_1PT", kpi_name="单点KPI") + _seed_history(db, kpi, [100.0]) + assert engine.forecast_kpi(1, "F_TEST_1PT", db) is None + + def test_forecast_kpi_no_history(self, db: Session): + create_test_kpi(db, kpi_code="F_TEST_NOHIST", kpi_name="无历史KPI") + assert engine.forecast_kpi(1, "F_TEST_NOHIST", db) is None + + def test_forecast_kpi_unknown_code(self, db: Session): + assert engine.forecast_kpi(1, "F_NOT_EXIST", db) is None + + def test_forecast_kpi_zero_periods(self, db: Session): + kpi = create_test_kpi(db, kpi_code="F_TEST_ZP", kpi_name="零期KPI") + _seed_history(db, kpi, [100, 110, 120]) + r = engine.forecast_kpi(1, "F_TEST_ZP", db, periods=0) + assert r is not None + assert r["forecast"] == [] + assert r["next_target"] is None + + +# ── API 端点测试 ── + +class TestKpiForecastApi: + def test_single_forecast(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_API_REV", kpi_name="API收入", + dimension="finance", unit="万元") + _seed_history(db, kpi, [100, 108, 116, 124, 132]) + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_API_REV", "periods": 3, + }) + assert resp.status_code == 200 + data = resp.json() + assert data["kpi"]["code"] == "F_API_REV" + assert data["trend"] in ("up", "down", "flat") + assert len(data["forecast"]) == 3 + assert data["next_target"] is not None + assert data["summary"] + + def test_moving_average_param(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_API_MA", kpi_name="API费用", unit="元") + _seed_history(db, kpi, [5, 6, 7, 6, 8]) + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_API_MA", "model": "moving_average", + }) + assert resp.status_code == 200 + assert resp.json()["model"] == "moving_average" + + def test_insufficient_history_404(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_API_LESS", kpi_name="数据不足") + _seed_history(db, kpi, [100.0]) + resp = client.get(f"{BASE}/kpi-forecast", params={"entity_id": 1, "kpi_code": "F_API_LESS"}) + assert resp.status_code == 404 + + def test_no_history_404(self, client: TestClient, db: Session): + create_test_kpi(db, kpi_code="F_API_EMPTY", kpi_name="空历史") + resp = client.get(f"{BASE}/kpi-forecast", params={"entity_id": 1, "kpi_code": "F_API_EMPTY"}) + assert resp.status_code == 404 + + def test_zero_periods(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_API_ZERO", kpi_name="零期") + _seed_history(db, kpi, [100, 110, 120]) + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_API_ZERO", "periods": 0, + }) + assert resp.status_code == 200 + data = resp.json() + assert data["forecast"] == [] + assert data["next_target"] is None + + def test_bad_model_400(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_API_BADM", kpi_name="坏模型") + _seed_history(db, kpi, [1, 2, 3, 4]) + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_API_BADM", "model": "lstm", + }) + assert resp.status_code == 400 + + def test_bad_periods_400(self, client: TestClient, db: Session): + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_API_BADM", "periods": -1, + }) + assert resp.status_code == 400 + + +class TestKpiForecastFinanceApi: + def test_batch_finance(self, client: TestClient, db: Session): + """只返回财务维度且历史≥3条的KPI，按可预测性排序""" + good1 = create_test_kpi(db, kpi_code="F_BATCH_REV", kpi_name="批量收入", dimension="finance") + _seed_history(db, good1, [100, 105, 110, 115, 120, 125, 130, 135]) # 8条, 完美上升 → high + good2 = create_test_kpi(db, kpi_code="F_BATCH_COST", kpi_name="批量成本", dimension="finance") + _seed_history(db, good2, [50, 52, 51, 53]) # 4条 → low/medium + short = create_test_kpi(db, kpi_code="F_BATCH_SHORT", kpi_name="数据不足", dimension="finance") + _seed_history(db, short, [10, 20]) # 2条 → 应被排除 + non_fin = create_test_kpi(db, kpi_code="C_BATCH_SAT", kpi_name="客户满意", dimension="customer") + _seed_history(db, non_fin, [80, 82, 84, 86]) # 非财务 → 应被排除 + + resp = client.get(f"{BASE}/kpi-forecast/finance", params={"entity_id": 1}) + assert resp.status_code == 200 + data = resp.json() + assert data["total"] == 2 + codes = [d["kpi"]["code"] for d in data["data"]] + assert "F_BATCH_REV" in codes and "F_BATCH_COST" in codes + assert "F_BATCH_SHORT" not in codes + assert "C_BATCH_SAT" not in codes + # 可预测性排序：F_BATCH_REV(8条 high) 应排前面 + assert data["data"][0]["kpi"]["code"] == "F_BATCH_REV" + + def test_batch_zero_periods(self, client: TestClient, db: Session): + kpi = create_test_kpi(db, kpi_code="F_BATCH_ZP", kpi_name="零期批量", dimension="finance") + _seed_history(db, kpi, [1, 2, 3, 4]) + resp = client.get(f"{BASE}/kpi-forecast/finance", params={"entity_id": 1, "periods": 0}) + assert resp.status_code == 200 + assert resp.json()["data"][0]["forecast"] == [] + + +class TestEntityIsolation: + def test_entity2_kpi_invisible_to_entity1(self, client: TestClient, db: Session): + """entity2 的KPI在 entity1 下查询 → 404（不串数据）""" + _seed_entity2(db) + kpi2 = create_test_kpi(db, kpi_code="F_ISO_E2", kpi_name="博海专属KPI", + dimension="finance", entity_id=2) + _seed_history(db, kpi2, [100, 110, 120, 130], entity_id=2) + # 无token时 entity_id 走 query 参数 → entity1 + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 1, "kpi_code": "F_ISO_E2", + }) + assert resp.status_code == 404 + # entity2 自己能查到 + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 2, "kpi_code": "F_ISO_E2", + }) + assert resp.status_code == 200 + assert resp.json()["entity_id"] == 2 + + def test_batch_isolation(self, client: TestClient, db: Session): + """批量预测：entity1 结果不含 entity2 的财务KPI""" + _seed_entity2(db) + kpi1 = create_test_kpi(db, kpi_code="F_ISO_E1", kpi_name="酣客收入", dimension="finance") + _seed_history(db, kpi1, [100, 110, 120, 130, 140], entity_id=1) + kpi2 = create_test_kpi(db, kpi_code="F_ISO_E2B", kpi_name="博海收入", dimension="finance", entity_id=2) + _seed_history(db, kpi2, [200, 210, 220, 230], entity_id=2) + + resp = client.get(f"{BASE}/kpi-forecast/finance", params={"entity_id": 1}) + codes = [d["kpi"]["code"] for d in resp.json()["data"]] + assert "F_ISO_E1" in codes + assert "F_ISO_E2B" not in codes + + def test_token_binds_entity(self, client: TestClient, db: Session): + """token绑定entity1时显式传entity_id=2 → 403 越权拦截""" + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="F_ISO_TK", kpi_name="TokenKPI") + _seed_history(db, kpi, [1, 2, 3, 4]) + resp = client.get(f"{BASE}/kpi-forecast", params={ + "entity_id": 2, "kpi_code": "F_ISO_TK", + }, headers=auth_header(token)) + assert resp.status_code == 403 diff --git a/frontend/components.d.ts b/frontend/components.d.ts index dab88de1..f850d29b 100644 --- a/frontend/components.d.ts +++ b/frontend/components.d.ts @@ -21,6 +21,8 @@ declare module 'vue' { ElCheckbox: typeof import('element-plus/es')['ElCheckbox'] ElCheckboxGroup: typeof import('element-plus/es')['ElCheckboxGroup'] ElCol: typeof import('element-plus/es')['ElCol'] + ElCollapse: typeof import('element-plus/es')['ElCollapse'] + ElCollapseItem: typeof import('element-plus/es')['ElCollapseItem'] ElContainer: typeof import('element-plus/es')['ElContainer'] ElDatePicker: typeof import('element-plus/es')['ElDatePicker'] ElDescriptions: typeof import('element-plus/es')['ElDescriptions'] diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index e5aa7353..9d3faf3c 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -220,6 +220,9 @@ export const predictApi = { forecastAccuracy: (params?: any) => api.get('/predict/accuracy', { params }), scenarioSuggestions: (params?: any) => api.get('/predict/scenario-suggestions', { params }), generateSuggestion: (data: any) => api.post('/predict/scenario-suggestion/generate', data), + // KPI趋势预测（预测性成本智能） + kpiForecast: (params?: any) => api.get('/predict/kpi-forecast', { params }), + kpiForecastFinance: (params?: any) => api.get('/predict/kpi-forecast/finance', { params }), } export const deviationPushApi = { diff --git a/frontend/src/views/PredictDashboard.vue b/frontend/src/views/PredictDashboard.vue index 1f2067d4..c1efdd5f 100644 --- a/frontend/src/views/PredictDashboard.vue +++ b/frontend/src/views/PredictDashboard.vue @@ -321,12 +321,87 @@ + + + + + +
{{ kfSummary }}
+ + + + + + + + + + + + + + + + + + + + + + + + + +