feat: P0-02/P0-03/P0-05 数据库时间语义与约束

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 精度
This commit is contained in:
shangfangjian
2026-09-15 02:04:25 +08:00
parent c657e04679
commit 0980c2242a
9 changed files with 122 additions and 13 deletions
@@ -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')
+7 -2
View File
@@ -1,11 +1,15 @@
"""回测路由。""" """回测路由。"""
from __future__ import annotations from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from src.llm.backtest import run_backtest from src.llm.backtest import run_backtest
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1", tags=["backtest"]) router = APIRouter(prefix="/api/v1", tags=["backtest"])
@@ -38,7 +42,8 @@ async def backtest(req: BacktestRequest):
model=req.model, model=req.model,
) )
except Exception as e: except Exception as e:
raise HTTPException(500, f"backtest failed: {e}") logger.exception("backtest failed")
raise HTTPException(500, "回测执行失败,请查看服务器日志")
return { return {
"summary": { "summary": {
@@ -46,7 +51,7 @@ async def backtest(req: BacktestRequest):
"scored": summary.scored, "scored": summary.scored,
"accuracy_1x2": summary.accuracy_1x2, "accuracy_1x2": summary.accuracy_1x2,
"avg_score_rmse": summary.avg_score_rmse, "avg_score_rmse": summary.avg_score_rmse,
"avg_subjective_confidence": summary.avg_confidence, "avg_subjective_confidence": summary.avg_subjective_confidence,
"calibration": summary.calibration, "calibration": summary.calibration,
}, },
"results": [ "results": [
+6
View File
@@ -14,6 +14,7 @@ import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
from collections.abc import Iterable from collections.abc import Iterable
from datetime import datetime, timezone
from sqlalchemy import select from sqlalchemy import select
@@ -223,6 +224,7 @@ class BzzoiroSource:
await db.flush() await db.flush()
existing_matches[match_key] = m # 防止同批重复 existing_matches[match_key] = m # 防止同批重复
if nm.home_xg is not None or nm.away_xg is not None: if nm.home_xg is not None or nm.away_xg is not None:
now = datetime.now(timezone.utc)
stats = MatchStats( stats = MatchStats(
match_id=m.id, match_id=m.id,
home_xg=nm.home_xg, home_xg=nm.home_xg,
@@ -238,6 +240,10 @@ class BzzoiroSource:
away_yellow_cards=nm.away_yellow_cards, away_yellow_cards=nm.away_yellow_cards,
home_red_cards=nm.home_red_cards, home_red_cards=nm.home_red_cards,
away_red_cards=nm.away_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) db.add(stats)
league_r["inserted"] += 1 league_r["inserted"] += 1
+4 -3
View File
@@ -190,11 +190,13 @@ async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> li
match_date: 比赛日期 match_date: 比赛日期
as_of: 截止时间(cutoff)。只返回 retrieved_at <= as_of 的记录。 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 from src.db.models import Injury
# 只处理 match_date:去掉时间部分,仅比较日期
if hasattr(match_date, "date"): if hasattr(match_date, "date"):
match_date = 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 时间点之前已采集的数据
# 注意: as_of 保持 datetime,不截断为 date,避免错误排除同日合法数据
if as_of is not None: 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.is_not(None))
stmt = stmt.where(Injury.retrieved_at <= as_of) stmt = stmt.where(Injury.retrieved_at <= as_of)
+9 -1
View File
@@ -10,6 +10,7 @@ import json
import logging import logging
import random import random
import re import re
from datetime import datetime, timezone
from sqlalchemy import func, select from sqlalchemy import func, select
@@ -136,7 +137,14 @@ class UnderstatSource:
# 回填 xG # 回填 xG
if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None): 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) db.add(existing.stats)
await db.flush() await db.flush()
if existing.stats is not None: if existing.stats is not None:
+12 -2
View File
@@ -115,6 +115,11 @@ class MatchStats(Base):
home_red_cards: Mapped[int | None] = mapped_column(Integer) home_red_cards: Mapped[int | None] = mapped_column(Integer)
away_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) 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") match: Mapped[Match] = relationship(back_populates="stats")
@@ -163,8 +168,12 @@ class Prediction(Base):
# multi-agent 模式: 各专家报告 # multi-agent 模式: 各专家报告
mode: Mapped[str] = mapped_column(String(20), nullable=False, default="single") mode: Mapped[str] = mapped_column(String(20), nullable=False, default="single")
agent_outputs: Mapped[dict | None] = mapped_column(JSONB) agent_outputs: Mapped[dict | None] = mapped_column(JSONB)
# 回测/可复现性 # 预测状态: success / failed / degraded
cutoff_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) 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)) input_hash: Mapped[str | None] = mapped_column(String(64))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
actual_home_goals: Mapped[int | None] = mapped_column(Integer) 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("subjective_confidence >= 0 AND subjective_confidence <= 1", name="ck_confidence_range"),
CheckConstraint("pred_1x2 IN ('1', 'X', '2')", name="ck_pred_1x2_enum"), CheckConstraint("pred_1x2 IN ('1', 'X', '2')", name="ck_pred_1x2_enum"),
CheckConstraint("mode IN ('single', 'multi')", name="ck_mode_enum"), CheckConstraint("mode IN ('single', 'multi')", name="ck_mode_enum"),
CheckConstraint("status IN ('success', 'failed', 'degraded')", name="ck_status_enum"),
) )
+8 -2
View File
@@ -7,6 +7,7 @@ import json
import logging import logging
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timezone
from src.core.config import settings from src.core.config import settings
from src.db.base import AsyncSessionLocal from src.db.base import AsyncSessionLocal
@@ -166,7 +167,9 @@ async def predict_match_multi(
# 1. 比赛头(各 agent 共享;不存在则 404) # 1. 比赛头(各 agent 共享;不存在则 404)
header = await load_match_header(match_id) 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. 并行专家 # 2. 并行专家
specialist_provider = _get_specialist_provider() specialist_provider = _get_specialist_provider()
@@ -215,7 +218,10 @@ async def predict_match_multi(
reasoning=validated.reasoning, reasoning=validated.reasoning,
raw_response=final, raw_response=final,
agent_outputs=[r.to_dict() for r in reports], 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, input_hash=input_hash,
) )
session.add(pred) session.add(pred)
+11 -1
View File
@@ -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") 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 @dataclass
class MatchContext: class MatchContext:
match_id: int 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 gf += fm.home_goals if side == "home" else fm.away_goals
ga += fm.away_goals if side == "home" else fm.home_goals ga += fm.away_goals if side == "home" else fm.home_goals
n += 1 n += 1
if fm.stats: # 只使用 cutoff 之前已可用的统计数据
if fm.stats and _is_stats_available(fm.stats, before):
if fm.stats.home_shots is not None: if fm.stats.home_shots is not None:
shots += fm.stats.home_shots if side == "home" else fm.stats.away_shots 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 sot += fm.stats.home_shots_on_target if side == "home" else fm.stats.away_shots_on_target
+7 -2
View File
@@ -116,7 +116,9 @@ async def _predict_single(
ctx = await build_context(match_id) ctx = await build_context(match_id)
# 1.5 计算快照元数据(用于可复现性) # 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() input_hash = hashlib.sha256(ctx.text.encode("utf-8")).hexdigest()
# 2. 拼 prompt(指定版本) # 2. 拼 prompt(指定版本)
@@ -165,7 +167,10 @@ async def _predict_single(
subjective_confidence=validated.subjective_confidence, subjective_confidence=validated.subjective_confidence,
reasoning=validated.reasoning, reasoning=validated.reasoning,
raw_response=resp.raw, 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, input_hash=input_hash,
) )
session.add(pred) session.add(pred)