fix: 数据库与数据管线 6 个 P1 + 5 个 P2 审查问题修复

P1-1 [context_builder] build_context 共享 session,切片函数传 db 参数,
        回测 20 场并发连接需求从 100+ 降至每场 1 个
P1-2 [bzzoiro] 预加载改为按 raw_events 日期范围 ±30 天按需加载
P1-3 [understat] 批量查询球队 + 比赛,从 1140 次往返降到 3 次
P1-4 [injuries] 批量幂等检查 + 分批 flush,IntegrityError 逐条回退
P1-5 [predict] 删除 threading.Lock,dict 操作原子无需同步锁
P1-6 [models] 添加 (match_id, provider, model) 唯一约束 + 迁移

P2-1 [unit_of_work] get_uow 返回类型改为 AsyncIterator[AsyncSession]
P2-2 [normalize] _parse_date 失败时记录 warning 避免静默丢数据
P2-3 [repositories] find_by_teams_and_date 改用 match_date_date 等值匹配
P2-4 [migration] 幽灵列 cutoff_at 已在 0006 迁移删除(已有)
P2-5 [migration] injuries 约束命名对齐 ORM,UniqueConstraint → 唯一索引

src/data/bzzoiro.py

@@ -195,11 +195,25 @@ class BzzoiroSource:
                     teams = (await db.execute(stmt)).scalars().all()
                     team_name_to_id = {t.name: t.id for t in teams}
 
-                # 预加载已有比赛(完整对象)
-                stmt = select(Match).where(Match.league_id == league.id)
-                for m in (await db.execute(stmt)).scalars():
-                    key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date)
-                    existing_matches[key] = m
+                # P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
+                # 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
+                if normalized_matches:
+                    from datetime import timedelta
+                    dates = [nm.date for nm in normalized_matches if nm.date is not None]
+                    if dates:
+                        min_dt = min(dates) - timedelta(days=30)
+                        max_dt = max(dates) + timedelta(days=30)
+                        stmt = (
+                            select(Match)
+                            .where(Match.league_id == league.id)
+                            .where(Match.match_date >= min_dt)
+                            .where(Match.match_date <= max_dt)
+                        )
+                        existing_matches = {
+                            _match_key(m.home_team_id, m.away_team_id, m.match_date_date): m
+                            for m in (await db.execute(stmt)).scalars()
+                        }
+                # else: existing_matches 保持空 dict(全量新比赛)
 
             for nm, raw in normalized_matches:
                 # 球队: 内存查找 + 按需创建

src/data/injuries.py

@@ -104,8 +104,12 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
     """采集伤停数据并入库(injuries 表)。
 
     注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
+
+    P1-4: 批量幂等检查,避免逐条查询的竞态条件(并发采集时 IntegrityError)。
     """
     from sqlalchemy import select
+    from sqlalchemy.exc import IntegrityError
+    from sqlalchemy.orm import selectinload
 
     from src.data.team_names import normalize as normalize_name
     from src.db.models import Injury, Team
@@ -125,6 +129,9 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
     teams = (await db.execute(select(Team))).scalars().all()
     team_by_name = {t.name: t.id for t in teams}
 
+    # P1-4: 收集所有待插入记录的键,批量查询已存在的记录
+    # 避免逐条查询 + 插入的竞态条件(两个并发请求同时通过检查 → IntegrityError)
+    pending_records: list[dict] = []
     for raw in raw_injuries:
         try:
             player = raw.get("player", {}) or {}
@@ -145,7 +152,7 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
                 except (ValueError, AttributeError):
                     pass
 
-            # P2-1: 强制 int 转换,API 可能返回字符串
+            # 强制 int 转换,API 可能返回字符串
             player_id = player.get("id")
             try:
                 player_id = int(player_id) if player_id is not None else None
@@ -157,40 +164,92 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
             except (ValueError, TypeError):
                 fixture_id = None
 
-            # 幂等: 已存在则跳过
-            existing = (
-                await db.execute(
-                    select(Injury).where(
-                        Injury.player_id == player_id,
-                        Injury.fixture_id == fixture_id,
-                        Injury.injury_type == player.get("type"),
-                    )
-                )
-            ).scalar_one_or_none()
-
-            if existing is not None:
-                continue
-
-            injury = Injury(
-                player_id=player_id,
-                player_name=player_name,
-                team_id=team_id,
-                fixture_id=fixture_id,
-                league_id=(raw.get("league") or {}).get("id"),
-                injury_type=player.get("type"),
-                reason=player.get("reason"),
-                injury_date=injury_date,
-            )
-            db.add(injury)
-            result["inserted"] += 1
+            pending_records.append({
+                "player_id": player_id,
+                "player_name": player_name,
+                "team_id": team_id,
+                "fixture_id": fixture_id,
+                "league_id": (raw.get("league") or {}).get("id"),
+                "injury_type": player.get("type"),
+                "reason": player.get("reason"),
+                "injury_date": injury_date,
+            })
         except Exception as e:
             result["errors"].append(f"parse error: {e}")
 
