fix: 全方位审查问题修复（P0~P3）

P0: ORM-迁移同步
- 新增 RawEvent/IngestFailure/DataQualityCheck/DataLineage 4 个模型类
- MatchStats 补充 xg_source/xg_updated_at/xg_source_record_id 字段
- 修复 server_default=_utcnow → func.now()（4 处）
- BigInteger 导入

P1:
- log_buffer.py 移除 threading.Lock（asyncio 单线程下无需锁）
- 确认 context_builder/understat/injuries 等已有修复

P2:
- Dockerfile 新增非 root 用户 + .dockerignore
- 前端 fetchDashboard 修复 total 字段（改用 items.length）
- 前端 fetchSystemConfig 改用真实 /admin/settings 端点
- 修复 validation.py return self"" → return self

P3:
- predict.py 缓存加 _CACHE_MAX_SIZE=200 淘汰
- eval.py get_eval_summary 加 limit 参数（默认 1000）+ 返回 total_settled
- orchestrator.py agent provider 配置缓存 60s

src/api/routes/eval.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import logging
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 
 from src.api.deps import require_admin
 from src.api.schemas import EvalSummaryOut, SettleRequest
@@ -30,6 +30,6 @@ async def settle(req: SettleRequest, db: AsyncSession = Depends(get_db)):
 
 
 @router.get("/eval/summary", response_model=EvalSummaryOut, dependencies=[Depends(require_admin)])
-async def eval_summary():
-    """提供商/模型准确率对比。"""
-    return await get_eval_summary()
+async def eval_summary(limit: int = Query(1000, ge=1, le=10000, description="最大评估条数")):
+    """提供商/模型准确率对比。P3-4: 默认评估最近 1000 条,可通过 limit 调整。"""
+    return await get_eval_summary(limit=limit)

src/core/log_buffer.py

@@ -6,11 +6,9 @@
 from __future__ import annotations
 
 import logging
-import threading
 from collections import deque
 
 _BUFFER: deque[dict] = deque(maxlen=2000)
-_LOCK = threading.Lock()
 
 _LEVEL_ORDER = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50}
 
@@ -31,8 +29,7 @@ class MemoryLogHandler(logging.Handler):
                 "logger": record.name,
                 "message": self.format(record),
             }
-            with _LOCK:
-                _BUFFER.append(entry)
+            _BUFFER.append(entry)
         except Exception:  # noqa: BLE001  日志采集绝不影响业务
             self.handleError(record)
 
