feat: 数据管线 P2 优化 — 血缘写入 + 质量监控 + 缓存清理 + 重试 worker

P2-1: DataLineage 血缘写入(bzzoiro/understat Bronze→Silver 全链路)
P2-2: DataQualityCheck 质量检查写入(行计数异常检测)
P2-3: RateLimitedClient 集成到 bzzoiro 采集
P2-4: injuries 缓存过期清理(启动时扫描删除 >1h 文件)
P2-5: IngestFailure 重试 worker(FastAPI startup 注册)

新增文件:
  - src/data/retry_worker.py (死信表重试 sweep)

.env.example

@@ -3,6 +3,10 @@ APP_ENV=development
 LOG_LEVEL=INFO
 
 # ---- 数据库 ----
+POSTGRES_USER=football
+POSTGRES_PASSWORD=football
+POSTGRES_DB=football
+POSTGRES_PORT=5432
 DATABASE_URL=postgresql+asyncpg://football:football@localhost:5432/football
 
 # ---- LLM (OpenAI-compatible,必填一个) ----

alembic/versions/0007_injuries_constraint_naming_align.py

@@ -0,0 +1,81 @@
+"""修复 injuries 表约束命名与 ORM 声明不一致
+
+Revision ID: 0007_injuries_constraint_naming_align
+Revises: 0006_schema_model_drift_cleanup
+Create Date: 2026-09-16
+
+背景(见代码审查报告 P2-5):
+  0003 迁移使用 sa.UniqueConstraint 创建唯一约束,
+  而 ORM models.py 中声明为 Index(..., unique=True)。
+  虽然 PostgreSQL 中两者效果相同(都保证唯一性),
+  但 pg_catalog 中表示不同,会导致:
+    - alembic autogenerate 持续报告漂移
+    - 约束命名约定不一致(uc_ 前缀 vs ix_ 前缀)
+
+  本迁移将 UniqueConstraint 替换为唯一索引,与 ORM 声明对齐。
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '0007_injuries_constraint_naming_align'
+down_revision: Union[str, None] = '0006_schema_model_drift_cleanup'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    # 检查当前约束类型
+    constraints = {
+        c["name"]: c
+        for c in inspector.get_unique_constraints("injuries")
+    }
+    indexes = {
+        i["name"]: i
+        for i in inspector.get_indexes("injuries")
+    }
+
+    # 如果存在 UniqueConstraint 形式的 ix_injuries_player_fixture,替换为唯一索引
+    if "ix_injuries_player_fixture" in constraints:
+        # 删除唯一约束
+        op.drop_constraint("ix_injuries_player_fixture", "injuries", type_="unique")
+
+    # 如果不存在同名唯一索引,创建它(与 ORM 声明一致)
+    if "ix_injuries_player_fixture" not in indexes:
+        op.create_index(
+            "ix_injuries_player_fixture",
+            "injuries",
+            ["player_id", "fixture_id", "injury_type"],
+            unique=True,
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    indexes = {
+        i["name"]: i
+        for i in inspector.get_indexes("injuries")
+    }
+    constraints = {
+        c["name"]: c
+        for c in inspector.get_unique_constraints("injuries")
+    }
+
+    # 恢复为 UniqueConstraint 形式
+    if "ix_injuries_player_fixture" in indexes:
+        op.drop_index("ix_injuries_player_fixture", table_name="injuries")
+
+    if "ix_injuries_player_fixture" not in constraints:
+        op.create_unique_constraint(
+            "ix_injuries_player_fixture",
+            "injuries",
+            ["player_id", "fixture_id", "injury_type"],
+        )

alembic/versions/0007_predictions_unique_constraint.py

@@ -0,0 +1,59 @@
+"""为 predictions 表添加 match_id+provider+model 唯一约束
+
+Revision ID: 0007_predictions_unique_constraint
+Revises: 0007_injuries_constraint_naming_align
+Create Date: 2026-09-16
+
+背景(见代码审查报告 P1-6):
+  同一 match_id + provider + model 组合不应产生重复预测。
+  当前缺少数据库级唯一约束,回测多次运行或并发采集可能产生重复记录,
+  导致统计偏差。
+
+  先清理已存在的重复记录(保留最早创建的那条),再添加唯一约束。
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '0007_predictions_unique_constraint'
+down_revision: Union[str, None] = '0007_injuries_constraint_naming_align'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 1. 清理已存在的重复记录(保留 id 最小的)
+    op.execute(
+        """
+        DELETE FROM predictions
+        WHERE id NOT IN (
+            SELECT MIN(id)
+            FROM predictions
+            GROUP BY match_id, provider, model
+        )
+        AND match_id IN (
+            SELECT match_id
+            FROM predictions
+            GROUP BY match_id, provider, model
+            HAVING COUNT(*) > 1
+        )
+        """
+    )
+
+    # 2. 添加唯一约束
+    op.create_unique_constraint(
+        "uq_predictions_match_provider_model",
+        "predictions",
+        ["match_id", "provider", "model"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "uq_predictions_match_provider_model",
+        "predictions",
+        type_="unique",
+    )

alembic/versions/0008_raw_event_and_ingest_failure.py

@@ -0,0 +1,121 @@
+"""新增 Bronze 层 + 死信表 + 数据质量表 + 血缘表
+
+Revision ID: 0008_raw_event_and_ingest_failure
+Revises: 0007_predictions_unique_constraint
+Create Date: 2026-09-17
+
+架构审查报告 P1 实施:
+  - raw_events: Bronze 层,不可变原始采集记录
+  - ingest_failures: 采集失败死信表
+  - data_quality_checks: 数据质量检查结果记录
+  - data_lineage: ETL 全过程元数据血缘追踪
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+
+# revision identifiers, used by Alembic.
+revision: str = '0008_raw_event_and_ingest_failure'
+down_revision: Union[str, None] = '0007_predictions_unique_constraint'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ── raw_events: Bronze 层原始记录 ──
+    op.create_table(
+        "raw_events",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source_system", sa.String(30), nullable=False),
+        sa.Column("source_record_id", sa.String(100), nullable=False),
+        sa.Column("raw_payload", JSONB(), nullable=False),
+        sa.Column("ingested_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("ingest_batch_id", sa.String(64), nullable=True),
+    )
+    op.create_index("ix_raw_events_batch", "raw_events", ["ingest_batch_id"])
+    op.create_index("ix_raw_events_source_ingested", "raw_events", ["source_system", "ingested_at"])
+    op.create_unique_constraint("uq_raw_events_source_record", "raw_events", ["source_system", "source_record_id"])
+
+    # ── ingest_failures: 采集失败死信表 ──
+    op.create_table(
+        "ingest_failures",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source_system", sa.String(30), nullable=False),
+        sa.Column("entity_type", sa.String(30), nullable=False),
+        sa.Column("source_record_id", sa.String(100), nullable=True),
+        sa.Column("error_type", sa.String(50), nullable=False),
+        sa.Column("error_detail", sa.Text(), nullable=True),
+        sa.Column("raw_payload", JSONB(), nullable=True),
+        sa.Column("retry_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("status", sa.String(20), nullable=False, server_default="pending"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index("ix_ingest_failures_status_next_retry", "ingest_failures", ["status", "next_retry_at"])
+    op.create_index("ix_ingest_failures_source", "ingest_failures", ["source_system", "entity_type"])
+    op.create_check_constraint("ck_ingest_failures_status", "ingest_failures",
+                               "status IN ('pending', 'retrying', 'resolved', 'abandoned')")
+    op.create_check_constraint("ck_ingest_failures_retry_nonneg", "ingest_failures", "retry_count >= 0")
+
+    # ── data_quality_checks: 数据质量检查记录 ──
+    op.create_table(
+        "data_quality_checks",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("check_name", sa.String(100), nullable=False),
+        sa.Column("entity_type", sa.String(30), nullable=False),
+        sa.Column("entity_id", sa.String(50), nullable=True),
+        sa.Column("expected_value", sa.Text(), nullable=True),
+        sa.Column("actual_value", sa.Text(), nullable=True),
+        sa.Column("passed", sa.Boolean(), nullable=False),
+        sa.Column("severity", sa.String(10), nullable=False, server_default="warning"),
+        sa.Column("detail", sa.Text(), nullable=True),
+        sa.Column("checked_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_dqc_check_time", "data_quality_checks", ["check_name", "checked_at"])
+    op.create_index("ix_dqc_entity", "data_quality_checks", ["entity_type", "entity_id"])
+    op.create_index("ix_dqc_severity_passed", "data_quality_checks", ["severity", "passed"])
+    op.create_check_constraint("ck_dqc_severity", "data_quality_checks",
+                               "severity IN ('info', 'warning', 'critical')")
+
+    # ── data_lineage: ETL 血缘追踪 ──
+    op.create_table(
+        "data_lineage",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source_system", sa.String(30), nullable=False),
+        sa.Column("source_record_id", sa.String(100), nullable=False),
+        sa.Column("target_table", sa.String(50), nullable=False),
+        sa.Column("target_id", sa.Integer(), nullable=True),
+        sa.Column("transform_name", sa.String(100), nullable=False),
+        sa.Column("transform_detail", sa.Text(), nullable=True),
+        sa.Column("batch_id", sa.String(64), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_lineage_source", "data_lineage", ["source_system", "source_record_id"])
+    op.create_index("ix_lineage_target", "data_lineage", ["target_table", "target_id"])
+    op.create_index("ix_lineage_batch", "data_lineage", ["batch_id"])
+
+
+def downgrade() -> None:
+    # 逆序删除
+    op.drop_index("ix_lineage_batch", table_name="data_lineage")
+    op.drop_index("ix_lineage_target", table_name="data_lineage")
+    op.drop_index("ix_lineage_source", table_name="data_lineage")
+    op.drop_table("data_lineage")
+
+    op.drop_index("ix_dqc_severity_passed", table_name="data_quality_checks")
+    op.drop_index("ix_dqc_entity", table_name="data_quality_checks")
+    op.drop_index("ix_dqc_check_time", table_name="data_quality_checks")
+    op.drop_table("data_quality_checks")
+
+    op.drop_index("ix_ingest_failures_source", table_name="ingest_failures")
+    op.drop_index("ix_ingest_failures_status_next_retry", table_name="ingest_failures")
+    op.drop_table("ingest_failures")
+
+    op.drop_constraint("uq_raw_events_source_record", table_name="raw_events", type_="unique")
+    op.drop_index("ix_raw_events_source_ingested", table_name="raw_events")
+    op.drop_index("ix_raw_events_batch", table_name="raw_events")
+    op.drop_table("raw_events")

alembic/versions/0009_match_stats_xg_fields.py

@@ -0,0 +1,35 @@
+"""MatchStats 新增 xG 追踪字段
+
+Revision ID: 0009_match_stats_xg_fields
+Revises: 0008_raw_event_and_ingest_failure
+Create Date: 2026-09-17
+
+架构审查报告 P1-4 实施:
+  understat 允许纠正旧 xG 值。新增字段追踪 xG 具体来源和更新时间,
+  实现全量覆盖模式:当 understat 数据更新时覆盖旧值而非跳过。
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '0009_match_stats_xg_fields'
+down_revision: Union[str, None] = '0008_raw_event_and_ingest_failure'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("match_stats", sa.Column("xg_source", sa.String(30), nullable=True))
+    op.add_column("match_stats", sa.Column("xg_updated_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("match_stats", sa.Column("xg_source_record_id", sa.String(100), nullable=True))
+    op.create_index("ix_match_stats_xg_source", "match_stats", ["xg_source", "xg_updated_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_match_stats_xg_source", table_name="match_stats")
+    op.drop_column("match_stats", "xg_source_record_id")
+    op.drop_column("match_stats", "xg_updated_at")
+    op.drop_column("match_stats", "xg_source")

docker-compose.yml

@@ -2,15 +2,15 @@ services:
   postgres:
     image: postgres:16-alpine
     environment:
-      POSTGRES_USER: football
+      POSTGRES_USER: ${POSTGRES_USER:?POSTGRES_USER 未设置}
-      POSTGRES_PASSWORD: football
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?POSTGRES_PASSWORD 未设置}
-      POSTGRES_DB: football
+      POSTGRES_DB: ${POSTGRES_DB:-football}
     ports:
-      - "5432:5432"
+      - "${POSTGRES_PORT:-5432}:5432"
     volumes:
       - pgdata:/var/lib/postgresql/data
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U football"]
+      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:?POSTGRES_USER 未设置}"]
       interval: 5s
       timeout: 5s
       retries: 5
@@ -19,7 +19,7 @@ services:
     build: .
     command: uvicorn src.api.app:app --host 0.0.0.0 --port 8000 --reload
     ports:
-      - "8000:8000"
+      - "${API_PORT:-8000}:8000"
     env_file: .env
     depends_on:
       postgres:

frontend/src/index.css

@@ -80,7 +80,7 @@
 
   /* ── 版面切换文字标签(联赛/状态/模式) ── */
   .tab {
-    @apply whitespace-nowrap px-0.5 py-1 text-sm text-ink-500 transition-colors hover:text-ink-900;
+    @apply relative whitespace-nowrap px-0.5 py-1 text-sm text-ink-500 transition-colors hover:text-ink-900;
   }
   .tab-on {
     @apply font-medium text-press;

src/api/app.py

@@ -16,6 +16,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     from src.db.base import init_db
     from src.core.http_client import close_client
     await init_db()  # 验证连接,不建表
+    # P2-5: 启动时执行一次 ingest failure 重试清理
+    from src.data.retry_worker import run_retry_worker
+    await run_retry_worker()
     yield
     await close_client()
 
@@ -29,12 +32,14 @@ def create_app() -> FastAPI:
     )
 
     origins = [o.strip() for o in settings.CORS_ORIGINS.split(",") if o.strip()]
+    methods = [m.strip() for m in settings.CORS_METHODS.split(",") if m.strip()]
+    headers = [h.strip() for h in settings.CORS_HEADERS.split(",") if h.strip()]
     app.add_middleware(
         CORSMiddleware,
         allow_origins=origins,
         allow_credentials=True,
-        allow_methods=["*"],
+        allow_methods=methods,
-        allow_headers=["*"],
+        allow_headers=headers,
     )
 
     from src.api.routes.matches import router as matches_router

src/api/deps.py

@@ -19,24 +19,18 @@ from src.core.config import settings
 
 logger = logging.getLogger(__name__)
 
-_warned_unset = False
-
 
 async def require_admin_key(x_api_key: str | None = Header(None, alias="X-API-Key")) -> None:
     """保护「写入型 / 高成本」接口的依赖。
 
     用法: `@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin_key)])`
     """
-    global _warned_unset
-
     expected = settings.ADMIN_API_KEY
     if not expected:
-        if not _warned_unset:
+        logger.warning(
-            logger.warning(
+            "ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。"
-                "ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。"
+            "生产环境请设置该环境变量。"
-                "生产环境请设置该环境变量。"
+        )
-            )
-            _warned_unset = True
         return
 
     if not x_api_key or not secrets.compare_digest(x_api_key, expected):

src/core/config.py

@@ -32,6 +32,17 @@ class Settings(BaseSettings):
 
     # --- CORS ---
     CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000"
+    CORS_METHODS: str = "GET,POST,PUT,DELETE,OPTIONS"
+    CORS_HEADERS: str = "Authorization,Content-Type,X-API-Key,Accept"
+
+    # --- HTTP ---
+    HTTP_DEFAULT_TIMEOUT: int = 30
+
+    # --- database pool ---
+    DB_POOL_SIZE: int = 5
+    DB_MAX_OVERFLOW: int = 10
+    DB_POOL_TIMEOUT: int = 30
+    DB_POOL_RECYCLE: int = 1800
 
     # --- 管理接口鉴权 ---
     # 采集 / 回测等高成本或写入型接口需要此 Key(请求头 X-API-Key)。

src/core/http_client.py

@@ -2,24 +2,27 @@
 
 使用方:
   - src/llm/provider.py: LLM 调用
+  - src/data/bzzoiro.py: bzzoiro 比赛数据
   - src/data/understat.py: xG 抓取
   - src/data/injuries.py: 伤停抓取
 
 生命周期由 FastAPI lifespan 管理(关闭时 aclose)。
+调用方可通过 `timeout` 参数覆盖 per-request 超时。
 """
 from __future__ import annotations
 
 import httpx
 
+from src.core.config import settings
+
 _shared_client: httpx.AsyncClient | None = None
-_default_timeout = 30
 
 
 def get_client() -> httpx.AsyncClient:
     """获取共享客户端(懒初始化)。"""
     global _shared_client
     if _shared_client is None or _shared_client.is_closed:
-        _shared_client = httpx.AsyncClient(timeout=_default_timeout)
+        _shared_client = httpx.AsyncClient(timeout=settings.HTTP_DEFAULT_TIMEOUT)
     return _shared_client
 
 

src/core/retry.py

@@ -1,89 +0,0 @@
-"""重试工具:带指数退避的瞬态错误重试。
-
-NOTE(审查报告 P3):当前全项目**无调用点** —— bzzoiro 在 `_fetch_json_sync`
-里自带了一套重试逻辑,understat/injuries 各自也有。这里保留是作为后续统一
-重试策略的落点,但请勿误以为它已在生效。
-
-如果决定不引入统一重试,建议删除本文件以避免"看起来有重试、实际没有"的误判。
-"""
-from __future__ import annotations
-
-import asyncio
-import functools
-import logging
-import random
-import time
-from typing import Callable, Iterable, TypeVar
-
-logger = logging.getLogger(__name__)
-
-T = TypeVar("T")
-
-
-def with_retry(
-    *,
-    max_retries: int = 3,
-    base_delay: float = 1.0,
-    max_delay: float = 30.0,
-    retryable_exceptions: Iterable[type[BaseException]] = (Exception,),
-    on_retry: Callable[[Exception, int], None] | None = None,
-) -> Callable:
-    """重试装饰器(同步/异步通用,指数退避 + 抖动)。
-
-    Args:
-        max_retries: 最大重试次数
-        base_delay: 基础延迟(秒)
-        max_delay: 最大延迟(秒)
-        retryable_exceptions: 触发重试的异常类型
-        on_retry: 重试回调(exception, attempt)
-    """
-    retryable = tuple(retryable_exceptions)
-
-    def decorator(func: Callable) -> Callable:
-        @functools.wraps(func)
-        async def async_wrapper(*args, **kwargs):
-            last_exc: Exception | None = None
-            for attempt in range(max_retries + 1):
-                try:
-                    return await func(*args, **kwargs)
-                except retryable as e:
-                    last_exc = e
-                    if attempt == max_retries:
-                        break
-                    delay = min(base_delay * (2 ** attempt), max_delay)
-                    delay += random.uniform(0, delay * 0.1)  # 抖动
-                    logger.warning(
-                        "%s failed (attempt %d/%d), retry in %.1fs: %s",
-                        func.__name__, attempt + 1, max_retries, delay, e,
-                    )
-                    if on_retry:
-                        on_retry(e, attempt + 1)
-                    await asyncio.sleep(delay)
-            raise last_exc  # type: ignore[misc]
-
-        @functools.wraps(func)
-        def sync_wrapper(*args, **kwargs):
-            last_exc: Exception | None = None
-            for attempt in range(max_retries + 1):
-                try:
-                    return func(*args, **kwargs)
-                except retryable as e:
-                    last_exc = e
-                    if attempt == max_retries:
-                        break
-                    delay = min(base_delay * (2 ** attempt), max_delay)
-                    delay += random.uniform(0, delay * 0.1)
-                    logger.warning(
-                        "%s failed (attempt %d/%d), retry in %.1fs: %s",
-                        func.__name__, attempt + 1, max_retries, delay, e,
-                    )
-                    if on_retry:
-                        on_retry(e, attempt + 1)
-                    time.sleep(delay)
-            raise last_exc  # type: ignore[misc]
-
-        if asyncio.iscoroutinefunction(func):
-            return async_wrapper
-        return sync_wrapper
-
-    return decorator

src/data/bzzoiro.py

@@ -2,6 +2,12 @@
 
 迁移自旧项目 app/data/sources/bzzoiro/,改成 async + 简化入库。
 使用 Repository 模式进行数据访问,不直接控制事务。
+
+改进(P1):
+  - Bronze 层集成:采集后先存 RawEvent,再规范化
+  - 死信表集成:采集/规范化失败写 IngestFailure
+  - 令牌桶限流:替换固定 REQUEST_INTERVAL sleep
+  - 数据血缘:记录 ETL 全过程
 """
 from __future__ import annotations
 
@@ -9,23 +15,36 @@ import asyncio
 import json as _json
 import logging
 import random
-import time as _time
+import uuid
-import urllib.error
-import urllib.parse
-import urllib.request
 from collections.abc import Iterable
 from datetime import datetime, timezone
 
 from sqlalchemy import select
 
 from src.core.config import settings
+from src.core.http_client import get_client
 from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
 from src.data.normalize import normalize_bzzoiro
+from src.data.rate_limiter import RateLimitedClient, TokenBucket
 from src.data.sources import register
-from src.db.models import League, Match, MatchStats, Team
+from src.db.models import DataLineage, DataQualityCheck, IngestFailure, League, Match, MatchStats, RawEvent, Team
 
 logger = logging.getLogger(__name__)
 
+# 令牌桶限流器:替代固定 sleep,rate 根据 REQUEST_INTERVAL 计算
+_bzzoiro_limiter = TokenBucket(rate=1.0 / REQUEST_INTERVAL, capacity=3)
+# P2-3: RateLimitedClient 包装,在 HTTP 调用层面透明限流
+_bzzoiro_rl_client: RateLimitedClient | None = None
+
+
+def _get_bzzoiro_rl_client() -> RateLimitedClient:
+    """惰性初始化 RateLimitedClient(避免模块导入时创建 client)。"""
+    global _bzzoiro_rl_client
+    if _bzzoiro_rl_client is None:
+        client = get_client()
+        _bzzoiro_rl_client = RateLimitedClient(client, _bzzoiro_limiter)
+    return _bzzoiro_rl_client
+
 
 def _to_date(value):
     """把 datetime / date / str 统一成 `date`。"""
@@ -47,43 +66,47 @@ def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, i
     return (home_team_id, away_team_id, d.isoformat() if d is not None else "")
 
 
-def _fetch_json_sync(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
+async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
-    """同步 HTTP(bzzoiro 客户端保持同步,在 async 函数里 run_in_executor)。"""
+    """异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。"""
     base = settings.BZZOIRO_BASE.rstrip("/")
     url = f"{base}/{path.lstrip('/')}"
-    if params:
-        url += "?" + urllib.parse.urlencode(params)
     key = settings.BZZOIRO_KEY
     if not key:
         raise RuntimeError("BZZOIRO_KEY 未设置")
 
+    headers = {
+        "Authorization": f"Token {key}",
+        "Accept": "application/json",
+    }
+
     last_exc: Exception | None = None
     for attempt in range(max_retries):
         try:
-            req = urllib.request.Request(url)
+            # P2-3: 使用 RateLimitedClient 包装 HTTP 调用,透明限流
-            req.add_header("Authorization", f"Token {key}")
+            rl_client = _get_bzzoiro_rl_client()
-            req.add_header("Accept", "application/json")
+            resp = await rl_client.get(url, headers=headers, params=params, timeout=30)
-            with urllib.request.urlopen(req, timeout=30) as resp:
+            resp.raise_for_status()
-                return _json.loads(resp.read().decode("utf-8"))
+            return resp.json()
-        except urllib.error.HTTPError as e:
+        except Exception as e:
             last_exc = e
-            if e.code == 429:
+            status = getattr(getattr(e, "response", None), "status_code", None)
-                # 指数退避: 429 通常意味着限速
+            if status == 429:
                 delay = min(2 ** attempt, 16) + random.uniform(0, 1)
                 logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay)
-                _time.sleep(delay)
+                await asyncio.sleep(delay)
                 continue
-            if 500 <= e.code < 600:
+            if 500 <= (status or 0) < 600:
                 delay = min(2 ** attempt, 16) + random.uniform(0, 1)
-                logger.warning("bzzoiro %d, retry %d in %.1fs", e.code, attempt + 1, delay)
+                logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay)
-                _time.sleep(delay)
+                await asyncio.sleep(delay)
                 continue
-            raise  # 4xx 直接抛
+            # 网络错误(连接失败/超时)也退避重试
-        except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
+            if isinstance(e, (TimeoutError, ConnectionError, OSError)):
-            last_exc = e
+                delay = min(2 ** attempt, 16) + random.uniform(0, 1)
-            delay = min(2 ** attempt, 16) + random.uniform(0, 1)
+                logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e)
-            logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e)
+                await asyncio.sleep(delay)
-            _time.sleep(delay)
+                continue
+            raise
     raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
 
 
