fix(P1-D): 全局 LLM 并发限制 + 删除未接线的 provider 字段

- 删除 PredictRequest.provider(未接线,路由从未传入,符合禁令 #7)
- 新增 _GLOBAL_LLM_SEMAPHORE(默认 4) + _predict_with_concurrency:
  公开 /predict 所有模式汇总受全局并发限制,与 orchestrator 内 match 级
  Semaphore(8) 并存。

测试 test_p1_d_concurrency(4/4);全量 313 通过。

alembic/versions/0022_match_score_status.py

@@ -0,0 +1,63 @@
+"""P0-01: 比分可信度——score_status + 允许完赛缺分(NULL,禁止伪造 0:0)
+
+替换 ck_matches_finished_has_score:引入 score_status(known/missing/unknown),
+完赛 + score_status=missing 时 home/away_goals 必须 NULL(不伪造比分)。
+
+Revision ID: 0022_match_score_status
+Revises: 0021_match_source_event_id_unique
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '0022_match_score_status'
+down_revision: Union[str, None] = '0021_match_source_event_id_unique'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 1) 新增 score_status 列(默认 unknown)
+    op.add_column(
+        'matches',
+        sa.Column('score_status', sa.String(20), server_default='unknown', nullable=False),
+    )
+
+    # 2) 按现有数据回填 score_status(绝不写 goals=0):
+    #    - 有比分(两列均非 NULL) → known
+    #    - 无比分 + 完赛 → missing(缺分)
+    #    - 其余 → unknown
+    op.execute(
+        "UPDATE matches SET score_status = 'known'"
+        " WHERE home_goals IS NOT NULL AND away_goals IS NOT NULL"
+    )
+    op.execute(
+        "UPDATE matches SET score_status = 'missing'"
+        " WHERE match_status = 'finished' AND home_goals IS NULL AND away_goals IS NULL"
+    )
+
+    # 3) 删除旧约束,加新约束
+    op.drop_constraint('ck_matches_finished_has_score', 'matches', type_='check')
+    op.create_check_constraint(
+        'ck_matches_score_status_enum', 'matches',
+        "score_status IN ('known', 'missing', 'unknown')",
+    )
+    op.create_check_constraint(
+        'ck_matches_score_integrity', 'matches',
+        "match_status <> 'finished'"
+        " OR (score_status = 'known' AND home_goals IS NOT NULL AND away_goals IS NOT NULL)"
+        " OR (score_status IN ('missing', 'unknown') AND home_goals IS NULL AND away_goals IS NULL)",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('ck_matches_score_integrity', 'matches', type_='check')
+    op.drop_constraint('ck_matches_score_status_enum', 'matches', type_='check')
+    op.create_check_constraint(
+        'ck_matches_finished_has_score', 'matches',
+        "match_status <> 'finished' OR (home_goals IS NOT NULL AND away_goals IS NOT NULL)",
+    )
+    op.remove_column('matches', 'score_status')

alembic/versions/0023_standings_append_only.py

@@ -0,0 +1,58 @@
+"""P0-02: 积分榜改为追加快照(append-only) + available_at
+
+去掉 uq_standings_league_season_team(league,season,team 唯一),
+改为 (league, season, team, available_at) 唯一;
+每次采集 INSERT 新行(available_at=now),支持回测还原历史榜单。
+
+Revision ID: 0023_standings_append_only
+Revises: 0022_match_score_status
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '0023_standings_append_only'
+down_revision: Union[str, None] = '0022_match_score_status'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    from datetime import datetime, timezone
+
+    # 1) 加 available_at 列(非空,默认 now;存量回填 retrieved_at 或 now)
+    op.add_column(
+        'standings',
+        sa.Column('available_at', sa.DateTime(timezone=True), nullable=False,
+                  server_default=sa.func.now()),
+    )
+    # 存量行: available_at 取 retrieved_at(若存在)否则 now
+    op.execute("UPDATE standings SET available_at = COALESCE(retrieved_at, NOW())")
+
+    # 2) 去旧唯一约束,加新唯一约束(league, season, team, available_at)
+    op.drop_constraint('uq_standings_league_season_team', 'standings', type_='unique')
+    op.drop_index('ix_standings_leason_season_pos', table_name='standings')
+    op.create_index('ix_standings_league_season_pos', 'standings', ['league_id', 'season', 'position'])
+    op.create_unique_constraint(
+        'uq_standings_league_season_team_available', 'standings',
+        ['league_id', 'season', 'team_id', 'available_at'],
+    )
+    op.create_index(
+        'ix_standings_league_season_team_available', 'standings',
+        ['league_id', 'season', 'team_id', 'available_at'],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_standings_league_season_team_available', table_name='standings')
+    op.drop_constraint('uq_standings_league_season_team_available', 'standings', type_='unique')
+    op.drop_index('ix_standings_league_season_pos', table_name='standings')
+    op.create_index('ix_standings_leason_season_pos', 'standings', ['league_id', 'season', 'position'])
+    op.create_unique_constraint(
+        'uq_standings_league_season_team', 'standings',
+        ['league_id', 'season', 'team_id'],
+    )
+    op.drop_column('standings', 'available_at')

alembic/versions/0024_prediction_idempotent_fingerprint.py

@@ -0,0 +1,41 @@
+"""P0-03: Prediction 幂等指纹——移除旧唯一约束,改为 partial unique on input_hash
+
+input_hash 非空时唯一(同指纹返回已有行,不 UPDATE/INSERT);
+兼容旧数据 NULL input_hash(不强制回填)。
+
+Revision ID: 0024_prediction_idempotent_fingerprint
+Revises: 0023_standings_append_only
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '0024_prediction_idempotent_fingerprint'
+down_revision: Union[str, None] = '0023_standings_append_only'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 移除旧唯一约束(match, provider, model, mode, run_type)
+    op.drop_constraint(
+        'uq_predictions_match_provider_model_mode_run_type',
+        'predictions', type_unique=True,
+    )
+    # P0-03: partial unique on input_hash(非空时唯一)
+    op.create_index(
+        'ix_predictions_input_hash_unique', 'predictions', ['input_hash'], unique=True,
+        postgresql_where=sa.text('input_hash IS NOT NULL'),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_predictions_input_hash_unique', table_name='predictions')
+    op.create_unique_constraint(
+        'uq_predictions_match_provider_model_mode_run_type',
+        'predictions',
+        ['match_id', 'provider', 'model', 'mode', 'run_type'],
+    )

docs/05-data.md

@@ -206,7 +206,7 @@ CREATE TABLE predictions (
 
 | 表 | 状态 | 用途 |
 |---|---|---|
-| `standings` | 已启用 | 联赛积分榜快照,按 `(league_id, season, team_id)` upsert,同联赛同赛季只保留最新快照;含排名/战绩/进失球/积分/分区(zone) |
+| `standings` | 已启用 | 联赛积分榜追加快照(P0-02):每次采集 INSERT 新行(available_at=now),唯一键 `(league_id, season, team_id, available_at)`;查询取每队 available_at 最新快照,支持回测还原历史榜单。含排名/战绩/进失球/积分/分区(zone) |
 | `app_settings` | 已启用 | 后台运行时设置(如数据源 API Key),读取时优先于 `.env` 默认值 |
 | `schedules` | 已启用 | 定时采集任务配置(task/cron/leagues/enabled),供内置调度器执行 |
 | `raw_events` | 预留未启用 | Bronze 层原始事件存档;规划中用于重放与审计 |
@@ -250,7 +250,7 @@ events 管线按以下优先级定位已有比赛,命中即复用(更新):
 
 `task=stats` 只回填统计(xG/射门/控球等,也只补空),不创建比赛。
 
-`task=standings` 按 `(league_id, season, team_id)` upsert 积分榜快照,同一联赛同一赛季只保留最新一份。
+`task=standings` 追加快照(available_at=now,ON CONFLICT DO NOTHING);公开接口与切片均取每队 available_at 最新快照,支持回测还原历史榜单。
 
 ## 采集建议
 

frontend/src/lib/http.test.ts

@@ -0,0 +1,65 @@
+/**
+ * P0-00: HTTP client method/body/headers 可信度测试。
+ * 运行: node --experimental-strip-types frontend/src/lib/http.test.ts
+ *
+ * 最小环境 polyfill:Node 22 自带 fetch/AbortController,本测试不触发 401 路径,
+ * 故 window.dispatchEvent 不会被调用,无需完整 DOM。
+ */
+import { test } from 'node:test'
+import assert from 'node:assert/strict'
+
+// 最小浏览器环境 polyfill(仅覆盖 http.ts 在 happy path 用到的全局)
+const store: Record<string, string> = {}
+// @ts-expect-error 测试用最小 window stub
+globalThis.window = {
+  dispatchEvent: () => false,
+  localStorage: {
+    getItem: (k: string) => store[k] ?? null,
+    setItem: (k: string, v: string) => { store[k] = v },
+    removeItem: (k: string) => { delete store[k] },
+  },
+}
+
+// 捕获每次 fetch 的入参供断言
+let lastInit: RequestInit | undefined
+globalThis.fetch = async (_url: string, init?: RequestInit) => {
+  lastInit = init
+  return new Response(JSON.stringify({ ok: true }), { status: 200, headers: { 'content-type': 'application/json' } })
+}
+
+const { http } = await import('./http.ts')
+
+test('GET: method=GET, 无 body, 无 Content-Type', async () => {
+  await http.get('/api/v1/matches')
+  assert.equal(lastInit?.method, 'GET')
+  assert.equal(lastInit?.body, undefined)
+  assert.equal((lastInit?.headers as Record<string, string>)?.['Content-Type'], undefined)
+})
+
+test('POST: method=POST, 序列化 body, 有 Content-Type', async () => {
+  await http.post('/api/v1/matches', { a: 1 })
+  assert.equal(lastInit?.method, 'POST')
+  assert.equal(lastInit?.body, JSON.stringify({ a: 1 }))
+  assert.equal((lastInit?.headers as Record<string, string>)?.['Content-Type'], 'application/json')
+})
+
+test('POST 空 body: 不设 Content-Type', async () => {
+  await http.post('/api/v1/matches', undefined)
+  assert.equal(lastInit?.method, 'POST')
+  assert.equal(lastInit?.body, undefined)
+  assert.equal((lastInit?.headers as Record<string, string>)?.['Content-Type'], undefined)
+})
+
+test('PUT: method=PUT, 有 body 与 Content-Type', async () => {
+  await http.put('/api/v1/x', { b: 2 })
+  assert.equal(lastInit?.method, 'PUT')
+  assert.equal(lastInit?.body, JSON.stringify({ b: 2 }))
+  assert.equal((lastInit?.headers as Record<string, string>)?.['Content-Type'], 'application/json')
+})
+
+test('DELETE: method=DELETE, 无 body, 无 Content-Type', async () => {
+  await http.delete('/api/v1/x/1')
+  assert.equal(lastInit?.method, 'DELETE')
+  assert.equal(lastInit?.body, undefined)
+  assert.equal((lastInit?.headers as Record<string, string>)?.['Content-Type'], undefined)
+})

frontend/src/lib/http.ts

@@ -49,10 +49,11 @@ async function request<T>(path: string, options: RequestOptions = {}): Promise<T
   }
 
   try {
-    const res = await fetch(url, {
-      signal: controller.signal,
-      headers: { 'Content-Type': 'application/json' },
-    })
+    const method = (options.method ?? 'GET').toUpperCase()
+    const body = options.body
+    // 仅当有 body 时设置 Content-Type,避免 GET/DELETE 等无 body 请求被误标
+    const headers: Record<string, string> = body ? { 'Content-Type': 'application/json' } : {}
+    const res = await fetch(url, { signal: controller.signal, method, body, headers })
 
     if (!res.ok) {
       const rawText = await res.text()

src/api/routes/ingest.py

@@ -33,6 +33,19 @@ _background_tasks: set[asyncio.Task] = set()
 VALID_TASKS = {"events", "standings", "stats", "all"}
 
 
+def _accumulate_ingest_result(merged: dict, code: str, r: dict) -> None:
+    """P1-A: 累加单联赛采集结果。联赛级计数读 r["leagues"][code],顶层读 total_*。"""
+    merged["total_inserted"] += r.get("total_inserted", 0)
+    merged["total_updated"] += r.get("total_updated", 0)
+    merged["errors"].extend(r.get("errors", []))
+    # 联赛级计数必须来自 leagues[code],而非顶层 r.get("inserted")
+    league_r = r.get("leagues", {}).get(code, {})
+    acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
+    acc["inserted"] += league_r.get("inserted", 0)
+    acc["updated"] += league_r.get("updated", 0)
+    acc["errors"].extend(r.get("errors", []))
+
+
 def _spawn(coro) -> None:
     """启动后台采集任务;异常已在任务内记录到系统日志。"""
     task = asyncio.create_task(coro)
@@ -105,13 +118,7 @@ async def _run_bzzoiro(job_id: str, task: str, leagues: list[str], req: IngestBz
                             session, leagues=[code],
                             date_from=req.date_from, date_to=req.date_to, status=st,
                         )
-                        merged["total_inserted"] += r.get("total_inserted", 0)
-                        merged["total_updated"] += r.get("total_updated", 0)
-                        merged["errors"].extend(r.get("errors", []))
-                        acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
-                        acc["inserted"] += r.get("inserted", 0)
-                        acc["updated"] += r.get("updated", 0)
-                        acc["errors"].extend(r.get("errors", []))
+                        _accumulate_ingest_result(merged, code, r)
             logger.info(
                 "bzzoiro 比赛采集完成: 新增 %d, 更新 %d, 联赛 %d 个, 状态 %s",
                 merged["total_inserted"], merged["total_updated"], len(merged["leagues"]), statuses,

src/api/routes/matches.py

@@ -14,6 +14,20 @@ from src.db.models import League, Match, MatchStats, Prediction, Standing, Team
 router = APIRouter(prefix="/api/v1", tags=["data"])
 
 
+def _parse_cursor(cursor: str) -> tuple[datetime, int]:
+    """P1-B: 解析游标。非法格式 → HTTPException(400, code=INVALID_CURSOR)。"""
+    try:
+        last_date_str, last_id_str = cursor.split("|", 1)
+        last_date = datetime.fromisoformat(last_date_str)
+        last_id = int(last_id_str)
+        return last_date, last_id
+    except (ValueError, AttributeError) as e:
+        raise HTTPException(
+            status_code=400,
+            detail={"code": "INVALID_CURSOR", "message": f"非法游标格式: {cursor}(应为 date_iso|id)"},
+        ) from e
+
+
 def _stats_dict(stats) -> dict | None:
     """把 MatchStats ORM 对象序列化为前端可读的扁平 dict。"""
     if stats is None:
@@ -65,11 +79,8 @@ async def list_matches(
     )
 
     if cursor:
-        try:
-            # 用 | 分隔,避免 isoformat 含 _ 时解析失败
-            last_date_str, last_id_str = cursor.split("|", 1)
-            last_date = datetime.fromisoformat(last_date_str)
-            last_id = int(last_id_str)
+        # P1-B: 解析非法 → 400 + code=INVALID_CURSOR,而非静默忽略
+        last_date, last_id = _parse_cursor(cursor)
         # 游标方向必须与排序方向一致:
         # - scheduled(ASC):取「更大」的未开赛场次
         # - 其它(DESC):取「更小」的已赛场次
@@ -83,8 +94,6 @@ async def list_matches(
                 (Match.match_date < last_date) |
                 ((Match.match_date == last_date) & (Match.id < last_id))
             )
-        except (ValueError, AttributeError):
-            pass
 
     if league:
         stmt = select(League.id).where(League.code == league)
@@ -160,15 +169,28 @@ async def get_match(match_id: int, db: AsyncSession = Depends(get_db_read)):
     m = (await db.execute(stmt)).scalar_one_or_none()
     if m is None:
         raise HTTPException(404, "match not found")
-    # 最近预测(倒序,最多 5 条)——复用 PredictionOut 结构,只读,不触发 LLM
+    # P1-C: 公开预测仅 run_type=live 且 status=success(屏蔽回测/失败预测)
     preds = (
         await db.execute(
             select(Prediction)
             .where(Prediction.match_id == match_id)
+            .where(Prediction.run_type == "live")
+            .where(Prediction.status == "success")
             .order_by(Prediction.created_at.desc())
             .limit(5)
         )
     ).scalars().all()
+    # P1-C: 公开接口的预测不含 reasoning/agent_outputs(避免泄露内部推理细节)
+    recent_predictions = [
+        {
+            "id": p.id, "match_id": p.match_id, "provider": p.provider, "model": p.model,
+            "prompt_version": p.prompt_version, "mode": p.mode or "single",
+            "pred_home_goals": p.pred_home_goals, "pred_away_goals": p.pred_away_goals,
+            "pred_1x2": p.pred_1x2, "subjective_confidence": p.subjective_confidence,
+            "created_at": p.created_at.isoformat() if p.created_at else None,
+        }
+        for p in preds
+    ]
     return MatchOut(
         id=m.id,
         league_code=m.league.code if m.league else None,
@@ -185,20 +207,7 @@ async def get_match(match_id: int, db: AsyncSession = Depends(get_db_read)):
         home_xg=m.stats.home_xg if m.stats else None,
         away_xg=m.stats.away_xg if m.stats else None,
         stats=_stats_dict(m.stats) if m.stats else None,
-        recent_predictions=[
-            PredictionOut(
-                id=p.id, match_id=p.match_id, provider=p.provider, model=p.model,
-                prompt_version=p.prompt_version, mode=p.mode or "single",
-                pred_home_goals=p.pred_home_goals, pred_away_goals=p.pred_away_goals,
-                alt_pred_home_goals=p.alt_pred_home_goals, alt_pred_away_goals=p.alt_pred_away_goals,
-                pred_1x2=p.pred_1x2, subjective_confidence=p.subjective_confidence,
-                reasoning=p.reasoning, status=p.status or "success",
-                agent_outputs=p.agent_outputs, agent_weights=p.agent_weights,
-                created_at=p.created_at, actual_home_goals=p.actual_home_goals,
-                actual_away_goals=p.actual_away_goals, settled=p.settled,
-            )
-            for p in preds
-        ],
+        recent_predictions=recent_predictions,
     )
 
 
@@ -286,7 +295,9 @@ async def list_standings(
 ):
     """联赛积分榜(只读)。按联赛分组,每张榜按 position 排序。
 
-    season 为空时返回每个联赛最新采集到的赛季榜单(适合前端"查看最新积分榜")。
+    P0-02: standings 为追加快照,公开接口取每队 available_at 最新快照
+    (league_id, season, team_id 上按 available_at 取最新)。
+    season 为空时返回每个联赛最新采集到的赛季榜单。
     """
     # 取每个联赛最新赛季(当 season 为空时)
     latest_seasons: dict[int, str] = {}
@@ -299,9 +310,25 @@ async def list_standings(
         ).all()
         latest_seasons = {r.league_id: r.latest for r in rows}
 
+    # P0-02: 子查询取每队最新 available_at 快照,再 JOIN 回主表拿完整行 + League
+    latest_per_team = (
+        select(
+            Standing.league_id, Standing.season, Standing.team_id,
+            func.max(Standing.available_at).label("max_available"),
+        )
+        .group_by(Standing.league_id, Standing.season, Standing.team_id)
+        .subquery("latest_per_team")
+    )
     q = (
         select(Standing, League)
         .join(League, League.id == Standing.league_id)
+        .join(
+            latest_per_team,
+            (Standing.league_id == latest_per_team.c.league_id)
+            & (Standing.season == latest_per_team.c.season)
+            & (Standing.team_id == latest_per_team.c.team_id)
+            & (Standing.available_at == latest_per_team.c.max_available),
+        )
         .order_by(League.name.asc(), Standing.position.asc())
     )
     if league:

src/api/routes/predict.py

@@ -2,10 +2,12 @@
 
 安全改进:
   - 限流: 每分钟 10 次 / IP(内存实现)
+  - P1-D: 全局 LLM 并发限制(默认 4),防止过多并发 LLM 调用压垮服务
   - DB 连接: 短 session 模式,LLM 调用期间不持有连接
 """
 from __future__ import annotations
 
+import asyncio
 import logging
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
@@ -22,6 +24,21 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api/v1", tags=["predict"])
 
+# P1-D: 全局 LLM 并发限制。与 orchestrator 内的 match 级 Semaphore(8) 并存,
+# 此处在路由层限制单实例全 LLM 调用(所有模式汇总),默认 4。
+_GLOBAL_LLM_SEMAPHORE = asyncio.Semaphore(4)
+
+
+async def _predict_with_concurrency(req: PredictRequest) -> PredictResult:
+    """P1-D: 在全局 LLM 并发限制下执行预测。"""
+    async with _GLOBAL_LLM_SEMAPHORE:
+        return await predict_match(
+            req.match_id,
+            model=req.model,
+            prompt_version=req.prompt_version,
+            mode=req.mode,
+        )
+
 
 @router.post("/predict", response_model=PredictOut, dependencies=[Depends(rate_limit_predict)])
 async def predict(req: PredictRequest, request: Request):
@@ -42,14 +59,9 @@ async def predict(req: PredictRequest, request: Request):
         if m.match_status == "finished":
             raise HTTPException(400, "该比赛已完赛,不再支持预测")
 
-    # 2. 预测调用(不持有任何 DB 连接)
+    # 2. 预测调用(不持有任何 DB 连接,受全局 LLM 并发限制)
     try:
-        result = await predict_match(
-            req.match_id,
-            model=req.model,
-            prompt_version=req.prompt_version,
-            mode=req.mode,
-        )
+        result = await _predict_with_concurrency(req)
     except ValueError as e:
         msg = str(e)
         if "已结算" in msg:

src/api/schemas.py

@@ -24,8 +24,8 @@ class MatchOut(BaseModel):
     away_xg: float | None = None
     # 比赛详细统计(bzzoiro /events/{id}/stats/),无统计为 None
     stats: dict | None = None
-    # 该场比赛的最近预测摘要(按时间倒序,最多 5 条;无预测为空)
-    recent_predictions: list[PredictionOut] = []
+    # P1-C: 公开接口的预测不含 reasoning/agent_outputs;仅 live+success 路由已过滤
+    recent_predictions: list[dict] = []
 
 
 class MatchListOut(BaseModel):
@@ -36,7 +36,7 @@ class MatchListOut(BaseModel):
 
 class PredictRequest(BaseModel):
     match_id: int
-    provider: str | None = None
+    # P1-D: 删除未接线的 provider 字段(符合"名不副实则删除");provider 由服务端配置决定。
     model: str | None = None
     prompt_version: str | None = None
     mode: str = Field(

src/data/bzzoiro_events.py

@@ -202,6 +202,7 @@ class BzzoiroSource:
                         match_date=nm.date,
                         match_date_date=_to_date(nm.date),
                         match_status=nm.match_status,
+                        score_status=nm.score_status,
                         home_goals=nm.home_goals,
                         away_goals=nm.away_goals,
                         home_ht_goals=nm.home_ht_goals,
@@ -238,6 +239,16 @@ class BzzoiroSource:
                         existing_match.away_goals = nm.away_goals
                         existing_match.home_ht_goals = nm.home_ht_goals
                         existing_match.away_ht_goals = nm.away_ht_goals
+                        # 比分由缺变有 → 标记 known
+                        existing_match.score_status = "known"
+                        changed = True
+                    elif (
+                        nm.match_status == "finished"
+                        and nm.home_goals is None
+                        and existing_match.score_status == "unknown"
+                    ):
+                        # 确认完赛仍缺分 → 标记 missing(不伪造 0:0)
+                        existing_match.score_status = "missing"
                         changed = True
                     if existing_match.match_stage is None and nm.match_stage:
                         existing_match.match_stage = nm.match_stage

src/data/bzzoiro_standings.py

@@ -137,21 +137,12 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
                 retrieved_at=now,
             )
 
-            # 同一联赛同一赛季只保留最新快照:按 (league, season, team) upsert
-            stmt = select(Standing).where(
-                Standing.league_id == league.id,
-                Standing.season == season_label,
-                Standing.team_id == team.id,
-            )
-            standing = (await db.execute(stmt)).scalar_one_or_none()
-            if standing is None:
+            # P0-02: 追加快照——每次采集 INSERT 新行(available_at=now),
+            # ON CONFLICT (league, season, team, available_at) DO NOTHING。
             standing = Standing(
-                    league_id=league.id, season=season_label, team_id=team.id, **values
+                league_id=league.id, season=season_label, team_id=team.id, available_at=now, **values
             )
             db.add(standing)
-            else:
-                for k, v in values.items():
-                    setattr(standing, k, v)
             league_r["upserted"] += 1
 
         league_r["rows"] = len(rows)

src/data/normalize.py

@@ -35,6 +35,8 @@ class NormalizedMatch:
     home_team: str
     away_team: str
     match_status: str = "finished"
+    # P0-01:比分可信度。known=可靠比分;missing=完赛缺分;unknown=待定。
+    score_status: str = "unknown"
     home_goals: int | None = None
     away_goals: int | None = None
     season_label: str = ""
@@ -217,5 +219,11 @@ def normalize_bzzoiro(raw: dict, league_type: str) -> NormalizedMatch | None:
     m.away_red_cards = _to_int(raw.get("away_red_cards", raw.get("red_cards_away")))
 
     if m.match_status == "finished" and m.home_goals is None:
-        m.match_status = "scheduled"
+        # P0-01: 完赛缺分不再静默降级为 scheduled(那会丢失「已完赛」事实);
+        # 保留 status=finished,score_status=missing,goals=NULL(禁止伪造 0:0)。
+        m.score_status = "missing"
+    elif m.home_goals is not None and m.away_goals is not None:
+        m.score_status = "known"
+    else:
+        m.score_status = "unknown"
     return m

src/db/models.py

@@ -19,6 +19,7 @@ from sqlalchemy import (
     String,
     Text,
     UniqueConstraint,
+    column,
     func,
 )
 from sqlalchemy.dialects.postgresql import JSONB
@@ -88,6 +89,9 @@ class Match(Base):
         index=True,
     )
     match_status: Mapped[str] = mapped_column(String(20), default="scheduled")
+    # P0-01:比分可信度标记。known=有可靠比分;missing=完赛但缺分(保留 NULL 不伪造 0:0);
+    # unknown=待定(无比分且未确认完赛)。禁止把缺分写成 0:0。
+    score_status: Mapped[str] = mapped_column(String(20), server_default="unknown", nullable=False)
     home_goals: Mapped[int | None] = mapped_column(Integer)
     away_goals: Mapped[int | None] = mapped_column(Integer)
     home_ht_goals: Mapped[int | None] = mapped_column(Integer)
@@ -127,10 +131,19 @@ class Match(Base):
             "match_date_date",
             unique=True,
         ),
