fix(HEL-461): CLI/后台强制重发改为整组边界切换

eod-refresh --force 与管理后台补数不再单数据集发布，统一走 force_republish_boundary，避免绕过 A/B 完整边界。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
2026-09-05 11:26:27 +08:00
parent 32f565ecb9
commit 75c2e33b68
7 changed files with 140 additions and 46 deletions
@@ -115,11 +115,13 @@ 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 整批边界。

 ## 备份

@@ -269,7 +269,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;
@@ -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
@@ -143,8 +143,14 @@ 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.
+                result = self.pipeline.force_republish_boundary(dataset, day)
+                failures = self.pipeline.eod_failures(result)
+                if failures:
+                    raise ApiError("FAILED_PRECONDITION", "; ".join(failures))
            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

@@ -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)
@@ -666,17 +666,33 @@ class Pipeline:
        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)
+    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_release_group(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 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 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.

@@ -689,11 +705,11 @@ class Pipeline:
        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.
+        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] = {}
@@ -702,7 +718,7 @@ class Pipeline:
        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:
+        if not force and not missing_official and not stocks_missing:
            for dataset in datasets:
                results[dataset] = {
                    "dataset": dataset,
@@ -719,8 +735,8 @@ class Pipeline:
                }
            return results

-        # Incomplete boundary → restage every official member together.
-        pending = list(datasets) if missing_official else []
+        # Incomplete or forced boundary → restage every official member together.
+        pending = list(datasets)

        for dataset in pending:
            if failure is not None:
@@ -743,10 +759,12 @@ class Pipeline:
                }

        # 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):
+        # 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=bool(missing_official))
+                stocks_plan = self._stage_stocks_snapshot(day, force=rebuild_official)
            except Exception as exc:
                failure = f"{STOCKS_DATASET}: {exc}"
                results[STOCKS_DATASET] = {
@@ -809,7 +827,8 @@ class Pipeline:
            self.audit(
                "pipeline", "release-group", f"eod:{day}",
                json.dumps(
-                    {"state": "ok", "switched": sorted(staged)}, ensure_ascii=False
+                    {"state": "ok", "switched": sorted(staged), "force": bool(force)},
+                    ensure_ascii=False,
                ),
            )
        return results
@@ -311,5 +311,62 @@ class StocksRefreshAtomicTests(unittest.TestCase):
        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")
+
+
 if __name__ == "__main__":
    unittest.main()
@@ -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__":