From 1166a157ec7645eb580087e5f6b8d88ade84db16 Mon Sep 17 00:00:00 2001
From: shangfangjian <shangfangjian@bilidili.cn>
Date: Wed, 9 Sep 2026 21:39:56 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E5=9B=9E=E6=B5=8B?=
 =?UTF-8?q?=E6=A1=86=E6=9E=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 新增 src/llm/backtest.py: 回测核心逻辑
  - 查询历史已完赛比赛
  - 逐场预测(自动防未来信息泄漏)
  - 实际比分回填 + 统计
  - 1X2 准确率 / 比分 RMSE / 置信度校准
- 新增 POST /api/v1/backtest 路由
- 支持按联赛/日期范围/模式/模型筛选
---
 src/api/app.py             |   2 +
 src/api/routes/backtest.py |  69 +++++++++++++
 src/llm/backtest.py        | 204 +++++++++++++++++++++++++++++++++++++
 3 files changed, 275 insertions(+)
 create mode 100644 src/api/routes/backtest.py
 create mode 100644 src/llm/backtest.py

diff --git a/src/api/app.py b/src/api/app.py
index 4fff88e..dbd33d0 100644
--- a/src/api/app.py
+++ b/src/api/app.py
@@ -41,11 +41,13 @@ def create_app() -> FastAPI:
     from src.api.routes.predict import router as predict_router
     from src.api.routes.ingest import router as ingest_router
     from src.api.routes.eval import router as eval_router
+    from src.api.routes.backtest import router as backtest_router
 
     app.include_router(matches_router)
     app.include_router(predict_router)
     app.include_router(ingest_router)
     app.include_router(eval_router)
+    app.include_router(backtest_router)
 
     @app.get("/health")
     async def health():