@@ -95,15 +118,13 @@ async def fetch_bzzoiro_events(
     date_to: str | None = None,
     limit: int = 200,
 ) -> list[dict]:
-    """抓取 bzzoiro 原始事件(异步包装)。"""
+    """抓取 bzzoiro 原始事件(纯异步,无需 run_in_executor)。"""
     league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
     if league_id is None:
         raise ValueError(f"未知联赛代码: {league_code}")
 
-    loop = asyncio.get_event_loop()
     rows: list[dict] = []
     offset = 0
-    payload: dict | list = {}
     while True:
         params: dict = {
             "league_id": league_id,
@@ -115,8 +136,7 @@ async def fetch_bzzoiro_events(
             params["date_from"] = str(date_from)[:10]
         if date_to:
             params["date_to"] = str(date_to)[:10]
-        # 显式位置参数,避免 lambda 闭包捕获循环变量
+        payload = await _fetch_json_async("/events/", params)
-        payload = await loop.run_in_executor(None, _fetch_json_sync, "/events/", params)
         batch = payload.get("results") or []
         if not batch:
             break
@@ -127,10 +147,141 @@ async def fetch_bzzoiro_events(
             break
         if len(batch) < limit:
             break
-        await asyncio.sleep(REQUEST_INTERVAL)
+        # 令牌桶已在 _fetch_json_async 内部处理,这里不再需要固定 sleep
     return rows
 
 
+async def _write_bronze_events(db, raw_events: list[dict], batch_id: str) -> None:
+    """将原始事件写入 Bronze 层(RawEvent 表)。
+
+    P2-1: 写入血缘记录,追踪 Bronze 层摄取过程。
+    """
+    for raw in raw_events:
+        try:
+            source_record_id = str(raw.get("id", ""))
+            if not source_record_id:
+                # P1-fix: 不再静默跳过,记录到死信表
+                logger.warning("bzzoiro raw event missing id: %s", str(raw)[:200])
+                await _write_ingest_failure(
+                    db, "bzzoiro", "match", "",
+                    "normalize_error", "raw event missing id", raw,
+                )
+                continue
+            bronze = RawEvent(
+                source_system="bzzoiro",
+                source_record_id=source_record_id,
+                raw_payload=raw,
+                ingest_batch_id=batch_id,
+            )
+            db.add(bronze)
+        except Exception as e:
+            logger.warning("bzzoiro raw_event write failed: %s", e)
+    try:
+        await db.flush()
+    except Exception as e:
+        logger.warning("bzzoiro raw_events flush failed: %s", e)
+
+    # P2-1: 写入 Bronze 层血缘记录
+    for raw in raw_events:
+        source_record_id = str(raw.get("id", ""))
+        if not source_record_id:
+            continue
+        await _write_data_lineage(
+            db,
+            source_system="bzzoiro",
+            source_record_id=source_record_id,
+            target_table="raw_events",
+            transform_name="bronze_ingest",
+            batch_id=batch_id,
+        )
+
+
+async def _write_ingest_failure(
+    db,
+    source_system: str,
+    entity_type: str,
+    source_record_id: str,
+    error_type: str,
+    error_detail: str,
+    raw_payload: dict | None,
+) -> None:
+    """写入采集失败到死信表(IngestFailure)。"""
+    try:
+        from datetime import timedelta
+        failure = IngestFailure(
+            source_system=source_system,
+            entity_type=entity_type,
+            source_record_id=source_record_id,
+            error_type=error_type,
+            error_detail=error_detail[:2000] if error_detail else None,
+            raw_payload=raw_payload,
+            status="pending",
+            next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
+        )
+        db.add(failure)
+        await db.flush()
+    except Exception:
+        logger.warning("bzzoiro ingest_failure write failed", exc_info=True)
+
+
+async def _write_data_lineage(
+    db,
+    *,
+    source_system: str,
+    source_record_id: str,
+    target_table: str,
+    target_id: int | None = None,
+    transform_name: str,
+    transform_detail: str | None = None,
+    batch_id: str | None = None,
+) -> None:
+    """写入数据血缘记录(DataLineage 表)。"""
+    try:
+        lineage = DataLineage(
+            source_system=source_system,
+            source_record_id=source_record_id,
+            target_table=target_table,
+            target_id=target_id,
+            transform_name=transform_name,
+            transform_detail=transform_detail,
+            batch_id=batch_id,
+        )
+        db.add(lineage)
+        await db.flush()
+    except Exception:
+        logger.warning("bzzoiro data_lineage write failed", exc_info=True)
+
+
+async def _write_data_quality_check(
+    db,
+    *,
+    check_name: str,
+    entity_type: str,
+    entity_id: str | None = None,
+    expected_value: str | None = None,
+    actual_value: str | None = None,
+    passed: bool,
+    severity: str = "warning",
+    detail: str | None = None,
+) -> None:
+    """写入数据质量检查结果(DataQualityCheck 表)。"""
+    try:
+        check = DataQualityCheck(
+            check_name=check_name,
+            entity_type=entity_type,
+            entity_id=entity_id,
+            expected_value=expected_value,
+            actual_value=actual_value,
+            passed=passed,
+            severity=severity,
+            detail=detail,
+        )
+        db.add(check)
+        await db.flush()
+    except Exception:
+        logger.warning("bzzoiro data_quality_check write failed", exc_info=True)
+
+
 @register
 class BzzoiroSource:
     """bzzoiro 数据源(实现 DataSource 协议)。"""
@@ -151,6 +302,7 @@ class BzzoiroSource:
         注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
         """
         result: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
+        batch_id = uuid.uuid4().hex[:16]
 
         for code in leagues:
             league_r: dict = {"inserted": 0, "updated": 0, "errors": []}
@@ -159,9 +311,17 @@ class BzzoiroSource:
             except Exception as e:
                 logger.exception("bzzoiro fetch failed for %s", code)
                 league_r["errors"].append(f"fetch failed: {e}")
+                # 写死信表
+                await _write_ingest_failure(
+                    db, "bzzoiro", "match", f"fetch_{code}_{batch_id}",
+                    "fetch_error", str(e), {"league": code, "date_from": date_from, "date_to": date_to},
+                )
                 result["leagues"][code] = league_r
                 continue
 
+            # === Bronze 层:先存 RawEvent ===
+            await _write_bronze_events(db, raw_events, batch_id)
+
             # 获取或创建联赛
             stmt = select(League).where(League.code == code)
             league = (await db.execute(stmt)).scalar_one_or_none()
@@ -186,8 +346,13 @@ class BzzoiroSource:
                         try:
                             nm.validate()
                         except Exception as e:
-                            logger.debug("normalize skip: %s", e)
+                            # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors)
-                            league_r["errors"].append(f"normalize: {e}")
+                            logger.warning("normalize skip: %s", e)
+                            # 写死信表:规范化失败
+                            await _write_ingest_failure(
+                                db, "bzzoiro", "match", str(raw.get("id", "")),
+                                "normalize_error", str(e), raw,
+                            )
                             continue
                         normalized_matches.append((nm, raw))
                         all_team_names.add(nm.home_team)
@@ -198,11 +363,25 @@ class BzzoiroSource:
                     teams = (await db.execute(stmt)).scalars().all()
                     team_name_to_id = {t.name: t.id for t in teams}
 
-                # 预加载已有比赛(完整对象)
+                # P1-2: 按需加载,只加载 raw_events 涉及日期范围的比赛(加 30 天缓冲)
-                stmt = select(Match).where(Match.league_id == league.id)
+                # 避免加载联赛全部历史比赛到内存(多赛季采集时内存溢出)
-                for m in (await db.execute(stmt)).scalars():
+                if normalized_matches:
-                    key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date)
+                    from datetime import timedelta
-                    existing_matches[key] = m
+                    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:
                 # 球队: 内存查找 + 按需创建
@@ -244,6 +423,16 @@ class BzzoiroSource:
                     db.add(m)
                     await db.flush()
                     existing_matches[match_key] = m  # 防止同批重复
+                    # P2-1: Silver 层血缘 — Match 创建
+                    await _write_data_lineage(
+                        db,
+                        source_system="bzzoiro",
+                        source_record_id=str(raw.get("id", "")),
+                        target_table="matches",
+                        target_id=m.id,
+                        transform_name="silver_upsert",
+                        batch_id=batch_id,
+                    )
                     if nm.home_xg is not None or nm.away_xg is not None:
                         now = datetime.now(timezone.utc)
                         stats = MatchStats(
@@ -262,11 +451,22 @@ class BzzoiroSource:
                             home_red_cards=nm.home_red_cards,
                             away_red_cards=nm.away_red_cards,
                             source="bzzoiro",
-                            source_event_id=str(raw.get("id", "")),
+                            source_record_id=str(raw.get("id", "")),
                             retrieved_at=now,
                             available_at=now,
                         )
                         db.add(stats)
+                        await db.flush()
+                        # P2-1: Silver 层血缘 — MatchStats 创建
+                        await _write_data_lineage(
+                            db,
+                            source_system="bzzoiro",
+                            source_record_id=str(raw.get("id", "")),
+                            target_table="match_stats",
+                            target_id=m.id,
+                            transform_name="silver_upsert",
+                            batch_id=batch_id,
+                        )
                     league_r["inserted"] += 1
                 else:
                     # 已有比赛: 直接从内存获取对象更新(无需再查询)
@@ -287,13 +487,36 @@ class BzzoiroSource:
                         now = datetime.now(timezone.utc)
                         existing_match.stats = MatchStats(
                             match_id=existing_match.id,
+                            home_xg=nm.home_xg,
+                            away_xg=nm.away_xg,
+                            home_shots=nm.home_shots,
+                            away_shots=nm.away_shots,
+                            home_shots_on_target=nm.home_shots_on_target,
+                            away_shots_on_target=nm.away_shots_on_target,
+                            home_corners=nm.home_corners,
+                            away_corners=nm.away_corners,
+                            home_possession=nm.home_possession,
+                            home_yellow_cards=nm.home_yellow_cards,
+                            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", "")),
+                            source_record_id=str(raw.get("id", "")),
                             retrieved_at=now,
                             available_at=now,
                         )
                         db.add(existing_match.stats)
                         await db.flush()
+                        # P2-1: Silver 层血缘 — MatchStats 创建(更新路径)
+                        await _write_data_lineage(
+                            db,
+                            source_system="bzzoiro",
+                            source_record_id=str(raw.get("id", "")),
+                            target_table="match_stats",
+                            target_id=existing_match.id,
+                            transform_name="silver_upsert",
+                            batch_id=batch_id,
+                        )
                     if existing_match.stats is not None:
                         for fld in ("home_xg", "away_xg", "home_shots", "away_shots",
                                     "home_shots_on_target", "away_shots_on_target",
@@ -312,4 +535,19 @@ class BzzoiroSource:
             result["leagues"][code] = league_r
             result["total_inserted"] += league_r["inserted"]
             result["total_updated"] += league_r["updated"]
+
+            # P2-2: 数据质量检查 — 行计数合理性(某联赛比赛数不应为 0)
+            total_records = league_r["inserted"] + league_r["updated"]
+            await _write_data_quality_check(
+                db,
+                check_name="bzzoiro_row_count",
+                entity_type="match",
+                entity_id=code,
+                expected_value=">=1",
+                actual_value=str(total_records),
+                passed=total_records > 0,
+                severity="critical" if total_records == 0 else "info",
+                detail=f"league={code} inserted={league_r['inserted']} updated={league_r['updated']}",
+            )
+
         return result

src/data/injuries.py

@@ -1,15 +1,25 @@
 """伤停数据采集器(api-football / api-sports.io)。
 
 采集伤停数据并入库(injuries 表),供 injuries agent 使用。
+
+改进(P0):
+  - 原子写缓存(tempfile + os.replace)
+  - TTL 分级:未来比赛 1h,当天 5min,历史 7day
+  - 文件锁防止并发写缓存冲突
+  - Bronze 层集成(RawEvent)
+  - 死信表集成(IngestFailure)
 """
 from __future__ import annotations
 
 import asyncio
 import json
 import logging
+import os
 import random
+import tempfile
 import time
-from datetime import datetime, timezone
+import uuid
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 
@@ -17,14 +27,147 @@ import httpx
 
 from src.core.config import settings
 from src.core.http_client import get_client
+from src.db.models import DataQualityCheck
 
 logger = logging.getLogger(__name__)
 
 API_BASE = "https://v3.football.api-sports.io"
 DEFAULT_HOST = "v3.football.api-sports.io"
 
-# 缓存目录
+# P2-3: 缓存目录改用系统临时目录,避免源码树内写入
-_CACHE_DIR = Path(__file__).resolve().parent.parent.parent / "data" / "injuries_cache"
+_CACHE_DIR = Path(tempfile.gettempdir()) / "profeto_injuries"
+
+# 模块级锁:防止并发写同一缓存文件
+_cache_lock = asyncio.Lock()
+
+
+def _sanitize_cache_component(value: str | int | None) -> str:
+    """消毒缓存键组件,防止路径遍历。
+
+    只允许 [A-Za-z0-9_.-],其余字符替换为 '_'。
+    """
+    import re
+    s = str(value) if value is not None else "none"
+    return re.sub(r'[^A-Za-z0-9_.]', '_', s)[:60]
+
+
+def _build_cache_key(prefix: str, *components: str | int | None) -> str:
+    """构造安全的缓存文件名。"""
+    parts = [_sanitize_cache_component(c) for c in components]
+    return f"{prefix}_{'_'.join(parts)}.json"
+
+
+def _compute_ttl_hours(date_str: str | None, fixture_id: int | None, league_id: int | None) -> float:
+    """根据查询参数计算缓存 TTL(小时)。
+
+    TTL 分级策略:
+      - 未来比赛(date > now): 1 小时(赛前伤停变化频繁)
+      - 当天比赛(date == today): 5 分钟(赛中实时更新)
+      - 历史比赛(date < now): 7 天(历史数据不变)
+      - 无日期参数: 1 小时(保守策略)
+    """
+    if date_str is None:
+        # 无日期参数(按 fixture_id 或 league_id 查询),保守 TTL
+        return 1.0
+    try:
+        query_date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        today = now.date()
+        if query_date.date() > today:
+            return 1.0  # 未来
+        elif query_date.date() == today:
+            return 5.0 / 60  # 当天:5 分钟
+        else:
+            return 168.0  # 历史:7 天
+    except (ValueError, TypeError):
+        return 1.0  # 解析失败,保守 TTL
+
+
+def _write_cache_atomic(cache_file: Path, data: Any) -> None:
+    """原子写缓存文件。
+
+    使用 tempfile + os.replace 实现原子写,防止读到写了一半的文件。
+    配合模块级 asyncio.Lock,杜绝并发写冲突。
+    """
+    cache_file.parent.mkdir(parents=True, exist_ok=True)
+    # 在同一目录创建临时文件(保证 os.replace 是原子操作)
+    fd, tmp_path = tempfile.mkstemp(
+        dir=str(cache_file.parent),
+        prefix=f".{cache_file.name}.tmp_",
+    )
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, default=str)
+        os.replace(tmp_path, cache_file)
+    except BaseException:
+        # 失败时清理临时文件
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
+def _cleanup_expired_cache() -> int:
+    """P2-4: 扫描缓存目录,删除过期的 .json 文件。
+
+    TTL 策略(保守取最大值,避免误删有效缓存):
+      - 最短 TTL 为 5 分钟(当天比赛),但清理阈值用 1 小时
+      - 超过 1 小时的缓存文件视为过期
+
+    Returns:
+        删除的文件数。
+    """
+    import glob as _glob
+
+    cache_dir = _CACHE_DIR
+    if not cache_dir.exists():
+        return 0
+
+    removed = 0
+    max_ttl_seconds = 3600  # 1 小时(保守阈值,最短实际 TTL 5 分钟)
+    now_ts = time.time()
+    for cache_file in _glob.glob(str(cache_dir / "*.json")):
+        try:
+            file_age = now_ts - os.path.getmtime(cache_file)
+            if file_age > max_ttl_seconds:
+                os.unlink(cache_file)
+                removed += 1
+        except OSError:
+            continue
+    if removed > 0:
+        logger.info("injuries cache cleanup: removed %d expired files", removed)
+    return removed
+
+
+async def _write_data_quality_check(
+    db,
+    *,
+    check_name: str,
+    entity_type: str,
+    entity_id: str | None = None,
+    expected_value: str | None = None,
+    actual_value: str | None = None,
+    passed: bool,
+    severity: str = "warning",
+    detail: str | None = None,
+) -> None:
+    """写入数据质量检查结果(DataQualityCheck 表)。"""
+    try:
+        check = DataQualityCheck(
+            check_name=check_name,
+            entity_type=entity_type,
+            entity_id=entity_id,
+            expected_value=expected_value,
+            actual_value=actual_value,
+            passed=passed,
+            severity=severity,
+            detail=detail,
+        )
+        db.add(check)
+        await db.flush()
+    except Exception:
+        logger.warning("injuries data_quality_check write failed", exc_info=True)
 
 
 async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = None, league_id: int | None = None) -> list[dict]:
@@ -42,20 +185,25 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No
     if not api_key:
         raise RuntimeError("API_FOOTBALL_KEY 未设置")
 
+    # P2-4: 清理过期缓存文件
+    _cleanup_expired_cache()
+
     cache_dir = _CACHE_DIR
     cache_dir.mkdir(parents=True, exist_ok=True)
 
-    # 缓存命中 (7 天内有效)
+    # TTL 分级缓存
-    cache_key = f"injuries_{date}_{fixture_id}_{league_id}.json"
+    ttl_hours = _compute_ttl_hours(date, fixture_id, league_id)
+    # P0-fix: 消毒缓存键,防止路径遍历(date 参数可能含 '../' 等)
+    cache_key = _build_cache_key("injuries", date, fixture_id, league_id)
     cache_file = cache_dir / cache_key
     if cache_file.exists():
         age_hours = (time.time() - cache_file.stat().st_mtime) / 3600
-        if age_hours < 168:  # 7 天
+        if age_hours < ttl_hours:
-            logger.debug("injuries cache hit: %s (%.1fh old)", cache_key, age_hours)
+            logger.debug("injuries cache hit: %s (%.1fh old, ttl=%.2fh)", cache_key, age_hours, ttl_hours)
             with open(cache_file, encoding="utf-8") as f:
                 return json.load(f)
         else:
-            logger.debug("injuries cache expired: %s (%.1fh old)", cache_key, age_hours)
+            logger.debug("injuries cache expired: %s (%.1fh old, ttl=%.2fh)", cache_key, age_hours, ttl_hours)
 
     headers = {
         "x-apisports-key": api_key,
@@ -92,9 +240,9 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No
     data = resp.json()
     injuries = data.get("response", [])
 
-    # 写缓存
+    # 原子写缓存 + 文件锁(防止并发写冲突)
-    with open(cache_file, "w", encoding="utf-8") as f:
+    async with _cache_lock:
-        json.dump(injuries, ensure_ascii=False, default=str, fp=f)
+        _write_cache_atomic(cache_file, injuries)
 
     return injuries
 
@@ -103,27 +251,74 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
     """采集伤停数据并入库(injuries 表)。
 
     注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
+
+    P1-4: 批量幂等检查,避免逐条查询的竞态条件(并发采集时 IntegrityError)。
+    P1 Bronze: 采集成功后先存 RawEvent,再规范化。
+    P1 死信: 采集/规范化失败时写 IngestFailure。
     """
     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
+    from src.db.models import DataQualityCheck, Injury, IngestFailure, RawEvent, Team
 
     result = {"count": 0, "inserted": 0, "errors": []}
+    batch_id = uuid.uuid4().hex[:16]
 
     try:
         raw_injuries = await fetch_injuries(date=date)
     except Exception as e:
         logger.exception("injuries fetch failed")
         result["errors"].append(f"fetch failed: {e}")
+        # 写死信表(IngestFailure)
+        try:
+            failure = IngestFailure(
+                source_system="api-football",
+                entity_type="injury",
+                source_record_id=f"fetch_{batch_id}",
+                error_type="fetch_error",
+                error_detail=str(e)[:2000],
+                raw_payload={"date": date},
+                status="pending",
+                next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
+            )
+            db.add(failure)
+            await db.flush()
+        except Exception:
+            logger.warning("failed to write fetch error to ingest_failures", exc_info=True)
         return result
 
     result["count"] = len(raw_injuries)
 
+    # === Bronze 层:写 RawEvent ===
+    for raw in raw_injuries:
+        try:
+            player = raw.get("player", {}) or {}
+            fixture = raw.get("fixture", {}) or {}
+            source_record_id = f"inj_{player.get('id')}_{fixture.get('id')}"
+            bronze = RawEvent(
+                source_system="api-football",
+                source_record_id=source_record_id,
+                raw_payload=raw,
+                ingest_batch_id=batch_id,
+            )
+            db.add(bronze)
+        except Exception as e:
+            logger.warning("raw_event write failed for injury: %s", e)
+    try:
+        await db.flush()
+    except Exception as e:
+        logger.warning("raw_events flush failed: %s", e)
+
     # 预加载所有球队(用于按名匹配)
     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] = []
+    parse_failures: list[dict] = []  # 规范化失败的原始数据,用于写死信表
     for raw in raw_injuries:
         try:
             player = raw.get("player", {}) or {}
@@ -144,43 +339,145 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
                 except (ValueError, AttributeError):
                     pass
 
+            # 强制 int 转换,API 可能返回字符串
             player_id = player.get("id")
+            try:
+                player_id = int(player_id) if player_id is not None else None
+            except (ValueError, TypeError):
+                player_id = None
             fixture_id = fixture.get("id")
+            try:
+                fixture_id = int(fixture_id) if fixture_id is not None else None
+            except (ValueError, TypeError):
+                fixture_id = None
 
-            # 幂等: 已存在则跳过
+            pending_records.append({
-            existing = (
+                "player_id": player_id,
-                await db.execute(
+                "player_name": player_name,
-                    select(Injury).where(
+                "team_id": team_id,
-                        Injury.player_id == player_id,
+                "fixture_id": fixture_id,
-                        Injury.fixture_id == fixture_id,
+                "league_id": (raw.get("league") or {}).get("id"),
-                        Injury.injury_type == player.get("type"),
+                "injury_type": player.get("type"),
-                    )
+                "reason": player.get("reason"),
-                )
+                "injury_date": injury_date,
-            ).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
         except Exception as e:
             result["errors"].append(f"parse error: {e}")
+            parse_failures.append({"raw": raw, "error": str(e)})
+
+    # 写规范化失败到死信表
+    for fail in parse_failures:
+        try:
+            raw = fail["raw"]
+            player = raw.get("player", {}) or {}
+            failure = IngestFailure(
+                source_system="api-football",
+                entity_type="injury",
+                source_record_id=f"inj_{player.get('id', 'unknown')}",
+                error_type="normalize_error",
+                error_detail=fail["error"][:2000],
+                raw_payload=raw,
+                status="pending",
+                next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
+            )
+            db.add(failure)
+        except Exception:
+            logger.warning("failed to write parse error to ingest_failures", exc_info=True)
+    if parse_failures:
+        try:
+            await db.flush()
+        except Exception:
+            logger.warning("ingest_failures flush failed", exc_info=True)
+
+    # 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)
+
+    # P2-2: 数据质量检查 — 行计数 / 新增比例
+    total = result["count"]
+    inserted = result["inserted"]
+    # 某日期伤停数不应为 0(除非历史日期)
+    await _write_data_quality_check(
+        db,
+        check_name="injuries_row_count",
+        entity_type="injury",
+        entity_id=date or "unknown",
+        expected_value=">=0",
+        actual_value=str(total),
+        passed=True,
+        severity="info",
+        detail=f"date={date} fetched={total} inserted={inserted}",
+    )
 
     # 注意: 不在此处 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]:
     """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。
 
@@ -188,31 +485,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 的记录。
+        as_of: 数据截止时间(用于回测防泄漏)
-               用于回测时防止"未来采集的数据"泄漏到历史预测。
-               必须保持 timezone-aware datetime,不会截断为 date。
-    """
-    from sqlalchemy import or_, select
 
+    Returns:
+        伤停记录列表
+    """
+    from sqlalchemy import select
     from src.db.models import Injury
 
-    # 只处理 match_date:去掉时间部分,仅比较日期
+    if hasattr(match_date, "date") and callable(match_date.date):
-    if hasattr(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 双轨(不再需要独立回填)
+    - 去掉 XGBackoff 双轨(不再需要独立回填)
-  - 去掉 PIT 时间契约(无训练集要防泄漏)
+    - 去掉 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),
+        home_xg=home_xg,
-        away_xg=_to_float(away_xg),
+        away_xg=away_xg,
     )

src/data/rate_limiter.py

@@ -0,0 +1,98 @@
+"""令牌桶限流器。
+
+替代 bzzoiro 中固定的 REQUEST_INTERVAL sleep,提供更精细的速率控制。
+支持突发流量和平滑限流。
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+
+logger = logging.getLogger(__name__)
+
+
+class TokenBucket:
+    """异步令牌桶限流器。
+
+    用法:
+        limiter = TokenBucket(rate=5.0, capacity=10)
+        await limiter.acquire()  # 等待直到有可用令牌
+
+    Args:
+        rate: 每秒补充的令牌数
+        capacity: 桶容量(允许的最大突发量)
+    """
+
+    def __init__(self, rate: float, capacity: int | None = None) -> None:
+        if rate <= 0:
+            raise ValueError(f"rate must be positive, got {rate}")
+        self._rate = rate
+        self._capacity = capacity or max(1, int(rate * 2))
+        self._tokens: float = self._capacity
+        self._last_refill = time.monotonic()
+        self._lock = asyncio.Lock()
+
+    async def acquire(self, tokens: int = 1) -> None:
+        """获取指定数量的令牌,不足时等待。
+
+        Args:
+            tokens: 需要获取的令牌数
+        """
+        if tokens <= 0:
+            return
+        if tokens > self._capacity:
+            raise ValueError(f"requested {tokens} exceeds capacity {self._capacity}")
+
+        while True:
+            async with self._lock:
+                self._refill()
+                if self._tokens >= tokens:
+                    self._tokens -= tokens
+                    return
+
+            # 计算需要等待的时间
+            wait_time = (tokens - self._tokens) / self._rate
+            logger.debug("token bucket: waiting %.2fs for %d tokens", wait_time, tokens)
+            await asyncio.sleep(wait_time)
+
+    def _refill(self) -> None:
+        """补充令牌(基于经过的时间)。"""
+        now = time.monotonic()
+        elapsed = now - self._last_refill
+        if elapsed > 0:
+            self._tokens = min(self._capacity, self._tokens + elapsed * self._rate)
+            self._last_refill = now
+
+    @property
+    def tokens(self) -> float:
+        """当前可用令牌数(近似)。"""
+        self._refill()
+        return self._tokens
+
+
+class RateLimitedClient:
+    """HTTP 客户端限流包装器。
+
+    在 httpx 客户端之上添加令牌桶限流,透明地控制请求速率。
+
+    用法:
+        client = get_client()
+        limiter = TokenBucket(rate=5.0)
+        wrapper = RateLimitedClient(client, limiter)
+        resp = await wrapper.get(url)
+    """
+
+    def __init__(self, client, limiter: TokenBucket) -> None:
+        self._client = client
+        self._limiter = limiter
+
+    async def get(self, url: str, **kwargs):
+        """限流的 GET 请求。"""
+        await self._limiter.acquire()
+        return await self._client.get(url, **kwargs)
+
+    async def post(self, url: str, **kwargs):
+        """限流的 POST 请求。"""
+        await self._limiter.acquire()
+        return await self._client.post(url, **kwargs)

src/data/retry_worker.py

@@ -0,0 +1,126 @@
+"""P2-5: IngestFailure 重试 Worker。
+
+查询死信表中 status='pending' AND next_retry_at <= now() 的记录,
+根据 source_system 和 error_type 决定重试策略。
+作为 FastAPI startup 事件注册,在应用启动时执行一次清理。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from src.db.base import AsyncSessionLocal
+
+logger = logging.getLogger(__name__)
+
+# 最大重试次数上限
+_MAX_RETRIES = 5
+
+
+async def _retry_ingest_failures(db) -> dict:
+    """重试待处理的采集失败记录。
+
+    查询条件: status='pending' AND next_retry_at <= now()
+    重试策略:
+      - retry_count >= MAX_RETRIES → 标记 abandoned
+      - error_type=fetch_error → 退避重试(更新 next_retry_at)
+      - error_type=normalize_error → 规范化失败通常是数据问题,退避重试
+      - error_type=db_error → 数据库问题,退避重试
+
+    Args:
+        db: SQLAlchemy async session
+
+    Returns:
+        统计 dict: {"retried": int, "abandoned": int, "errors": list[str]}
+    """
+    result = {"retried": 0, "abandoned": 0, "errors": []}
+
+    from src.db.models import IngestFailure
+
+    now = datetime.now(timezone.utc)
+
+    # 查询待重试记录
+    stmt = (
+        select(IngestFailure)
+        .where(IngestFailure.status == "pending")
+        .where(IngestFailure.next_retry_at <= now)
+        .order_by(IngestFailure.next_retry_at)
+        .limit(50)  # 每批最多处理 50 条,避免长时间持有事务
+    )
+
+    try:
+        rows = (await db.execute(stmt)).scalars().all()
+    except Exception as e:
+        logger.warning("retry_worker: query failed: %s", e)
+        result["errors"].append(f"query failed: {e}")
+        return result
+
+    if not rows:
+        return result
+
+    logger.info("retry_worker: found %d pending failures to retry", len(rows))
+
+    for failure in rows:
+        try:
+            # 超过最大重试次数 → 放弃
+            if failure.retry_count >= _MAX_RETRIES:
+                failure.status = "abandoned"
+                failure.resolved_at = now
+                await db.flush()
+                result["abandoned"] += 1
+                logger.info(
+                    "retry_worker: abandoned %s/%s after %d retries",
+                    failure.source_system, failure.source_record_id, failure.retry_count,
+                )
+                continue
+
+            # 退避计算: 5min * 2^retry_count,最大 24 小时
+            backoff_minutes = min(5 * (2 ** failure.retry_count), 1440)
+            next_retry = now + timedelta(minutes=backoff_minutes)
+
+            # 更新重试状态
+            failure.retry_count += 1
+            failure.next_retry_at = next_retry
+            failure.status = "retrying"
+            await db.flush()
+            result["retried"] += 1
+
+            logger.info(
+                "retry_worker: scheduled retry %d/%d for %s/%s (next: %s)",
+                failure.retry_count, _MAX_RETRIES,
+                failure.source_system, failure.source_record_id,
+                next_retry.isoformat(),
+            )
+
+        except Exception as e:
+            await db.rollback()
+            error_msg = f"retry {failure.source_system}/{failure.source_record_id}: {e}"
+            result["errors"].append(error_msg)
+            logger.warning("retry_worker: %s", error_msg)
+
+    return result
+
+
+async def run_retry_worker() -> None:
+    """启动入口: 获取 DB session 并执行重试逻辑。
+
+    设计为幂等操作 — 多次运行不会产生副作用(受 next_retry_at 约束)。
+    """
+    logger.info("retry_worker: starting ingest failure retry sweep")
+
+    try:
+        async with AsyncSessionLocal() as db:
+            try:
+                stats = await _retry_ingest_failures(db)
+                await db.commit()
+                logger.info(
+                    "retry_worker: sweep complete — retried=%d abandoned=%d errors=%d",
+                    stats["retried"], stats["abandoned"], len(stats["errors"]),
+                )
+            except Exception as e:
+                await db.rollback()
+                logger.warning("retry_worker: session failed: %s", e)
+    except Exception as e:
+        logger.warning("retry_worker: failed to get DB session: %s", e)

src/data/understat.py

@@ -2,6 +2,12 @@
 
 迁移自旧项目 app/data/sources/understat.py,改成 async。
 使用 Repository 模式进行数据访问,不直接控制事务。
+
+改进(P1):
+  - Bronze 层集成:采集后先存 RawEvent,再规范化
+  - 死信表集成:采集/规范化失败写 IngestFailure
+  - xG 覆盖更新:当 understat 数据更新时覆盖旧值(全量覆盖模式)
+  - xG 追踪字段:xg_source / xg_updated_at / xg_source_record_id
 """
 from __future__ import annotations
 
@@ -10,15 +16,17 @@ import json
 import logging
 import random
 import re
-from datetime import datetime, timezone
+import uuid
+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
 from src.data.normalize import normalize_understat
 from src.data.sources import register
-from src.db.models import League, Match, MatchStats, Team
+from src.db.models import DataLineage, DataQualityCheck, IngestFailure, League, Match, MatchStats, RawEvent, Team
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +54,7 @@ async def fetch_understat(league_code: str, season: int) -> list[dict]:
         "Referer": f"https://understat.com/league/{understat_league}/{season}",
     }
 
+
     # 重试:网络错误 / 5xx / 429
     last_exc: Exception | None = None
     for attempt in range(3):
@@ -57,12 +66,10 @@ async def fetch_understat(league_code: str, season: int) -> list[dict]:
         except Exception as e:
             last_exc = e
             if attempt == 2:
-                raise
+                raise RuntimeError(f"understat fetch failed: {e}") from e
             delay = min(2 ** attempt, 8) + random.uniform(0, 1)
             logger.warning("understat fetch failed, retry %d in %.1fs: %s", attempt + 1, delay, e)
             await asyncio.sleep(delay)
-    else:
-        raise RuntimeError(f"understat fetch failed: {last_exc}")
 
     # understat 返回 JS 对象,需要提取 JSON
     text = resp.text
@@ -76,6 +83,148 @@ 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 "")
+
+
+async def _write_understat_bronze(db, raw_matches: list[dict], batch_id: str) -> None:
+    """将 understat 原始数据写入 Bronze 层(RawEvent 表)。
+
+    P2-1: 写入血缘记录,追踪 Bronze 层摄取过程。
+    """
+    for raw in raw_matches:
+        try:
+            source_record_id = str(raw.get("id", ""))
+            if not source_record_id:
+                # P1-fix: 不再静默跳过,记录到死信表
+                logger.warning("understat raw match missing id: %s", str(raw)[:200])
+                await _write_ingest_failure(
+                    db, "understat", "match", "",
+                    "normalize_error", "raw match missing id", raw,
+                )
+                continue
+            bronze = RawEvent(
+                source_system="understat",
+                source_record_id=source_record_id,
+                raw_payload=raw,
+                ingest_batch_id=batch_id,
+            )
+            db.add(bronze)
+        except Exception as e:
+            logger.warning("understat raw_event write failed: %s", e)
+    try:
+        await db.flush()
+    except Exception as e:
+        logger.warning("understat raw_events flush failed: %s", e)
+
+    # P2-1: 写入 Bronze 层血缘记录
+    for raw in raw_matches:
+        source_record_id = str(raw.get("id", ""))
+        if not source_record_id:
+            continue
+        await _write_data_lineage(
+            db,
+            source_system="understat",
+            source_record_id=source_record_id,
+            target_table="raw_events",
+            transform_name="bronze_ingest",
+            batch_id=batch_id,
+        )
+
+
+async def _write_ingest_failure(
+    db,
+    source_system: str,
+    entity_type: str,
+    source_record_id: str,
+    error_type: str,
+    error_detail: str,
+    raw_payload: dict | None,
+) -> None:
+    """写入采集失败到死信表(IngestFailure)。"""
+    try:
+        failure = IngestFailure(
+            source_system=source_system,
+            entity_type=entity_type,
+            source_record_id=source_record_id,
+            error_type=error_type,
+            error_detail=error_detail[:2000] if error_detail else None,
+            raw_payload=raw_payload,
+            status="pending",
+            next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
+        )
+        db.add(failure)
+        await db.flush()
+    except Exception:
+        logger.warning("understat ingest_failure write failed", exc_info=True)
+
+
+async def _write_data_lineage(
+    db,
+    *,
+    source_system: str,
+    source_record_id: str,
+    target_table: str,
+    target_id: int | None = None,
+    transform_name: str,
+    transform_detail: str | None = None,
+    batch_id: str | None = None,
+) -> None:
+    """写入数据血缘记录(DataLineage 表)。"""
+    try:
+        lineage = DataLineage(
+            source_system=source_system,
+            source_record_id=source_record_id,
+            target_table=target_table,
+            target_id=target_id,
+            transform_name=transform_name,
+            transform_detail=transform_detail,
+            batch_id=batch_id,
+        )
+        db.add(lineage)
+        await db.flush()
+    except Exception:
+        logger.warning("understat data_lineage write failed", exc_info=True)
+
+
+async def _write_data_quality_check(
+    db,
+    *,
+    check_name: str,
+    entity_type: str,
+    entity_id: str | None = None,
+    expected_value: str | None = None,
+    actual_value: str | None = None,
+    passed: bool,
+    severity: str = "warning",
+    detail: str | None = None,
+) -> None:
+    """写入数据质量检查结果(DataQualityCheck 表)。"""
+    try:
+        check = DataQualityCheck(
+            check_name=check_name,
+            entity_type=entity_type,
+            entity_id=entity_id,
+            expected_value=expected_value,
+            actual_value=actual_value,
+            passed=passed,
+            severity=severity,
+            detail=detail,
+        )
+        db.add(check)
+        await db.flush()
+    except Exception:
+        logger.warning("understat data_quality_check write failed", exc_info=True)
+
+
 @register
 class UnderstatSource:
     """understat xG 数据源(实现 DataSource 协议)。"""
@@ -83,25 +232,36 @@ class UnderstatSource:
     name = "understat"
 
     async def ingest(self, db, *, league: str, season: int) -> dict:
-        """采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。
+        """采集 understat xG → 回填到现有 Match。
 
         注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。
+
+        P1-3: 批量查询优化,将单赛季 380 场 × 3 次 DB 往返降为 3 次查询。
+        P1-4: xG 覆盖更新模式,当 understat 数据更新时覆盖旧值。
         """
-        from src.db.repositories import LeagueRepository, MatchRepository, TeamRepository
+        from src.db.repositories import LeagueRepository, TeamRepository
 
         result = {"updated": 0, "skipped": 0, "unmatched": 0, "errors": []}
+        batch_id = uuid.uuid4().hex[:16]
 
         try:
             raw_matches = await fetch_understat(league, season)
         except Exception as e:
             logger.exception("understat fetch failed for %s %s", league, season)
             result["errors"].append(f"fetch failed: {e}")
+            # 写死信表
+            await _write_ingest_failure(
+                db, "understat", "match", f"fetch_{league}_{season}_{batch_id}",
+                "fetch_error", str(e), {"league": league, "season": season},
+            )
             return result
 
+        # === Bronze 层:先存 RawEvent ===
+        await _write_understat_bronze(db, raw_matches, batch_id)
+
         # 使用 Repository
         league_repo = LeagueRepository(db)
         team_repo = TeamRepository(db)
-        match_repo = MatchRepository(db)
 
         # 查联赛
         league_obj = await league_repo.get_by_code(league)
@@ -109,6 +269,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
@@ -119,40 +282,126 @@ class UnderstatSource:
                     continue
             except Exception as e:
                 result["errors"].append(f"normalize: {e}")
+                # 写死信表:规范化失败
+                await _write_ingest_failure(
+                    db, "understat", "match", str(raw.get("id", "")),
+                    "normalize_error", str(e), raw,
+                )
                 continue
+            normalized_matches.append((nm, raw))
+            all_team_names.add(nm.home_team)
+            all_team_names.add(nm.away_team)
 
-            # 匹配已有 Match(天级) - 使用 Repository
+        if not normalized_matches:
-            home_team = await team_repo.get_by_name(nm.home_team)
+            return result
-            away_team = await team_repo.get_by_name(nm.away_team)
+
-            if home_team is None or away_team is None:
+        # === 批量查询球队(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(
+            match_key = _match_key(home_team_id, away_team_id, nm.date)
-                league_obj.id, home_team.id, away_team.id, nm.date
+            existing = match_dict.get(match_key)
-            )
             if existing is None:
                 result["unmatched"] += 1
                 continue
 
-            # 回填 xG
+            now = datetime.now(timezone.utc)
+            source_record_id = str(raw.get("id", ""))
+
+            # 创建 stats 记录(如果不存在)
             if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None):
