feat(HEL-478): 估值发布后晚间复核并原子追补上游修订

盘后成功发布后继续轻量比对 daily_basic 网站字段，发现修订才走质量门与整组原子切换，避免 17:10 快照落后于晚间上游改写。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

backend/data/datahub/bridge.py

@@ -122,10 +122,12 @@ class DatahubBridge:
                 legacy_rows = legacy_query(api_name, params, fields)
             except Exception as exc:
                 if flags.read and hub_rows is not None and hub_error is None:
-                    self._emit_shadow(compare_rows(dataset, [], hub_canonical, hub_meta, self._error_text(exc)))
+                    self._emit_shadow(
+                        compare_rows(dataset, [], hub_canonical, hub_meta, self._error_text(exc), fields)
+                    )
                     return project_fields(hub_rows, fields)
                 raise
-            self._emit_shadow(compare_rows(dataset, legacy_rows, hub_canonical, hub_meta, hub_error))
+            self._emit_shadow(compare_rows(dataset, legacy_rows, hub_canonical, hub_meta, hub_error, fields))
             if flags.read and hub_rows is not None and hub_error is None:
                 return project_fields(hub_rows, fields)
             return legacy_rows

backend/data/datahub/compare.py

@@ -5,6 +5,7 @@ from typing import Any
 from backend.data.datahub.native import SCALE_FIELDS, row_key, to_canonical_row, yyyymmdd
 
 NUMERIC_TOLERANCE = 1e-4