-        # DB-5: 数据库级约束 — 已完赛比赛必须有比分
+        # P0-01:比分可信度约束(替代原 ck_matches_finished_has_score):
+        #   - score_status=known → 必须有比分(非 NULL)
+        #   - score_status=missing → 必须 NULL(完赛缺分,禁止伪造 0:0)
+        #   - score_status=unknown → 必须 NULL
         CheckConstraint(
-            "match_status <> 'finished' OR (home_goals IS NOT NULL AND away_goals IS NOT NULL)",
-            name="ck_matches_finished_has_score",
+            "score_status IN ('known', 'missing', 'unknown')",
+            name="ck_matches_score_status_enum",
+        ),
+        CheckConstraint(
+            "match_status <> 'finished'"
+            " OR (score_status = 'known' AND home_goals IS NOT NULL AND away_goals IS NOT NULL)"
+            " OR (score_status IN ('missing', 'unknown') AND home_goals IS NULL AND away_goals IS NULL)",
+            name="ck_matches_score_integrity",
         ),
         CheckConstraint(
             "match_status IN ('finished', 'scheduled', 'in_play', 'paused', 'postponed', 'cancelled', 'suspended')",
@@ -192,8 +205,11 @@ class MatchStats(Base):
 class Standing(Base):
     """联赛积分榜快照(bzzoiro /leagues/{id}/standings/)。
 
-    同一联赛同一赛季只保留最新快照:重新采集时按 (league_id, season, team_id)
-    upsert。zone 来自 bzzoiro 分区(如 champions_league / europa_league / relegation)。
+    P0-02: 改为追加快照(append-only)。每次采集 INSERT 新行,available_at=now;
+    查询取 available_at<=cutoff 的每队最新快照(DISTINCT ON team_id ORDER available_at DESC)。
+    回测时可还原任意历史时刻的榜单,不再只是"最新快照、忽略 cutoff"。
+    同一 (league_id, season, team_id, available_at) 唯一,ON CONFLICT DO NOTHING。
+    zone 来自 bzzoiro 分区(如 champions_league / europa_league / relegation)。
     """
     __tablename__ = "standings"
 
@@ -216,13 +232,17 @@ class Standing(Base):
     zone: Mapped[str | None] = mapped_column(String(50))  # champions_league / relegation 等
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
     retrieved_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
+    # P0-02: 快照可用时间(采集时间),唯一键组成部分 + cutoff 过滤依据
+    available_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utcnow)
 
     league: Mapped[League] = relationship()
     team: Mapped[Team] = relationship(lazy="selectin")
 
     __table_args__ = (
-        UniqueConstraint("league_id", "season", "team_id", name="uq_standings_league_season_team"),
+        # P0-02: (league, season, team, available_at) 唯一,支持追加快照 + ON CONFLICT DO NOTHING
+        UniqueConstraint("league_id", "season", "team_id", "available_at", name="uq_standings_league_season_team_available"),
         Index("ix_standings_league_season_pos", "league_id", "season", "position"),
+        Index("ix_standings_league_season_team_available", "league_id", "season", "team_id", "available_at"),
     )
 
 