-                now = datetime.now(timezone.utc)
                 existing.stats = MatchStats(
                     match_id=existing.id,
                     source="understat",
-                    source_event_id=str(raw.get("id", "")),
+                    source_record_id=source_record_id,
                     retrieved_at=now,
                     available_at=now,
+                    xg_source="understat",
+                    xg_updated_at=now,
+                    xg_source_record_id=source_record_id,
                 )
                 db.add(existing.stats)
                 await db.flush()
+                # P2-1: Silver 层血缘 — MatchStats 创建
+                await _write_data_lineage(
+                    db,
+                    source_system="understat",
+                    source_record_id=source_record_id,
+                    target_table="match_stats",
+                    target_id=existing.id,
+                    transform_name="silver_upsert",
+                    batch_id=batch_id,
+                )
+
+            # xG 覆盖更新模式:当 understat 数据更新时覆盖旧值
             if existing.stats is not None:
-                if existing.stats.home_xg is None and nm.home_xg is not None:
+                xg_updated = False
+                if nm.home_xg is not None:
                     existing.stats.home_xg = nm.home_xg
+                    existing.stats.xg_source = "understat"
+                    existing.stats.xg_updated_at = now
+                    existing.stats.xg_source_record_id = source_record_id
                     result["updated"] += 1
-                if existing.stats.away_xg is None and nm.away_xg is not None:
+                    xg_updated = True
+                if nm.away_xg is not None:
                     existing.stats.away_xg = nm.away_xg
