diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index e474b29..607e299 100644 --- a/xiaobai-datahub/README.md +++ b/xiaobai-datahub/README.md @@ -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 整批边界。 ## 备份 diff --git a/xiaobai-datahub/admin/app.js b/xiaobai-datahub/admin/app.js index 119796f..5e19f6e 100644 --- a/xiaobai-datahub/admin/app.js +++ b/xiaobai-datahub/admin/app.js @@ -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; diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index b9bf4e9..fb284f5 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/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 @@ -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 diff --git a/xiaobai-datahub/datahub/cli.py b/xiaobai-datahub/datahub/cli.py index daea873..b94e025 100644 --- a/xiaobai-datahub/datahub/cli.py +++ b/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) diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 599b24e..9c9fa81 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -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 diff --git a/xiaobai-datahub/tests/test_atomic_release.py b/xiaobai-datahub/tests/test_atomic_release.py index 71c3dbc..49a1ffc 100644 --- a/xiaobai-datahub/tests/test_atomic_release.py +++ b/xiaobai-datahub/tests/test_atomic_release.py @@ -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() diff --git a/xiaobai-datahub/tests/test_quality_gates.py b/xiaobai-datahub/tests/test_quality_gates.py index c772703..ecc1db6 100644 --- a/xiaobai-datahub/tests/test_quality_gates.py +++ b/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__":