feat: 数据管线架构优化 — Bronze 层 + 死信队列 + 令牌桶 + 血缘追踪

P0: injuries 缓存原子写(tempfile+os.replace) + TTL 分级 + asyncio.Lock 并发安全
P1: 引入 Bronze 层 RawEvent 表(原始事件存档,不可变)
P1: IngestFailure 死信表(失败持久化,支持重试恢复)
P1: understat 改为 xG 覆盖更新模式 + xG 来源追踪字段
P1: 令牌桶限流器(TokenBucket)替换固定 sleep
P1: DataQualityCheck 数据质量监控表
P1: DataLineage 血缘追踪表
P2: 回测并发控制 Semaphore(3)

新增文件:
  - src/data/rate_limiter.py (令牌桶限流器)
  - alembic/versions/0008 (4 张新表迁移)
  - alembic/versions/0009 (MatchStats xG 字段迁移)

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")

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,6 +15,7 @@ import asyncio
 import json as _json
 import logging
 import random
+import uuid
 from collections.abc import Iterable
 from datetime import datetime, timezone
 
@@ -18,11 +25,15 @@ 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 TokenBucket
 from src.data.sources import register
-from src.db.models import League, Match, MatchStats, Team
+from src.db.models import IngestFailure, League, Match, MatchStats, RawEvent, Team
 
 logger = logging.getLogger(__name__)
 
+# 令牌桶限流器:替代固定 sleep,rate 根据 REQUEST_INTERVAL 计算
+_bzzoiro_limiter = TokenBucket(rate=1.0 / REQUEST_INTERVAL, capacity=3)
+
 
 def _to_date(value):
     """把 datetime / date / str 统一成 `date`。"""
@@ -60,6 +71,8 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
     last_exc: Exception | None = None
     for attempt in range(max_retries):
         try:
+            # 令牌桶限流:在发起请求前获取令牌
+            await _bzzoiro_limiter.acquire()
             client = get_client()
             resp = await client.get(url, headers=headers, params=params, timeout=30)
             resp.raise_for_status()
@@ -124,10 +137,58 @@ 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 表)。"""
+    for raw in raw_events:
+        try:
+            source_record_id = str(raw.get("id", ""))
+            if not source_record_id:
+                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)
+
+
+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",
+        )
+        db.add(failure)
+        await db.flush()
+    except Exception:
+        logger.warning("bzzoiro ingest_failure write failed", exc_info=True)
+
+
 @register
 class BzzoiroSource:
     """bzzoiro 数据源(实现 DataSource 协议)。"""
@@ -148,6 +209,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": []}
@@ -156,9 +218,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()
@@ -185,6 +255,11 @@ class BzzoiroSource:
                         except Exception as e:
                             # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors)
                             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)
@@ -273,7 +348,7 @@ 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,
                         )
@@ -298,8 +373,21 @@ 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,
                         )

src/data/injuries.py

@@ -1,16 +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
 
@@ -27,6 +36,60 @@ DEFAULT_HOST = "v3.football.api-sports.io"
 # P2-3: 缓存目录改用系统临时目录,避免源码树内写入
 _CACHE_DIR = Path(tempfile.gettempdir()) / "profeto_injuries"
 
+# 模块级锁:防止并发写同一缓存文件
+_cache_lock = asyncio.Lock()
+
+
+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
+
 
 async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = None, league_id: int | None = None) -> list[dict]:
     """采集伤停数据。
@@ -46,17 +109,18 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No
     cache_dir = _CACHE_DIR
     cache_dir.mkdir(parents=True, exist_ok=True)
 
-    # 缓存命中 (7 天内有效)
+    # TTL 分级缓存
+    ttl_hours = _compute_ttl_hours(date, fixture_id, league_id)
     cache_key = f"injuries_{date}_{fixture_id}_{league_id}.json"
     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,
@@ -93,9 +157,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
 
@@ -106,25 +170,63 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
     注意: 本方法不控制事务(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 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",
+            )
+            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}
@@ -132,6 +234,7 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
     # P1-4: 收集所有待插入记录的键,批量查询已存在的记录
     # 避免逐条查询 + 插入的竞态条件(两个并发请求同时通过检查 → IntegrityError)
     pending_records: list[dict] = []
+    parse_failures: list[dict] = []  # 规范化失败的原始数据,用于写死信表
     for raw in raw_injuries:
         try:
             player = raw.get("player", {}) or {}
@@ -176,6 +279,30 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict:
             })
         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",
+            )
+            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()

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/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,6 +16,7 @@ import json
 import logging
 import random
 import re
+import uuid
 from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select
@@ -19,7 +26,7 @@ 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 IngestFailure, League, Match, MatchStats, RawEvent, Team
 
 logger = logging.getLogger(__name__)
 
@@ -89,6 +96,54 @@ def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, i
     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 表)。"""
+    for raw in raw_matches:
+        try:
+            source_record_id = str(raw.get("id", ""))
+            if not source_record_id:
+                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)
+
+
+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",
+        )
+        db.add(failure)
+        await db.flush()
+    except Exception:
+        logger.warning("understat ingest_failure write failed", exc_info=True)
+
+
 @register
 class UnderstatSource:
     """understat xG 数据源(实现 DataSource 协议)。"""
@@ -96,23 +151,33 @@ 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, 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)
@@ -136,6 +201,11 @@ 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)
@@ -167,7 +237,7 @@ class UnderstatSource:
                 key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date)
                 match_dict[key] = m
 
-        # === 内存匹配 + 回填 xG ===
+        # === 内存匹配 + 回填 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)
@@ -181,24 +251,37 @@ class UnderstatSource:
                 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()
+
+            # xG 覆盖更新模式:当 understat 数据更新时覆盖旧值
             if existing.stats is not None:
-                if existing.stats.home_xg is None and nm.home_xg is not None:
+                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:
+                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
 
         # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务
         return result

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
@@ -131,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"),
     )
 
 
@@ -216,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/llm/backtest.py

@@ -150,8 +150,8 @@ async def run_backtest(
 
     summary = BacktestSummary(total=len(candidates), scored=0)
 
-    # P1-6: 并发控制,同时最多 8 场预测(避免 LLM API 限流)
+    # P2: 并发控制,同时最多 3 场预测(避免 LLM API 限流,保护下游服务)
-    sem = asyncio.Semaphore(8)
+    sem = asyncio.Semaphore(3)
 
     async def _one(c: BacktestCandidate) -> BacktestMatchResult | None:
         async with sem: