fix: 补充 Alembic 迁移 + bzzoiro 批量优化 + validation 兼容层

- 新增 0004_snapshot_and_constraints.py: 重命名 confidence → subjective_confidence,
  新增 cutoff_at/input_hash, 添加 CHECK 约束
- bzzoiro.py: 预加载完整 Match 对象到内存,更新路径不再重复查询
- validation.py: 旧字段 confidence 兼容并打 deprecation 日志

alembic/versions/0004_snapshot_and_constraints.py

@@ -0,0 +1,49 @@
+"""add snapshot fields and check constraints
+
+Revision ID: 0004_snapshot_and_constraints
+Revises: 0003_injuries
+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 = '0004_snapshot_and_constraints'
+down_revision: Union[str, None] = '0003_injuries'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 1. 重命名 confidence → subjective_confidence
+    op.alter_column('predictions', 'confidence', new_column_name='subjective_confidence')
+
+    # 2. 新增快照字段
+    op.add_column('predictions', sa.Column('cutoff_at', sa.DateTime(timezone=True), nullable=True))
+    op.add_column('predictions', sa.Column('input_hash', sa.String(length=64), nullable=True))
+
+    # 3. 新增 CHECK 约束
+    op.create_check_constraint('ck_pred_home_goals_nonneg', 'predictions', 'pred_home_goals >= 0')
+    op.create_check_constraint('ck_pred_away_goals_nonneg', 'predictions', 'pred_away_goals >= 0')
+    op.create_check_constraint('ck_confidence_range', 'predictions', 'subjective_confidence >= 0 AND subjective_confidence <= 1')
+    op.create_check_constraint('ck_pred_1x2_enum', 'predictions', "pred_1x2 IN ('1', 'X', '2')")
+    op.create_check_constraint('ck_mode_enum', 'predictions', "mode IN ('single', 'multi')")
+
+
+def downgrade() -> None:
+    # 1. 删除 CHECK 约束
+    op.drop_constraint('ck_mode_enum', 'predictions', type_='check')
+    op.drop_constraint('ck_pred_1x2_enum', 'predictions', type_='check')
+    op.drop_constraint('ck_confidence_range', 'predictions', type_='check')
+    op.drop_constraint('ck_pred_away_goals_nonneg', 'predictions', type_='check')
+    op.drop_constraint('ck_pred_home_goals_nonneg', 'predictions', type_='check')
+
+    # 2. 删除快照字段
+    op.drop_column('predictions', 'input_hash')
+    op.drop_column('predictions', 'cutoff_at')
+
+    # 3. 恢复列名
+    op.alter_column('predictions', 'subjective_confidence', new_column_name='confidence')

src/data/bzzoiro.py

@@ -151,7 +151,7 @@ class BzzoiroSource:
 
             # === 批量优化: 预加载球队和已有比赛到内存 ===
             team_name_to_id: dict[str, int] = {}
-            existing_match_keys: set[tuple[int, int, str]] = set()
+            existing_matches: dict[tuple[int, int, str], Match] = {}  # 完整对象,避免重复查询
             normalized_matches: list = []  # 缓存规范化结果,避免重复调用
 
             if raw_events:
@@ -175,15 +175,11 @@ class BzzoiroSource:
                     teams = (await db.execute(stmt)).scalars().all()
                     team_name_to_id = {t.name: t.id for t in teams}
 
-                # 预加载已有比赛
+                # 预加载已有比赛(完整对象)
-                from sqlalchemy import func
+                stmt = select(Match).where(Match.league_id == league.id)
-                stmt = (
+                for m in (await db.execute(stmt)).scalars():
-                    select(Match.home_team_id, Match.away_team_id, func.date(Match.match_date).label("d"))
+                    key = (m.home_team_id, m.away_team_id, str(m.match_date_date))
-                    .where(Match.league_id == league.id)
+                    existing_matches[key] = m
-                )
-                rows = (await db.execute(stmt)).all()
-                for row in rows:
-                    existing_match_keys.add((row.home_team_id, row.away_team_id, str(row.d)))
 
             for nm in normalized_matches:
                 # 球队: 内存查找 + 按需创建
@@ -206,9 +202,9 @@ class BzzoiroSource:
                 # 查找已有比赛: 内存查找
                 date_key = nm.date.date().isoformat() if hasattr(nm.date, "date") else str(nm.date)
                 match_key = (home_team_id, away_team_id, date_key)
-                existing = None if match_key not in existing_match_keys else "exists"
+                existing_match = existing_matches.get(match_key)
 
-                if existing is None:
+                if existing_match is None:
                     m = Match(
                         league_id=league.id,
                         season=nm.season_label or None,
@@ -225,7 +221,7 @@ class BzzoiroSource:
                     )
                     db.add(m)
                     await db.flush()
-                    existing_match_keys.add(match_key)  # 防止同批重复
+                    existing_matches[match_key] = m  # 防止同批重复
                     if nm.home_xg is not None or nm.away_xg is not None:
                         stats = MatchStats(
                             match_id=m.id,
@@ -246,17 +242,7 @@ class BzzoiroSource:
                         db.add(stats)
                     league_r["inserted"] += 1
                 else:
-                    # 已有比赛: 需要查询对象来更新
+                    # 已有比赛: 直接从内存获取对象更新(无需再查询)
-                    from sqlalchemy import func
-                    stmt = (
-                        select(Match)
-                        .where(Match.league_id == league.id)
-                        .where(Match.home_team_id == home_team_id)
-                        .where(Match.away_team_id == away_team_id)
-                        .where(func.date(Match.match_date) == date_key)
-                    )
-                    existing_match = (await db.execute(stmt)).scalar_one()
-
                     changed = False
                     if existing_match.match_status != nm.match_status and nm.match_status == "finished":
                         existing_match.match_status = nm.match_status

src/llm/validation.py

@@ -4,8 +4,12 @@
 """
 from __future__ import annotations
 
+import logging
+
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+logger = logging.getLogger(__name__)
+
 
 class AgentReportSchema(BaseModel):
     """单个专家 Agent 输出的校验 schema。"""
@@ -92,11 +96,17 @@ def validate_agent_output(raw: dict) -> AgentReportSchema:
 
 def validate_prediction_output(raw: dict) -> PredictionOutputSchema:
     """校验最终预测输出。"""
+    # 优先新字段,旧字段仅兼容并打日志
+    conf = raw.get("subjective_confidence")
+    if conf is None and "confidence" in raw:
+        logger.warning("Deprecated field 'confidence' used, prefer 'subjective_confidence'")
+        conf = raw["confidence"]
+
     return PredictionOutputSchema(
         pred_home_goals=float(raw.get("pred_home_goals", 0)),
         pred_away_goals=float(raw.get("pred_away_goals", 0)),
         pred_1x2=raw.get("1x2") or raw.get("pred_1x2", "X"),
-        subjective_confidence=float(raw.get("confidence", 0.5)),
+        subjective_confidence=float(conf if conf is not None else 0.5),
         reasoning=str(raw.get("reasoning", ""))[:1000],
     )