"""回测框架:在历史数据上运行预测并评估 LLM 预测质量。 核心机制: - build_context 已内置 before=match_date,天然防未来信息泄漏 - 对历史比赛跑预测 → 用实际比分 settle → 统计准确率 - 并发控制: asyncio.Semaphore 限制同时 LLM 调用数 """ from __future__ import annotations import asyncio import logging from dataclasses import dataclass, field from datetime import datetime from sqlalchemy import select from sqlalchemy.orm import selectinload from src.db.models import Match from src.db.unit_of_work import get_uow from src.llm.eval import settle_prediction from src.llm.predict import predict_match from src.llm.utils import actual_1x2 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 subjective_confidence: float | None correct_1x2: bool prediction_id: int @dataclass class BacktestCandidate: """回测候选比赛(字段快照,不持有 ORM 对象)。 session 关闭后仍可安全读取:所有需要的关系字段已在查询时物化为普通值, 避免在 session 之外访问惰性加载的关系属性(会抛 MissingGreenlet)。 """ match_id: int league_code: str | None home_team: str away_team: str match_date: datetime home_goals: int away_goals: int @dataclass class BacktestSummary: """回测汇总统计。""" total: int scored: int accuracy_1x2: float | None = None avg_score_rmse: float | None = None avg_subjective_confidence: float | None = None calibration: list[dict] = field(default_factory=list) results: list[BacktestMatchResult] = field(default_factory=list) 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[BacktestCandidate]: """查询已完赛且有比分的比赛(回测候选)。 返回普通值快照而非 ORM 对象:调用方在 session 关闭后仍需使用这些字段, 而 league / home_team / away_team 是惰性加载关系,在 async 下于 session 之外访问会抛 MissingGreenlet。这里用 selectinload 预加载后立即物化。 """ stmt = ( select(Match) .options( selectinload(Match.league), selectinload(Match.home_team), selectinload(Match.away_team), ) .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) # 在 session 内物化为纯数据,切断与 ORM 会话的耦合 return [ BacktestCandidate( 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, home_goals=m.home_goals, away_goals=m.away_goals, ) for m in 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 get_uow() as session: candidates = await _get_historical_matches( session, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit ) summary = BacktestSummary(total=len(candidates), scored=0) # P1-6: 并发控制,同时最多 8 场预测(避免 LLM API 限流) sem = asyncio.Semaphore(8) async def _one(c: BacktestCandidate) -> BacktestMatchResult | None: async with sem: try: result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False, backtest=True) await settle_prediction(result.prediction_id, c.home_goals, c.away_goals) actual = actual_1x2(c.home_goals, c.away_goals) return BacktestMatchResult( match_id=c.match_id, league_code=c.league_code, home_team=c.home_team, away_team=c.away_team, match_date=c.match_date.strftime("%Y-%m-%d") if c.match_date else "?", actual_home=c.home_goals, actual_away=c.away_goals, actual_1x2=actual, pred_home=result.pred_home_goals, pred_away=result.pred_away_goals, pred_1x2=result.pred_1x2, subjective_confidence=result.subjective_confidence, correct_1x2=result.pred_1x2 == actual, prediction_id=result.prediction_id, ) except Exception: logger.exception("backtest match %s failed", c.match_id) return None # 并行执行,保持结果顺序 results = await asyncio.gather(*[_one(c) for c in candidates]) for r in results: if r is not None: summary.results.append(r) summary.scored += 1 # 汇总统计 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.subjective_confidence for r in summary.results if r.subjective_confidence is not None] if confs: summary.avg_subjective_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.subjective_confidence is None: continue for key, b in buckets.items(): lo, hi = b["range"] if lo <= r.subjective_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 ]