+CANONICAL_ALIASES = {"volume": "vol"}
 
 
 def compare_rows(
@@ -13,8 +14,10 @@ def compare_rows(
     hub_rows: list[dict[str, Any]] | None,
     hub_meta: dict[str, Any] | None = None,
     hub_error: str | None = None,
+    fields: str = "",
 ) -> dict[str, Any]:
     hub = hub_rows or []
+    requested = _requested_fields(fields)
     legacy_map = {row_key(dataset, row): row for row in legacy_rows}
     hub_map = {row_key(dataset, _align_hub_row(row)): row for row in hub}
     missing_hub = sorted(key for key in legacy_map if key not in hub_map)
@@ -26,7 +29,7 @@ def compare_rows(
         hub_row = hub_map.get(key)
         if hub_row is None:
             continue
-        field_report = _compare_fields(dataset, legacy, hub_row)
+        field_report = _compare_fields(dataset, legacy, hub_row, requested)
         if field_report["unit_conversion"]:
             unit_conversion.append({"key": list(key), "fields": field_report["unit_conversion"]})
         if field_report["value_diff"]:
@@ -53,6 +56,7 @@ def compare_rows(
         "published_at": (hub_meta or {}).get("published_at"),
         "trade_date": yyyymmdd((hub_meta or {}).get("trade_date")),
         "hub_error": hub_error,
+        "fields_compared": sorted(requested) if requested is not None else None,
         "equal": (
             not hub_error
             and not missing_hub
@@ -71,13 +75,36 @@ def _align_hub_row(row: dict[str, Any]) -> dict[str, Any]:
     return aligned
 
 
-def _compare_fields(dataset: str, legacy: dict[str, Any], hub: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
+def _requested_fields(fields: str) -> list[str] | None:
+    """Fields the website actually asked for; None means "no projection"."""
+    keys = [item.strip() for item in str(fields or "").split(",") if item.strip()]
+    if not keys:
+        return None
+    seen: list[str] = []
+    for key in keys:
+        canonical = CANONICAL_ALIASES.get(key, key)
+        if canonical not in seen:
+            seen.append(canonical)
+    return seen
+
+
+def _compare_fields(
+    dataset: str,
+    legacy: dict[str, Any],
+    hub: dict[str, Any],
+    requested: list[str] | None = None,
+) -> dict[str, list[dict[str, Any]]]:
     canonical_legacy = to_canonical_row(dataset, legacy)
     hub_canonical = _hub_canonical(dataset, hub)
     native_hub = _align_hub_row(hub)
     value_diff: list[dict[str, Any]] = []
     unit_conversion: list[dict[str, Any]] = []
     keys = (set(canonical_legacy) | set(hub_canonical)) - {"batch_id", "updated_at", "volume"}
+    if requested is not None:
+        # Compare only what the website asked for. Extra hub columns are
+        # transport detail, not business differences; a requested field still
+        # alarms when it is missing or holds a different value.
+        keys = set(requested) - {"batch_id", "updated_at", "volume"}
     scales = SCALE_FIELDS.get(dataset) or {}
     for field in sorted(keys):
         left = canonical_legacy.get(field)

tests/test_datahub_bridge.py

@@ -201,6 +201,87 @@ class DatahubBridgeTests(unittest.TestCase):
         skew = compare_rows("daily", [LEGACY_DAILY], [HUB_DAILY], {"stale": False, "staleness_seconds": 12})
         self.assertTrue(skew["time_skew"])
 
+    def test_shadow_extra_hub_columns_are_not_false_diffs_when_projected(self) -> None:
+        hub_full = {**HUB_DAILY, "adj_factor": 1.1}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close")}
+        report = compare_rows(
+            "daily", [legacy_close_only], [hub_full],
+            {"stale": False, "staleness_seconds": 0},
+            fields="ts_code,trade_date,close",
+        )
+        self.assertTrue(report["equal"])
+        self.assertEqual(report["value_diff_count"], 0)
+        self.assertEqual(report["fields_compared"], ["close", "trade_date", "ts_code"])
+        # without projection the same pair shows the historic false diff
+        unprojected = compare_rows("daily", [legacy_close_only], [hub_full])
+        self.assertFalse(unprojected["equal"])
+
+        legacy_stocks = {"ts_code": "600000.SH", "name": "浦发银行"}
+        hub_stocks = {
+            "ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+            "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110",
+        }
+        stocks = compare_rows("stocks", [legacy_stocks], [hub_stocks], {}, fields="ts_code,name")
+        self.assertTrue(stocks["equal"])
+
+        legacy_cal = {"cal_date": "20240902", "is_open": 1}
+        hub_cal = {
+            "cal_date": "20240902", "is_open": True,
+            "pretrade_date": "20240830", "prev_open": "20240830",
+        }
+        calendar = compare_rows(
+            "calendar", [legacy_cal], [hub_cal], {}, fields="cal_date,is_open"
+        )
+        self.assertTrue(calendar["equal"])
+
+    def test_shadow_projection_still_alarms_on_requested_field_problems(self) -> None:
+        hub_missing_field = {k: v for k, v in HUB_DAILY.items() if k != "close"}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close")}
+        lost = compare_rows(
+            "daily", [legacy_close_only], [hub_missing_field], fields="ts_code,trade_date,close"
+        )
+        self.assertFalse(lost["equal"])
+        self.assertEqual(lost["value_diff_count"], 1)
+
+        changed = compare_rows(
+            "daily", [legacy_close_only], [{**HUB_DAILY, "close": 99.0}],
+            fields="ts_code,trade_date,close",
+        )
+        self.assertFalse(changed["equal"])
+        self.assertEqual(changed["value_diff_count"], 1)
+        self.assertEqual(changed["value_diffs"][0]["fields"][0]["field"], "close")
+
+        gone = compare_rows("daily", [LEGACY_DAILY], [], fields="ts_code,trade_date,close")
+        self.assertEqual(gone["missing_hub_count"], 1)
+        self.assertFalse(gone["equal"])
+
+        unit = compare_rows(
+            "daily", [LEGACY_DAILY], [{**HUB_DAILY, "amount": 2000.0, "volume": 1000.0}],
+            fields="ts_code,trade_date,vol,amount",
+        )
+        self.assertGreater(unit["unit_conversion_count"], 0)
+        self.assertFalse(unit["equal"])
+
+    def test_bridge_shadow_report_uses_website_request_fields(self) -> None:
+        hub_full = {**HUB_DAILY, "adj_factor": 1.1}
+        legacy_close_only = {k: LEGACY_DAILY[k] for k in ("ts_code", "trade_date", "close", "vol", "amount")}
+        reports: list[dict[str, Any]] = []
+        client = FakeClient(
+            response=DatahubResponse(
+                data=[hub_full],
+                meta={"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0},
+            )
+        )
+        wrapped = DatahubAwareTushareClient(
+            FakeLegacy([legacy_close_only]),
+            DatahubBridge(flags(daily=(False, True)), client, shadow_sink=reports.append),
+        )
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,trade_date,close,vol,amount")
+        self.assertEqual(rows[0]["close"], 10.20)
+        self.assertEqual(rows[0]["vol"], 1000.0)
+        self.assertTrue(reports[0]["equal"])
+        self.assertEqual(reports[0]["matched"], 1)
+
     def test_native_roundtrip_matches_known_scales(self) -> None:
         native = to_native_row("daily", HUB_DAILY)
         self.assertEqual(native["vol"], 1000.0)

xiaobai-datahub/README.md

@@ -83,6 +83,16 @@ python -m datahub history-backfill
 
 `hub-quality.config.json` 的 `field_gates` 按数据集配置关键字段：非空率下限（支持按字段覆盖，如 `dv_ttm` 合法高空值）、非有限值比例上限、以及相对上一已发布批次的非空率塌陷保护。字段大面积为空的批次会被拒绝发布、保留上一份正常正式数据，失败原因逐字段写入 `batches.error` / `quality_json`。被拒后数据集仍视为缺失，盘后自动重试（HEL-435 机制）会继续尝试直到成功或截止。配置对任意数据集生效，不写死单日或单字段。
 
+## 整批原子发布（release group）
+
+盘后发布/重发（eod_a、eod_retry、`eod-refresh`、跨数据集重发）不再逐数据集各自切换，而是走整批原子可见机制：
+
+- 一致性边界：日 K、估值、资金流、竞价同属 A 组整批；指数日 K 为 B 组；当日股票主档快照随 A 组一同切换（主档 `stock_master` 的 UPSERT 与快照发布同一事务，不会出现主档先行/滞后）。
+- 流程：组内全部成员先在暂存表完成拉取、字段质量门、覆盖检查和跨数据集交叉校验（`cross_gates` 配置 ts_code 覆盖重叠率下限），全部达标后才在**一个 SQLite 事务**里复制正式表并翻转全部 `publications` 指针。
+- 任一成员失败（拉取失败、质量门拒绝、交叉校验不过、切换事务中断）→ 整批不切换，对外继续提供上一份完整正式版本，失败原因写入 `batches.error` 与 `audit_log`（`action=release-group`），等待晚间自动重试。
+- 读取侧任何时刻只会看到"旧完整版本"或"新完整版本"：发布指针在单事务内统一翻转，容器重启/事务中断自动回滚，不暴露字段残缺或跨数据集混合版本。
+- 幂等：仅当一致性边界内全部成员都已发布时才整组跳过；边界内任有缺失则整组重暂存后统一切换，避免旧批次与新批次混在同一次重发中。重复执行、并发重试不会在完整边界已就绪时生成重复批次（调度器另有 EOD 互斥锁）。
+
 ## 股票主档每日刷新与发布
 
 交易日 20:00 与 23:10（`stocks_refresh_times` 可配）自动刷新股票主档并发布版本化快照（`eod_stocks` + `publications.dataset='stocks'`），覆盖当日新上市、证券简称变化和上市首日 N/C 前缀摘除；无变化则跳过，重复执行幂等。`/v1/stocks` 从最新已发布快照提供数据并带 `batch_id` / `published_at`；`/v1/datasets/status` 同步展示 stocks 状态。
@@ -105,11 +115,25 @@ python -m datahub moneyflow-backfill        # --trading-days 60 --end-date --for
 
 ```bash
 cd xiaobai-datahub
-python -m datahub eod-refresh --trade-date 20260904                  # 只补缺失数据集
+python -m datahub eod-refresh --trade-date 20260904                  # 补不完整的 A/B 边界
 python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
-# 强制重取重发：仍走全部质量门，生成新批次，上一批次保留可回滚
+# --force 按一致性边界整组重发：valuation/daily/moneyflow/auction/stocks → A 组；
+# index_daily → B 组。不可再单独切换某一个正式数据集。
 ```
 
+管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
+
+## 估值发布后复核与自动追补
+
+Tushare `daily_basic` 会在盘后继续改当日字段。HEL-423 在 2026-09-07 观察到：中枢 17:10 发布 `003021.SZ turnover_rate=1.3565`，21:05 上游/旧链路已是 `1.3572`；其余 7 类观察对象当日一致。日 K、资金流、竞价、指数没有同类晚间修订证据，股票主档已有 20:00/23:10 刷新，因此默认只复核估值，不盲目全量重拉。
+
+窗口（可配）：交易日 **20:00–23:20**，每 30 分钟一次轻量比对（对齐网站 21:00 / 23:30 观察）。只拉取 `daily_basic`，按网站真实请求字段精确比较，无误差豁免。
+
+- 无变化：不产生新批次，状态「已追平」。
+- 发现修订：重新走字段质量门、覆盖检查和 A 组整批原子发布；读者全程只能看到上一完整版本或新完整版本。
+- 上游空 / 接口失败 / 不完整 / 质量门拒绝：保留上一完整版本，状态「复核失败」。
+- 23:20 截止后停止当晚复核；下一自然日盘前对上一交易日再做一次安全追赶。
+- 与 `eod_a` / `eod_retry` 共用互斥锁；容器重启会在窗口内立即补一次。
 
 ## 备份
 

xiaobai-datahub/admin/app.js

@@ -105,6 +105,7 @@ async function render() {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
     const eod = data.eod_status || {};
+    const rev = data.revision_status || {};
     const eodLabels = {
       pending_first_attempt: "等待首次尝试",
       waiting_upstream: "等待上游",
@@ -112,6 +113,14 @@ async function render() {
       cutoff_failed: "已截止失败",
       closed_day: "休市",
     };
+    const revLabels = {
+      waiting_review: "等待复核",
+      review_failed: "复核失败",
+      aligned: "已追平",
+      cutoff: "已截止",
+      pending_publish: "待发布",
+      closed_day: "休市",
+    };
     const eodExtra = [];
     if (eod.state === "waiting_upstream") {
       eodExtra.push(`已试 ${eod.attempts} 次`);
@@ -121,12 +130,16 @@ async function render() {
     if (eod.state === "cutoff_failed" && eod.missing_datasets) {
       eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
     }
+    const revExtra = [];
+    if (rev.detail) revExtra.push(esc(String(rev.detail)));
+    if (rev.window) revExtra.push(esc(String(rev.window)));
     page.innerHTML = `
       <div class="cards">
         <div class="card"><div class="muted">交易日</div><strong>${esc(data.trade_date)}</strong></div>
         <div class="card"><div class="muted">阶段</div><strong>${esc(data.session_phase)}</strong></div>
         <div class="card"><div class="muted">今日发布</div><strong>${data.publications.length}</strong></div>
         <div class="card"><div class="muted">盘后补跑</div><strong>${esc(eodLabels[eod.state] || eod.state || "-")}</strong><div class="muted">${eodExtra.join(" · ")}</div></div>
+        <div class="card"><div class="muted">估值复核</div><strong>${esc(revLabels[rev.state] || rev.state || "-")}</strong><div class="muted">${revExtra.join(" · ")}</div></div>
         <div class="card"><div class="muted">异常批次</div><strong class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
       </div>
       <h2>最近调用</h2>
@@ -269,7 +282,7 @@ function renderRelease(data) {
 
 async function dangerous(kind, dataset) {
   const date = ($("rel-date") && $("rel-date").value) || "";
-  const ds = dataset || prompt("数据集（daily / valuation / moneyflow / auction / index_daily / reference）", "daily");
+  const ds = dataset || prompt("数据集（daily/valuation/moneyflow/auction/stocks→A组整批；index_daily→B组；或 reference）", "daily");
   if (!ds) return;
   const password = prompt("二次确认：输入管理密码");
   if (!password) return;

xiaobai-datahub/config/hub-quality.config.json

@@ -17,11 +17,27 @@
   "eod_retry_start": "15:15",
   "eod_retry_interval_minutes": 30,
   "eod_retry_cutoff": "23:30",
+  "revision_review_datasets": ["valuation"],
+  "revision_review_start": "20:00",
+  "revision_review_interval_minutes": 30,
+  "revision_review_cutoff": "23:20",
   "moneyflow_history_trading_days": 60,
   "stocks_refresh_times": [
     "20:00",
     "23:10"
   ],
+  "cross_gates": [
+    {
+      "left": "daily",
+      "right": "valuation",
+      "min_key_overlap": 0.98
+    },
+    {
+      "left": "daily",
+      "right": "moneyflow",
+      "min_key_overlap": 0.98
+    }
+  ],
   "field_gates": {
     "valuation": {
       "fields": [

xiaobai-datahub/datahub/admin_api.py

@@ -6,7 +6,7 @@ from typing import Any
 from datahub.adapters import RESERVED
 from datahub.auth import AuthService
 from datahub.db import HubDB
-from datahub.pipeline import Pipeline
+from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
 from datahub.scheduler import Scheduler
 from datahub.serving import ApiError
 from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
@@ -39,6 +39,7 @@ class AdminAPI:
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
             "eod_status": self.scheduler.eod_status(today),
+            "revision_status": self.scheduler.revision_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -91,6 +92,7 @@ class AdminAPI:
                 {"id": "eod_a", "at": "15:05", "title": "盘后批 A daily/valuation/moneyflow/auction"},
                 {"id": "eod_b", "at": "15:10", "title": "盘后批 B index_daily"},
                 {"id": "eod_retry", "at": "15:15-23:30", "title": "盘后未出数自动重试（每 30 分钟，成功即停）"},
+                {"id": "eod_revise", "at": "20:00-23:20", "title": "估值发布后复核（轻量比对，有修订才整组原子追补）"},
                 {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"},
                 {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},
@@ -143,8 +145,22 @@ class AdminAPI:
             self._dangerous(password, confirm, f"{dataset}:{day}")
             if dataset == "reference":
                 result = self.pipeline.ingest_reference(day)
+            elif dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET:
+                # Manual same-day republish must rebuild the full A/B boundary.
+                # Gate failures and mid-switch exceptions both surface as
+                # FAILED_PRECONDITION so the admin API never leaks raw
+                # transaction errors to the client.
+                try:
+                    result = self.pipeline.force_republish_boundary(dataset, day)
+                    failures = self.pipeline.eod_failures(result)
+                    if failures:
+                        raise ApiError("FAILED_PRECONDITION", "; ".join(failures))
+                except ApiError:
+                    raise
+                except Exception as exc:
+                    raise ApiError("FAILED_PRECONDITION", str(exc)) from exc
             else:
-                result = self.pipeline.run_dataset(dataset, day)
+                raise ApiError("INVALID_ARGUMENT", f"unsupported backfill dataset: {dataset}")
         self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
         return result
 

xiaobai-datahub/datahub/cli.py

@@ -7,7 +7,7 @@ import json
 import sys
 
 from datahub.hub import build_hub
-from datahub.pipeline import OFFICIAL_DATASETS
+from datahub.pipeline import EOD_A_DATASETS, OFFICIAL_DATASETS, STOCKS_DATASET
 from datahub.settings import load_settings
 from datahub.timeutil import yyyymmdd
 
@@ -19,15 +19,15 @@ def main(argv: list[str] | None = None) -> int:
     history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
     history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
     history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
-    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已发布数据集，仍走质量门禁）")
+    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已完整发布的一致性边界，仍走质量门禁）")
     refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
     refresh.add_argument(
         "--force", action="store_true",
-        help="对 --dataset 指定的数据集强制重取重发（生成新批次，保留上一批次可回滚）",
+        help="强制重发 --dataset 所属的完整一致性边界（A 组或 B 组），生成新批次并保留上一批次可回滚",
     )
     refresh.add_argument(
         "--dataset", default=None,
-        help="配合 --force 使用：只强制重发该数据集（如 valuation）",
+        help="配合 --force：指定边界内任一成员（如 valuation→整组 A；index_daily→整组 B）",
     )
     stocks_refresh = sub.add_parser("stocks-refresh", help="刷新股票主档并发布正式快照（幂等：无变化则跳过）")
     stocks_refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
@@ -54,26 +54,27 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "eod-refresh":
         day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
         if args.force:
-            datasets = tuple(sorted({args.dataset} & OFFICIAL_DATASETS)) if args.dataset else ()
-            if args.dataset and not datasets:
-                parser.error(f"unknown dataset: {args.dataset}")
-            if not datasets:
+            allowed = set(OFFICIAL_DATASETS) | {STOCKS_DATASET}
+            if not args.dataset:
                 parser.error("--force requires --dataset (e.g. --dataset valuation)")
-            result = {}
-            for dataset in datasets:
-                result[dataset] = hub.pipeline.run_dataset(dataset, day)
+            if args.dataset not in allowed:
+                parser.error(f"unknown dataset: {args.dataset}")
+            result = hub.pipeline.force_republish_boundary(args.dataset, day)
+            boundary = "A" if args.dataset in EOD_A_DATASETS or args.dataset == STOCKS_DATASET else "B"
         else:
             result = hub.pipeline.run_eod_missing(day)
+            boundary = None
         hub.pipeline.audit("cli", "eod-refresh", f"eod:{day}", json.dumps(
-            {"force": bool(args.force), "dataset": args.dataset,
+            {"force": bool(args.force), "dataset": args.dataset, "boundary": boundary,
              **{name: item.get("state") for name, item in result.items() if isinstance(item, dict)}},
             ensure_ascii=False,
         ))
         if args.force:
-            payload = {"trade_date": day, "datasets": result}
+            failures = hub.pipeline.eod_failures(result)
+            payload = {"trade_date": day, "boundary": boundary, "datasets": result}
             json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
             sys.stdout.write("\n")
-            return 0
+            return 0 if not failures else 1
         missing = hub.pipeline.missing_official_datasets(day)
         payload = {"trade_date": day, "datasets": result, "missing_after": missing}
         json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)

xiaobai-datahub/datahub/db.py

@@ -238,6 +238,19 @@ CREATE TABLE IF NOT EXISTS eod_progress (
   updated_at TEXT NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS revision_progress (
+  trade_date TEXT PRIMARY KEY,
+  state TEXT NOT NULL,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at TEXT,
+  next_retry_at TEXT,
+  finished_at TEXT,
+  catchup_done INTEGER NOT NULL DEFAULT 0,
+  last_diff TEXT,
+  detail TEXT,
+  updated_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS audit_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,
   actor TEXT NOT NULL,

xiaobai-datahub/datahub/pipeline.py

@@ -15,6 +15,12 @@ from datahub.governance.ratelimit import TokenBucket
 from datahub.governance.retry import RetryError, retry_call
 from datahub.logutil import get_logger
 from datahub.normalize import finite_number, normalize_daily
+from datahub.revision import (
+    compare_fields,
+    diff_published_vs_upstream,
+    official_table,
+    revision_datasets,
+)
 from datahub.settings import Settings
 from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd
 
@@ -133,6 +139,95 @@ def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int:
     return int(row["n"] if row is not None else 0)
 
 
+def _staging_count_or_raise(connection: Any, dataset: str, trade_date: str, batch_id: str) -> int:
+    rows_out = _staging_row_count(connection, dataset, batch_id)
+    if rows_out <= 0:
+        report = {
+            "rows": 0,
+            "errors": [EMPTY_BATCH_ERROR],
+            "warnings": [],
+            "hard_fail": True,
+            "soft_fail": False,
+            "batch_id": batch_id,
+            "dataset": dataset,
+            "trade_date": trade_date,
+        }
+        LOGGER.warning(
+            "skip official publish for empty batch",
+            extra={
+                "hub": {
+                    "dataset": dataset,
+                    "trade_date": trade_date,
+                    "batch_id": batch_id,
+                    "rows_out": rows_out,
+                    "reason": "upstream_empty",
+                }
+            },
+        )
+        raise QualityError("empty batch cannot be officially published", report)
+    return rows_out
+
+
+def _upsert_publication(
+    connection: Any,
+    dataset: str,
+    trade_date: str,
+    batch_id: str,
+    state: str,
+    published_at: str,
+) -> None:
+    current = connection.execute(
+        "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
+        (dataset, trade_date),
+    ).fetchone()
+    prev = str(current["active_batch"]) if current else None
+    connection.execute(
+        """
+        INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
+        VALUES (?, ?, ?, ?, ?, ?)
+        ON CONFLICT(dataset, trade_date) DO UPDATE SET
+          prev_batch=excluded.prev_batch,
+          active_batch=excluded.active_batch,
+          state=excluded.state,
+          published_at=excluded.published_at
+        """,
+        (dataset, trade_date, batch_id, prev, state, published_at),
+    )
+
+
+def _record_publication_history(
+    connection: Any,
+    dataset: str,
+    trade_date: str,
+    batch_id: str,
+    published_at: str,
+    quality: dict[str, Any],
+) -> None:
+    max_gen = connection.execute(
+        "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
+        (dataset, trade_date),
+    ).fetchone()
+    generation = int(max_gen["g"]) + 1
+    connection.execute(
+        "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
+        (dataset, trade_date, batch_id, published_at, generation),
+    )
+    keep = int(quality.get("publication_generations") or 3)
+    stale = connection.execute(
+        """
+        SELECT batch_id FROM publication_history
+        WHERE dataset = ? AND trade_date = ?
+        ORDER BY generation DESC
+        """,
+        (dataset, trade_date),
+    ).fetchall()
+    for row in stale[keep:]:
+        connection.execute(
+            "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
+            (dataset, trade_date, row["batch_id"]),
+        )
+
+
 class QualityError(RuntimeError):
     def __init__(self, message: str, report: dict[str, Any]) -> None:
         super().__init__(message)
@@ -259,11 +354,20 @@ class Pipeline:
         is published; ``force`` re-publishes unconditionally. New listings,
         renames (incl. N/C prefix removal) and status changes all flow into
         the snapshot, which carries batch_id/published_at metadata.
+
+        The ``stock_master`` UPSERT happens inside the same publish
+        transaction as the snapshot switch — fetch / quality-gate / switch
+        failures leave the master on the previous complete values.
         """
         day = yyyymmdd(trade_date or self.clock())
+        try:
             rows = self._fetch_dataset(STOCKS_DATASET, day)
-        with self.db.write() as connection:
-            self._upsert_stock_master(connection, rows, isoformat(self.clock()))
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         if not force:
             active, snapshot = self.published_stock_snapshot(day)
             if active is not None:
@@ -280,7 +384,14 @@ class Pipeline:
                         "batch_id": active,
                         "rows": len(snapshot),
                     }
+        try:
             result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         self.audit(
             "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
             json.dumps({"batch_id": result["batch_id"], "rows": result["rows"]}, ensure_ascii=False),
@@ -548,43 +659,592 @@ class Pipeline:
         return [dataset for dataset in sorted(OFFICIAL_DATASETS) if dataset not in published]
 
     def run_eod_missing(self, trade_date: str) -> dict[str, Any]:
-        """Fetch/publish every official dataset still missing for the date.
+        """Republish every incomplete EOD consistency group for the date.
 
-        Idempotent: datasets with an existing publication are skipped, so
-        repeats never overwrite the current official batch. Per-dataset
-        failures are collected instead of aborting the remaining datasets.
+        A-group (daily/valuation/moneyflow/auction + stocks) and B-group
+        (index_daily) are separate boundaries. Within a group, either the
+        whole boundary is already published (idempotent skip) or every
+        member is re-staged and switched together — never fill only the
+        missing members on top of older batches from an earlier partial run.
         """
-        return self._run_eod_datasets(tuple(sorted(OFFICIAL_DATASETS)), trade_date)
+        results: dict[str, Any] = {}
+        results.update(self.run_eod_batch_a(trade_date))
+        results.update(self.run_eod_batch_b(trade_date))
+        return results
 
-    def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_A_DATASETS, trade_date)
+    def run_eod_batch_a(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True, force=force)
 
-    def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_B_DATASETS, trade_date)
+    def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_release_group(EOD_B_DATASETS, trade_date, force=force)
 
-    def _run_eod_datasets(self, datasets: tuple[str, ...], trade_date: str) -> dict[str, Any]:
+    def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Force-republish the full A/B consistency boundary that owns ``dataset``.
+
+        CLI ``eod-refresh --force`` and admin manual backfill must not publish a
+        single official member alone — that would mix old and new batches inside
+        the same trade date. Naming any A-group member (or stocks) rebuilds the
+        whole A group; naming ``index_daily`` rebuilds B.
+        """
+        name = str(dataset or "").strip()
+        if name in EOD_A_DATASETS or name == STOCKS_DATASET:
+            return self.run_eod_batch_a(trade_date, force=True)
+        if name in EOD_B_DATASETS:
+            return self.run_eod_batch_b(trade_date, force=True)
+        raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
+
+    def published_official_rows(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
+        day = yyyymmdd(trade_date)
+        batch_id = self.active_batch(dataset, day)
+        if not batch_id:
+            return []
+        fields = compare_fields(dataset)
+        table = official_table(dataset)
+        if fields:
+            columns = ",".join(fields)
+            return self.db.fetchall(
+                f"SELECT {columns} FROM {table} WHERE batch_id = ?",
+                (batch_id,),
+            )
+        return self.db.fetchall(f"SELECT * FROM {table} WHERE batch_id = ?", (batch_id,))
+
+    def compare_revision(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Light fetch of one revision-risk dataset vs the published official rows."""
+        day = yyyymmdd(trade_date)
+        published = self.published_official_rows(dataset, day)
+        if not published:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "skipped",
+                "reason": "not_published",
+            }
+        try:
+            upstream = self._fetch_dataset(dataset, day)
+        except Exception as exc:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_error",
+                "error": str(exc),
+            }
+        if not upstream:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_empty",
+                "error": "revision review upstream empty",
+                "published_rows": len(published),
+                "upstream_rows": 0,
+            }
+        listed = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
+        )
+        listed_n = int((listed or {}).get("n") or 0)
+        floor = float(self.settings.quality.get("daily_row_ratio") or 0.98)
+        if listed_n and len(upstream) / listed_n < floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"/ listed {listed_n} < {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        if len(upstream) < len(published) * floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"< published {len(published)} * {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        compared = diff_published_vs_upstream(dataset, published, upstream)
+        compared["trade_date"] = day
+        compared["state"] = "changed" if compared["changed"] else "unchanged"
+        compared["reason"] = "revised" if compared["changed"] else "unchanged"
+        return compared
+
+    def review_published_revisions(self, trade_date: str) -> dict[str, Any]:
+        """Evening/morning catch-up: compare website fields, republish only on change.
+
+        Unchanged → no new batch. Changed → full A/B boundary quality gate +
+        atomic switch (HEL-459/460/461). Empty/failed/incomplete upstream keeps
+        the previous complete official version.
+        """
         day = yyyymmdd(trade_date)
         results: dict[str, Any] = {}
+        for dataset in revision_datasets(self.settings.quality):
+            compared = self.compare_revision(dataset, day)
+            if compared.get("state") == "skipped":
+                results[dataset] = compared
+                continue
+            if compared.get("state") == "failed":
+                results[dataset] = compared
+                LOGGER.warning(
+                    "revision review kept previous official version",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": compared.get("reason"),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": compared.get("reason"),
+                            "error": compared.get("error"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            if not compared.get("changed"):
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aligned",
+                    "reason": "unchanged",
+                    "published_rows": compared.get("published_rows"),
+                    "upstream_rows": compared.get("upstream_rows"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps({"state": "aligned", "reason": "unchanged"}, ensure_ascii=False),
+                )
+                continue
+            LOGGER.info(
+                "revision review detected upstream rewrite, republishing boundary",
+                extra={
+                    "hub": {
+                        "dataset": dataset,
+                        "trade_date": day,
+                        "diffs": compared.get("diffs"),
+                        "event": "revision_review_changed",
+                    }
+                },
+            )
+            try:
+                published = self.force_republish_boundary(dataset, day)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "republish_error",
+                    "error": str(exc),
+                    "diffs": compared.get("diffs"),
+                }
+                LOGGER.warning(
+                    "revision republish failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": str(exc),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {"state": "failed", "reason": "republish_error", "error": str(exc)},
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            failures = self.eod_failures(published)
+            if failures:
+                results.update(published)
+                results[dataset] = {
+                    **(published.get(dataset) or {}),
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "quality_gate",
+                    "error": "; ".join(failures),
+                    "diffs": compared.get("diffs"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": "quality_gate",
+                            "error": "; ".join(failures),
+                            "diffs": compared.get("diffs"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            results.update(published)
+            results["review"] = {
+                "dataset": dataset,
+                "trade_date": day,
+                "state": "aligned",
+                "reason": "revised",
+                "diffs": compared.get("diffs"),
+                "missing_codes": compared.get("missing_codes"),
+                "extra_codes": compared.get("extra_codes"),
+            }
+            self.audit(
+                "pipeline", "revision-review", f"{dataset}:{day}",
+                json.dumps(
+                    {
+                        "state": "aligned",
+                        "reason": "revised",
+                        "diffs": compared.get("diffs"),
+                        "switched": sorted(
+                            name for name, item in published.items()
+                            if isinstance(item, dict) and item.get("state") == "published"
+                        ),
+                    },
+                    ensure_ascii=False,
+                ),
+            )
+        return results
+
+    def run_release_group(
+        self,
+        datasets: tuple[str, ...],
+        trade_date: str,
+        include_stocks: bool = False,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """One post-market publish/republish becomes one atomic visibility flip.
+
+        Consistency boundary: every member (official datasets, plus the daily
+        stocks snapshot when ``include_stocks``) is fetched, staged,
+        field-gated and cross-validated BEFORE any reader can see it. Only
+        when the whole group passes does a single SQLite transaction copy
+        all staging batches to the official tables and flip every
+        ``publications`` row at once. Any member failure aborts the group:
+        the previous complete official version keeps serving and the reason
+        is recorded on the batches and in the audit log.
+
+        Skip is all-or-nothing for the boundary unless ``force``: if every
+        official member (and stocks when required) is already published, the
+        group is skipped. If any official member is still missing — or
+        ``force`` is set — every official member is re-staged, so a retry or
+        manual republish never mixes old and new batches in one release.
+        """
+        day = yyyymmdd(trade_date)
+        results: dict[str, Any] = {}
+        staged: dict[str, dict[str, Any]] = {}
+        failure: str | None = None
+        missing_official = [dataset for dataset in datasets if self.active_batch(dataset, day) is None]
+        stocks_missing = include_stocks and self.active_batch(STOCKS_DATASET, day) is None
+
+        if not force and not missing_official and not stocks_missing:
             for dataset in datasets:
-            if self.active_batch(dataset, day) is not None:
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "skipped",
                     "reason": "already_published",
                 }
+            if include_stocks:
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+            return results
+
+        # Incomplete or forced boundary → restage every official member together.
+        pending = list(datasets)
+
+        for dataset in pending:
+            if failure is not None:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aborted",
+                    "reason": f"release group aborted: {failure}",
+                }
                 continue
             try:
-                results[dataset] = self.run_dataset(dataset, day)
+                staged[dataset] = self._stage_and_validate(dataset, day)
             except Exception as exc:
+                failure = f"{dataset}: {exc}"
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "failed",
                     "error": str(exc),
                 }
+
+        # Stocks join the same switch when the official boundary is being
+        # rebuilt (missing or forced), or when only the stocks snapshot is
+        # still missing.
+        rebuild_official = bool(force or missing_official)
+        if include_stocks and failure is None and (rebuild_official or stocks_missing):
+            try:
+                stocks_plan = self._stage_stocks_snapshot(day, force=rebuild_official)
+            except Exception as exc:
+                failure = f"{STOCKS_DATASET}: {exc}"
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": str(exc),
+                }
+            else:
+                if stocks_plan is not None:
+                    staged[STOCKS_DATASET] = stocks_plan
+
+        if failure is None and staged:
+            cross_errors = self._cross_gate_errors(staged)
+            if cross_errors:
+                failure = "; ".join(cross_errors)
+
+        if failure is not None:
+            for dataset, item in staged.items():
+                self._abandon_batch(item["batch_id"], f"release group not switched: {failure}")
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": f"release group not switched: {failure}",
+                    "batch_id": item["batch_id"],
+                }
+            LOGGER.warning(
+                "release group blocked, previous official version keeps serving",
+                extra={
+                    "hub": {
+                        "trade_date": day,
+                        "datasets": sorted(staged),
+                        "reason": failure,
+                        "event": "release_group_blocked",
+                    }
+                },
+            )
+            self.audit(
+                "pipeline", "release-group", f"eod:{day}",
+                json.dumps({"state": "failed", "reason": failure}, ensure_ascii=False),
+            )
             return results
 
+        if staged:
+            try:
+                self._switch_release_group(day, staged)
+            except Exception as exc:
+                reason = f"release group switch failed: {exc}"
+                for item in staged.values():
+                    self._abandon_batch(item["batch_id"], reason)
+                LOGGER.warning(
+                    "release group switch failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "trade_date": day,
+                            "datasets": sorted(staged),
+                            "reason": reason,
+                            "event": "release_group_switch_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "release-group", f"eod:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": reason,
+                            "switched": [],
+                            "force": bool(force),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                raise
+            for dataset, item in staged.items():
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": item["state"],
+                    "batch_id": item["batch_id"],
+                    "rows": item["rows"],
+                }
+            self.audit(
+                "pipeline", "release-group", f"eod:{day}",
+                json.dumps(
+                    {"state": "ok", "switched": sorted(staged), "force": bool(force)},
+                    ensure_ascii=False,
+                ),
+            )
+        return results
+
+    def _stage_and_validate(self, dataset: str, trade_date: str, attempts: int | None = None) -> dict[str, Any]:
+        """Fetch → stage → quality-gate one member without publishing it."""
+        day = yyyymmdd(trade_date)
+        batch_id = self.next_batch_id(dataset, day)
+        max_attempts = attempts or self.settings.max_publish_attempts
+        rows: list[dict[str, Any]] = []
+        self._set_batch(batch_id, dataset, day, "scheduled", 0)
+        try:
+            self._set_batch(batch_id, dataset, day, "fetching", 1)
+            rows = retry_call(
+                lambda: self._fetch_dataset(dataset, day),
+                attempts=max_attempts,
+                base_delay=0.05,
+                sleeper=lambda _d: time.sleep(_d),
+            )
+            self._stage(dataset, batch_id, rows)
+            self._set_batch(batch_id, dataset, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
+            self._set_batch(batch_id, dataset, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
+            report = self.validate(dataset, batch_id, day, rows)
+            if report["hard_fail"]:
+                self._reject_batch(batch_id, dataset, day, rows, report)
+                raise QualityError("integrity gate failed", report)
+        except RetryError as exc:
+            self._set_batch(batch_id, dataset, day, "failed", max_attempts, error=str(exc), finished=True)
+            raise
+        except QualityError as exc:
+            current = self.db.fetchone("SELECT state FROM batches WHERE batch_id = ?", (batch_id,))
+            if current and current["state"] not in {"staged", "failed"}:
+                self._reject_batch(batch_id, dataset, day, rows, exc.report)
+            raise
+        except Exception as exc:
+            self._set_batch(batch_id, dataset, day, "failed", 1, error=str(exc), finished=True)
+            raise
+        self._set_batch(
+            batch_id, dataset, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
+        )
+        return {
+            "dataset": dataset,
+            "trade_date": day,
+            "batch_id": batch_id,
+            "rows": len(rows),
+            "quality": report,
+            "state": "degraded" if report["soft_fail"] else "published",
+        }
+
+    def _stage_stocks_snapshot(self, trade_date: str, force: bool = False) -> dict[str, Any] | None:
+        """Stage the daily stocks snapshot for a release group switch.
+
+        Returns None when the published snapshot is already identical to
+        upstream (idempotent skip) unless ``force`` is set. The stock_master
+        upsert is deferred into the group switch / publish transaction so the
+        master never runs ahead of the published snapshot.
+        """
+        day = yyyymmdd(trade_date)
+        active, snapshot = self.published_stock_snapshot(day)
+        rows = self._fetch_dataset(STOCKS_DATASET, day)
+        if active is not None and not force:
+            upstream = sorted(
+                tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows
+            )
+            published = sorted(tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in snapshot)
+            if upstream == published:
+                return None
+        batch_id = self.next_batch_id(STOCKS_DATASET, day)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "scheduled", 0)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "fetching", 1)
+        self._stage(STOCKS_DATASET, batch_id, rows)
+        self._set_batch(batch_id, STOCKS_DATASET, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
+        self._set_batch(batch_id, STOCKS_DATASET, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
+        report = self.validate(STOCKS_DATASET, batch_id, day, rows)
+        if report["hard_fail"]:
+            self._reject_batch(batch_id, STOCKS_DATASET, day, rows, report)
+            raise QualityError("integrity gate failed", report)
+        self._set_batch(
+            batch_id, STOCKS_DATASET, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
+        )
+        return {
+            "dataset": STOCKS_DATASET,
+            "trade_date": day,
+            "batch_id": batch_id,
+            "rows": len(rows),
+            "row_values": rows,
+            "quality": report,
+            "state": "degraded" if report["soft_fail"] else "published",
+        }
+
+    def _cross_gate_errors(self, staged: dict[str, dict[str, Any]]) -> list[str]:
+        """Cross-dataset consistency checks on staged batches (交叉校验)."""
+        errors: list[str] = []
+        gates = self.settings.quality.get("cross_gates") or []
+        for gate in gates if isinstance(gates, list) else []:
+            if not isinstance(gate, dict):
+                continue
+            left = str(gate.get("left") or "")
+            right = str(gate.get("right") or "")
+            if not left or not right or left not in staged or right not in staged:
+                continue
+            floor = float(gate.get("min_key_overlap") or 0.98)
+            left_keys = self._staging_keys(left, staged[left]["batch_id"])
+            right_keys = self._staging_keys(right, staged[right]["batch_id"])
+            denom = max(len(left_keys), len(right_keys))
+            overlap = (len(left_keys & right_keys) / denom) if denom else 1.0
+            if overlap < floor:
+                errors.append(
+                    f"cross gate: {left} vs {right} key overlap {overlap:.4f} < {floor}"
+                )
+        return errors
+
+    def _staging_keys(self, dataset: str, batch_id: str) -> set[str]:
+        table = DATASET_TABLES[dataset][1]
+        rows = self.db.fetchall(
+            f"SELECT DISTINCT ts_code FROM {table} WHERE batch_id = ?",
+            (batch_id,),
+        )
+        return {str(row["ts_code"]) for row in rows}
+
+    def _switch_release_group(self, trade_date: str, members: dict[str, dict[str, Any]]) -> None:
+        """Single transaction: copy every member and flip every publication."""
+        day = yyyymmdd(trade_date)
+        published_at = isoformat(self.clock())
+        with self.db.write() as connection:
+            for dataset, item in members.items():
+                _staging_count_or_raise(connection, dataset, day, item["batch_id"])
+            for dataset, item in members.items():
+                connection.execute(EOD_COPY[dataset], (item["batch_id"],))
+                if dataset == STOCKS_DATASET:
+                    self._upsert_stock_master(connection, item["row_values"], published_at)
+            if self.before_commit:
+                self.before_commit()
+            for dataset, item in members.items():
+                _upsert_publication(connection, dataset, day, item["batch_id"], item["state"], published_at)
+                _record_publication_history(
+                    connection, dataset, day, item["batch_id"], published_at, self.settings.quality
+                )
+                connection.execute(
+                    "UPDATE batches SET state='published', finished_at=? WHERE batch_id=?",
+                    (published_at, item["batch_id"]),
+                )
+
+    def _abandon_batch(self, batch_id: str, reason: str) -> None:
+        row = self.db.fetchone("SELECT dataset, trade_date FROM batches WHERE batch_id = ?", (batch_id,))
+        if not row:
+            return
+        self._set_batch(
+            batch_id, str(row["dataset"]), str(row["trade_date"]), "failed", 1,
+            error=reason, finished=True,
+        )
+
     @staticmethod
     def eod_failures(results: dict[str, Any]) -> list[str]:
         return [
@@ -735,77 +1395,24 @@ class Pipeline:
         return batch_id, stats
 
     def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None:
-        copy_sql = EOD_COPY[dataset]
         published_at = isoformat(self.clock())
         with self.db.write() as connection:
-            rows_out = _staging_row_count(connection, dataset, batch_id)
-            if rows_out <= 0:
-                report = {
-                    "rows": 0,
-                    "errors": [EMPTY_BATCH_ERROR],
-                    "warnings": [],
-                    "hard_fail": True,
-                    "soft_fail": False,
-                    "batch_id": batch_id,
-                    "dataset": dataset,
-                    "trade_date": trade_date,
-                }
-                LOGGER.warning(
-                    "skip official publish for empty batch",
-                    extra={
-                        "hub": {
-                            "dataset": dataset,
-                            "trade_date": trade_date,
-                            "batch_id": batch_id,
-                            "rows_out": rows_out,
-                            "reason": "upstream_empty",
-                        }
-                    },
-                )
-                raise QualityError("empty batch cannot be officially published", report)
-            current = connection.execute(
-                "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
-                (dataset, trade_date),
-            ).fetchone()
-            prev = str(current["active_batch"]) if current else None
-            connection.execute(copy_sql, (batch_id,))
+            _staging_count_or_raise(connection, dataset, trade_date, batch_id)
+            connection.execute(EOD_COPY[dataset], (batch_id,))
+            if dataset == STOCKS_DATASET:
+                staging = DATASET_TABLES[STOCKS_DATASET][1]
+                stock_rows = [
+                    dict(row)
+                    for row in connection.execute(
+                        f"SELECT * FROM {staging} WHERE batch_id = ?",
+                        (batch_id,),
+                    ).fetchall()
+                ]
+                self._upsert_stock_master(connection, stock_rows, published_at)
             if self.before_commit:
                 self.before_commit()
-            connection.execute(
-                """
-                INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
-                VALUES (?, ?, ?, ?, ?, ?)
-                ON CONFLICT(dataset, trade_date) DO UPDATE SET
-                  prev_batch=excluded.prev_batch,
-                  active_batch=excluded.active_batch,
-                  state=excluded.state,
-                  published_at=excluded.published_at
-                """,
-                (dataset, trade_date, batch_id, prev, state, published_at),
-            )
-            max_gen = connection.execute(
-                "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
-                (dataset, trade_date),
-            ).fetchone()
-            generation = int(max_gen["g"]) + 1
-            connection.execute(
-                "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
-                (dataset, trade_date, batch_id, published_at, generation),
-            )
-            keep = int(self.settings.quality.get("publication_generations") or 3)
-            stale = connection.execute(
-                """
-                SELECT batch_id FROM publication_history
-                WHERE dataset = ? AND trade_date = ?
-                ORDER BY generation DESC
-                """,
-                (dataset, trade_date),
-            ).fetchall()
-            for row in stale[keep:]:
-                connection.execute(
-                    "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
-                    (dataset, trade_date, row["batch_id"]),
-                )
+            _upsert_publication(connection, dataset, trade_date, batch_id, state, published_at)
+            _record_publication_history(connection, dataset, trade_date, batch_id, published_at, self.settings.quality)
 
     def rollback(self, dataset: str, trade_date: str, actor: str = "admin") -> dict[str, Any]:
         trade_date = yyyymmdd(trade_date)

xiaobai-datahub/datahub/revision.py

@@ -0,0 +1,111 @@
+"""Post-publish revision review for datasets whose upstream may rewrite T-day fields.
+
+HEL-423 field evidence, not a whitelist of tolerated diffs:
+
+- 2026-09-07 valuation/daily_basic: hub published 003021.SZ turnover_rate=1.3565
+  at 17:10; website legacy and a direct Tushare read at 21:05 both showed 1.3572.
+  The other seven observed objects (daily, moneyflow, auction, stocks, status,
+  index_daily, calendar) matched. Hub had already stopped the day after the
+  first successful publish, so the revision never self-healed.
+- 2026-09-02: same dataset, opposite direction (hub already held the later
+  value). Confirms daily_basic is rewritten after the first complete dump.
+
+Daily bars, moneyflow, auction and index_daily have no same-evening field
+revision evidence. Stocks already refreshes at 20:00/23:10. Review therefore
+fetches only configured revision-risk datasets (default: valuation) and
+compares the website-requested field set. No numeric tolerance.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from datahub.db import DATASET_TABLES
+from datahub.normalize import VALUATION_FIELDS
+from datahub.numbers import finite_number, round4
+
+# Datasets with proven same-evening upstream rewrites. Config may replace this
+# list; it must not silently expand to a full EOD re-pull.
+DEFAULT_REVISION_DATASETS = ("valuation",)
+
+# Website daily_basic request (HEL-423): ts_code/trade_date plus the eight
+# value fields used by the old link and field_gates.
+WEBSITE_COMPARE_FIELDS: dict[str, tuple[str, ...]] = {
+    "valuation": VALUATION_FIELDS,
+}
+
+REVISION_STATES = ("waiting_review", "review_failed", "aligned", "cutoff")
+
+
+def revision_datasets(quality: dict[str, Any] | None) -> tuple[str, ...]:
+    raw = (quality or {}).get("revision_review_datasets")
+    if isinstance(raw, (list, tuple)) and raw:
+        names = tuple(str(item) for item in raw if str(item))
+        if names:
+            return names
+    return DEFAULT_REVISION_DATASETS
+
+
+def compare_fields(dataset: str) -> tuple[str, ...]:
+    fields = WEBSITE_COMPARE_FIELDS.get(dataset)
+    if fields:
+        return fields
+    gate = {}
+    return tuple(str(item) for item in (gate.get("fields") or []) if str(item))
+
+
+def _norm_value(field: str, value: Any) -> Any:
+    if field in {"ts_code", "trade_date"}:
+        return str(value or "")
+    number = round4(finite_number(value))
+    return number
+
+
+def row_signature(row: dict[str, Any], fields: tuple[str, ...]) -> tuple[Any, ...]:
+    return tuple(_norm_value(field, row.get(field)) for field in fields)
+
+
+def diff_published_vs_upstream(
+    dataset: str,
+    published: list[dict[str, Any]],
+    upstream: list[dict[str, Any]],
+    *,
+    max_diffs: int = 20,
+) -> dict[str, Any]:
+    """Exact compare on website-requested fields. No tolerance / exemption."""
+    fields = compare_fields(dataset)
+    if not fields:
+        fields = tuple(sorted({key for row in published + upstream for key in row if key != "batch_id"}))
+    pub_map = {str(row.get("ts_code") or "").upper(): row for row in published}
+    up_map = {str(row.get("ts_code") or "").upper(): row for row in upstream}
+    missing = sorted(code for code in pub_map if code not in up_map)
+    extra = sorted(code for code in up_map if code not in pub_map)
+    diffs: list[dict[str, Any]] = []
+    for code in sorted(set(pub_map) & set(up_map)):
+        left = row_signature(pub_map[code], fields)
+        right = row_signature(up_map[code], fields)
+        if left == right:
+            continue
+        for field, old, new in zip(fields, left, right):
+            if old == new:
+                continue
+            diffs.append({"ts_code": code, "field": field, "published": old, "upstream": new})
+            if len(diffs) >= max_diffs:
+                break
+        if len(diffs) >= max_diffs:
+            break
+    changed = bool(diffs or missing or extra)
+    return {
+        "changed": changed,
+        "dataset": dataset,
+        "fields": list(fields),
+        "published_rows": len(published),
+        "upstream_rows": len(upstream),
+        "missing_codes": missing[:max_diffs],
+        "extra_codes": extra[:max_diffs],
+        "diffs": diffs,
+    }
+
+
+def official_table(dataset: str) -> str:
+    return DATASET_TABLES[dataset][0]

xiaobai-datahub/datahub/scheduler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import threading
 from collections.abc import Callable
 from datetime import datetime, time, timedelta
@@ -8,13 +9,14 @@ from typing import Any
 from datahub.db import HubDB
 from datahub.logutil import get_logger
 from datahub.pipeline import Pipeline
+from datahub.revision import revision_datasets
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
-EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry", "eod_revise"}
 
 
 def is_open_day(db: HubDB, day: str) -> bool:
@@ -27,6 +29,20 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def previous_open_day(db: HubDB, day: str) -> str | None:
+    row = db.fetchone(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date < ?
+        ORDER BY cal_date DESC LIMIT 1
+        """,
+        (day,),
+    )
+    if row is None:
+        return None
+    return str(row["cal_date"])
+
+
 def _hhmm(value: str) -> time:
     return datetime.strptime(value, "%H:%M").time()
 
@@ -49,6 +65,7 @@ class Scheduler:
             "eod_a": self._eod_a,
             "eod_b": self._eod_b,
             "eod_retry": self._eod_retry,
+            "eod_revise": self._eod_revise,
             "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
@@ -118,6 +135,8 @@ class Scheduler:
             if job_id in {"eod_a", "eod_b"}:
                 self._settle_eod(day)
         ran.extend(self._eod_retry_tick(now, day, open_day))
+        ran.extend(self._revision_review_tick(now, day, open_day))
+        ran.extend(self._revision_catchup_tick(now, day))
         return ran
 
     # ------------------------------------------------------------------
@@ -219,6 +238,201 @@ class Scheduler:
             "detail": (row or {}).get("detail"),
         }
 
+    def revision_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM revision_progress WHERE trade_date = ?", (day,))
+
+    def revision_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """等待复核 / 复核失败 / 已追平 / 已截止."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        row = self.revision_progress(day)
+        open_day = is_open_day(self.db, day)
+        published = self._revision_ready(day)
+        if row and row["state"] in {"aligned", "review_failed", "cutoff", "waiting_review"}:
+            state = str(row["state"])
+        elif not open_day:
+            state = "closed_day"
+        elif not published:
+            state = "pending_publish"
+        else:
+            state = "waiting_review"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "datasets": list(revision_datasets(self.pipeline.settings.quality)),
+            "attempts": int((row or {}).get("attempts") or 0),
+            "last_attempt_at": (row or {}).get("last_attempt_at"),
+            "next_retry_at": (row or {}).get("next_retry_at") if state in {"waiting_review", "review_failed"} else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "catchup_done": bool(int((row or {}).get("catchup_done") or 0)),
+            "detail": (row or {}).get("detail"),
+            "window": f"{self.pipeline.settings.revision_review_start}-{self.pipeline.settings.revision_review_cutoff}",
+        }
+
+    def _revision_ready(self, day: str) -> bool:
+        return all(
+            self.pipeline.active_batch(dataset, day)
+            for dataset in revision_datasets(self.pipeline.settings.quality)
+        )
+
+    def _revision_due(self, now: datetime, row: dict[str, Any] | None) -> bool:
+        if row is None or not row.get("last_attempt_at"):
+            return True
+        try:
+            last = datetime.fromisoformat(str(row["last_attempt_at"]))
+        except ValueError:
+            return True
+        interval = timedelta(minutes=self.pipeline.settings.revision_review_interval_minutes)
+        return now_shanghai(last).replace(tzinfo=None) + interval <= now.replace(tzinfo=None)
+
+    def _revision_review_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day or not self._revision_ready(day):
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.revision_review_start)
+        cutoff = _hhmm(settings.revision_review_cutoff)
+        row = self.revision_progress(day)
+        if current < start:
+            if row is None:
+                self._save_revision_progress(day, state="waiting_review")
+            return []
+        if current >= cutoff:
+            if row is None or row["state"] not in {"aligned", "cutoff"}:
+                detail = "复核窗口已截止"
+                self._save_revision_progress(
+                    day, state="cutoff", finished_at=isoformat(now), detail=detail,
+                )
+                with self.db.write() as connection:
+                    connection.execute(
+                        "INSERT INTO job_runs(job_id, state, started_at, finished_at, error, attempt, detail)"
+                        " VALUES ('eod_revise','failed',?,?,?,?,?)",
+                        (
+                            isoformat(now), isoformat(now), detail,
+                            int((row or {}).get("attempts") or 0), "revision cutoff reached",
+                        ),
+                    )
+            elif row["state"] == "aligned" and not row.get("finished_at"):
+                self._save_revision_progress(day, finished_at=isoformat(now))
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(day, now, catchup=False)
+
+    def _revision_catchup_tick(self, now: datetime, day: str) -> list[str]:
+        prev = previous_open_day(self.db, day)
+        if prev is None or prev >= day:
+            return []
+        if not self._revision_ready(prev):
+            return []
+        row = self.revision_progress(prev)
+        if row and int(row.get("catchup_done") or 0):
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(prev, now, catchup=True)
+
+    def _run_revision_job(self, day: str, now: datetime, catchup: bool) -> list[str]:
+        attempts = int((self.revision_progress(day) or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.revision_review_interval_minutes
+        self._save_revision_progress(
+            day,
+            state="waiting_review",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+        ran: list[str] = []
+        try:
+            out = self.run_job("eod_revise", day)
+        except Exception as exc:
+            LOGGER.warning("revision review failed for %s: %s", day, exc)
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+            )
+            ran.append("eod_revise")
+            return ran
+        ran.append("eod_revise")
+        if out.get("state") == "skipped":
+            return ran
+        result = out.get("result") if isinstance(out.get("result"), dict) else {}
+        failed = [
+            name for name, item in result.items()
+            if isinstance(item, dict) and item.get("state") == "failed"
+        ]
+        review = result.get("review") if isinstance(result.get("review"), dict) else None
+        watched = [
+            result[name]
+            for name in revision_datasets(self.pipeline.settings.quality)
+            if isinstance(result.get(name), dict)
+        ]
+        diff_blob = None
+        if review and review.get("diffs"):
+            diff_blob = json.dumps(review.get("diffs"), ensure_ascii=False)
+        else:
+            for item in watched:
+                if item.get("diffs"):
+                    diff_blob = json.dumps(item.get("diffs"), ensure_ascii=False)
+                    break
+        revised = bool(review and review.get("reason") == "revised")
+        matched = any(item.get("reason") == "unchanged" or item.get("state") == "aligned" for item in watched)
+        if failed:
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+                last_diff=diff_blob,
+            )
+        elif revised or matched:
+            fields: dict[str, Any] = {
+                "state": "aligned",
+                "finished_at": isoformat(now),
+                "detail": "已追平" if revised else "已追平（无变化）",
+                "last_diff": diff_blob,
+            }
+            if catchup:
+                fields["catchup_done"] = 1
+            self._save_revision_progress(day, **fields)
+        return ran
+
+    def _save_revision_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "catchup_done", "last_diff", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM revision_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({
+                    "trade_date": day,
+                    "state": "waiting_review",
+                    "attempts": 0,
+                    "catchup_done": 0,
+                })
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO revision_progress({','.join(columns)}) VALUES ({placeholders})",
+                    tuple(payload[name] for name in columns),
+                )
+            else:
+                assignments = ", ".join(f"{name} = ?" for name in fields)
+                connection.execute(
+                    f"UPDATE revision_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
     def _record_eod_attempt(self, day: str, now: datetime) -> None:
         row = self.eod_progress(day)
         attempts = int((row or {}).get("attempts") or 0) + 1
@@ -320,6 +534,9 @@ class Scheduler:
     def _eod_retry(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_missing(trade_date)
 
+    def _eod_revise(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.review_published_revisions(trade_date)
+
     def _stocks_refresh(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.refresh_stocks(trade_date)
 

xiaobai-datahub/datahub/settings.py

@@ -79,6 +79,20 @@ class Settings:
     def eod_retry_cutoff(self) -> str:
         return str(self.quality.get("eod_retry_cutoff") or "23:30")
 
+    @property
+    def revision_review_start(self) -> str:
+        # Before the 21:00 website shadow observation.
+        return str(self.quality.get("revision_review_start") or "20:00")
+
+    @property
+    def revision_review_interval_minutes(self) -> int:
+        return int(self.quality.get("revision_review_interval_minutes") or 30)
+
+    @property
+    def revision_review_cutoff(self) -> str:
+        # Last light review ~23:00; cutoff before the 23:30 observation.
+        return str(self.quality.get("revision_review_cutoff") or "23:20")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/tests/test_atomic_release.py

@@ -0,0 +1,408 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.settings import Settings
+from datahub.serving import V1API
+from tests.fixtures import TRADE_DATE, fake_transport
+
+GROUP_A = ("daily", "valuation", "moneyflow", "auction")
+
+
+class GroupTransport:
+    """fake_transport with per-API degradation switches for release-group tests."""
+
+    def __init__(self) -> None:
+        self.empty: set[str] = set()
+        self.keep_rows: dict[str, int] = {}
+        self.stocks: list[dict] | None = None
+        self.calls: list[str] = []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        if api_name in self.empty:
+            return []
+        if api_name == "stock_basic" and self.stocks is not None:
+            return [dict(row) for row in self.stocks]
+        rows = fake_transport(api_name, params, fields)
+        keep = self.keep_rows.get(api_name)
+        if keep is not None:
+            return rows[:keep]
+        return rows
+
+
+def make_pipe(transport: GroupTransport, quality_extra: dict | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    adapter = TushareAdapter("test-token", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.98,
+        "null_rate_max": 0.01,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        admin_password="admin-pass",
+        tushare_token="test-token",
+        db_path=db.path,
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    pipe._tmp = tmp
+    return pipe, db
+
+
+def publications_map(db: HubDB, day: str) -> dict[str, str]:
+    rows = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+    return {str(row["dataset"]): str(row["active_batch"]) for row in rows}
+
+
+class ReleaseGroupSwitchTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+
+    def test_whole_group_switches_in_one_publish_instant(self) -> None:
+        results = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(set(results), {*GROUP_A, "stocks"})
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (TRADE_DATE,))
+        self.assertEqual(len(pubs), 5)
+        self.assertEqual(len({row["published_at"] for row in pubs}), 1)
+        # official rows copied and serving resolves the new batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        payload = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(payload["meta"]["batch_id"], results["daily"]["batch_id"])
+        stocks = api.handle("/v1/stocks", {})
+        self.assertEqual(stocks["meta"]["batch_id"], results["stocks"]["batch_id"])
+
+    def test_any_member_failure_blocks_entire_group(self) -> None:
+        self.transport.empty = {"daily_basic"}  # valuation upstream returns nothing
+        results = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["valuation"]["state"], "failed")
+        self.assertEqual(results["moneyflow"]["state"], "aborted")
+        self.assertEqual(results["auction"]["state"], "aborted")
+        self.assertEqual(results["daily"]["state"], "failed")  # staged fine, then abandoned
+        # nothing became visible, and the reason is recorded
+        self.assertEqual(publications_map(self.db, TRADE_DATE), {})
+        abandoned = self.db.fetchall(
+            "SELECT * FROM batches WHERE trade_date = ? AND state = 'failed'",
+            (TRADE_DATE,),
+        )
+        self.assertTrue(any("release group not switched" in str(row["error"] or "") for row in abandoned))
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIn("valuation", str(audit["detail"]))
+        # still missing → evening retries keep trying
+        self.assertIn("daily", self.pipe.missing_official_datasets(TRADE_DATE))
+
+    def test_failure_keeps_previous_complete_version_serving(self) -> None:
+        first = self.pipe.run_dataset("daily", TRADE_DATE)
+        self.transport.empty = {"daily_basic"}
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # incomplete A-group restages daily with the others; valuation fails → no A switch
+        self.assertEqual(results["daily"]["state"], "failed")
+        self.assertEqual(results["valuation"]["state"], "failed")
+        # the already-published daily batch is untouched and keeps serving
+        self.assertEqual(self.pipe.active_batch("daily", TRADE_DATE), first["batch_id"])
+        pubs = publications_map(self.db, TRADE_DATE)
+        self.assertEqual(pubs["daily"], first["batch_id"])
+        self.assertNotIn("valuation", pubs)
+        self.assertNotIn("moneyflow", pubs)
+        self.assertNotIn("auction", pubs)
+        # B-group is an independent boundary and may still publish
+        self.assertEqual(results["index_daily"]["state"], "published")
+        payload = V1API(self.db, self.pipe, self.pipe.settings).handle(
+            "/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]}
+        )
+        self.assertEqual(payload["meta"]["batch_id"], first["batch_id"])
+
+    def test_partial_group_retry_does_not_mix_batches(self) -> None:
+        """Already-published A members must be restaged with missing ones."""
+        first_daily = self.pipe.run_dataset("daily", TRADE_DATE)
+        first_moneyflow = self.pipe.run_dataset("moneyflow", TRADE_DATE)
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # A-group switched as one boundary; B-group (index) also published
+        for name in (*GROUP_A, "stocks"):
+            self.assertEqual(results[name]["state"], "published", name)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        pubs = self.db.fetchall(
+            "SELECT dataset, active_batch, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        # old partial batches replaced — no cross-batch mix of the first wave
+        self.assertNotEqual(by_ds["daily"]["active_batch"], first_daily["batch_id"])
+        self.assertNotEqual(by_ds["moneyflow"]["active_batch"], first_moneyflow["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in (*GROUP_A, "stocks")}
+        self.assertEqual(len(a_times), 1)
+        # serving resolves the new complete A-group batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        daily = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(daily["meta"]["batch_id"], results["daily"]["batch_id"])
+        self.assertEqual(daily["meta"]["batch_id"], by_ds["daily"]["active_batch"])
+
+    def test_reads_during_switch_see_old_state_until_commit(self) -> None:
+        snapshots: list[dict] = []
+
+        def watcher() -> None:
+            with self.db.connect() as connection:
+                rows = connection.execute(
+                    "SELECT dataset, active_batch FROM publications WHERE trade_date = ?",
+                    (TRADE_DATE,),
+                ).fetchall()
+                snapshots.append({str(row["dataset"]): row["active_batch"] for row in rows})
+
+        self.pipe.before_commit = watcher
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        # inside the switch transaction the group was still invisible
+        self.assertEqual(snapshots[0], {})
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertEqual(set(after), {*GROUP_A, "stocks"})
+
+    def test_switch_crash_rolls_back_whole_group(self) -> None:
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(publications_map(self.db, TRADE_DATE), {})
+        for table in ("eod_bars", "eod_valuation", "eod_moneyflow", "eod_auction", "eod_stocks"):
+            rows = self.db.fetchall(f"SELECT * FROM {table} WHERE trade_date = ?", (TRADE_DATE,))
+            self.assertEqual(rows, [], table)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        detail = str(audit["detail"])
+        self.assertIn("killed mid-switch", detail)
+        self.assertIn("failed", detail)
+
+    def test_duplicate_runs_are_idempotent(self) -> None:
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+        batches_before = {
+            str(row["batch_id"])
+            for row in self.db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (TRADE_DATE,))
+        }
+        calls_before = len(self.transport.calls)
+        again = self.pipe.run_eod_missing(TRADE_DATE)
+        self.assertEqual({item["state"] for item in again.values()}, {"skipped"})
+        self.assertEqual({item["reason"] for item in again.values()}, {"already_published"})
+        batches_after = {
+            str(row["batch_id"])
+            for row in self.db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (TRADE_DATE,))
+        }
+        self.assertEqual(batches_after, batches_before)
+        self.assertEqual(len(self.transport.calls), calls_before)
+        self.assertEqual(self.pipe.missing_official_datasets(TRADE_DATE), [])
+
+    def test_cross_gate_failure_blocks_switch(self) -> None:
+        transport = GroupTransport()
+        pipe, db = make_pipe(
+            transport,
+            quality_extra={"cross_gates": [
+                {"left": "daily", "right": "moneyflow", "min_key_overlap": 1.0},
+            ]},
+        )
+        pipe.ingest_reference(TRADE_DATE)
+        transport.keep_rows["moneyflow"] = 1  # moneyflow covers only half the market
+        results = pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["moneyflow"]["state"], "failed")
+        self.assertIn("cross gate", str(results["moneyflow"]["error"]))
+        self.assertEqual(publications_map(db, TRADE_DATE), {})
+
+    def test_stocks_master_and_snapshot_switch_together_or_not_at_all(self) -> None:
+        original = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+        renamed = [dict(original[0]), {**original[1], "name": "金钛股份"}]
+        self.transport.stocks = renamed
+        self.pipe.run_eod_batch_a(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "金钛股份")
+        stocks_pub = self.db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset = 'stocks' AND trade_date = ?",
+            (TRADE_DATE,),
+        )
+        self.assertIsNotNone(stocks_pub)
+
+        # failure path: rename staged but the group is blocked → master stays untouched
+        transport = GroupTransport()
+        transport.stocks = original
+        pipe, db = make_pipe(
+            transport,
+            quality_extra={"cross_gates": [
+                {"left": "daily", "right": "moneyflow", "min_key_overlap": 1.0},
+            ]},
+        )
+        pipe.ingest_reference(TRADE_DATE)  # master seeded with "N金钛"
+        transport.stocks = renamed
+        transport.keep_rows["moneyflow"] = 1
+        results = pipe.run_eod_batch_a(TRADE_DATE)
+        self.assertEqual(results["stocks"]["state"], "failed")
+        master = db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        stocks_pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset = 'stocks' AND trade_date = ?",
+            (TRADE_DATE,),
+        )
+        self.assertIsNone(stocks_pub)
+
+
+class StocksRefreshAtomicTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(first["state"], "published")
+        self.first_batch = first["batch_id"]
+
+    def test_refresh_keeps_master_when_snapshot_publish_fails(self) -> None:
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "金钛股份", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+
+        def explode() -> None:
+            raise RuntimeError("snapshot switch killed")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("snapshot switch killed", str(audit["detail"]))
+
+    def test_refresh_keeps_master_when_quality_gate_rejects(self) -> None:
+        self.transport.stocks = []  # empty → hard fail before publish
+        with self.assertRaises(Exception):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+
+
+class ForceBoundaryEntryTests(unittest.TestCase):
+    """CLI force / admin backfill must rebuild the full A/B boundary."""
+
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.first = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+
+    def test_force_republish_valuation_rebuilds_whole_a_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+            self.assertEqual(after[name], results[name]["batch_id"], name)
+        # B-group left alone
+        self.assertEqual(after["index_daily"], before["index_daily"])
+        pubs = self.db.fetchall(
+            "SELECT dataset, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        a_times = {row["published_at"] for row in pubs if row["dataset"] in {*GROUP_A, "stocks"}}
+        self.assertEqual(len(a_times), 1)
+
+    def test_force_republish_index_rebuilds_only_b_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("index_daily", TRADE_DATE)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertNotEqual(after["index_daily"], before["index_daily"])
+        for name in GROUP_A:
+            self.assertEqual(after[name], before[name], name)
+
+    def test_admin_backfill_official_dataset_uses_boundary(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+        result = admin.backfill("moneyflow", TRADE_DATE, "StartPass1", f"moneyflow:{TRADE_DATE}", "tester")
+        self.assertEqual(result["moneyflow"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+        # bad password / wrong confirm still rejected
+        with self.assertRaises(ApiError):
+            admin.backfill("daily", TRADE_DATE, "wrong", f"daily:{TRADE_DATE}", "tester")
+
+    def test_admin_backfill_switch_crash_is_failed_precondition(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(ApiError) as ctx:
+            admin.backfill("valuation", TRADE_DATE, "StartPass1", f"valuation:{TRADE_DATE}", "tester")
+        self.assertEqual(ctx.exception.code, "FAILED_PRECONDITION")
+        self.assertIn("killed mid-switch", ctx.exception.message)
+        # previous complete A/B versions keep serving
+        self.assertEqual(publications_map(self.db, TRADE_DATE), before)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("killed mid-switch", str(audit["detail"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_eod_retry.py

@@ -111,15 +111,24 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(progress["state"], "done")
         self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
 
-        # success stops all further same-day requests
+        # success stops further eod_retry; revision window has not started yet
         batches_before = len(self._batches(db, day))
         eod_calls_before = len(self._eod_calls(transport))
         sched.tick(clock_at(day, 17, 0))
-        sched.tick(clock_at(day, 23, 0))
         self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 0)
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(self._eod_calls(transport)), eod_calls_before)
 
+        # 23:00 is inside the valuation review window: light daily_basic only, no new batch
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 1)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        extra = [name for name in self._eod_calls(transport)[eod_calls_before:]]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
+
     def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
         day = "20240902"
         db, transport, pipe, sched = self._make(set())
@@ -174,6 +183,7 @@ class EodRetryTests(unittest.TestCase):
         self.assertIn("eod_a", ran)
         self.assertIn("eod_b", ran)
         self.assertNotIn("eod_retry", ran)
+        self.assertIn("eod_revise", ran)
         self.assertEqual(self._published(db, day), OFFICIAL)
         after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
         self.assertEqual(
@@ -181,7 +191,9 @@ class EodRetryTests(unittest.TestCase):
             active_map,
         )
         self.assertEqual(set(official_batches()), batches_before)  # no duplicate batches
-        self.assertEqual(self._eod_calls(transport), calls_before)  # no duplicate upstream EOD calls
+        extra = self._eod_calls(transport)[len(calls_before):]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
         self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 21, 0))["state"], "done")
 
     def test_restart_with_partial_publish_only_fetches_missing(self) -> None:
@@ -206,6 +218,7 @@ class EodRetryTests(unittest.TestCase):
         for hh, mm in ((15, 5), (15, 10), (15, 40), (16, 10), (20, 0), (23, 40)):
             ran = sched.tick(clock_at(day, hh, mm))
             self.assertNotIn("eod_retry", ran)
+            self.assertNotIn("eod_revise", ran)
         eod_runs = db.fetchall("SELECT * FROM job_runs WHERE job_id LIKE 'eod%'")
         self.assertEqual(eod_runs, [])
         self.assertIsNone(db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,)))
@@ -226,12 +239,18 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(transport.calls), calls_before)
 
+        revised = sched.run_job("eod_revise", day)
+        self.assertEqual(revised["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+
         sched._eod_lock.acquire()  # simulate an in-flight EOD job
         try:
             busy = sched.run_job("eod_retry", day)
             self.assertEqual(busy["state"], "skipped")
             busy_a = sched.run_job("eod_a", day)
             self.assertEqual(busy_a["state"], "skipped")
+            busy_r = sched.run_job("eod_revise", day)
+            self.assertEqual(busy_r["state"], "skipped")
         finally:
             sched._eod_lock.release()
         self.assertEqual(len(self._batches(db, day)), batches_before)

xiaobai-datahub/tests/test_quality_gates.py

@@ -228,25 +228,34 @@ class GateRetryInterplayTests(unittest.TestCase):
 
 
 class ForceRepublishTests(unittest.TestCase):
-    def test_run_dataset_over_published_keeps_prev_for_rollback(self) -> None:
+    def test_force_boundary_republish_keeps_prev_for_rollback(self) -> None:
         transport = ValuationTransport()
         pipe, db = make_pipe(transport)
         pipe.ingest_reference(TRADE_DATE)
-        first = pipe.run_dataset("valuation", TRADE_DATE)
+        first = pipe.run_eod_batch_a(TRADE_DATE)
+        first_val = first["valuation"]["batch_id"]
+        first_daily = first["daily"]["batch_id"]
         transport.mode = "vr_all_null"
-        with self.assertRaises(QualityError):
-            pipe.run_dataset("valuation", TRADE_DATE)  # gate holds: bad re-publish refused
+        blocked = pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual(blocked["valuation"]["state"], "failed")
+        self.assertEqual(pipe.active_batch("valuation", TRADE_DATE), first_val)
+        self.assertEqual(pipe.active_batch("daily", TRADE_DATE), first_daily)
         transport.mode = "ok"
-        second = pipe.run_dataset("valuation", TRADE_DATE)  # CLI --force path
-        self.assertNotEqual(first["batch_id"], second["batch_id"])
-        pub = db.fetchone(
-            "SELECT * FROM publications WHERE dataset='valuation' AND trade_date=?",
+        second = pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual(second["valuation"]["state"], "published")
+        self.assertNotEqual(second["valuation"]["batch_id"], first_val)
+        self.assertNotEqual(second["daily"]["batch_id"], first_daily)
+        pubs = db.fetchall(
+            "SELECT dataset, active_batch, prev_batch, published_at FROM publications WHERE trade_date=?",
             (TRADE_DATE,),
         )
-        self.assertEqual(pub["active_batch"], second["batch_id"])
-        self.assertEqual(pub["prev_batch"], first["batch_id"])
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        a_times = {by_ds[name]["published_at"] for name in ("daily", "valuation", "moneyflow", "auction", "stocks")}
+        self.assertEqual(len(a_times), 1)
+        self.assertEqual(by_ds["valuation"]["active_batch"], second["valuation"]["batch_id"])
+        self.assertEqual(by_ds["valuation"]["prev_batch"], first_val)
         rolled = pipe.rollback("valuation", TRADE_DATE, actor="cli")
-        self.assertEqual(rolled["active_batch"], first["batch_id"])
+        self.assertEqual(rolled["active_batch"], first_val)
 
 
 if __name__ == "__main__":

xiaobai-datahub/tests/test_revision_review.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.base import AdapterError
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.scheduler import Scheduler
+from datahub.serving import V1API
+from datahub.settings import Settings
+from datahub.timeutil import SHANGHAI
+from tests.fixtures import RAW, TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+from tests.test_quality_gates import FIELD_GATES
+
+SAMPLE_DAY = "20260907"
+NEXT_DAY = "20260908"
+SAMPLE_CODE = "003021.SZ"
+
+
+def _dated(row: dict, day: str) -> dict:
+    item = dict(row)
+    if "trade_date" in item:
+        item["trade_date"] = day
+    return item
+
+
+class RevisingTransport:
+    """Fixture transport that can rewrite daily_basic after the first publish."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, extra_calendar: list[dict] | None = None) -> None:
+        self.calls: list[str] = []
+        self.fail_daily_basic = False
+        self.empty_daily_basic = False
+        self.null_volume_ratio = False
+        self.turnover_by_code: dict[str, float] = {}
+        self.extra_calendar = extra_calendar or []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        day = str(params.get("trade_date") or "")
+        if api_name == "trade_cal":
+            rows = fake_transport(api_name, params, fields)
+            extra = [
+                row for row in self.extra_calendar
+                if str(params.get("start_date") or "") <= row["cal_date"] <= str(params.get("end_date") or "99999999")
+            ]
+            return rows + extra
+        if self.fail_daily_basic and api_name == "daily_basic":
+            raise AdapterError("tushare daily_basic unavailable")
+        if self.empty_daily_basic and api_name == "daily_basic":
+            return []
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+            if day:
+                rows = [_dated(row, day) for row in rows]
+            return rows
+        rows = fake_transport(api_name, params, fields)
+        if api_name == "stock_basic":
+            rows = list(rows)
+            rows.append({
+                "ts_code": SAMPLE_CODE, "symbol": "003021", "name": "兆威机电",
+                "area": "广东", "industry": "元器件", "market": "主板",
+                "list_status": "L", "list_date": "20201202",
+            })
+            return rows
+        if api_name in self.DATE_APIS:
+            template = RAW.get(api_name) or []
+            if not day:
+                return [_dated(row, TRADE_DATE) for row in template]
+            out = [_dated(row, day) for row in template]
+            extra = copy.deepcopy(template[0])
+            extra["ts_code"] = SAMPLE_CODE
+            extra["trade_date"] = day
+            if api_name == "daily_basic":
+                extra["turnover_rate"] = self.turnover_by_code.get(SAMPLE_CODE, extra.get("turnover_rate"))
+                if self.null_volume_ratio:
+                    extra["volume_ratio"] = None
+                    for row in out:
+                        row["volume_ratio"] = None
+            out.append(extra)
+            if api_name == "daily_basic":
+                for row in out:
+                    code = str(row.get("ts_code") or "")
+                    if code in self.turnover_by_code:
+                        row["turnover_rate"] = self.turnover_by_code[code]
+            return out
+        return rows
+
+
+def make_revision_env(quality_extra: dict | None = None, extra_calendar: list[dict] | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    transport = RevisingTransport(extra_calendar=extra_calendar)
+    adapter = TushareAdapter("x", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.5,
+        "null_rate_max": 0.5,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+        "field_gates": FIELD_GATES,
+        "revision_review_start": "20:00",
+        "revision_review_interval_minutes": 30,
+        "revision_review_cutoff": "23:20",
+        "revision_review_datasets": ["valuation"],
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        db_path=db.path,
+        backup_dir=Path(tmp.name) / "backups",
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    sched = Scheduler(db, pipe)
+    return tmp, db, transport, pipe, sched
+
+
+SAMPLE_CALENDAR = [
+    {"exchange": "SSE", "cal_date": SAMPLE_DAY, "is_open": 1, "pretrade_date": "20260906"},
+    {"exchange": "SSE", "cal_date": NEXT_DAY, "is_open": 1, "pretrade_date": SAMPLE_DAY},
+]
+
+
+class RevisionReviewTests(unittest.TestCase):
+    def _publish(self, pipe: Pipeline, day: str) -> None:
+        pipe.ingest_reference(day)
+        pipe.run_eod_batch_a(day)
+        pipe.run_eod_batch_b(day)
+
+    def _turnover(self, db: HubDB, day: str, code: str = SAMPLE_CODE) -> float | None:
+        pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (day,),
+        )
+        row = db.fetchone(
+            "SELECT turnover_rate FROM eod_valuation WHERE batch_id=? AND ts_code=?",
+            (pub["active_batch"], code),
+        )
+        return None if row is None else row["turnover_rate"]
+
+    def _batch_ids(self, db: HubDB, day: str) -> set[str]:
+        return {str(row["batch_id"]) for row in db.fetchall("SELECT batch_id FROM batches WHERE trade_date=?", (day,))}
+
+    def test_no_change_does_not_create_a_new_batch(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), before)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertIn("无变化", progress["detail"])
+        status = sched.revision_status(TRADE_DATE, clock=clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(status["state"], "aligned")
+
+    def test_hel423_20260907_single_field_revision_is_caught_up(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+        first = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        seen: list[float | None] = []
+
+        def watch() -> None:
+            seen.append(self._turnover(db, SAMPLE_DAY))
+
+        pipe.before_commit = watch
+        ran = sched.tick(clock_at(SAMPLE_DAY, 20, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(seen, [1.3565])  # readers still see the previous complete version mid-switch
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        second = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+        self.assertNotEqual(second, first)
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.valuation({"date": SAMPLE_DAY, "code": SAMPLE_CODE})
+        row = next(item for item in payload["data"] if item["ts_code"] == SAMPLE_CODE)
+        self.assertEqual(row["turnover_rate"], 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(progress["detail"], "已追平")
+        audit = db.fetchone(
+            "SELECT * FROM audit_log WHERE action='revision-review' ORDER BY id DESC"
+        )
+        self.assertIn("1.3572", str(audit["detail"]))
+        self.assertIn(SAMPLE_CODE, str(audit["detail"]))
+
+    def test_empty_or_failed_upstream_keeps_previous_version(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        batches = self._batch_ids(db, TRADE_DATE)
+
+        transport.empty_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+        transport.empty_daily_basic = False
+        transport.fail_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 30))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), batches)
+
+    def test_quality_gate_rejects_catchup_and_keeps_previous(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        transport.turnover_by_code[SAMPLE_CODE] = 9.9999
+        transport.null_volume_ratio = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+    def test_repeat_ticks_after_align_do_not_republish(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        after_fix = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 10))  # inside interval
+        self.assertEqual(len(db.fetchall("SELECT * FROM job_runs WHERE job_id='eod_revise'")), 1)
+        sched.tick(clock_at(TRADE_DATE, 20, 30))  # next light compare, no change
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), after_fix)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_restart_catches_up_inside_window(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(TRADE_DATE, 21, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_cutoff_stops_evening_reviews_and_morning_catchup_runs(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        sched.tick(clock_at(SAMPLE_DAY, 23, 25))  # past 23:20 cutoff, no review yet
+        cutoff = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(cutoff["state"], "cutoff")
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(SAMPLE_DAY, 23, 50))  # still same calendar day, no catch-up
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        ran = sched.tick(clock_at(NEXT_DAY, 8, 45))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(int(progress["catchup_done"]), 1)
+
+        batches = self._batch_ids(db, SAMPLE_DAY)
+        sched.tick(clock_at(NEXT_DAY, 8, 50))
+        self.assertEqual(self._batch_ids(db, SAMPLE_DAY), batches)
+
+    def test_only_valuation_is_light_fetched(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = [name for name in transport.calls]
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        extra = transport.calls[len(before):]
+        self.assertIn("daily_basic", extra)
+        self.assertNotIn("daily", extra)
+        self.assertNotIn("moneyflow", extra)
+        self.assertNotIn("stk_auction", extra)
+        self.assertNotIn("index_daily", extra)
+
+
+if __name__ == "__main__":
+    unittest.main()