fix(HEL-461): 整批发布按完整边界重暂存，主档与快照同事务

边界内任有缺失则整组重暂存后统一切换，避免旧新批次混发；
refresh_stocks 失败时主档保持旧值，并补齐回归测试。

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

xiaobai-datahub/datahub/pipeline.py

@@ -348,11 +348,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())
-        rows = self._fetch_dataset(STOCKS_DATASET, day)
-        with self.db.write() as connection:
-            self._upsert_stock_master(connection, rows, isoformat(self.clock()))
+        try:
+            rows = self._fetch_dataset(STOCKS_DATASET, day)
+        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:
@@ -369,7 +378,14 @@ class Pipeline:
                         "batch_id": active,
                         "rows": len(snapshot),
                     }
-        result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        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),
@@ -637,14 +653,18 @@ 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. All missing
-        members stage first and switch in one atomic release group; a single
-        member failure keeps the previous complete official version serving.
+        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_release_group(tuple(sorted(OFFICIAL_DATASETS)), trade_date, include_stocks=True)
+        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_release_group(EOD_A_DATASETS, trade_date, include_stocks=True)
@@ -661,29 +681,46 @@ class Pipeline:
         """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.
+        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: if every official member
+        (and stocks when required) is already published, the group is
+        skipped. If any official member is still missing, every official
+        member is re-staged — including ones that already had a publication
+        — so a retry 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
-        pending: list[str] = []
-        for dataset in datasets:
-            if self.active_batch(dataset, day) is not 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 missing_official and not stocks_missing:
+            for dataset in datasets:
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "skipped",
                     "reason": "already_published",
                 }
-            else:
-                pending.append(dataset)
+            if include_stocks:
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+            return results
+
+        # Incomplete boundary → restage every official member together.
+        pending = list(datasets) if missing_official else []
 
         for dataset in pending:
             if failure is not None:
@@ -705,9 +742,11 @@ class Pipeline:
                     "error": str(exc),
                 }
 
-        if include_stocks and failure is None and self.active_batch(STOCKS_DATASET, day) is None:
+        # Stocks join the same switch when the official boundary is being
+        # rebuilt, or when only the stocks snapshot is still missing.
+        if include_stocks and failure is None and (missing_official or stocks_missing):
             try:
-                stocks_plan = self._stage_stocks_snapshot(day)
+                stocks_plan = self._stage_stocks_snapshot(day, force=bool(missing_official))
             except Exception as exc:
                 failure = f"{STOCKS_DATASET}: {exc}"
                 results[STOCKS_DATASET] = {
@@ -820,18 +859,18 @@ class Pipeline:
             "state": "degraded" if report["soft_fail"] else "published",
         }
 
-    def _stage_stocks_snapshot(self, trade_date: str) -> dict[str, Any] | None:
+    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). The stock_master upsert is deferred into
-        the group switch transaction so the master never runs ahead of the
-        published snapshot.
+        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:
+        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
             )
@@ -1077,6 +1116,16 @@ class Pipeline:
         with self.db.write() as connection:
             _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()
             _upsert_publication(connection, dataset, trade_date, batch_id, state, published_at)