fix(HEL-459): 影子比较按请求字段投影，盘后整批原子发布

比较侧只对网站本次请求字段计业务差异，忽略数据中枢额外列；
盘后 A/B/重发改为先整批暂存与交叉校验，再单事务切换公开版本。

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 状态。

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

@@ -22,6 +22,18 @@
     "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/pipeline.py

@@ -133,6 +133,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)
@@ -551,20 +640,40 @@ class Pipeline:
         """Fetch/publish every official dataset still missing 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.
+        repeats never overwrite the current official batch. All missing
+        members stage first and switch in one atomic release group; a single
+        member failure keeps the previous complete official version serving.
         """
-        return self._run_eod_datasets(tuple(sorted(OFFICIAL_DATASETS)), trade_date)
+        return self.run_release_group(tuple(sorted(OFFICIAL_DATASETS)), trade_date, include_stocks=True)
 
     def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_A_DATASETS, trade_date)
+        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True)
 
     def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
-        return self._run_eod_datasets(EOD_B_DATASETS, trade_date)
+        return self.run_release_group(EOD_B_DATASETS, trade_date)
 
-    def _run_eod_datasets(self, datasets: tuple[str, ...], trade_date: str) -> dict[str, Any]:
+    def run_release_group(
+        self,
+        datasets: tuple[str, ...],
+        trade_date: str,
+        include_stocks: 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`` and not yet published) 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.
+        """
         day = yyyymmdd(trade_date)
         results: dict[str, Any] = {}
+        staged: dict[str, dict[str, Any]] = {}
+        failure: str | None = None
+        pending: list[str] = []
         for dataset in datasets:
             if self.active_batch(dataset, day) is not None:
                 results[dataset] = {
@@ -573,18 +682,247 @@ class Pipeline:
                     "state": "skipped",
                     "reason": "already_published",
                 }
+            else:
+                pending.append(dataset)
+
+        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),
                 }
+
+        if include_stocks and failure is None and self.active_batch(STOCKS_DATASET, day) is None:
+            try:
+                stocks_plan = self._stage_stocks_snapshot(day)
+            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:
+                for item in staged.values():
+                    self._abandon_batch(item["batch_id"], f"release group switch failed: {exc}")
+                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)}, 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) -> 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). The stock_master upsert is deferred into
+        the group switch 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:
+            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 +1073,14 @@ 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 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/tests/test_atomic_release.py

@@ -0,0 +1,235 @@
+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] = []
+        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:
+            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)
+        self.assertEqual(results["daily"]["state"], "skipped")
+        self.assertEqual(results["valuation"]["state"], "failed")
+        # the already-published complete batch is untouched and keeps serving
+        self.assertEqual(self.pipe.active_batch("daily", TRADE_DATE), first["batch_id"])
+        self.assertEqual(
+            publications_map(self.db, TRADE_DATE),
+            {"daily": first["batch_id"]},
+        )
+        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_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)
+
+    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)
+
+
+if __name__ == "__main__":
+    unittest.main()