diff --git a/src/api/routes/backtest.py b/src/api/routes/backtest.py
new file mode 100644
index 0000000..cbd1123
--- /dev/null
+++ b/src/api/routes/backtest.py
@@ -0,0 +1,69 @@
+"""回测路由。"""
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from src.llm.backtest import run_backtest
+
+router = APIRouter(prefix="/api/v1", tags=["backtest"])
+
+
+class BacktestRequest(BaseModel):
+    league_id: int | None = Field(None, description="联赛 ID")
+    date_from: str | None = Field(None, description="起始日期 YYYY-MM-DD")
+    date_to: str | None = Field(None, description="结束日期 YYYY-MM-DD")
+    mode: str = Field("single", description="预测模式: single(快) / multi(多 agent)")
+    limit: int = Field(20, ge=1, le=200, description="最大回测场数")
+    model: str | None = Field(None, description="指定模型 (空=默认)")
+
+
+@router.post("/backtest")
+async def backtest(req: BacktestRequest):
+    """对历史比赛运行回测。
+
+    对每场已完赛比赛:
+      1. 用比赛之前的数据构建上下文 (防未来信息泄漏)
+      2. 调 LLM 预测
+      3. 用实际比分回填
+      4. 统计准确率 / RMSE / 校准度
+    """
+    try:
+        summary = await run_backtest(
+            league_id=req.league_id,
+            date_from=req.date_from,
+            date_to=req.date_to,
+            mode=req.mode,
+            limit=req.limit,
+            model=req.model,
+        )
+    except Exception as e:
+        raise HTTPException(500, f"backtest failed: {e}")
+
+    return {
+        "summary": {
+            "total": summary.total,
+            "scored": summary.scored,
+            "accuracy_1x2": summary.accuracy_1x2,
+            "avg_score_rmse": summary.avg_score_rmse,
+            "avg_confidence": summary.avg_confidence,
+            "calibration": summary.calibration,
+        },
+        "results": [
+            {
+                "match_id": r.match_id,
+                "league_code": r.league_code,
+                "home_team": r.home_team,
+                "away_team": r.away_team,
+                "match_date": r.match_date,
+                "actual_score": f"{r.actual_home}-{r.actual_away}",
+                "actual_1x2": r.actual_1x2,
+                "pred_home": r.pred_home,
+                "pred_away": r.pred_away,
+                "pred_1x2": r.pred_1x2,
+                "confidence": r.confidence,
+                "correct_1x2": r.correct_1x2,
+            }
+            for r in summary.results
+        ],
+    }
diff --git a/src/llm/backtest.py b/src/llm/backtest.py
new file mode 100644
index 0000000..24ae305
--- /dev/null
+++ b/src/llm/backtest.py
@@ -0,0 +1,204 @@
+"""回测框架:在历史数据上运行预测并评估 LLM 预测质量。
+
+核心机制:
+  - build_context 已内置 before=match_date,天然防未来信息泄漏
+  - 对历史比赛跑预测 → 用实际比分 settle → 统计准确率
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+from sqlalchemy import and_, select
+
+from src.db.base import AsyncSessionLocal
+from src.db.models import League, Match, Prediction
+from src.llm.eval import settle_prediction
+from src.llm.predict import predict_match
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class BacktestMatchResult:
+    """单场回测结果。"""
+    match_id: int
+    league_code: str | None
+    home_team: str
+    away_team: str
+    match_date: str
+    actual_home: int
+    actual_away: int
+    actual_1x2: str
+    pred_home: float | None
+    pred_away: float | None
+    pred_1x2: str | None
+    confidence: float | None
+    correct_1x2: bool
+    prediction_id: int
+
+
+@dataclass
+class BacktestSummary:
+    """回测汇总统计。"""
+    total: int
+    scored: int
+    accuracy_1x2: float | None = None
+    avg_score_rmse: float | None = None
+    avg_confidence: float | None = None
+    calibration: list[dict] = field(default_factory=list)
+    results: list[BacktestMatchResult] = field(default_factory=list)
+
+
+def _actual_1x2(home: int, away: int) -> str:
+    """实际比分 → 胜平负。"""
+    if home > away:
+        return "1"
+    if home == away:
+        return "X"
+    return "2"
+
+
+async def _get_historical_matches(
+    db,
+    *,
+    league_id: int | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    limit: int = 50,
+) -> list[Match]:
+    """查询已完赛且有比分的比赛(回测候选)。"""
+    stmt = (
+        select(Match)
+        .where(Match.match_status == "finished")
+        .where(Match.home_goals.is_not(None))
+        .where(Match.away_goals.is_not(None))
+    )
+    if league_id is not None:
+        stmt = stmt.where(Match.league_id == league_id)
+    if date_from:
+        stmt = stmt.where(Match.match_date >= date_from)
+    if date_to:
+        stmt = stmt.where(Match.match_date <= date_to)
+
+    stmt = stmt.order_by(Match.match_date.desc()).limit(limit)
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def run_backtest(
+    *,
+    league_id: int | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    mode: str = "single",
+    limit: int = 50,
+    model: str | None = None,
+) -> BacktestSummary:
+    """运行回测。
+
+    Args:
+        league_id: 联赛 ID
+        date_from: 起始日期 (YYYY-MM-DD)
+        date_to: 结束日期 (YYYY-MM-DD)
+        mode: 预测模式 (single/multi)
+        limit: 最大回测场数
+        model: 指定模型 (None=默认)
+
+    Returns:
+        BacktestSummary 含逐场结果 + 汇总统计
+    """
+    async with AsyncSessionLocal() as db:
+        matches = await _get_historical_matches(
+            db, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit
+        )
+
+    summary = BacktestSummary(total=len(matches), scored=0)
+
+    for m in matches:
+        try:
+            # 预测 (build_context 内部已用 before=match_date 防泄漏)
+            result = await predict_match(m.id, mode=mode, model=model)
+
+            # 用实际比分 settle
+            await settle_prediction(result.prediction_id, m.home_goals, m.away_goals)
+
+            actual = _actual_1x2(m.home_goals, m.away_goals)
+            correct = result.pred_1x2 == actual
+
+            bt = BacktestMatchResult(
+                match_id=m.id,
+                league_code=m.league.code if m.league else None,
+                home_team=m.home_team.name if m.home_team else "?",
+                away_team=m.away_team.name if m.away_team else "?",
+                match_date=m.match_date.strftime("%Y-%m-%d") if m.match_date else "?",
+                actual_home=m.home_goals,
+                actual_away=m.away_goals,
+                actual_1x2=actual,
+                pred_home=result.pred_home_goals,
+                pred_away=result.pred_away_goals,
+                pred_1x2=result.pred_1x2,
+                confidence=result.confidence,
+                correct_1x2=correct,
+                prediction_id=result.prediction_id,
+            )
+            summary.results.append(bt)
+            summary.scored += 1
+
+        except Exception as e:
+            logger.warning("backtest match %s failed: %s", m.id, e)
+
+    # 汇总统计
+    if summary.scored > 0:
+        correct_count = sum(1 for r in summary.results if r.correct_1x2)
+        summary.accuracy_1x2 = round(correct_count / summary.scored * 100, 1)
+
+        # 比分 RMSE
+        errors = []
+        for r in summary.results:
+            if r.pred_home is not None and r.pred_away is not None:
+                err = ((r.pred_home - r.actual_home) ** 2 + (r.pred_away - r.actual_away) ** 2) ** 0.5
+                errors.append(err)
+        if errors:
+            summary.avg_score_rmse = round(sum(errors) / len(errors), 2)
+
+        # 平均置信度
+        confs = [r.confidence for r in summary.results if r.confidence is not None]
+        if confs:
+            summary.avg_confidence = round(sum(confs) / len(confs), 2)
+
+        # 校准:按置信度分桶,看实际准确率是否匹配
+        summary.calibration = _compute_calibration(summary.results)
+
+    return summary
+
+
+def _compute_calibration(results: list[BacktestMatchResult]) -> list[dict]:
+    """置信度校准:分桶统计实际准确率。"""
+    buckets: dict[str, dict] = {
+        "0.9-1.0": {"range": (0.9, 1.0), "total": 0, "correct": 0},
+        "0.7-0.9": {"range": (0.7, 0.9), "total": 0, "correct": 0},
+        "0.5-0.7": {"range": (0.5, 0.7), "total": 0, "correct": 0},
+        "0.3-0.5": {"range": (0.3, 0.5), "total": 0, "correct": 0},
+        "0.0-0.3": {"range": (0.0, 0.3), "total": 0, "correct": 0},
+    }
+    for r in results:
+        if r.confidence is None:
+            continue
+        for key, b in buckets.items():
+            lo, hi = b["range"]
+            if lo <= r.confidence <= hi:
+                b["total"] += 1
+                if r.correct_1x2:
+                    b["correct"] += 1
+                break
+
+    return [
+        {
+            "bucket": key,
+            "total": b["total"],
+            "accuracy": round(b["correct"] / b["total"] * 100, 1) if b["total"] else None,
+        }
+        for key, b in buckets.items()
+        if b["total"] > 0
+    ]