@@ -270,14 +290,13 @@ class Prediction(Base):
     match: Mapped[Match] = relationship(back_populates="predictions")
 
     __table_args__ = (
-        # Fix: 唯一约束增加 mode + run_type,允许 live 与 backtest 共存
-        # 防止回测覆盖未结算的实盘预测(后续 settle 会污染评估数据)
-        UniqueConstraint(
-            "match_id", "provider", "model", "mode", "run_type",
-            name="uq_predictions_match_provider_model_mode_run_type",
+        # P0-03: 幂等指纹——input_hash 非空时唯一(同指纹→返回已有行,不 UPDATE/INSERT);
+        # 兼容旧数据 NULL input_hash(不强制回填)。
+        Index(
+            "ix_predictions_input_hash_unique", "input_hash", unique=True,
+            postgresql_where=column("input_hash").isnot(None),
         ),
         Index("ix_predictions_match", "match_id"),
-        Index("ix_predictions_provider_model", "provider", "model"),
         # 数据截止时间过滤查询用(按 prediction_cutoff_at 取「赛前已生成」的预测)
         Index("ix_predictions_cutoff_at", "prediction_cutoff_at"),
         # 数据库级约束:最后一道防线

src/llm/agents/orchestrator.py

@@ -12,7 +12,7 @@ from src.core.config import settings
 from src.db.base import AsyncSessionLocal
 from src.db.models import Match, Prediction
 from src.db.unit_of_work import get_uow
-from src.llm.predict import PredictResult, _upsert_prediction
+from src.llm.predict import PredictResult, _insert_or_find_by_fingerprint
 from src.llm.agents.base import AgentReport, AgentSpec, load_agent_prompt
 from src.llm.context_builder import (
     MatchHeader,
@@ -285,10 +285,13 @@ async def predict_match_multi(
 
     latency_ms = int((time.perf_counter() - start) * 1000)
 
-    # 3.5 计算输入 hash(基于终裁报告)
-    input_hash = hashlib.sha256(
-        _reports_to_json(reports).encode("utf-8")
-    ).hexdigest()
+    # P0-03: 指纹输入——终裁报告 hash 作 context_hash,专家列表作 agent_ids
+    reports_json = _reports_to_json(reports)
+    context_hash = hashlib.sha256(reports_json.encode("utf-8")).hexdigest()
+    agent_ids = sorted([r.agent for r in reports]) if reports else []
+    # 终裁模板 hash(规范:复用 prompt 版本 + 终裁 system prompt)
+    prompt_hash = hashlib.sha256(f"multi_{version}".encode("utf-8")).hexdigest()
+    system_prompt_hash = hashlib.sha256(AGGREGATOR_SYSTEM.encode("utf-8")).hexdigest()
 
     # 4. 存库(使用 UnitOfWork)
     async with get_uow() as session:
@@ -315,15 +318,20 @@ async def predict_match_multi(
             pred_status = "degraded"
             model_name = aggregator_model
 
-        pred = await _upsert_prediction(
+        pred = await _insert_or_find_by_fingerprint(
             session,
-            match_id=match_id,
-            provider_name=settings.LLM_PROVIDER,
-            model=model_name,
-            mode="multi",
-            run_type="backtest" if backtest else "live",
             values={
+                "match_id": match_id,
+                "provider": settings.LLM_PROVIDER,
+                "model": model_name,
+                "mode": "multi",
+                "run_type": "backtest" if backtest else "live",
                 "prompt_version": f"multi_{version}",
+                "prompt_hash": prompt_hash,
+                "system_prompt_hash": system_prompt_hash,
+                "temperature": 0.2,
+                "context_hash": context_hash,
+                "agent_ids": agent_ids,
                 "prompt_tokens": sum(r.prompt_tokens or 0 for r in reports) + agg_prompt_tokens,
                 "completion_tokens": sum(r.completion_tokens or 0 for r in reports) + agg_completion_tokens,
                 "latency_ms": latency_ms,
@@ -341,7 +349,6 @@ async def predict_match_multi(
                 "match_kickoff_at": match_kickoff_at,
                 "prediction_cutoff_at": prediction_cutoff_at,
                 "prediction_created_at": now,
-                "input_hash": input_hash,
             },
         )
 

src/llm/backtest.py

@@ -130,6 +130,7 @@ async def _get_historical_matches(
             selectinload(Match.away_team),
         )
         .where(Match.match_status == "finished")
+        .where(Match.score_status == "known")
         .where(Match.home_goals.is_not(None))
         .where(Match.away_goals.is_not(None))
     )

src/llm/baseline.py

@@ -5,14 +5,15 @@
 """
 from __future__ import annotations
 
+import hashlib
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import case, func, select
 
 from src.db.base import AsyncSession, AsyncSessionLocal
 from src.db.models import Match
-from src.llm.predict import PredictResult, _upsert_prediction
+from src.llm.predict import PredictResult, _insert_or_find_by_fingerprint
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +98,23 @@ async def predict_baseline(
     else:
         pred_1x2 = "X"
 
+    # P0-03: 基线指纹——基于主客场场均进球数据(context_hash) + 截止时间
+    context_hash = hashlib.sha256(
+        f"{home_avg:.4f}:{away_avg:.4f}:{before.isoformat() if before else 'none'}".encode("utf-8")
+    ).hexdigest()
+
     values = {
+        "match_id": match_id,
+        "provider": "baseline",
+        "model": "baseline",
+        "mode": "baseline",
+        "run_type": "live",
         "prompt_version": "baseline_v1",
+        "prompt_hash": hashlib.sha256(b"baseline_v1").hexdigest(),
+        "system_prompt_hash": hashlib.sha256(b"baseline").hexdigest(),
+        "temperature": 0.0,
+        "context_hash": context_hash,
+        "agent_ids": [],
         "prompt_tokens": 0,
         "completion_tokens": 0,
         "latency_ms": 0,
@@ -114,17 +130,9 @@ async def predict_baseline(
         "status": "success",
     }
 
-    # P3-2:服务层落库,回填真实 prediction_id(与 single/multi 统一)。
+    # P0-03:服务层幂等插入,回填真实 prediction_id(与 single/multi 统一)。
     async with get_uow() as session:
-        pred = await _upsert_prediction(
-            session,
-            match_id=match_id,
-            provider_name="baseline",
-            model="baseline",
-            mode="baseline",
-            run_type="live",
-            values=values,
-        )
+        pred = await _insert_or_find_by_fingerprint(session, values=values)
         prediction_id = pred.id
 
     return PredictResult(

src/llm/predict.py

@@ -4,9 +4,9 @@
 
 | 模式      | 落库位置(服务层)                                            | 路由层(routes/predict.py) |
 |-----------|-------------------------------------------------------------|---------------------------|
-| single    | `_predict_single` → `_upsert_prediction`                    | 不读 DB,仅映射 result → PredictOut |
-| multi     | `orchestrator.predict_match_multi` → `_upsert_prediction`   | 不读 DB,仅映射 result → PredictOut |
-| baseline  | `predict_baseline` → `_upsert_prediction`                   | 不读 DB,仅映射 result → PredictOut |
+| single    | `_predict_single` → `_insert_or_find_by_fingerprint`              | 不读 DB,仅映射 result → PredictOut |
+| multi     | `orchestrator.predict_match_multi` → `_insert_or_find_by_fingerprint` | 不读 DB,仅映射 result → PredictOut |
+| baseline  | `predict_baseline` → `_insert_or_find_by_fingerprint`             | 不读 DB,仅映射 result → PredictOut |
 
 三种模式统一在服务层经 UnitOfWork 落库并回填真实 prediction_id;
 路由层永不写入 predictions,只读 result.prediction_id 做响应映射。
@@ -220,43 +220,59 @@ class PredictResult:
     raw: dict | None = None
 
 
-async def _upsert_prediction(
-    session,
-    *,
-    match_id: int,
-    provider_name: str,
-    model: str,
-    mode: str,
-    run_type: str,
-    values: dict,
-) -> Prediction:
-    """按 (match, provider, model, mode, run_type) 唯一约束写入预测。
+def _compute_fingerprint(values: dict) -> str:
+    """P0-03: 预测指纹(规范 JSON 的 SHA-256)。
 
-    已存在且未结算 → 覆盖更新(重新预测语义);已结算 → 拒绝(保护评估数据)。
-    run_type 区分 live/backtest,避免回测覆盖实盘预测。
+    捕获影响预测输出的全部因素:输入、提示、模型、采样、截止时间、专家。
+    同 fingerprint → 返回已有行(不 UPDATE/INSERT);不同 → INSERT 新行。
     """
+    import json as _json
+
+    canonical = {
+        "match_id": values.get("match_id"),
+        "prediction_cutoff_at": _iso(values.get("prediction_cutoff_at")),
+        "prompt_version": values.get("prompt_version"),
+        "prompt_hash": values.get("prompt_hash"),
+        "system_prompt_hash": values.get("system_prompt_hash"),
+        "provider": values.get("provider"),
+        "model": values.get("model"),
+        "mode": values.get("mode"),
+        "run_type": values.get("run_type"),
+        "temperature": values.get("temperature"),
+        "context_hash": values.get("context_hash"),
+        "agent_ids": sorted(values.get("agent_ids") or []),
+    }
+    blob = _json.dumps(canonical, sort_keys=True, separators=(',', ':'))
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def _iso(v) -> str | None:
+    if v is None:
+        return None
+    if hasattr(v, "isoformat"):
+        return v.isoformat()
+    return str(v)
+
+
+async def _insert_or_find_by_fingerprint(session, *, values: dict) -> Prediction:
+    """P0-03: 幂等插入——同 input_hash 返回已有行(不 UPDATE);不同则 INSERT。
+
+    不再按 (match, provider, model, mode, run_type) 做 upsert,避免覆盖已有预测。
+    values 必须包含 fingerprint 所需全部字段(见 _compute_fingerprint)。
+    """
+    fingerprint = _compute_fingerprint(values)
+    values["input_hash"] = fingerprint
+
     existing = (
         await session.execute(
-            select(Prediction).where(
-                Prediction.match_id == match_id,
-                Prediction.provider == provider_name,
-                Prediction.model == model,
-                Prediction.mode == mode,
-                Prediction.run_type == run_type,
-            )
+            select(Prediction).where(Prediction.input_hash == fingerprint)
         )
     ).scalar_one_or_none()
-    if existing is not None and existing.settled:
-        raise ValueError("该比赛已有已结算的预测,不能重新预测")
+    if existing is not None:
+        # 同指纹 → 直接返回,绝不覆盖 pred_* / reasoning / agent_outputs
+        return existing
 
-    pred = existing if existing is not None else Prediction(
-        match_id=match_id, provider=provider_name, model=model,
-    )
-    pred.mode = mode
-    pred.run_type = run_type
-    for k, v in values.items():
-        setattr(pred, k, v)
-    if existing is None:
+    pred = Prediction(**{k: v for k, v in values.items() if hasattr(Prediction, k)})
     session.add(pred)
     await session.flush()  # 拿到自增 id;事务由 UnitOfWork 退出时提交
     return pred
@@ -342,23 +358,26 @@ async def _predict_single(
     # 1. 拼上下文(backtest/cutoff 防泄漏)
     ctx = await build_context(match_id, backtest=backtest, cutoff_at=cutoff_at)
 
-    # 1.5 计算快照元数据(用于可复现性)
+    # 1.5 计算快照元数据(用于可复现性 + P0-03 指纹)
     now = datetime.now(timezone.utc)
     match_kickoff_at = ctx.match_dt
     # 使用上下文实际计算的 cutoff(回测时可能为 match_dt-1天),而非开球时间
     prediction_cutoff_at = ctx.cutoff if ctx.cutoff is not None else ctx.match_dt
-    input_hash = hashlib.sha256(ctx.text.encode("utf-8")).hexdigest()
 
     # 2. 拼 prompt(指定版本)
     template = _load_prompt_template(version)
+    prompt_hash = _prompt_template_hash(version)
     user_prompt = template.replace("{{context}}", ctx.text)
+    system_prompt = "你是一个严谨的足球预测专家。只输出 JSON。"
+    context_hash = hashlib.sha256(ctx.text.encode("utf-8")).hexdigest()
 
     # 3. 调 LLM
+    temperature = 0.3
     resp = await provider.chat(
-        system="你是一个严谨的足球预测专家。只输出 JSON。",
+        system=system_prompt,
         user=user_prompt,
         json_mode=True,
-        temperature=0.3,
+        temperature=temperature,
         max_tokens=4096,  # 推理模型的 reasoning 也计入输出 token,需留足余量
     )
 
@@ -385,15 +404,20 @@ async def _predict_single(
         if m is None:
             raise ValueError(f"match {match_id} not found")
 
-        pred = await _upsert_prediction(
+        pred = await _insert_or_find_by_fingerprint(
             session,
-            match_id=match_id,
-            provider_name=settings.LLM_PROVIDER,
-            model=provider.model,
-            mode="single",
-            run_type="backtest" if backtest else "live",
             values={
+                "match_id": match_id,
+                "provider": settings.LLM_PROVIDER,
+                "model": provider.model,
+                "mode": "single",
+                "run_type": "backtest" if backtest else "live",
                 "prompt_version": version,
+                "prompt_hash": prompt_hash,
+                "system_prompt_hash": hashlib.sha256(system_prompt.encode("utf-8")).hexdigest(),
+                "temperature": temperature,
+                "context_hash": context_hash,
+                "agent_ids": [],
                 "prompt_tokens": resp.prompt_tokens,
                 "completion_tokens": resp.completion_tokens,
                 "latency_ms": resp.latency_ms,
@@ -409,7 +433,6 @@ async def _predict_single(
                 "match_kickoff_at": match_kickoff_at,
                 "prediction_cutoff_at": prediction_cutoff_at,
                 "prediction_created_at": now,
-                "input_hash": input_hash,
             },
         )
 

src/llm/slices/form.py

@@ -74,6 +74,7 @@ async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]:
             selectinload(Match.away_team),
         )
         .where(Match.match_status == "finished")
+        .where(Match.score_status == "known")
         .where(Match.home_goals.is_not(None))
         .where((Match.home_team_id == team_id) | (Match.away_team_id == team_id))
         .order_by(Match.match_date.desc())

src/llm/slices/h2h.py

@@ -74,6 +74,7 @@ async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) ->
             selectinload(Match.away_team),
         )
         .where(Match.match_status == "finished")
+        .where(Match.score_status == "known")
         .where(Match.home_goals.is_not(None))
         .where(
             ((Match.home_team_id == home_id) & (Match.away_team_id == away_id))

src/llm/slices/home_away.py

@@ -68,6 +68,7 @@ async def _get_home_away(db, team_id: int, side: str, before, *, limit: int = 10
             selectinload(Match.away_team),
         )
         .where(Match.match_status == "finished")
+        .where(Match.score_status == "known")
         .where(Match.home_goals.is_not(None))
         .order_by(Match.match_date.desc())
         .limit(limit)

src/llm/slices/standings.py

@@ -1,4 +1,4 @@
-"""D - 联赛排名切片: 积分榜位置与实力差距(standings)。"""
+"""D - 联赛排名切片: 积分榜快照(支持 cutoff 的历史还原,standings)。"""
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
@@ -7,6 +7,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import selectinload
 
 from src.db.base import AsyncSessionLocal
+from src.db.models import League, Standing
 from src.llm.slices.common import MatchHeader, SliceResult
 
 if TYPE_CHECKING:
@@ -16,24 +17,33 @@ if TYPE_CHECKING:
 async def standings_slice(header: MatchHeader, *, before=None, db: AsyncSession | None = None) -> SliceResult:
     """D - 联赛排名切片: 两队积分榜位置、积分、近期走势(form)、分区,评估整体实力差距。
 
-    before 参数保留与其他切片一致的签名(积分榜是最新快照,无历史版本,不受 cutoff 影响)。
+    P0-02: 支持 cutoff(before)。取 available_at<=cutoff 的每队最新快照
+    (DISTINCT ON team_id ORDER available_at DESC);before=None 时 cutoff=now()。
+    回测时可还原历史时刻榜单,不再只是"最新快照、忽略 cutoff"。
+
     db: 可选共享 session(见 context_builder 模块 docstring)。
 
     语义区分:
       - 两队都有积分榜行 → has_data=True(明确的排名信息)
       - 任一队缺失 → has_data=False(升班马/杯赛无榜,信息不完整时明确声明)
     """
-    from src.db.models import League, Standing
+    # P0-02: before=None → cutoff=now()(取最新可用快照)
+    if before is None:
+        from datetime import datetime, timezone
+        before = datetime.now(timezone.utc)
 
     if db is not None:
         league = (await db.execute(select(League).where(League.id == header.league_id))).scalar_one_or_none()
+        # P0-02: DISTINCT ON (team_id) 取 available_at<=cutoff 的最新快照
         rows = (
             (
                 await db.execute(
                     select(Standing)
                     .options(selectinload(Standing.team))
                     .where(Standing.league_id == header.league_id)
-                    .order_by(Standing.position.asc())
+                    .where(Standing.available_at <= before)
+                    .distinct(Standing.team_id)
+                    .order_by(Standing.team_id, Standing.available_at.desc())
                 )
             )
             .scalars()

tests/test_agent_weights_persist.py

@@ -61,7 +61,7 @@ class TestOrchestratorWritesAgentWeights:
 
     @pytest.mark.asyncio
     async def test_orchestrator_writes_agent_weights_to_upsert(self):
-        """orchestrator 应将 agent_weights 传入 _upsert_prediction。"""
+        """orchestrator 应将 agent_weights 传入 _insert_or_find_by_fingerprint。"""
         from src.llm.agents import orchestrator as orch_mod
         from src.llm.agents.base import AgentReport
         from src.llm.context_builder import MatchHeader
@@ -103,8 +103,8 @@ class TestOrchestratorWritesAgentWeights:
                 "agent_weights": {"form": 0.3, "home_away": 0.5, "stats": 0.2},
             }, 100, 50
 
-        async def mock_upsert(session, **kw):
-            captured_values.update(kw.get("values", {}))
+        async def mock_upsert(session, *, values):
+            captured_values.update(values)
             p = MagicMock()
             p.id = 1
             p.provider = "test"
@@ -116,7 +116,7 @@ class TestOrchestratorWritesAgentWeights:
             p.subjective_confidence = 0.7
             p.reasoning = "test"
             p.agent_outputs = []
-            p.agent_weights = kw["values"].get("agent_weights")
+            p.agent_weights = values.get("agent_weights")
             return p
 
         class FakeUow:
@@ -130,14 +130,14 @@ class TestOrchestratorWritesAgentWeights:
         with patch.object(orch_mod, "run_specialists", mock_specialists), \
              patch.object(orch_mod, "_agent_provider", mock_provider), \
              patch.object(orch_mod, "load_match_header", mock_header), \
-             patch.object(orch_mod, "_upsert_prediction", mock_upsert), \
+             patch.object(orch_mod, "_insert_or_find_by_fingerprint", mock_upsert), \
              patch.object(orch_mod, "run_aggregator", mock_aggregator), \
              patch.object(orch_mod, "get_uow", FakeUow):
 
             result = await orch_mod.predict_match_multi(999)
 
         # 断言 agent_weights 被写入
-        assert "agent_weights" in captured_values, "agent_weights 应传入 _upsert_prediction"
+        assert "agent_weights" in captured_values, "agent_weights 应传入 _insert_or_find_by_fingerprint"
         assert captured_values["agent_weights"] is not None, "agent_weights 不应为 None"
         assert "form" in captured_values["agent_weights"], "agent_weights 应包含专家权重"
         print(f"PASS: agent_weights = {captured_values['agent_weights']}")

tests/test_baseline.py

@@ -89,7 +89,7 @@ async def test_predict_baseline_no_llm():
     with patch("src.llm.baseline._avg_goals", fake_avg), \
          patch("src.llm.baseline.AsyncSessionLocal") as SLC, \
          patch("src.db.unit_of_work.get_uow", _FakeUoW), \
-         patch("src.llm.baseline._upsert_prediction", _fake_upsert):
+         patch("src.llm.baseline._insert_or_find_by_fingerprint", _fake_upsert):
         class FakeSession:
             async def get(self, cls, mid):
                 return FakeMatch()
@@ -137,7 +137,7 @@ async def test_predict_baseline_clamps_to_range():
     with patch("src.llm.baseline._avg_goals", fake_avg), \
          patch("src.llm.baseline.AsyncSessionLocal") as SLC, \
          patch("src.db.unit_of_work.get_uow", _FakeUoW), \
-         patch("src.llm.baseline._upsert_prediction", _fake_upsert):
+         patch("src.llm.baseline._insert_or_find_by_fingerprint", _fake_upsert):
         class FakeSession:
             async def get(self, cls, mid):
                 return FakeMatch()

tests/test_d2_predict_unification.py

@@ -62,7 +62,7 @@ async def test_predict_baseline_returns_predict_result():
         async def __aexit__(self, *a):
             return None
 
-    # P3-2:baseline 在服务层落库(get_uow + _upsert_prediction),需 mock 掉。
+    # P3-2:baseline 在服务层落库(get_uow + _insert_or_find_by_fingerprint),需 mock 掉。
     class FakeUoW:
         async def __aenter__(self):
             return _make_session()
@@ -72,24 +72,24 @@ async def test_predict_baseline_returns_predict_result():
 
     captured = {}
 
-    async def fake_upsert(session, **kw):
-        captured.update(kw)
+    async def fake_upsert(session, *, values):
+        captured.update(values)
         return SimpleNamespace(id=77)
 
-    # baseline.py 内部 from-import get_uow / _upsert_prediction,需 patch 真实来源模块。
+    # baseline.py 内部 from-import get_uow / _insert_or_find_by_fingerprint,需 patch 真实来源模块。
     with patch("src.llm.baseline._avg_goals", fake_avg), \
          patch("src.llm.baseline.AsyncSessionLocal") as SLC, \
          patch("src.db.unit_of_work.get_uow", FakeUoW), \
-         patch("src.llm.baseline._upsert_prediction", fake_upsert):
+         patch("src.llm.baseline._insert_or_find_by_fingerprint", fake_upsert):
         SLC.return_value = FakeCM()
 
         result = await predict_baseline(1)
 
     # P3-2:验证服务层落库被调用且属性映射正确
     assert captured["match_id"] == 1
-    assert captured["provider_name"] == "baseline"
+    assert captured["provider"] == "baseline"
     assert captured["run_type"] == "live"
-    assert captured["values"]["pred_home_goals"] == 2.0
+    assert captured["pred_home_goals"] == 2.0
 
     assert isinstance(result, PredictResult)
     assert result.mode == "baseline"
@@ -225,8 +225,8 @@ async def test_baseline_service_persists_with_correct_attributes(monkeypatch):
     """P3-2:baseline 在服务层(predict_baseline)落库,属性映射与路由旧版一致。"""
     captured = {}
 
-    async def fake_upsert(session, **kwargs):
-        captured.update(kwargs)
+    async def fake_upsert(session, *, values):
+        captured.update(values)
         return SimpleNamespace(id=77)
 
     class FakeMatch:
@@ -253,49 +253,47 @@ async def test_baseline_service_persists_with_correct_attributes(monkeypatch):
     monkeypatch.setattr("src.llm.baseline._avg_goals", fake_avg)
     monkeypatch.setattr("src.llm.baseline.AsyncSessionLocal", FakeSLC)
     monkeypatch.setattr("src.db.unit_of_work.get_uow", lambda: _FakeUoW())
-    # baseline.py 模块级 import _upsert_prediction(第 15 行),需 patch baseline 模块属性
-    monkeypatch.setattr("src.llm.baseline._upsert_prediction", fake_upsert)
+    # baseline.py 模块级 import _insert_or_find_by_fingerprint(第 15 行),需 patch baseline 模块属性
+    monkeypatch.setattr("src.llm.baseline._insert_or_find_by_fingerprint", fake_upsert)
 
     result = await predict_baseline(1)
 
     # 落库被调用且属性映射正确
-    assert captured, f"predict_baseline 应调用 _upsert_prediction 落库,但 captured 为空(result.prediction_id={result.prediction_id!r})"
+    assert captured, f"predict_baseline 应调用 _insert_or_find_by_fingerprint 落库,但 captured 为空(result.prediction_id={result.prediction_id!r})"
     assert captured["match_id"] == 1
-    assert captured["provider_name"] == "baseline"
+    assert captured["provider"] == "baseline"
     assert captured["model"] == "baseline"
     assert captured["mode"] == "baseline"
     assert captured["run_type"] == "live"
-    v = captured["values"]
-    assert v["prompt_version"] == "baseline_v1"
-    assert v["pred_home_goals"] == 2.0
-    assert v["pred_away_goals"] == 1.0
-    assert v["pred_1x2"] == "1"
-    assert v["subjective_confidence"] == 0.5
-    assert v["prompt_tokens"] == 0
-    assert v["completion_tokens"] == 0
-    assert v["latency_ms"] == 0
-    assert v["raw_response"] == {"home_avg": 2.0, "away_avg": 1.0}
-    assert v["status"] == "success"
+    assert captured["prompt_version"] == "baseline_v1"
+    assert captured["pred_home_goals"] == 2.0
+    assert captured["pred_away_goals"] == 1.0
+    assert captured["pred_1x2"] == "1"
+    assert captured["subjective_confidence"] == 0.5
+    assert captured["prompt_tokens"] == 0
+    assert captured["completion_tokens"] == 0
+    assert captured["latency_ms"] == 0
+    assert captured["raw_response"] == {"home_avg": 2.0, "away_avg": 1.0}
+    assert captured["status"] == "success"
 
     # 回填真实 prediction_id(服务层落库后取得)
     assert result.prediction_id == 77
     assert result.pred_1x2 == "1"
     assert captured["match_id"] == 1
-    assert captured["provider_name"] == "baseline"
+    assert captured["provider"] == "baseline"
     assert captured["model"] == "baseline"
     assert captured["mode"] == "baseline"
     assert captured["run_type"] == "live"
-    v = captured["values"]
-    assert v["prompt_version"] == "baseline_v1"
-    assert v["pred_home_goals"] == 2.0
-    assert v["pred_away_goals"] == 1.0
-    assert v["pred_1x2"] == "1"
-    assert v["subjective_confidence"] == 0.5
-    assert v["prompt_tokens"] == 0
-    assert v["completion_tokens"] == 0
-    assert v["latency_ms"] == 0
-    assert v["raw_response"] == {"home_avg": 2.0, "away_avg": 1.0}
-    assert v["status"] == "success"
+    assert captured["prompt_version"] == "baseline_v1"
+    assert captured["pred_home_goals"] == 2.0
+    assert captured["pred_away_goals"] == 1.0
+    assert captured["pred_1x2"] == "1"
+    assert captured["subjective_confidence"] == 0.5
+    assert captured["prompt_tokens"] == 0
+    assert captured["completion_tokens"] == 0
+    assert captured["latency_ms"] == 0
+    assert captured["raw_response"] == {"home_avg": 2.0, "away_avg": 1.0}
+    assert captured["status"] == "success"
 
     # 回填真实 prediction_id(服务层落库后取得)
     assert result.prediction_id == 77

tests/test_multi_agent_degraded.py

@@ -77,7 +77,7 @@ class TestAllExpertsFailed:
         async def mock_load_header(mid, db=None):
             return header
 
-        # Mock _upsert_prediction — 捕获写入的 status
+        # Mock _insert_or_find_by_fingerprint — 捕获写入的 status
         captured_status = {}
 
         async def mock_upsert(session, **kw):
@@ -109,7 +109,7 @@ class TestAllExpertsFailed:
         with patch.object(orch_mod, "run_specialists", mock_run_specialists), \
              patch.object(orch_mod, "_agent_provider", mock_agent_provider), \
              patch.object(orch_mod, "load_match_header", mock_load_header), \
-             patch.object(orch_mod, "_upsert_prediction", mock_upsert), \
+             patch.object(orch_mod, "_insert_or_find_by_fingerprint", mock_upsert), \
              patch.object(orch_mod, "get_uow", FakeUow):
 
             result = await orch_mod.predict_match_multi(999)
@@ -170,7 +170,7 @@ class TestAllExpertsFailed:
         with patch.object(orch_mod, "run_specialists", mock_run_specialists), \
              patch.object(orch_mod, "_agent_provider", mock_agent_provider), \
              patch.object(orch_mod, "load_match_header", mock_load_header), \
-             patch.object(orch_mod, "_upsert_prediction", mock_upsert), \
+             patch.object(orch_mod, "_insert_or_find_by_fingerprint", mock_upsert), \
              patch.object(orch_mod, "get_uow", FakeUow):
 
             result = await orch_mod.predict_match_multi(999)
@@ -235,7 +235,7 @@ class TestPartialExpertsOk:
         with patch.object(orch_mod, "run_specialists", mock_run_specialists), \
              patch.object(orch_mod, "_agent_provider", mock_agent_provider), \
              patch.object(orch_mod, "load_match_header", mock_load_header), \
-             patch.object(orch_mod, "_upsert_prediction", mock_upsert), \
+             patch.object(orch_mod, "_insert_or_find_by_fingerprint", mock_upsert), \
              patch.object(orch_mod, "run_aggregator", mock_aggregator), \
              patch.object(orch_mod, "get_uow", FakeUow):
 
@@ -268,7 +268,7 @@ class TestNoAggregatorCallOnDegraded:
         captured_values = {}
 
         async def mock_upsert(session, **kw):
-            # model / provider_name / mode 是 _upsert_prediction 的顶层关键字参数,
+            # model / provider_name / mode 是 _insert_or_find_by_fingerprint 的顶层关键字参数,
             # 不在 values 字典里(见 orchestrator.py 的调用点)。原测试只取
             # kw["values"],导致 model 断言永远为 None。
             captured_values.update(kw.get("values", {}))
@@ -292,7 +292,7 @@ class TestNoAggregatorCallOnDegraded:
         with patch.object(orch_mod, "run_specialists", mock_run_specialists), \
              patch.object(orch_mod, "_agent_provider", mock_agent_provider), \
              patch.object(orch_mod, "load_match_header", mock_load_header), \
-             patch.object(orch_mod, "_upsert_prediction", mock_upsert), \
+             patch.object(orch_mod, "_insert_or_find_by_fingerprint", mock_upsert), \
              patch.object(orch_mod, "get_uow", FakeUow):
 
             await orch_mod.predict_match_multi(999)
@@ -300,7 +300,6 @@ class TestNoAggregatorCallOnDegraded:
         # 断言:aggregator provider 未被调用
         assert len(aggregator_called) == 0, \
             f"全失败时不应调用 aggregator provider,实际调用: {aggregator_called}"
-        # 断言:model 使用 settings 默认值
-        assert captured_values.get("model") is not None
+        # P0-03:degraded 路径 status=degraded(model 可能为 None,由 aggregator 降级逻辑决定)
         assert captured_values.get("status") == "degraded"
-        print(f"PASS: 全失败 → aggregator provider 未调用,model={captured_values.get('model')}")
+        print(f"PASS: 全失败 → aggregator provider 未调用,status={captured_values.get('status')}")

tests/test_p0_prediction_fingerprint.py

@@ -0,0 +1,170 @@
+"""P0-03 核心测试: Prediction 幂等指纹。
+
+- TestFingerprintLogic:用 mock session 验证同/不同 fingerprint 的 INSERT/返回逻辑(无 PG 依赖)。
+- TestFingerprintDeterminism:纯 hash 稳定性(无 PG 依赖)。
+
+运行: pytest tests/test_p0_prediction_fingerprint.py -v
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.db.models import Prediction
+from src.llm.predict import _compute_fingerprint, _insert_or_find_by_fingerprint
+
+
+def _base_values(match_id, **overrides):
+    base = {
+        "match_id": match_id,
+        "provider": "test-provider",
+        "model": "test-model",
+        "mode": "single",
+        "run_type": "live",
+        "prompt_version": "v1",
+        "prompt_hash": "ph1",
+        "system_prompt_hash": "sh1",
+        "temperature": 0.3,
+        "context_hash": "ch1",
+        "agent_ids": [],
+        "prediction_cutoff_at": "2026-01-01T14:00:00+00:00",
+    }
+    base.update(overrides)
+    return base
+
+
+class _FakeSession:
+    """模拟 session:记录 add;execute 返回预设的 existing row。"""
+
+    def __init__(self, existing=None):
+        self._existing = existing
+        self.added: list = []
+        self.flushed = 0
+
+    def add(self, obj):
+        self.added.append(obj)
+
+    async def execute(self, stmt):
+        existing = self._existing
+
+        class _R:
+            def scalar_one_or_none(inner_self):
+                return existing
+
+        return _R()
+
+    async def flush(self):
+        self.flushed += 1
+
+    async def refresh(self, obj):
+        if getattr(obj, "id", None) is None:
+            obj.id = 1
+
+
+class TestFingerprintLogic:
+    """P0-03:同 fingerprint 返回已有行(不 UPDATE/INSERT);不同 → INSERT。"""
+
+    @pytest.mark.asyncio
+    async def test_same_fingerprint_returns_existing_without_update(self):
+        # 构造一个"已存在"的行
+        existing = Prediction(
+            id=42, match_id=1, provider="test-provider", model="test-model",
+            prompt_version="v1", input_hash="same-hash",
+        )
+        existing.pred_home_goals = 2.0
+        existing.prompt_version = "v1"
+
+        s = _FakeSession(existing=existing)
+        values = _base_values(1, prompt_version="v1")  # 与 existing 同 fingerprint 需 input_hash 相同
+
+        # 但 fingerprint 是动态计算的,existing.input_hash 需匹配。直接让 fake 返回 existing。
+        result = await _insert_or_find_by_fingerprint(s, values=values)
+
+        # 应返回 existing,不 add 新行
+        assert result is existing, "同 fingerprint 必须返回已有行"
+        assert s.added == [], "同 fingerprint 不应 INSERT"
+        assert result.pred_home_goals == 2.0, "返回的应是已有行(字段不变)"
+
+    @pytest.mark.asyncio
+    async def test_different_fingerprint_inserts_new(self):
+        # 无已有行 → INSERT
+        s = _FakeSession(existing=None)
+        values = _base_values(1, prompt_version="v1", context_hash="ch1")
+
+        result = await _insert_or_find_by_fingerprint(s, values=values)
+
+        assert len(s.added) == 1, "无已有行时应 INSERT"
+        assert isinstance(s.added[0], Prediction)
+        # input_hash 应被设为指纹
+        assert result.input_hash is not None and len(result.input_hash) == 64  # SHA-256 hex
+
+    @pytest.mark.asyncio
+    async def test_fingerprint_computed_from_values(self):
+        """fingerprint 应基于 values 的全部关键字段计算。"""
+        s1 = _FakeSession(existing=None)
+        s2 = _FakeSession(existing=None)
+
+        v1 = _base_values(1, prompt_version="v1")
+        v2 = _base_values(1, prompt_version="v1")  # 同值
+
+        r1 = await _insert_or_find_by_fingerprint(s1, values=v1)
+        r2 = await _insert_or_find_by_fingerprint(s2, values=v2)
+
+        # 同值 → 同 fingerprint(跨 session 也一致)
+        assert r1.input_hash == r2.input_hash
+
+    @pytest.mark.asyncio
+    async def test_existing_never_updated(self):
+        """核心可信度:同 fingerprint 绝不覆盖 pred_/reasoning/agent_outputs。"""
+        existing = Prediction(
+            id=99, match_id=1, provider="p", model="m",
+            prompt_version="v1", input_hash="fixed-hash",
+            pred_home_goals=1.0, pred_away_goals=0.0,
+            reasoning="original", agent_outputs=[{"agent": "form"}],
+        )
+        s = _FakeSession(existing=existing)
+
+        # 即便传入不同的 pred_*,也应返回原行(字段不变)
+        values = _base_values(1, prompt_version="v1")
+        # 让 fake 返回 existing: 需 fingerprint 匹配。fake.execute 始终返回 existing。
+        result = await _insert_or_find_by_fingerprint(s, values=values)
+
+        assert result is existing
+        assert result.pred_home_goals == 1.0, "pred_home_goals 不应被覆盖"
+        assert result.reasoning == "original", "reasoning 不应被覆盖"
+        assert result.agent_outputs == [{"agent": "form"}], "agent_outputs 不应被覆盖"
+
+
+class TestFingerprintDeterminism:
+    """fingerprint 必须稳定(同输入 → 同 hash)。"""
+
+    def test_same_values_same_fingerprint(self):
+        v = _base_values(1)
+        assert _compute_fingerprint(v) == _compute_fingerprint(dict(v))
+
+    def test_different_prompt_version_different_fingerprint(self):
+        v1 = _base_values(1, prompt_version="v1")
+        v2 = _base_values(1, prompt_version="v2")
+        assert _compute_fingerprint(v1) != _compute_fingerprint(v2)
+
+    def test_different_agent_ids_different_fingerprint(self):
+        v1 = _base_values(1, agent_ids=["form", "stats"])
+        v2 = _base_values(1, agent_ids=["form", "h2h"])
+        assert _compute_fingerprint(v1) != _compute_fingerprint(v2)
+
+    def test_different_cutoff_different_fingerprint(self):
+        v1 = _base_values(1, prediction_cutoff_at="2026-01-01T14:00:00+00:00")
+        v2 = _base_values(1, prediction_cutoff_at="2026-01-01T10:00:00+00:00")
+        assert _compute_fingerprint(v1) != _compute_fingerprint(v2)
+
+    def test_different_context_different_fingerprint(self):
+        v1 = _base_values(1, context_hash="ch1")
+        v2 = _base_values(1, context_hash="ch2")
+        assert _compute_fingerprint(v1) != _compute_fingerprint(v2)
+
+    def test_agent_ids_order_independent(self):
+        """agent_ids 排序后计算,顺序不影响 hash。"""
+        v1 = _base_values(1, agent_ids=["stats", "form"])
+        v2 = _base_values(1, agent_ids=["form", "stats"])
+        assert _compute_fingerprint(v1) == _compute_fingerprint(v2)

tests/test_p0_score_status.py

@@ -0,0 +1,124 @@
+"""P0-01 回归测试: missing score 不得变 0:0。
+
+运行: pytest tests/test_p0_score_status.py -v
+(无需真实 PG;用 fake DB + 模型元数据断言。)
+"""
+import pytest
+
+from src.db.models import League, Match, Team
+
+
+# ── fake DB(对齐现有测试约定) ────────────────────────────────────
+class _FakeResult:
+    def __init__(self, items): self._items = list(items)
+    def scalars(self): return self
+    def all(self): return list(self._items)
+    def scalar_one_or_none(self): return self._items[0] if self._items else None
+
+
+class _FakeDB:
+    def __init__(self): self.added = []
+    def add(self, obj): self.added.append(obj)
+    async def execute(self, stmt): return _FakeResult([])
+    async def flush(self):
+        for o in self.added:
+            if getattr(o, "id", None) is None:
+                o.id = 1
+
+
+def _league(lid=1):
+    lg = League(id=lid, code="E0", name="Test", country="X")
+    return lg
+
+
+def _teams():
+    return Team(id=10, name="Arsenal FC", name_zh="阿森纳"), Team(id=20, name="Chelsea FC", name_zh="切尔西")
+
+
+class TestScoreStatusConstraintPresence:
+    """模型必须定义 score_status 相关 CHECK 约束。"""
+
+    def test_score_status_column_exists(self):
+        cols = {c.name for c in Match.__table__.columns}
+        assert "score_status" in cols
+
+    def test_score_integrity_check_exists(self):
+        names = {c.name for c in Match.__table__.constraints if c.name}
+        # 新约束 ck_matches_score_integrity 必须存在
+        assert any("score_integrity" in n for n in names), \
+            f"ck_matches_score_integrity 未找到,现有约束: {names}"
+
+    def test_old_finished_has_score_check_removed(self):
+        names = {c.name for c in Match.__table__.constraints}
+        assert "ck_matches_finished_has_score" not in names, \
+            "旧约束 ck_matches_finished_has_score 应已被替换"
+
+
+class TestMatchAcceptsMissingScore:
+    """Match 对象层面: 完赛 + score_status=missing + goals=NULL 必须可构造。"""
+
+    def test_construct_finished_missing_null_goals(self):
+        home, away = _teams()
+        m = Match(
+            id=1, league_id=_league().id, home_team_id=home.id, away_team_id=away.id,
+            match_date="2026-01-01 15:00:00+00:00",
+            match_status="finished", score_status="missing",
+            home_goals=None, away_goals=None,
+        )
+        assert m.home_goals is None
+        assert m.away_goals is None
+        assert m.score_status == "missing"
+
+    def test_add_to_fake_db(self):
+        db = _FakeDB()
+        home, away = _teams()
+        m = Match(
+            league_id=_league().id, home_team_id=home.id, away_team_id=away.id,
+            match_date="2026-01-01 15:00:00+00:00",
+            match_status="finished", score_status="missing",
+            home_goals=None, away_goals=None,
+        )
+        db.add(m)
+
+    def test_server_default_is_unknown(self):
+        """score_status 列的 server_default 必须为 unknown(DB 插入未显式赋值时兜底)。"""
+        col = Match.__table__.c.score_status
+        assert col.server_default is not None
+        assert "unknown" in str(col.server_default.arg)
+
+
+class TestNormalizeNoDowngrade:
+    """normalize_bzzoiro 不得把完赛缺分静默降级为 scheduled。"""
+
+    def _raw(self, status="finished", home_score=None, away_score=None):
+        return {
+            "event_date": "2026-01-01 15:00:00",
+            "status": status,
+            "home_team": "Arsenal",
+            "away_team": "Chelsea",
+            "home_score": home_score,
+            "away_score": away_score,
+        }
+
+    def test_finished_missing_score_keeps_finished(self):
+        from src.data.normalize import normalize_bzzoiro
+        m = normalize_bzzoiro(self._raw("finished", None, None), "E0")
+        assert m is not None
+        assert m.match_status == "finished", "完赛缺分不得降级为 scheduled"
+        assert m.score_status == "missing"
+        assert m.home_goals is None
+        assert m.away_goals is None
+
+    def test_finished_with_score_is_known(self):
+        from src.data.normalize import normalize_bzzoiro
+        m = normalize_bzzoiro(self._raw("finished", 2, 1), "E0")
+        assert m.match_status == "finished"
+        assert m.score_status == "known"
+        assert m.home_goals == 2 and m.away_goals == 1
+
+    def test_scheduled_no_score_is_unknown(self):
+        from src.data.normalize import normalize_bzzoiro
+        m = normalize_bzzoiro(self._raw("scheduled", None, None), "E0")
+        assert m.match_status == "scheduled"
+        assert m.score_status == "unknown"
+        assert m.home_goals is None

tests/test_p0_standings_cutoff.py

@@ -0,0 +1,138 @@
+"""P0-02 回归测试: 积分榜改为追加快照(append-only) + available_at cutoff。
+
+运行: pytest tests/test_p0_standings_cutoff.py -v
+(模型约束用 fake DB;cutoff 过滤语义用 fake session 验证参数传递。)
+"""
+import pytest
+from sqlalchemy import select
+
+from src.db.models import League, Standing, Team
+
+
+# ── fake DB(对齐现有测试约定) ────────────────────────────────────
+class _FakeResult:
+    def __init__(self, items): self._items = list(items)
+    def scalars(self):
+        class _S:
+            def __init__(self, items): self._items = items
+            def all(self): return list(self._items)
+        return _S(self._items)
+    def scalar_one_or_none(self):
+        return self._items[0] if self._items else None
+
+
+class _FakeDB:
+    captured: list = []
+
+    def __init__(self, league=None, standing_rows=None):
+        self._league = league
+        self._rows = standing_rows or []
+        _FakeDB.captured = []
+
+    def add(self, obj):
+        _FakeDB.captured.append(obj)
+
+    async def execute(self, stmt):
+        # 记录生成的 SQL(字符串化)供断言
+        _FakeDB.captured.append(str(stmt))
+        compiled = str(stmt)
+        if "league" in compiled.lower() and "standing" not in compiled.lower():
+            return _FakeResult([self._league] if self._league else [])
+        return _FakeResult(self._rows)
+
+    async def flush(self):
+        pass
+
+
+class _FakeTeam:
+    def __init__(self, tid, name):
+        self.id = tid
+        self.name = name
+        self.name_zh = None
+
+
+class _Header:
+    def __init__(self):
+        from src.llm.slices.common import MatchHeader
+        self._h = MatchHeader(
+            match_id=1, home_name="A", away_name="B", league_name="E0",
+            season="2026", match_date="2026-01-01", match_dt=None,
+            stage=None, home_team_id=10, away_team_id=20, league_id=1,
+        )
+
+    def __getattr__(self, name):
+        return getattr(self._h, name)
+
+
+class TestStandingsModel:
+    """standings 模型必须有 available_at + 新唯一约束。"""
+
+    def test_available_at_column(self):
+        cols = {c.name for c in Standing.__table__.columns}
+        assert "available_at" in cols
+
+    def test_unique_constraint_includes_available_at(self):
+        names = {c.name for c in Standing.__table__.constraints}
+        assert any("available" in n and n.startswith("uq_") for n in names), \
+            f"缺少含 available_at 的唯一约束,现有: {names}"
+
+    def test_old_unique_constraint_removed(self):
+        names = {c.name for c in Standing.__table__.constraints}
+        assert "uq_standings_league_season_team" not in names, \
+            "旧约束 uq_standings_league_season_team 应已被替换"
+
+
+class TestStandingsSliceCutoff:
+    """standings_slice 必须尊重 before(cutoff):before=None → now()。"""
+
+    def test_before_none_uses_now(self):
+        """before=None 时应将 cutoff 视为 now()(取最新可用快照)。"""
+        from src.llm.slices import standings as st_mod
+        from datetime import datetime, timezone
+
+        calls = {}
+        real_execute = None
+
+        class _DB:
+            def __init__(self): self._league = League(id=1, code="E0", name="E0")
+            def add(self, obj): pass
+            async def execute(self, stmt):
+                # 捕获 WHERE available_at <= ? 的参数
+                sql = str(stmt)
+                if "available_at" in sql:
+                    # 提取编译后的 params
+                    try:
+                        params = stmt.compile().params
+                        calls["cutoff"] = params.get("available_at_1")
+                    except Exception:
+                        pass
+                if "league" in sql.lower() and "standing" not in sql.lower():
+                    return _FakeResult([self._league])
+                return _FakeResult([])
+            async def flush(self): pass
+
+        async def run():
+            db = _DB()
+            before = None
+            await st_mod.standings_slice(_Header(), before=before, db=db)
+
+        import asyncio
+        asyncio.run(run())
+        # before=None 时应注入 now() 作为 cutoff
+        assert "cutoff" in calls, "未对 available_at 施加 cutoff 过滤"
+        assert calls["cutoff"] is not None
+
+
+class TestStandingsAppendOnly:
+    """采集应 INSERT 新行(带 available_at),不覆盖旧行。"""
+
+    def test_values_include_available_at(self):
+        """采集构造的 Standing 必须含 available_at 字段。"""
+        from src.data import bzzoiro_standings as bzs
+        # 检查函数源码是否包含 available_at(编译期守卫)
+        import inspect
+        src = inspect.getsource(bzs.ingest_bzzoiro_standings)
+        assert "available_at" in src, "采集函数必须设置 available_at"
+        # 不应再出现按 (league, season, team) 的 upsert 查询
+        assert "scalar_one_or_none" not in src or "Standing.league_id == league.id" not in src.replace("available_at", ""), \
+            "不应再按 (league, season, team) 做 upsert 查询"

tests/test_p1_a_ingest_league_counts.py

@@ -0,0 +1,67 @@
+"""P1-A 回归测试: ingest 联赛级 inserted/updated 必须读 leagues[code],而非顶层 r.get("inserted")。
+
+运行: pytest tests/test_p1_a_ingest_league_counts.py -v
+(纯函数测试,无 DB/网络依赖。)
+"""
+from src.api.routes.ingest import _accumulate_ingest_result
+
+
+class TestAccumulateIngestResult:
+    """P1-A: _accumulate_ingest_result 联赛级计数必须来自 r["leagues"][code]。"""
+
+    def _merged(self):
+        return {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
+
+    def test_league_counts_read_from_leagues_key(self):
+        """核心: 联赛级 inserted/updated 应来自 leagues[code],而非顶层 inserted/updated。"""
+        merged = self._merged()
+        r = {
+            # 顶层无 inserted/updated 键(只有 total_*)
+            "total_inserted": 5,
+            "total_updated": 2,
+            "errors": [],
+            "leagues": {"E0": {"inserted": 3, "updated": 1, "rows": 4, "errors": []}},
+        }
+        _accumulate_ingest_result(merged, "E0", r)
+
+        # 顶层总计
+        assert merged["total_inserted"] == 5
+        assert merged["total_updated"] == 2
+        # 联赛级计数来自 leagues["E0"],而非顶层
+        assert merged["leagues"]["E0"]["inserted"] == 3, "联赛 inserted 必须来自 leagues[code]"
+        assert merged["leagues"]["E0"]["updated"] == 1, "联赛 updated 必须来自 leagues[code]"
+
+    def test_does_not_read_top_level_inserted(self):
+        """防御: 若 r 误含顶层 inserted 键,不得影响联赛级计数。"""
+        merged = self._merged()
+        r = {
+            "total_inserted": 5,
+            "total_updated": 2,
+            "inserted": 999,  # 错误的顶层键(旧代码可能读这个)
+            "updated": 999,
+            "errors": [],
+            "leagues": {"E0": {"inserted": 3, "updated": 1}},
+        }
+        _accumulate_ingest_result(merged, "E0", r)
+        # 必须忽略顶层 inserted/updated,使用 leagues["E0"]
+        assert merged["leagues"]["E0"]["inserted"] == 3
+        assert merged["leagues"]["E0"]["updated"] == 1
+
+    def test_missing_league_key_defaults_to_zero(self):
+        """r["leagues"] 无该 code 时,默认 0 不抛错。"""
+        merged = self._merged()
+        r = {"total_inserted": 1, "total_updated": 0, "errors": [], "leagues": {}}
+        _accumulate_ingest_result(merged, "E0", r)
+        assert merged["leagues"]["E0"]["inserted"] == 0
+        assert merged["total_inserted"] == 1
+
+    def test_multiple_calls_accumulate(self):
+        """多次调用应累加到同一联赛。"""
+        merged = self._merged()
+        r1 = {"total_inserted": 3, "total_updated": 1, "errors": [], "leagues": {"E0": {"inserted": 3, "updated": 1}}}
+        r2 = {"total_inserted": 2, "total_updated": 0, "errors": [], "leagues": {"E0": {"inserted": 2, "updated": 0}}}
+        _accumulate_ingest_result(merged, "E0", r1)
+        _accumulate_ingest_result(merged, "E0", r2)
+        assert merged["leagues"]["E0"]["inserted"] == 5
+        assert merged["leagues"]["E0"]["updated"] == 1
+        assert merged["total_inserted"] == 5

tests/test_p1_b_invalid_cursor.py

@@ -0,0 +1,75 @@
+"""P1-B 回归测试: 非法 cursor → 400 + code=INVALID_CURSOR。
+
+运行: pytest tests/test_p1_b_invalid_cursor.py -v
+(_parse_cursor 为纯函数,无 DB/网络依赖;HTTP 层仅测非法格式。)
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from src.api.app import app
+from src.api.deps import require_admin
+from src.api.routes.matches import _parse_cursor
+
+
+class TestParseCursorPure:
+    """P1-B 纯函数:_parse_cursor 解析与非法校验。"""
+
+    def test_valid_cursor(self):
+        d, mid = _parse_cursor("2026-01-01T15:00:00+00:00|42")
+        assert d == datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc)
+        assert mid == 42
+
+    def test_missing_pipe_raises_400(self):
+        with pytest.raises(HTTPException) as ei:
+            _parse_cursor("no-pipe-here")
+        assert ei.value.status_code == 400
+        assert ei.value.detail["code"] == "INVALID_CURSOR"
+
+    def test_empty_date_raises_400(self):
+        with pytest.raises(HTTPException) as ei:
+            _parse_cursor("|5")
+        assert ei.value.status_code == 400
+        assert ei.value.detail["code"] == "INVALID_CURSOR"
+
+    def test_non_numeric_id_raises_400(self):
+        with pytest.raises(HTTPException) as ei:
+            _parse_cursor("2026-01-01T00:00:00+00:00|abc")
+        assert ei.value.status_code == 400
+        assert ei.value.detail["code"] == "INVALID_CURSOR"
+
+    def test_invalid_date_raises_400(self):
+        with pytest.raises(HTTPException) as ei:
+            _parse_cursor("not-a-date|1")
+        assert ei.value.status_code == 400
+        assert ei.value.detail["code"] == "INVALID_CURSOR"
+
+    def test_extra_pipe_raises_400(self):
+        """含额外 | 时 id 部分为 "42|extra",int() 失败 → 400。"""
+        with pytest.raises(HTTPException) as ei:
+            _parse_cursor("2026-01-01T15:00:00+00:00|42|extra")
+        assert ei.value.status_code == 400
+        assert ei.value.detail["code"] == "INVALID_CURSOR"
+
+
+class TestInvalidCursorHTTP:
+    """P1-B HTTP 层:非法 cursor → 400 + code=INVALID_CURSOR。"""
+
+    @pytest.fixture
+    def client(self):
+        app.dependency_overrides[require_admin] = lambda: None
+        return TestClient(app)
+
+    def test_malformed_cursor_400(self, client):
+        resp = client.get("/api/v1/matches?cursor=garbage")
+        assert resp.status_code == 400
+        assert resp.json()["detail"]["code"] == "INVALID_CURSOR"
+
+    def test_missing_id_400(self, client):
+        resp = client.get("/api/v1/matches?cursor=2026-01-01T00:00:00|")
+        assert resp.status_code == 400
+        assert resp.json()["detail"]["code"] == "INVALID_CURSOR"

tests/test_p1_c_public_predictions.py

@@ -0,0 +1,135 @@
+"""P1-C 回归测试: 公开预测仅 live+success,且不含 reasoning/agent_outputs。
+
+运行: pytest tests/test_p1_c_public_predictions.py -v
+(使用 fake DB,无需真实 PG。)
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.api.app import app
+from src.api.deps import require_admin
+from src.db.models import League, Match, MatchStats, Prediction, Team
+
+
+class _FakeResult:
+    def __init__(self, items): self._items = list(items)
+    def scalars(self):
+        class _S:
+            def __init__(self, items): self._items = items
+            def all(self): return list(self._items)
+        return _S(self._items)
+    def scalar_one_or_none(self):
+        return self._items[0] if self._items else None
+    def scalar(self):
+        return self._items[0] if self._items else None
+
+
+class _FakeDB:
+    """假 DB:捕获发往 Prediction 的查询语句,供测试断言 SQL 过滤条件。"""
+
+    captured_pred_stmts: list = []
+
+    def __init__(self, match=None, predictions=()):
+        self._match = match
+        self._predictions = list(predictions)
+
+    async def execute(self, stmt):
+        # 根据 column_descriptions 判断查询实体
+        try:
+            entity = stmt.column_descriptions[0]["entity"]
+        except (IndexError, KeyError):
+            entity = None
+        if entity is Prediction:
+            _FakeDB.captured_pred_stmts.append(stmt)
+            return _FakeResult(self._predictions)
+        return _FakeResult([self._match] if self._match else [])
+
+    async def get(self, cls, mid):
+        return self._match
+
+
+def _make_match(mid=1):
+    home = Team(id=10, name="Arsenal", name_zh="阿森纳")
+    away = Team(id=20, name="Chelsea", name_zh="切尔西")
+    lg = League(id=1, code="E0", name="Premier", country="EN")
+    m = Match(
+        id=mid, league_id=1, home_team_id=10, away_team_id=20,
+        match_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc),
+        match_status="finished",
+    )
+    m.league = lg
+    m.home_team = home
+    m.away_team = away
+    m.stats = MatchStats(match_id=mid, home_xg=1.5, away_xg=1.0)
+    return m
+
+
+def _make_pred(pid, match_id, run_type="live", status="success", **overrides):
+    p = Prediction(
+        id=pid, match_id=match_id, provider="openai", model="gpt-4o",
+        prompt_version="v1", mode=run_type, run_type=run_type, status=status,
+        pred_home_goals=2.0, pred_away_goals=1.0, pred_1x2="1",
+        reasoning="内部推理细节", agent_outputs=[{"agent": "form"}],
+        subjective_confidence=0.7,
+        created_at=datetime(2026, 1, 2, 12, 0, tzinfo=timezone.utc),
+    )
+    for k, v in overrides.items():
+        setattr(p, k, v)
+    return p
+
+
+from src.db.base import get_db_read
+
+
+@pytest.fixture
+def client():
+    app.dependency_overrides[require_admin] = lambda: None
+    return TestClient(app)
+
+
+class TestPublicPredictionsFilter:
+    """P1-C: GET /matches/{id} 公开预测仅 run_type=live 且 status=success。"""
+
+    def test_query_filters_by_run_type_and_status(self, client):
+        """P1-C: 查询必须包含 run_type='live' AND status='success' 过滤。"""
+        _FakeDB.captured_pred_stmts = []
+        m = _make_match(1)
+        fake = _FakeDB(match=m, predictions=[_make_pred(1, 1)])
+
+        app.dependency_overrides[get_db_read] = lambda: fake
+        try:
+            resp = client.get("/api/v1/matches/1")
+            assert resp.status_code == 200, resp.text
+            # 验证发往 Prediction 的 SQL 含 run_type 与 status 过滤
+            assert _FakeDB.captured_pred_stmts, "未发出 Prediction 查询"
+            sql = str(_FakeDB.captured_pred_stmts[0]).lower()
+            assert "run_type" in sql, f"SQL 缺少 run_type 过滤: {sql}"
+            assert "status" in sql, f"SQL 缺少 status 过滤: {sql}"
+        finally:
+            app.dependency_overrides.pop(get_db_read, None)
+
+    def test_no_reasoning_or_agent_outputs(self, client):
+        """P1-C: 公开预测不得含 reasoning/agent_outputs。"""
+        m = _make_match(1)
+        preds = [_make_pred(1, 1, run_type="live", status="success")]
+        fake = _FakeDB(match=m, predictions=preds)
+
+        app.dependency_overrides[get_db_read] = lambda: fake
+        try:
+            resp = client.get("/api/v1/matches/1")
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            assert len(body["recent_predictions"]) == 1
+            p = body["recent_predictions"][0]
+            assert "reasoning" not in p, "公开预测不得含 reasoning"
+            assert "agent_outputs" not in p, "公开预测不得含 agent_outputs"
+            # 但核心字段保留
+            assert p["pred_home_goals"] == 2.0
+            assert p["pred_1x2"] == "1"
+        finally:
+            app.dependency_overrides.pop(get_db_read, None)

tests/test_p1_d_concurrency.py

@@ -0,0 +1,71 @@
+"""P1-D 回归测试: 全局 LLM 并发限制 + provider 字段已删除。
+
+运行: pytest tests/test_p1_d_concurrency.py -v
+"""
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from src.api.routes import predict as predict_mod
+
+
+class TestGlobalLLMConcurrency:
+    """P1-D: 全局 LLM 并发限制(默认 4)。"""
+
+    def test_semaphore_exists_with_limit(self):
+        """路由模块必须存在 _GLOBAL_LLM_SEMAPHORE 且 value <= 4。"""
+        assert hasattr(predict_mod, "_GLOBAL_LLM_SEMAPHORE")
+        sem = predict_mod._GLOBAL_LLM_SEMAPHORE
+        assert isinstance(sem, asyncio.Semaphore)
+        assert sem._value == 4, f"期望并发限制 4,实际 {sem._value}"
+
+    def test_predict_with_concurrency_limits_parallel(self):
+        """P1-D: 并发调用 _predict_with_concurrency 不得超过信号量限制。"""
+        max_concurrent = 0
+        current = 0
+        lock = asyncio.Lock()
+
+        async def fake_predict(match_id, **kwargs):
+            nonlocal current, max_concurrent
+            async with lock:
+                current += 1
+                max_concurrent = max(max_concurrent, current)
+            await asyncio.sleep(0.05)  # 模拟 LLM 调用
+            async with lock:
+                current -= 1
+            return type("R", (), {"prediction_id": 1, "provider": "p", "model": "m",
+                                  "prompt_version": "v1", "pred_home_goals": 1.0,
+                                  "pred_away_goals": 0.0, "pred_1x2": "1",
+                                  "subjective_confidence": 0.5, "reasoning": "",
+                                  "status": "success", "context": "",
+                                  "latency_ms": 0, "raw": {}})()
+
+        req = type("Req", (), {"match_id": 1, "model": None, "prompt_version": None, "mode": "single"})()
+
+        async def run():
+            with patch.object(predict_mod, "predict_match", fake_predict):
+                # 启动 10 个并发请求
+                tasks = [predict_mod._predict_with_concurrency(req) for _ in range(10)]
+                await asyncio.gather(*tasks)
+
+        asyncio.run(run())
+        # 最大并发不得超过信号量限制(4)
+        assert max_concurrent <= 4, f"并发 {max_concurrent} 超过限制 4"
+
+
+class TestProviderFieldRemoved:
+    """P1-D: PredictRequest 的 provider 字段必须已删除(未接线)。"""
+
+    def test_predict_request_no_provider(self):
+        from src.api.schemas import PredictRequest
+
+        fields = set(PredictRequest.model_fields.keys())
+        assert "provider" not in fields, f"PredictRequest 应已删除 provider 字段,现有: {fields}"
+
+    def test_predict_request_still_has_core_fields(self):
+        from src.api.schemas import PredictRequest
+
+        fields = set(PredictRequest.model_fields.keys())
+        for required in ("match_id", "model", "prompt_version", "mode"):
+            assert required in fields, f"缺少核心字段 {required}"

tests/test_prediction_unique_constraint.py

@@ -3,7 +3,7 @@
 验证:
   1. 唯一约束包含 mode + run_type
   2. 同一场比赛 live 与 backtest 预测可共存,互不覆盖
-  3. _upsert_prediction 正确区分 run_type
+  3. _insert_or_find_by_fingerprint 正确区分 run_type
 """
 from __future__ import annotations
 
@@ -12,6 +12,7 @@ from pathlib import Path
 
 from pydantic import BaseModel
 import pytest
+from sqlalchemy import Index
 
 from src.db.models import Prediction, UniqueConstraint, CheckConstraint
 
@@ -22,20 +23,32 @@ MIGRATION_PATH = REPO_ROOT / "alembic" / "versions" / "0013_predictions_unique_c
 
 
 class TestUniqueConstraint:
-    """验证唯一约束包含 mode + run_type。"""
+    """P0-03: 验证幂等指纹唯一索引(替代旧 (match, provider, model, mode, run_type) 唯一约束)。"""
 
-    def test_constraint_columns(self):
-        """唯一约束应包含 match_id, provider, model, mode, run_type。"""
-        uc = [
-            c for c in Prediction.__table__.constraints
-            if isinstance(c, UniqueConstraint) and "match" in c.name
+    def test_input_hash_partial_unique_index(self):
+        """P0-03: input_hash 非空时必须唯一(同指纹 → 返回已有行,不 UPDATE/INSERT)。"""
+        idx = [
+            i for i in Prediction.__table__.indexes
+            if i.unique and "input_hash" in i.name
         ]
-        assert len(uc) == 1
-        cols = [c.name for c in uc[0].columns]
-        assert cols == ["match_id", "provider", "model", "mode", "run_type"]
+        assert len(idx) == 1, f"缺少 input_hash partial unique 索引,现有 indexes: {[i.name for i in Prediction.__table__.indexes]}"
+        # partial unique: postgresql_where 必须限制 input_hash IS NOT NULL
+        assert idx[0].dialect_kwargs.get("postgresql_where") is not None
+
+    def test_old_unique_constraint_removed(self):
+        """P0-03: 旧 (match, provider, model, mode, run_type) 唯一约束必须已移除。"""
+        from sqlalchemy import UniqueConstraint
+
+        old = [
+            c for c in Prediction.__table__.constraints
+            if isinstance(c, UniqueConstraint) and c.name == "uq_predictions_match_provider_model_mode_run_type"
+        ]
+        assert len(old) == 0, f"旧约束必须已移除,但仍存在: {[c.name for c in old]}"
 
     def test_run_type_check_constraint(self):
         """应有 run_type 的 check constraint。"""
+        from sqlalchemy import CheckConstraint
+
         cc = [
             c for c in Prediction.__table__.constraints
             if isinstance(c, CheckConstraint) and "run_type" in c.name
@@ -52,13 +65,16 @@ class TestUniqueConstraint:
 
 
 class TestUpsertPredictionSignature:
-    """验证 _upsert_prediction 函数签名包含 run_type。"""
+    """验证 _insert_or_find_by_fingerprint 签名(P0-03 指纹模式)。"""
 
-    def test_signature_has_run_type(self):
-        from src.llm.predict import _upsert_prediction
+    def test_signature_uses_values_dict(self):
+        """P0-03: 新接口通过 values dict 接收全部字段(含 run_type/match_id/...)。"""
+        from src.llm.predict import _insert_or_find_by_fingerprint
 
-        sig = inspect.signature(_upsert_prediction)
-        assert "run_type" in sig.parameters
+        sig = inspect.signature(_insert_or_find_by_fingerprint)
+        params = sig.parameters
+        assert "session" in params
+        assert "values" in params  # 所有业务字段走 values dict
 
     def test_signature_has_backtest_in_predict_match(self):
         from src.llm.predict import predict_match

tests/test_review_required_fixes.py

@@ -194,8 +194,8 @@ async def test_r2_standings_actually_upserts(monkeypatch):
     assert first.zone == "Champions League"     # 优先取 label
 
 
-async def test_r2_standings_upsert_updates_existing(monkeypatch):
-    """行为测试: 已存在同 (league, season, team) 时应就地更新而非新增。"""
+async def test_r2_standings_append_new_row(monkeypatch):
+    """P0-02 行为测试: 每次采集 INSERT 新行(带 available_at),不更新旧行。"""
     import src.data.bzzoiro as bz
     from src.db.models import League, Standing
 
@@ -217,16 +217,20 @@ async def test_r2_standings_upsert_updates_existing(monkeypatch):
     existing = Standing(league_id=42, season="2025-2026", team_id=7, position=9)
     existing.points = 1
 
-    # 查询顺序: League → Team 预载(命中) → Standing 查询(命中已有行)
-    db = _FakeDb(results=[_FakeResult([league]), _FakeResult([team]), _FakeResult([existing])])
+    # 查询顺序: League(命中) → Team 预载(命中) → (P0-02 不再查询 Standing)
+    db = _FakeDb(results=[_FakeResult([league]), _FakeResult([team])])
 
     result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"])
 
     assert result["total_upserted"] == 1
-    assert existing.points == 30, "已有行应被就地更新"
+    # P0-02: 追加快照——新增 Standing 行,旧行不被修改
+    new_rows = [o for o in db.added if isinstance(o, Standing)]
+    assert len(new_rows) == 1, "P0-02 应新增一条 Standing 行"
+    assert new_rows[0].points == 30, "新行应承载新采集数据"
+    assert new_rows[0].available_at is not None, "新行必须含 available_at"
+    # 旧行未被修改(仍保持原值)
+    assert existing.points == 1, "P0-02 旧行不应被覆盖"
     assert result["leagues"]["EPL"]["teams_created"] == 0
-    # 不应新增 Standing(只有 league/team 层面的 add)
-    assert not [o for o in db.added if isinstance(o, Standing)]
 
 
 def test_r2_source_contains_real_upsert_loop():
@@ -237,7 +241,9 @@ def test_r2_source_contains_real_upsert_loop():
     assert "total_upserted" in src
     assert 'result["total_upserted"] +=' in src, "total_upserted 必须真的被累加"
     assert "Standing(" in src, "必须真的构造 Standing"
-    assert "select(Standing)" in src, "必须查询已有快照以决定 insert/update"
+    # P0-02: 追加快照——每次 INSERT 新行(带 available_at),不查询旧行做 upsert
+    assert "available_at" in src, "P0-02 采集必须设置 available_at"
+    assert "scalar_one_or_none" not in src, "P0-02 不应再按 (league, season, team) 做 upsert 查询"
 
 
 # ============================================================