@@ -54,8 +51,7 @@ def get_entries(
     """按条件查询缓冲日志,最新在前。"""
     min_no = _LEVEL_ORDER.get((min_level or "").upper(), 0)
     kw = (keyword or "").strip().lower()
-    with _LOCK:
-        items = list(_BUFFER)
+    items = list(_BUFFER)
     items.reverse()
     out: list[dict] = []
     for e in items:

src/db/models.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 from datetime import date, datetime, timezone
 
 from sqlalchemy import (
+    BigInteger,
     Boolean,
     CheckConstraint,
     Date,
@@ -131,6 +132,10 @@ 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 数据血缘:单独追踪 xG 字段的来源与更新时间(xG 可能独立于其他统计被更新)
+    xg_source: Mapped[str | None] = mapped_column(String(30))
+    xg_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    xg_source_record_id: Mapped[str | None] = mapped_column(String(100))
 
     match: Mapped[Match] = relationship(back_populates="stats")
 
@@ -228,3 +233,92 @@ class AppSetting(Base):
     key: Mapped[str] = mapped_column(String(100), primary_key=True)
     value: Mapped[str] = mapped_column(Text, nullable=False)
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
+
+
+# ── 数据管线基础设施(对应迁移 0008) ──────────────────────────────────
+# Bronze 层、死信、质量监控、血缘追踪 4 张表。
+
+
+class RawEvent(Base):
+    """Bronze 层:采集到的原始事件存档,便于重放与审计。"""
+    __tablename__ = "raw_events"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(50), nullable=False)
+    source_record_id: Mapped[str] = mapped_column(String(100), nullable=False)
+    raw_payload: Mapped[dict] = mapped_column(JSONB, nullable=False)
+    ingested_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    ingest_batch_id: Mapped[str | None] = mapped_column(String(36))
+
+    __table_args__ = (
+        UniqueConstraint("source_system", "source_record_id", name="uq_raw_event"),
+        Index("ix_raw_event_batch", "ingest_batch_id"),
+    )
+
+
+class IngestFailure(Base):
+    """采集失败死信:记录失败原因、重试次数与下次重试时间。"""
+    __tablename__ = "ingest_failures"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(50), nullable=False)
+    entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    source_record_id: Mapped[str | None] = mapped_column(String(100))
+    error_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    error_detail: Mapped[str | None] = mapped_column(Text)
+    raw_payload: Mapped[dict | None] = mapped_column(JSONB)
+    retry_count: Mapped[int] = mapped_column(Integer, server_default="0")
+    next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    status: Mapped[str] = mapped_column(String(20), server_default="pending")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    __table_args__ = (
+        Index("ix_ingest_failure_status", "status", "next_retry_at"),
+        CheckConstraint(
+            "status IN ('pending', 'retrying', 'resolved', 'abandoned')",
+            name="ck_ingest_failure_status",
+        ),
+    )
+
+
+class DataQualityCheck(Base):
+    """数据质量监控:记录每次质量检查的结果。"""
+    __tablename__ = "data_quality_checks"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    check_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    entity_id: Mapped[int | None] = mapped_column(Integer)
+    expected_value: Mapped[float | None] = mapped_column(Float)
+    actual_value: Mapped[float] = mapped_column(Float, nullable=False)
+    passed: Mapped[bool] = mapped_column(Boolean, nullable=False)
+    severity: Mapped[str] = mapped_column(String(10), server_default="warning")
+    detail: Mapped[dict | None] = mapped_column(JSONB)
+    checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+    __table_args__ = (
+        Index("ix_dqc_checked_at", "checked_at"),
+        Index("ix_dqc_entity", "entity_type", "entity_id"),
+    )
+
+
+class DataLineage(Base):
+    """ETL 血缘追踪:记录从源到目标的转换过程。"""
+    __tablename__ = "data_lineage"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    source_system: Mapped[str] = mapped_column(String(50), nullable=False)
+    source_record_id: Mapped[str] = mapped_column(String(100), nullable=False)
+    target_table: Mapped[str] = mapped_column(String(50), nullable=False)
+    target_id: Mapped[int | None] = mapped_column(Integer)
+    transform_name: Mapped[str] = mapped_column(String(50), nullable=False)
+    transform_detail: Mapped[dict | None] = mapped_column(JSONB)
+    batch_id: Mapped[str | None] = mapped_column(String(36))
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+
+    __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/agents/orchestrator.py

@@ -30,6 +30,11 @@ from src.llm.provider import LLMProvider, get_default_provider
 
 logger = logging.getLogger(__name__)
 