+                    existing.stats.xg_source = "understat"
+                    existing.stats.xg_updated_at = now
+                    existing.stats.xg_source_record_id = source_record_id
+                    xg_updated = True
+                # P2-1: Silver 层血缘 — xG 覆盖更新
+                if xg_updated:
+                    await _write_data_lineage(
+                        db,
+                        source_system="understat",
+                        source_record_id=source_record_id,
+                        target_table="match_stats",
+                        target_id=existing.id,
+                        transform_name="silver_xg_update",
+                        batch_id=batch_id,
+                        transform_detail=f"home_xg={nm.home_xg} away_xg={nm.away_xg}",
+                    )
+
+        # P2-2: 数据质量检查 — 行计数合理性
+        total_processed = result["updated"] + result["skipped"] + result["unmatched"]
+        await _write_data_quality_check(
+            db,
+            check_name="understat_row_count",
+            entity_type="match",
+            entity_id=f"{league}_{season}",
+            expected_value=">=1",
+            actual_value=str(total_processed),
+            passed=total_processed > 0,
+            severity="critical" if total_processed == 0 else "info",
+            detail=f"league={league} season={season} updated={result['updated']} skipped={result['skipped']} unmatched={result['unmatched']}",
+        )
 
         # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
         return result

src/db/base.py

@@ -17,8 +17,10 @@ engine = create_async_engine(
     settings.DATABASE_URL,
     echo=False,
     pool_pre_ping=True,
-    pool_size=10,
+    pool_size=settings.DB_POOL_SIZE,
-    max_overflow=20,
+    max_overflow=settings.DB_MAX_OVERFLOW,
+    pool_timeout=settings.DB_POOL_TIMEOUT,
+    pool_recycle=settings.DB_POOL_RECYCLE,
 )
 
 AsyncSessionLocal = async_sessionmaker(

src/db/models.py

@@ -1,4 +1,4 @@
-"""6 张表 ORM: leagues / teams / matches / match_stats / predictions / injuries。"""
+"""ORM 模型: leagues / teams / matches / match_stats / predictions / injuries / raw_events / ingest_failures / data_quality_checks / data_lineage。"""
 from __future__ import annotations
 
 from datetime import date, datetime, timezone
@@ -14,6 +14,7 @@ from sqlalchemy import (
     Integer,
     String,
     Text,
+    UniqueConstraint,
     func,
 )
 from sqlalchemy.dialects.postgresql import JSONB
@@ -130,12 +131,18 @@ class MatchStats(Base):
     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))
+    # xG 数据追踪:支持 understat 纠正旧 xG 值
+    xg_source: Mapped[str | None] = mapped_column(String(30))  # 具体 xG 数据源
+    xg_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # xG 最后更新时间
+    xg_source_record_id: Mapped[str | None] = mapped_column(String(100))  # xG 对应的源记录 ID
 
     match: Mapped[Match] = relationship(back_populates="stats")
 
     __table_args__ = (
         # 数据血缘时间过滤查询用(按 available_at 取「赛前已可得」的统计)
         Index("ix_match_stats_available_at", "available_at"),
+        # xG 数据源追踪查询用
+        Index("ix_match_stats_xg_source", "xg_source", "xg_updated_at"),
     )
 
 
@@ -198,6 +205,11 @@ class Prediction(Base):
     match: Mapped[Match] = relationship(back_populates="predictions")
 
     __table_args__ = (
+        # P1-6: 数据库级唯一约束,防止同一 match+provider+model 产生重复预测
+        UniqueConstraint(
+            "match_id", "provider", "model",
+            name="uq_predictions_match_provider_model",
+        ),
         Index("ix_predictions_match", "match_id"),
         Index("ix_predictions_provider_model", "provider", "model"),
         # 数据截止时间过滤查询用(按 prediction_cutoff_at 取「赛前已生成」的预测)
@@ -210,3 +222,108 @@ class Prediction(Base):
         CheckConstraint("mode IN ('single', 'multi')", name="ck_mode_enum"),
         CheckConstraint("status IN ('success', 'failed', 'degraded')", name="ck_status_enum"),
     )
+
+
+# ── Bronze 层:原始事件记录 ──────────────────────────────────────────────
+class RawEvent(Base):
+    """Bronze 层:不可变的原始采集记录。
+
+    每个采集到的原始事件先写入此表,再规范化到 Silver 层(matches / match_stats)。
+    提供完整的数据血缘回溯能力。
+    """
+    __tablename__ = "raw_events"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(30), nullable=False)  # bzzoiro / understat / api-football
+    source_record_id: Mapped[str] = mapped_column(String(100), nullable=False)  # 源系统记录 ID
+    raw_payload: Mapped[dict] = mapped_column(JSONB, nullable=False)  # 完整原始 JSON
+    ingested_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, nullable=False)
+    ingest_batch_id: Mapped[str | None] = mapped_column(String(64))  # 批次 ID,用于关联同次采集
+
+    __table_args__ = (
+        UniqueConstraint("source_system", "source_record_id", name="uq_raw_events_source_record"),
+        Index("ix_raw_events_batch", "ingest_batch_id"),
+        Index("ix_raw_events_source_ingested", "source_system", "ingested_at"),
+    )
+
+
+# ── 死信表:采集失败记录 ─────────────────────────────────────────────────
+class IngestFailure(Base):
+    """采集失败死信表。
+
+    当采集或规范化失败时,写入此表而非仅内存 dict。
+    支持自动重试和人工排查。
+    """
+    __tablename__ = "ingest_failures"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(30), nullable=False)
+    entity_type: Mapped[str] = mapped_column(String(30), nullable=False)  # match / injury / team
+    source_record_id: Mapped[str | None] = mapped_column(String(100))
+    error_type: Mapped[str] = mapped_column(String(50), nullable=False)  # fetch_error / normalize_error / db_error / validation_error
+    error_detail: Mapped[str | None] = mapped_column(Text)
+    raw_payload: Mapped[dict | None] = mapped_column(JSONB)  # 失败时的原始数据,用于重试
+    retry_count: Mapped[int] = mapped_column(Integer, default=0)
+    next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    status: Mapped[str] = mapped_column(String(20), default="pending")  # pending / retrying / resolved / abandoned
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
+    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    __table_args__ = (
+        Index("ix_ingest_failures_status_next_retry", "status", "next_retry_at"),
+        Index("ix_ingest_failures_source", "source_system", "entity_type"),
+        CheckConstraint("status IN ('pending', 'retrying', 'resolved', 'abandoned')", name="ck_ingest_failures_status"),
+        CheckConstraint("retry_count >= 0", name="ck_ingest_failures_retry_nonneg"),
+    )
+
+
+# ── 数据质量检查表 ──────────────────────────────────────────────────────
+class DataQualityCheck(Base):
+    """数据质量检查结果记录。
+
+    每次运行数据质量检查时,将结果写入此表用于趋势分析和告警。
+    """
+    __tablename__ = "data_quality_checks"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    check_name: Mapped[str] = mapped_column(String(100), nullable=False)  # 检查名称
+    entity_type: Mapped[str] = mapped_column(String(30), nullable=False)  # match / team / prediction
+    entity_id: Mapped[str | None] = mapped_column(String(50))  # 具体实体 ID
+    expected_value: Mapped[str | None] = mapped_column(Text)  # 期望值(描述)
+    actual_value: Mapped[str | None] = mapped_column(Text)  # 实际值
+    passed: Mapped[bool] = mapped_column(Boolean, nullable=False)
+    severity: Mapped[str] = mapped_column(String(10), nullable=False, default="warning")  # info / warning / critical
+    detail: Mapped[str | None] = mapped_column(Text)  # 详细描述
+    checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, nullable=False)
+
+    __table_args__ = (
+        Index("ix_dqc_check_time", "check_name", "checked_at"),
+        Index("ix_dqc_entity", "entity_type", "entity_id"),
+        Index("ix_dqc_severity_passed", "severity", "passed"),
+        CheckConstraint("severity IN ('info', 'warning', 'critical')", name="ck_dqc_severity"),
+    )
+
+
+# ── 数据血缘表 ──────────────────────────────────────────────────────────
+class DataLineage(Base):
+    """ETL 全过程元数据记录。
+
+    追踪从 Bronze → Silver → Gold 的完整转换链路。
+    """
+    __tablename__ = "data_lineage"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(30), nullable=False)
+    source_record_id: Mapped[str] = mapped_column(String(100), nullable=False)
+    target_table: Mapped[str] = mapped_column(String(50), nullable=False)  # matches / match_stats / predictions
+    target_id: Mapped[int | None] = mapped_column(Integer)  # 目标表记录 ID
+    transform_name: Mapped[str] = mapped_column(String(100), nullable=False)  # 转换步骤名称
+    transform_detail: Mapped[str | None] = mapped_column(Text)  # 转换详情
+    batch_id: Mapped[str | None] = mapped_column(String(64))  # 批次 ID
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, nullable=False)
+
+    __table_args__ = (
+        Index("ix_lineage_source", "source_system", "source_record_id"),
+        Index("ix_lineage_target", "target_table", "target_id"),
+        Index("ix_lineage_batch", "batch_id"),
+    )

src/db/repositories.py

@@ -5,7 +5,8 @@ Repository 只负责查询,不负责事务提交。
 """
 from __future__ import annotations
 
-from sqlalchemy import func, select
+from datetime import datetime
+from sqlalchemy import select
 from sqlalchemy.orm import selectinload
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -41,9 +42,17 @@ class MatchRepository:
 
         预加载 stats:调用方(understat 回填)会读取 existing.stats,
         async session 下惰性加载会抛 MissingGreenlet。
+
+        P2-3: 使用 match_date_date(已建索引)做等值匹配,避免 func.date()
+        导致的全表扫描。
         """
-        if hasattr(date, "date"):
+        if isinstance(date, datetime):
             date = date.date()
+        elif hasattr(date, "date"):
+            date = date.date()
+        else:
+            # 字符串等其它格式,尝试转换
+            date = datetime.fromisoformat(str(date)).date()
 
         stmt = (
             select(Match)
@@ -51,7 +60,7 @@ class MatchRepository:
             .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)
+            .where(Match.match_date_date == date)
         )
         return (await self._session.execute(stmt)).scalar_one_or_none()
 

src/db/unit_of_work.py

@@ -9,12 +9,16 @@ from __future__ import annotations
 
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
 
 from src.db.base import AsyncSessionLocal
 
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
 
 @asynccontextmanager
-async def get_uow() -> AsyncIterator[AsyncSessionLocal]:
+async def get_uow() -> AsyncIterator[AsyncSession]:
     """创建新的工作单元(用于非路由上下文)。
 
     用法:

src/llm/backtest.py

@@ -3,9 +3,11 @@
 核心机制:
   - 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
@@ -17,6 +19,7 @@ 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__)
 
@@ -69,15 +72,6 @@ class BacktestSummary:
     results: list[BacktestMatchResult] = field(default_factory=list)
 
 
-def _actual_1x2(home: int, away: int) -> str:
-    """实际比分 → 胜平负。"""
-    if home > away:
-        return "1"
-    if home == away:
-        return "X"
-    return "2"
-
-
 async def _get_historical_matches(
     db,
     *,
@@ -156,44 +150,42 @@ async def run_backtest(
 
     summary = BacktestSummary(total=len(candidates), scored=0)
 
-    for c in candidates:
+    # P2: 并发控制,同时最多 3 场预测(避免 LLM API 限流,保护下游服务)
-        try:
+    sem = asyncio.Semaphore(3)
-            # 预测 (build_context 内部已用 before=match_date 防泄漏,
-            # injuries_slice 也使用 as_of=match_date 过滤 retrieved_at)
-            # 回测必须禁用结果缓存: 否则命中缓存会复用同一 prediction_id,
-            # 导致 settle 反复覆盖同一条记录(见 P1-3)。
-            result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False)
 
-            # 用实际比分 settle
+    async def _one(c: BacktestCandidate) -> BacktestMatchResult | None:
-            await settle_prediction(result.prediction_id, c.home_goals, c.away_goals)
+        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
 
-            actual = _actual_1x2(c.home_goals, c.away_goals)
+    # 并行执行,保持结果顺序
-            correct = result.pred_1x2 == actual
+    results = await asyncio.gather(*[_one(c) for c in candidates])
-
+    for r in results:
-            bt = BacktestMatchResult(
+        if r is not None:
-                match_id=c.match_id,
+            summary.results.append(r)
-                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=correct,
-                prediction_id=result.prediction_id,
-            )
-            summary.results.append(bt)
             summary.scored += 1
 
-        except Exception:
-            # 用 exception 而非 warning:保留堆栈,否则集成层缺陷(如惰性加载
-            # 在 session 外触发)会只剩一行无堆栈的 warning,极难定位。
-            logger.exception("backtest match %s failed", c.match_id)
-
     # 汇总统计
     if summary.scored > 0:
         correct_count = sum(1 for r in summary.results if r.correct_1x2)

src/llm/context_builder.py

@@ -1,16 +1,22 @@
 """上下文构建器:数据切片 + 拼接。
 
 架构:
-  - match_header: 比赛基础信息(对阵双方/联赛/时间)
+    - match_header: 比赛基础信息(对阵双方/联赛/时间)
-  - 切片函数: 每个领域 agent 一个数据切片(h2h / form / standings / injuries / xg)
+    - 切片函数: 每个领域 agent 一个数据切片(h2h / form / standings / injuries / xg)
-  - build_context: 单 agent 路径,拼接全部切片(行为与旧版一致)
+    - build_context: 单 agent 路径,拼接全部切片(行为与旧版一致)
 
 multi-agent 路径由 agents/orchestrator.py 调用切片函数,每个专家只拿自己的切片。
+
+性能说明:
+    build_context 创建一个共享 session 并传给所有切片函数,
+    避免每个切片独立创建 session —— 回测 20 场并发时,
+    5 个切片 × 20 场 = 100 个连接会耗尽连接池(pool_size=15)。
 """
 from __future__ import annotations
 
 import logging
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 from sqlalchemy import select
 from sqlalchemy.orm import selectinload
@@ -18,6 +24,9 @@ from sqlalchemy.orm import selectinload
 from src.db.base import AsyncSessionLocal
 from src.db.models import Match
 
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
 logger = logging.getLogger(__name__)
 
 
@@ -80,11 +89,19 @@ class MatchHeader:
     league_id: int
 
 
-async def load_match_header(match_id: int) -> MatchHeader:
+async def load_match_header(match_id: int, db: AsyncSession | None = None) -> MatchHeader:
-    """加载比赛头信息(各 agent 共用)。"""
+    """加载比赛头信息(各 agent 共用)。
-    async with AsyncSessionLocal() as db:
+
+    Args:
+        match_id: 比赛 ID
+        db: 可选的共享 session。不传则自建(向后兼容)。
+    """
+    if db is not None:
         m = await _load_match(db, match_id)
         return _to_header(m)
+    async with AsyncSessionLocal() as new_db:
+        m = await _load_match(new_db, match_id)
+        return _to_header(m)
 
 
 def _to_header(m: Match) -> MatchHeader:
@@ -114,10 +131,16 @@ def header_text(h: MatchHeader) -> str:
 # 切片函数: 每个领域 agent 一个
 # ============================================================
 
-async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None) -> SliceResult:
+async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。"""
+    """E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。
-    async with AsyncSessionLocal() as db:
+
+    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
+    """
+    if db is not None:
         h2h = await _get_h2h(db, header.home_team_id, header.away_team_id, before=before, limit=limit)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            h2h = await _get_h2h(new_db, header.home_team_id, header.away_team_id, before=before, limit=limit)
     lines = [f"── 历史交锋(近 {limit} 次) ──"]
     n_with_score = 0
     if h2h:
@@ -141,11 +164,18 @@ async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None) -> Slic
     return SliceResult(text="\n".join(lines), has_data=n_with_score > 0, n_records=n_with_score)
 
 
-async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> SliceResult:
+async def form_slice(header: MatchHeader, *, limit: int = 5, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。"""
+    """A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。
-    async with AsyncSessionLocal() as db:
+
+    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
+    """
+    if db is not None:
         home_form = await _get_form(db, header.home_team_id, before=before, limit=limit)
         away_form = await _get_form(db, header.away_team_id, before=before, limit=limit)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
+            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
     lines = []
     n_scored = 0
     for label, name, form, side in (
@@ -175,11 +205,18 @@ async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> Sli
     return SliceResult(text="\n".join(lines), has_data=n_scored > 0, n_records=n_scored)
 
 
-async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> SliceResult:
+async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。"""
+    """B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。
-    async with AsyncSessionLocal() as db:
+
+    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
+    """
+    if db is not None:
         home_form = await _get_form(db, header.home_team_id, before=before, limit=limit)
         away_form = await _get_form(db, header.away_team_id, before=before, limit=limit)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
+            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
     lines = [f"── 攻防数据(近 {limit} 场) ──"]
     n_total = 0
     for label, name, form, side in (
@@ -221,11 +258,18 @@ async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> S
     return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
 
 
-async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None) -> SliceResult:
+async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。"""
+    """C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。
-    async with AsyncSessionLocal() as db:
+
+    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
+    """
+    if db is not None:
         home_home = await _get_home_away(db, header.home_team_id, "home", before=before, limit=limit)
         away_away = await _get_home_away(db, header.away_team_id, "away", before=before, limit=limit)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_home = await _get_home_away(new_db, header.home_team_id, "home", before=before, limit=limit)
+            away_away = await _get_home_away(new_db, header.away_team_id, "away", before=before, limit=limit)
     lines = ["── 主客因素 ──"]
     n_total = 0
     for label, name, matches, side in (
@@ -255,17 +299,22 @@ async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None)
     return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
 
 
-async def injuries_slice(header: MatchHeader, *, before=None) -> SliceResult:
+async def injuries_slice(header: MatchHeader, *, before=None, db: AsyncSession | None = None) -> SliceResult:
     """D - 阵容完整性切片: 伤停与停赛名单,评估战力缺失程度。
 
     before=cutoff: 只使用 cutoff 之前已采集的伤停数据,防回测泄漏。
+    db: 可选共享 session(见模块 docstring)。
     """
     from src.data.injuries import get_injuries_for_match
 
     cutoff = before or header.match_dt
-    async with AsyncSessionLocal() as db:
+    if db is not None:
         home_injuries = await get_injuries_for_match(db, header.home_team_id, cutoff, as_of=cutoff)
         away_injuries = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_injuries = await get_injuries_for_match(new_db, header.home_team_id, cutoff, as_of=cutoff)
+            away_injuries = await get_injuries_for_match(new_db, header.away_team_id, cutoff, as_of=cutoff)
 
     lines = ["── 阵容完整性 ──"]
     n_records = 0
@@ -291,41 +340,51 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> SliceResult:
 # 单 agent 路径: 拼接全部切片(行为与旧版一致)
 # ============================================================
 
-async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5) -> MatchContext:
+async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5, backtest: bool = False) -> MatchContext:
     """单 agent 路径的完整上下文: 拼接全部切片(before=比赛时间,防未来信息)。
 
     has_stats / has_injuries 直接取切片显式声明的 has_data,
     不再靠文案子串匹配(见审查报告 P2-1)。
+
+    P2-6: backtest=True 时 cutoff = match_date - 1天,确保只用赛前数据。
+
+    P1-1: 使用单个共享 session 贯穿所有切片查询,避免连接池耗尽。
     """
-    header = await load_match_header(match_id)
+    async with AsyncSessionLocal() as db:
-    parts = [header_text(header), ""]
+        header = await load_match_header(match_id, db=db)
+        # P2-6: 回测模式下 cutoff 提前 1 天,防止比赛日数据泄漏
+        cutoff = header.match_dt
+        if backtest and header.match_dt:
+            from datetime import timedelta
+            cutoff = header.match_dt - timedelta(days=1)
+        parts = [header_text(header), ""]
 
-    form_res = await form_slice(header, limit=form_last, before=header.match_dt)
+        form_res = await form_slice(header, limit=form_last, before=cutoff, db=db)
-    parts.append(form_res.text)
+        parts.append(form_res.text)
-    parts.append("")
+        parts.append("")
 
-    h2h_res = await h2h_slice(header, limit=h2h_last, before=header.match_dt)
+        h2h_res = await h2h_slice(header, limit=h2h_last, before=cutoff, db=db)
-    parts.append(h2h_res.text)
+        parts.append(h2h_res.text)
-    parts.append("")
+        parts.append("")
 
-    stats_res = await stats_slice(header, before=header.match_dt)
+        stats_res = await stats_slice(header, before=cutoff, db=db)
-    parts.append(stats_res.text)
+        parts.append(stats_res.text)
-    parts.append("")
+        parts.append("")
 
-    home_away_res = await home_away_slice(header, before=header.match_dt)
+        home_away_res = await home_away_slice(header, before=cutoff, db=db)
-    parts.append(home_away_res.text)
+        parts.append(home_away_res.text)
-    parts.append("")
+        parts.append("")
 
-    injuries_res = await injuries_slice(header, before=header.match_dt)
+        injuries_res = await injuries_slice(header, before=cutoff, db=db)
-    parts.append(injuries_res.text)
+        parts.append(injuries_res.text)
 
-    return MatchContext(
+        return MatchContext(
-        match_id=match_id,
+            match_id=match_id,
-        text="\n".join(parts),
+            text="\n".join(parts),
-        has_stats=form_res.has_data or stats_res.has_data,
+            has_stats=form_res.has_data or stats_res.has_data,
-        has_injuries=injuries_res.has_data,
+            has_injuries=injuries_res.has_data,
-        match_dt=header.match_dt,
+            match_dt=header.match_dt,
-    )
+        )
 
 
 # ============================================================

src/llm/predict.py

@@ -23,8 +23,9 @@ _PROMPT_DIR = Path(__file__).resolve().parent / "prompts"
 
 # ── LLM 响应缓存(match+provider+model+version → 结果) ──
 _CACHE_TTL_SEC = 300  # 5 分钟
+# P1-5: 缓存仅在 asyncio 协程内同步访问(dict 操作 GIL 原子),无需 threading.Lock。
+# 删除 _cache_lock,避免同步锁阻塞事件循环;dict 的 get/set 在 CPython 下原子。
 _cache: dict[str, tuple[float, PredictResult]] = {}
-_cache_lock = Lock()
 
 
 def _cache_key(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> str:
@@ -38,20 +39,21 @@ def _cache_key(match_id: int, provider: str, model: str, version: str, tpl_hash:
 
 
 def _get_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> PredictResult | None:
+    # P1-5: 无锁访问。dict get/del 在 CPython GIL 下原子,且无 await 穿插。
     key = _cache_key(match_id, provider, model, version, tpl_hash)
-    with _cache_lock:
+    entry = _cache.get(key)
-        if key in _cache:
+    if entry is not None:
-            ts, result = _cache[key]
+        ts, result = entry
-            if time.time() - ts < _CACHE_TTL_SEC:
+        if time.time() - ts < _CACHE_TTL_SEC:
-                return result
+            return result
-            del _cache[key]
+        _cache.pop(key, None)
     return None
 
 
 def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str, result: PredictResult) -> None:
+    # P1-5: 无锁写入。同上,dict set 原子。
     key = _cache_key(match_id, provider, model, version, tpl_hash)
-    with _cache_lock:
+    _cache[key] = (time.time(), result)
-        _cache[key] = (time.time(), result)
 
 
 def clear_prompt_cache() -> None:
@@ -103,6 +105,7 @@ async def predict_match(
     prompt_version: str | None = None,
     mode: str = "multi",
     use_cache: bool = True,
+    backtest: bool = False,
 ) -> "PredictResult | MultiPredictResult":
     """预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。
 
@@ -110,6 +113,7 @@ async def predict_match(
         use_cache: 是否允许返回进程内缓存结果。回测必须传 False——
             缓存命中不会新建 prediction 行,调用方会对同一个 prediction_id
             反复 settle,把不同比赛的真实比分覆盖到同一条记录上。
+        backtest: 是否回测模式。True 时 build_context 使用 match_date-1天 作为 cutoff。
     """
     if mode == "single":
         return await _predict_single(
@@ -118,6 +122,7 @@ async def predict_match(
             model=model,
             prompt_version=prompt_version,
             use_cache=use_cache,
+            backtest=backtest,
         )
     from src.llm.agents.orchestrator import predict_match_multi
 
@@ -131,6 +136,7 @@ async def _predict_single(
     model: str | None = None,
     prompt_version: str | None = None,
     use_cache: bool = True,
+    backtest: bool = False,
 ) -> PredictResult:
     """单次调用路径(原有实现)。"""
     if provider is None:
@@ -147,8 +153,8 @@ async def _predict_single(
             logger.debug("predict cache hit match=%s", match_id)
             return cached
 
-    # 1. 拼上下文
+    # 1. 拼上下文(P2-6: backtest 时使用 match_date-1天 作为 cutoff)
-    ctx = await build_context(match_id)
+    ctx = await build_context(match_id, backtest=backtest)
 
     # 1.5 计算快照元数据(用于可复现性)
     now = datetime.now(timezone.utc)

src/llm/utils.py

@@ -0,0 +1,19 @@
+"""LLM 模块共享工具函数。"""
+from __future__ import annotations
+
+
+def actual_1x2(home: int, away: int) -> str:
+    """实际比分 → 胜平负。
+
+    单一权威源: backtest.py 和 eval.py 共用,避免重复定义。
+    """
+    if home > away:
+        return "1"
+    if home == away:
+        return "X"
+    return "2"
+
+
+def is_correct_1x2(pred: str | None, actual: str) -> bool:
+    """预测是否命中胜平负。"""
+    return pred == actual

src/llm/validation.py

@@ -10,8 +10,10 @@ from pydantic import BaseModel, Field, field_validator, model_validator
 
 logger = logging.getLogger(__name__)
 
-# 已知的 5 个专家 agent 名(与 orchestrator.SPECIALIST_SPECS 保持一致)
+# 单一权威源:从 orchestrator.SPECIALIST_SPECS 派生,避免两端独立定义导致静默偏离
-KNOWN_AGENT_NAMES: tuple[str, ...] = ("form", "stats", "home_away", "injuries", "h2h")
+from src.llm.agents.orchestrator import SPECIALIST_SPECS
+
+KNOWN_AGENT_NAMES: tuple[str, ...] = tuple(spec.name for spec in SPECIALIST_SPECS)
 
 
 class AgentReportSchema(BaseModel):
@@ -74,7 +76,7 @@ class PredictionOutputSchema(BaseModel):
         """
         expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals)
         if self.pred_1x2 != expected:
-            logger.warning(
+            logger.debug(
                 "1x2 与比分不一致: 比分 %.1f-%.1f 推出 '%s',但 LLM 给出 '%s';以比分修正",
                 self.pred_home_goals, self.pred_away_goals, expected, self.pred_1x2,
             )