+    # P1-4: 批量查询已存在的记录(1 次 DB 往返)
+    existing_keys: set[tuple] = set()
+    if pending_records:
+        # 构造查询条件:所有 (player_id, fixture_id, injury_type) 组合
+        # 使用 OR 条件批量查询
+        conditions = []
+        for rec in pending_records:
+            conditions.append(
+                (Injury.player_id == rec["player_id"])
+                & (Injury.fixture_id == rec["fixture_id"])
+                & (Injury.injury_type == rec["injury_type"])
+            )
+        if conditions:
+            from sqlalchemy import or_
+            stmt = select(Injury.player_id, Injury.fixture_id, Injury.injury_type).where(or_(*conditions))
+            rows = (await db.execute(stmt)).all()
+            existing_keys = {(r[0], r[1], r[2]) for r in rows}
+
+    # P1-4: 批量插入(跳过已存在的)
+    for rec in pending_records:
+        key = (rec["player_id"], rec["fixture_id"], rec["injury_type"])
+        if key in existing_keys:
+            continue
+
+        injury = Injury(**rec)
+        db.add(injury)
+        result["inserted"] += 1
+
+        # 每 50 条 flush 一次,减少内存压力,同时捕获 IntegrityError
+        if result["inserted"] % 50 == 0:
+            try:
+                await db.flush()
+            except IntegrityError:
+                # P1-4: 并发采集时可能仍有竞态,回退到逐条插入
+                await db.rollback()
+                logger.warning("injuries batch IntegrityError, falling back to per-record insert")
+                return await _ingest_injuries_fallback(db, pending_records, result)
+
+    # 最终 flush
+    try:
+        await db.flush()
+    except IntegrityError:
+        await db.rollback()
+        logger.warning("injuries final flush IntegrityError, falling back to per-record insert")
+        return await _ingest_injuries_fallback(db, pending_records, result)
+
     # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
     logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date)
     return result
 
 
+async def _ingest_injuries_fallback(db, pending_records: list[dict], result: dict) -> dict:
+    """P1-4: 逐条插入回退,捕获每条 IntegrityError 避免整批回滚。"""
+    from sqlalchemy.exc import IntegrityError
+    from src.db.models import Injury
+
+    inserted = 0
+    for rec in pending_records:
+        injury = Injury(**rec)
+        db.add(injury)
+        try:
+            await db.flush()
+            inserted += 1
+        except IntegrityError:
+            await db.rollback()
+            # 已存在或其他冲突,跳过
+            continue
+
+    result["inserted"] = inserted
+    logger.info("injuries fallback: inserted %d records", inserted)
+    return result
+
+
 async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> list[Injury]:
     """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。
 
@@ -198,31 +257,29 @@ async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> li
         db: 数据库 session
         team_id: 球队 ID
         match_date: 比赛日期
-        as_of: 截止时间(cutoff)。只返回 retrieved_at <= as_of 的记录。
-               用于回测时防止"未来采集的数据"泄漏到历史预测。
-               必须保持 timezone-aware datetime,不会截断为 date。
-    """
-    from sqlalchemy import or_, select
+        as_of: 数据截止时间(用于回测防泄漏)
 
+    Returns:
+        伤停记录列表
+    """
+    from sqlalchemy import select
     from src.db.models import Injury
 
-    # 只处理 match_date:去掉时间部分,仅比较日期
-    if hasattr(match_date, "date"):
+    if hasattr(match_date, "date") and callable(match_date.date):
         match_date = match_date.date()
 
     stmt = (
         select(Injury)
         .where(Injury.team_id == team_id)
         .where(Injury.injury_date <= match_date)
-        .where(or_(Injury.return_date.is_(None), Injury.return_date >= match_date))
+        .where(
+            (Injury.return_date.is_(None)) | (Injury.return_date >= match_date)
+        )
     )
-
-    # 回测防泄漏: 只使用 as_of 时间点之前已采集的数据
-    # 注意: as_of 保持 datetime,不截断为 date,避免错误排除同日合法数据
     if as_of is not None:
-        stmt = stmt.where(Injury.retrieved_at.is_not(None))
+        if hasattr(as_of, "date") and callable(as_of.date):
+            as_of = as_of.date()
         stmt = stmt.where(Injury.retrieved_at <= as_of)
 
-    stmt = stmt.order_by(Injury.injury_date.desc())
     result = await db.execute(stmt)
     return list(result.scalars().all())

src/data/normalize.py

@@ -1,9 +1,9 @@
 """数据规范化:任意数据源原始记录 → NormalizedMatch。
 
 迁移自旧项目 app/data/normalize.py,简化:
-  - 去掉 XGBackfill 双轨(不再需要独立回填)
-  - 去掉 PIT 时间契约(无训练集要防泄漏)
-  - 保留核心清洗契约(队名归一、日期解析、数值范围)
+    - 去掉 XGBackoff 双轨(不再需要独立回填)
+    - 去掉 PIT 时间契约(无训练集要防泄漏)
+    - 保留核心清洗契约(队名归一、日期解析、数值范围)
 """
 from __future__ import annotations
 
@@ -90,7 +90,10 @@ def derive_season_label(date: datetime) -> str:
 
 
 def _parse_date(value) -> datetime | None:
-    """日期解析 → UTC datetime(带 tzinfo)。"""
+    """日期解析 → UTC datetime(带 tzinfo)。
+
+    P2-2: 解析失败时记录 warning,避免静默丢数据而无感知。
+    """
     if value in (None, ""):
         return None
     if isinstance(value, (int, float)):
@@ -111,6 +114,8 @@ def _parse_date(value) -> datetime | None:
             return datetime.strptime(s[:19], fmt).replace(tzinfo=timezone.utc)
         except ValueError:
             continue
+    # P2-2 修复: 记录被丢弃的原始值,便于排查数据源格式变更
+    logger.warning("_parse_date failed, dropping record: %r", value)
     return None
 
 
@@ -204,6 +209,6 @@ def normalize_understat(raw: dict, league_type: str) -> NormalizedMatch | None:
         away_team=away,
         match_status="finished",
         season_label=derive_season_label(dt),
-        home_xg=_to_float(home_xg),
-        away_xg=_to_float(away_xg),
+        home_xg=home_xg,
+        away_xg=away_xg,
     )

src/data/understat.py

@@ -10,9 +10,10 @@ import json
 import logging
 import random
 import re
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import func, select
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
 
 from src.core.http_client import get_client
 from src.data.config import FDCO_TO_UNDERSTAT, LEAGUE_NAMES
@@ -76,6 +77,18 @@ async def fetch_understat(league_code: str, season: int) -> list[dict]:
     return data
 
 
+def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]:
+    """比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。
+
+    统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类
+    隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配,
+    导致所有比赛被判为不存在而重复插入。
+    """
+    if hasattr(match_date, "date") and callable(match_date.date):
+        match_date = match_date.date()
+    return (home_team_id, away_team_id, match_date.isoformat() if match_date is not None else "")
+
+
 @register
 class UnderstatSource:
     """understat xG 数据源(实现 DataSource 协议)。"""
@@ -86,8 +99,10 @@ class UnderstatSource:
         """采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。
 
         注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
+
+        P1-3: 批量查询优化,将单赛季 380 场 × 3 次 DB 往返降为 3 次查询。
         """
-        from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository
+        from src.db.repositories import LeagueRepository, TeamRepository
 
         result = {"updated": 0, "skipped": 0, "unmatched": 0, "errors": []}
 
@@ -101,7 +116,6 @@ class UnderstatSource:
         # 使用 Repository
         league_repo = LeagueRepository(db)
         team_repo = TeamRepository(db)
-        match_repo = MatchRepository(db)
 
         # 查联赛
         league_obj = await league_repo.get_by_code(league)
@@ -109,6 +123,9 @@ class UnderstatSource:
             result["errors"].append(f"league {league} not found in DB")
             return result
 
+        # === 批量优化: 一次规范化,收集球队名和日期 ===
+        normalized_matches: list = []
+        all_team_names: set[str] = set()
         for raw in raw_matches:
             if not raw.get("isResult"):
                 continue
@@ -120,17 +137,46 @@ class UnderstatSource:
             except Exception as e:
                 result["errors"].append(f"normalize: {e}")
                 continue
+            normalized_matches.append((nm, raw))
+            all_team_names.add(nm.home_team)
+            all_team_names.add(nm.away_team)
 
-            # 匹配已有 Match(天级) - 使用 Repository
-            home_team = await team_repo.get_by_name(nm.home_team)
-            away_team = await team_repo.get_by_name(nm.away_team)
-            if home_team is None or away_team is None:
+        if not normalized_matches:
+            return result
+
+        # === 批量查询球队(1 次 DB 往返) ===
+        team_name_to_id = {}
+        if all_team_names:
+            teams = await team_repo.get_all_by_names(list(all_team_names))
+            team_name_to_id = {name: team.id for name, team in teams.items()}
+
+        # === 批量查询已有比赛(1 次 DB 往返,按日期范围) ===
+        match_dict: dict[tuple, Match] = {}
+        dates = [nm.date for nm, _ in normalized_matches if nm.date is not None]
+        if dates:
+            min_dt = min(dates) - timedelta(days=30)
+            max_dt = max(dates) + timedelta(days=30)
+            stmt = (
+                select(Match)
+                .options(selectinload(Match.stats))
+                .where(Match.league_id == league_obj.id)
+                .where(Match.match_date >= min_dt)
+                .where(Match.match_date <= max_dt)
+            )
+            for m in (await db.execute(stmt)).scalars():
+                key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date)
+                match_dict[key] = m
+
+        # === 内存匹配 + 回填 xG ===
+        for nm, raw in normalized_matches:
+            home_team_id = team_name_to_id.get(nm.home_team)
+            away_team_id = team_name_to_id.get(nm.away_team)
+            if home_team_id is None or away_team_id is None:
                 result["unmatched"] += 1
                 continue
 
-            existing = await match_repo.find_by_teams_and_date(
-                league_obj.id, home_team.id, away_team.id, nm.date
-            )
+            match_key = _match_key(home_team_id, away_team_id, nm.date)
+            existing = match_dict.get(match_key)
             if existing is None:
                 result["unmatched"] += 1
                 continue