+# P3-2: agent provider 配置缓存(TTL 60s),避免每次 _agent_provider 都多次查 DB
+_AGENT_PROVIDER_CACHE: dict[str, tuple[float, LLMProvider]] = {}
+_AGENT_PROVIDER_CACHE_TTL = 60.0
+
+
 # ── 5 个专家 agent 定义 ──
 # A=近期状态 B=攻防数据 C=主客因素 D=阵容完整性 E=历史交锋
 SPECIALIST_SPECS: list[AgentSpec] = [
@@ -103,7 +108,16 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider:
     覆盖优先级:
       模型: AGENT_MODEL_{ID}(运行时) → 层级默认(LLM_SPECIALIST/AGGREGATOR_MODEL) → 全局 LLM_MODEL
       地址/密钥: AGENT_BASE_URL_{ID} / AGENT_API_KEY_{ID}(运行时) → 全局 LLM_BASE_URL / LLM_API_KEY
+
+    P3-2: 结果缓存 60 秒,避免每次预测都多次查询运行时配置 DB。
     """
+    cache_key = f"{agent_id}:{tier}"
+    cached = _AGENT_PROVIDER_CACHE.get(cache_key)
+    if cached is not None:
+        ts, provider = cached
+        if time.time() - ts < _AGENT_PROVIDER_CACHE_TTL:
+            return provider
+
     pfx = f"AGENT_{agent_id.upper()}_"
     p = await get_default_provider()
     tier_model = settings.LLM_SPECIALIST_MODEL if tier == "specialist" else settings.LLM_AGGREGATOR_MODEL
@@ -118,6 +132,11 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider:
     key = await get_runtime_value(f"{pfx}API_KEY")
     if key:
         p.api_key = key
+
+    _AGENT_PROVIDER_CACHE[cache_key] = (time.time(), p)
+    # 简单淘汰:超过 20 条时清空(60s TTL 下不会累积太多)
+    if len(_AGENT_PROVIDER_CACHE) > 20:
+        _AGENT_PROVIDER_CACHE.clear()
     return p
 
 

src/llm/eval.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import logging
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from src.db.models import Prediction
 from src.db.unit_of_work import get_uow
@@ -32,12 +32,22 @@ def _actual_1x2(home: int, away: int) -> str:
     return "2"
 
 
-async def get_eval_summary() -> dict:
-    """按 provider × 模型聚合评估。"""
+async def get_eval_summary(limit: int = 1000) -> dict:
+    """按 provider × 模型聚合评估。
+
+    P3-4: 默认限制评估最近 1000 条已结算预测,避免全表加载导致内存压力。
+    可通过 eval 路由的 query 参数调整。
+    """
     async with get_uow() as session:
+        # 先统计全量已结算数,用于前端展示"共 X 条,评估 Y 条"
+        total_settled = (await session.execute(
+            select(func.count()).where(Prediction.settled == True)
+        )).scalar_one()
         stmt = (
             select(Prediction)
             .where(Prediction.settled == True)
+            .order_by(Prediction.id.desc())
+            .limit(limit)
         )
         result = await session.execute(stmt)
         rows = list(result.scalars().all())
@@ -76,4 +86,4 @@ async def get_eval_summary() -> dict:
                 "avg_score_rmse": round(avg_err, 2) if avg_err is not None else None,
                 "avg_subjective_confidence": round(avg_conf, 2) if avg_conf is not None else None,
             })
-        return {"summary": summary}
+        return {"summary": summary, "total_settled": total_settled, "evaluated": len(rows)}

src/llm/predict.py

@@ -8,7 +8,6 @@ import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from threading import Lock
 
 from src.core.config import settings
 from sqlalchemy import select
@@ -25,6 +24,7 @@ _PROMPT_DIR = Path(__file__).resolve().parent / "prompts"
 
 # ── LLM 响应缓存(match+provider+model+version → 结果) ──
 _CACHE_TTL_SEC = 300  # 5 分钟
+_CACHE_MAX_SIZE = 200  # P3-1: 有上限,避免长期运行内存无限增长
 # P1-5: 缓存仅在 asyncio 协程内同步访问(dict 操作 GIL 原子),无需 threading.Lock。
 # 删除 _cache_lock,避免同步锁阻塞事件循环;dict 的 get/set 在 CPython 下原子。
 _cache: dict[str, tuple[float, PredictResult]] = {}
@@ -56,6 +56,10 @@ def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash
     # P1-5: 无锁写入。同上,dict set 原子。
     key = _cache_key(match_id, provider, model, version, tpl_hash)
     _cache[key] = (time.time(), result)
+    # P3-1: 超过上限时淘汰最旧条目(按时间戳排序)
+    if len(_cache) > _CACHE_MAX_SIZE:
+        oldest_key = min(_cache, key=lambda k: _cache[k][0])
+        _cache.pop(oldest_key, None)
 
 
 def clear_prompt_cache() -> None:

src/llm/validation.py

@@ -75,7 +75,7 @@ class PredictionOutputSchema(BaseModel):
         ):
             self.alt_pred_home_goals = None
             self.alt_pred_away_goals = None
-        return self""
+        return self
 
     @field_validator("pred_1x2")
     @classmethod