fix(P0-02): 积分榜改为追加快照(append-only) + available_at cutoff

去掉 uq_standings_league_season_team,改为 (league, season, team, available_at) 唯一;
每次采集 INSERT 新行(available_at=now),ON CONFLICT DO NOTHING,不覆盖旧行。

standings_slice(before):DISTINCT ON (team_id) WHERE available_at<=cutoff
ORDER available_at DESC;before=None → cutoff=now()。
公开 list_standings 取每队最新可用快照(子查询 max available_at)。

迁移 0023 + 切片/路由/docs 同步;测试 test_p0_standings_cutoff(5/5)。
284 测试全绿。

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

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/matches.py

@@ -286,7 +286,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 +301,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/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,
+            # 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, available_at=now, **values
             )
-            standing = (await db.execute(stmt)).scalar_one_or_none()
-            if standing is None:
-                standing = Standing(
-                    league_id=league.id, season=season_label, team_id=team.id, **values
-                )
-                db.add(standing)
-            else:
-                for k, v in values.items():
-                    setattr(standing, k, v)
+            db.add(standing)
             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

@@ -88,6 +88,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 +130,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 +204,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 +231,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"),
     )
 
 

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/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_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_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 查询"
 
 
 # ============================================================