From 0980c2242a8ee490e1fb430b279c90c8a01fb683 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Tue, 15 Sep 2026 02:04:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20P0-02/P0-03/P0-05=20=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93=E6=97=B6=E9=97=B4=E8=AF=AD=E4=B9=89=E4=B8=8E=E7=BA=A6?= =?UTF-8?q?=E6=9D=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0-02: MatchStats 增加 source/source_record_id/retrieved_at/available_at context_builder 过滤统计数据时检查 available_at <= cutoff P0-03: Prediction 增加 match_kickoff_at/prediction_created_at/prediction_cutoff_at 明确区分比赛时间/预测创建时间/数据截止时间 P0-05: Prediction 增加 status 字段(success/failed/degraded) 数据库 CHECK 约束 Alembic: 0005_prediction_status_and_stats_provenance P1-02: injuries as_of 不再截断为 date,保持 datetime 精度 --- ..._prediction_status_and_stats_provenance.py | 58 +++++++++++++++++++ src/api/routes/backtest.py | 9 ++- src/data/bzzoiro.py | 6 ++ src/data/injuries.py | 7 ++- src/data/understat.py | 10 +++- src/db/models.py | 14 ++++- src/llm/agents/orchestrator.py | 10 +++- src/llm/context_builder.py | 12 +++- src/llm/predict.py | 9 ++- 9 files changed, 122 insertions(+), 13 deletions(-) create mode 100644 alembic/versions/0005_prediction_status_and_stats_provenance.py diff --git a/alembic/versions/0005_prediction_status_and_stats_provenance.py b/alembic/versions/0005_prediction_status_and_stats_provenance.py new file mode 100644 index 0000000..3538d1c --- /dev/null +++ b/alembic/versions/0005_prediction_status_and_stats_provenance.py @@ -0,0 +1,58 @@ +"""add prediction status/time semantics and MatchStats provenance + +Revision ID: 0005_prediction_status_and_stats_provenance +Revises: 0004_snapshot_and_constraints +Create Date: 2026-09-15 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = '0005_prediction_status_and_stats_provenance' +down_revision: Union[str, None] = '0004_snapshot_and_constraints' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # 1. Prediction 新增字段 + op.add_column('predictions', sa.Column('status', sa.String(length=20), nullable=False, server_default='success')) + op.add_column('predictions', sa.Column('match_kickoff_at', sa.DateTime(timezone=True), nullable=True)) + op.add_column('predictions', sa.Column('prediction_created_at', sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now())) + op.add_column('predictions', sa.Column('prediction_cutoff_at', sa.DateTime(timezone=True), nullable=True)) + + # 重命名 cutoff_at → 保留作为兼容,prediction_cutoff_at 为主字段 + # op.drop_column('predictions', 'cutoff_at') # 暂不删除,避免破坏现有数据 + + # 2. 新增 status CHECK 约束 + op.create_check_constraint('ck_status_enum', 'predictions', "status IN ('success', 'failed', 'degraded')") + + # 3. MatchStats 新增数据血缘字段 + op.add_column('match_stats', sa.Column('source', sa.String(length=30), nullable=True)) + op.add_column('match_stats', sa.Column('source_record_id', sa.String(length=100), nullable=True)) + op.add_column('match_stats', sa.Column('retrieved_at', sa.DateTime(timezone=True), nullable=True)) + op.add_column('match_stats', sa.Column('available_at', sa.DateTime(timezone=True), nullable=True)) + + # 4. 索引 + op.create_index('ix_match_stats_available_at', 'match_stats', ['available_at']) + op.create_index('ix_predictions_cutoff_at', 'predictions', ['prediction_cutoff_at']) + + +def downgrade() -> None: + op.drop_index('ix_predictions_cutoff_at', table_name='predictions') + op.drop_index('ix_match_stats_available_at', table_name='match_stats') + + op.drop_column('match_stats', 'available_at') + op.drop_column('match_stats', 'retrieved_at') + op.drop_column('match_stats', 'source_record_id') + op.drop_column('match_stats', 'source') + + op.drop_constraint('ck_status_enum', 'predictions', type_='check') + + op.drop_column('predictions', 'prediction_cutoff_at') + op.drop_column('predictions', 'prediction_created_at') + op.drop_column('predictions', 'match_kickoff_at') + op.drop_column('predictions', 'status') diff --git a/src/api/routes/backtest.py b/src/api/routes/backtest.py index 67d401f..48b96e4 100644 --- a/src/api/routes/backtest.py +++ b/src/api/routes/backtest.py @@ -1,11 +1,15 @@ """回测路由。""" from __future__ import annotations +import logging + from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field from src.llm.backtest import run_backtest +logger = logging.getLogger(__name__) + router = APIRouter(prefix="/api/v1", tags=["backtest"]) @@ -38,7 +42,8 @@ async def backtest(req: BacktestRequest): model=req.model, ) except Exception as e: - raise HTTPException(500, f"backtest failed: {e}") + logger.exception("backtest failed") + raise HTTPException(500, "回测执行失败,请查看服务器日志") return { "summary": { @@ -46,7 +51,7 @@ async def backtest(req: BacktestRequest): "scored": summary.scored, "accuracy_1x2": summary.accuracy_1x2, "avg_score_rmse": summary.avg_score_rmse, - "avg_subjective_confidence": summary.avg_confidence, + "avg_subjective_confidence": summary.avg_subjective_confidence, "calibration": summary.calibration, }, "results": [ diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index cc07a8e..e41cb98 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -14,6 +14,7 @@ import urllib.error import urllib.parse import urllib.request from collections.abc import Iterable +from datetime import datetime, timezone from sqlalchemy import select @@ -223,6 +224,7 @@ class BzzoiroSource: await db.flush() existing_matches[match_key] = m # 防止同批重复 if nm.home_xg is not None or nm.away_xg is not None: + now = datetime.now(timezone.utc) stats = MatchStats( match_id=m.id, home_xg=nm.home_xg, @@ -238,6 +240,10 @@ class BzzoiroSource: away_yellow_cards=nm.away_yellow_cards, home_red_cards=nm.home_red_cards, away_red_cards=nm.away_red_cards, + source="bzzoiro", + source_event_id=str(raw.get("id", "")), + retrieved_at=now, + available_at=now, ) db.add(stats) league_r["inserted"] += 1 diff --git a/src/data/injuries.py b/src/data/injuries.py index 747abf5..2b1da01 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -190,11 +190,13 @@ async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> li match_date: 比赛日期 as_of: 截止时间(cutoff)。只返回 retrieved_at <= as_of 的记录。 用于回测时防止"未来采集的数据"泄漏到历史预测。 + 必须保持 timezone-aware datetime,不会截断为 date。 """ - from sqlalchemy import and_, or_, select + from sqlalchemy import or_, select from src.db.models import Injury + # 只处理 match_date:去掉时间部分,仅比较日期 if hasattr(match_date, "date"): match_date = match_date.date() @@ -206,9 +208,8 @@ async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> li ) # 回测防泄漏: 只使用 as_of 时间点之前已采集的数据 + # 注意: as_of 保持 datetime,不截断为 date,避免错误排除同日合法数据 if as_of is not None: - if hasattr(as_of, "date"): - as_of = as_of.date() stmt = stmt.where(Injury.retrieved_at.is_not(None)) stmt = stmt.where(Injury.retrieved_at <= as_of) diff --git a/src/data/understat.py b/src/data/understat.py index 7d04574..3789293 100644 --- a/src/data/understat.py +++ b/src/data/understat.py @@ -10,6 +10,7 @@ import json import logging import random import re +from datetime import datetime, timezone from sqlalchemy import func, select @@ -136,7 +137,14 @@ class UnderstatSource: # 回填 xG if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None): - existing.stats = MatchStats(match_id=existing.id) + now = datetime.now(timezone.utc) + existing.stats = MatchStats( + match_id=existing.id, + source="understat", + source_event_id=str(raw.get("id", "")), + retrieved_at=now, + available_at=now, + ) db.add(existing.stats) await db.flush() if existing.stats is not None: diff --git a/src/db/models.py b/src/db/models.py index 55091ab..dac2e43 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -115,6 +115,11 @@ class MatchStats(Base): home_red_cards: Mapped[int | None] = mapped_column(Integer) away_red_cards: Mapped[int | None] = mapped_column(Integer) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow) + # 数据血缘:追踪统计数据的来源和可用时间 + source: Mapped[str | None] = mapped_column(String(30)) # bzzoiro / understat + source_record_id: Mapped[str | None] = mapped_column(String(100)) + retrieved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + available_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) match: Mapped[Match] = relationship(back_populates="stats") @@ -163,8 +168,12 @@ class Prediction(Base): # multi-agent 模式: 各专家报告 mode: Mapped[str] = mapped_column(String(20), nullable=False, default="single") agent_outputs: Mapped[dict | None] = mapped_column(JSONB) - # 回测/可复现性 - cutoff_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # 预测状态: success / failed / degraded + status: Mapped[str] = mapped_column(String(20), nullable=False, default="success") + # 时间语义:区分比赛时间、预测创建时间、数据截止时间 + match_kickoff_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + prediction_created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) + prediction_cutoff_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) input_hash: Mapped[str | None] = mapped_column(String(64)) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) actual_home_goals: Mapped[int | None] = mapped_column(Integer) @@ -182,4 +191,5 @@ class Prediction(Base): CheckConstraint("subjective_confidence >= 0 AND subjective_confidence <= 1", name="ck_confidence_range"), CheckConstraint("pred_1x2 IN ('1', 'X', '2')", name="ck_pred_1x2_enum"), CheckConstraint("mode IN ('single', 'multi')", name="ck_mode_enum"), + CheckConstraint("status IN ('success', 'failed', 'degraded')", name="ck_status_enum"), ) diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index cffdc5c..b4f9c74 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -7,6 +7,7 @@ import json import logging import time from dataclasses import dataclass +from datetime import datetime, timezone from src.core.config import settings from src.db.base import AsyncSessionLocal @@ -166,7 +167,9 @@ async def predict_match_multi( # 1. 比赛头(各 agent 共享;不存在则 404) header = await load_match_header(match_id) - cutoff_at = header.match_dt + match_kickoff_at = header.match_dt + prediction_cutoff_at = header.match_dt # 默认:比赛时间作为数据截止 + now = datetime.now(timezone.utc) # 2. 并行专家 specialist_provider = _get_specialist_provider() @@ -215,7 +218,10 @@ async def predict_match_multi( reasoning=validated.reasoning, raw_response=final, agent_outputs=[r.to_dict() for r in reports], - cutoff_at=cutoff_at, + status="success", + match_kickoff_at=match_kickoff_at, + prediction_cutoff_at=prediction_cutoff_at, + prediction_created_at=now, input_hash=input_hash, ) session.add(pred) diff --git a/src/llm/context_builder.py b/src/llm/context_builder.py index a7fcde8..6c77786 100644 --- a/src/llm/context_builder.py +++ b/src/llm/context_builder.py @@ -30,6 +30,15 @@ def _outcome(home_goals: int, away_goals: int, side: str) -> str: return "W" if away_goals > home_goals else ("D" if away_goals == home_goals else "L") +def _is_stats_available(stats, before) -> bool: + """检查统计数据在 cutoff 时间是否已可用。""" + if before is None: + return True + if stats.available_at is None: + return True # 无时间信息时保守处理:允许使用 + return stats.available_at <= before + + @dataclass class MatchContext: match_id: int @@ -162,7 +171,8 @@ async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> s gf += fm.home_goals if side == "home" else fm.away_goals ga += fm.away_goals if side == "home" else fm.home_goals n += 1 - if fm.stats: + # 只使用 cutoff 之前已可用的统计数据 + if fm.stats and _is_stats_available(fm.stats, before): if fm.stats.home_shots is not None: shots += fm.stats.home_shots if side == "home" else fm.stats.away_shots sot += fm.stats.home_shots_on_target if side == "home" else fm.stats.away_shots_on_target diff --git a/src/llm/predict.py b/src/llm/predict.py index c950e70..eb07b50 100644 --- a/src/llm/predict.py +++ b/src/llm/predict.py @@ -116,7 +116,9 @@ async def _predict_single( ctx = await build_context(match_id) # 1.5 计算快照元数据(用于可复现性) - cutoff_at = ctx.match_dt # 比赛时间 = 数据截止时间 + now = datetime.now(timezone.utc) + match_kickoff_at = ctx.match_dt + prediction_cutoff_at = ctx.match_dt # 默认:比赛时间作为数据截止 input_hash = hashlib.sha256(ctx.text.encode("utf-8")).hexdigest() # 2. 拼 prompt(指定版本) @@ -165,7 +167,10 @@ async def _predict_single( subjective_confidence=validated.subjective_confidence, reasoning=validated.reasoning, raw_response=resp.raw, - cutoff_at=cutoff_at, + status="success", + match_kickoff_at=match_kickoff_at, + prediction_cutoff_at=prediction_cutoff_at, + prediction_created_at=now, input_hash=input_hash, ) session.add(pred)