From 32f565ecb9557aca01b19570444a11a21d0f0e9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Sat, 5 Sep 2026 08:52:50 +0800 Subject: [PATCH] =?UTF-8?q?fix(HEL-461):=20=E6=95=B4=E6=89=B9=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E6=8C=89=E5=AE=8C=E6=95=B4=E8=BE=B9=E7=95=8C=E9=87=8D?= =?UTF-8?q?=E6=9A=82=E5=AD=98=EF=BC=8C=E4=B8=BB=E6=A1=A3=E4=B8=8E=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E5=90=8C=E4=BA=8B=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 边界内任有缺失则整组重暂存后统一切换，避免旧新批次混发； refresh_stocks 失败时主档保持旧值，并补齐回归测试。 Co-authored-by: Cursor Co-authored-by: multica-agent --- xiaobai-datahub/README.md | 2 +- xiaobai-datahub/datahub/pipeline.py | 107 ++++++++++++++----- xiaobai-datahub/tests/test_atomic_release.py | 96 +++++++++++++++-- 3 files changed, 167 insertions(+), 38 deletions(-) diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index be86a7a..e474b29 100644 --- a/xiaobai-datahub/README.md +++ b/xiaobai-datahub/README.md @@ -91,7 +91,7 @@ python -m datahub history-backfill - 流程：组内全部成员先在暂存表完成拉取、字段质量门、覆盖检查和跨数据集交叉校验（`cross_gates` 配置 ts_code 覆盖重叠率下限），全部达标后才在**一个 SQLite 事务**里复制正式表并翻转全部 `publications` 指针。 - 任一成员失败（拉取失败、质量门拒绝、交叉校验不过、切换事务中断）→ 整批不切换，对外继续提供上一份完整正式版本，失败原因写入 `batches.error` 与 `audit_log`（`action=release-group`），等待晚间自动重试。 - 读取侧任何时刻只会看到"旧完整版本"或"新完整版本"：发布指针在单事务内统一翻转，容器重启/事务中断自动回滚，不暴露字段残缺或跨数据集混合版本。 -- 幂等：已发布数据集自动跳过；重复执行、并发重试不会生成重复批次或覆盖正常版本（调度器另有 EOD 互斥锁）。 +- 幂等：仅当一致性边界内全部成员都已发布时才整组跳过；边界内任有缺失则整组重暂存后统一切换，避免旧批次与新批次混在同一次重发中。重复执行、并发重试不会在完整边界已就绪时生成重复批次（调度器另有 EOD 互斥锁）。 ## 股票主档每日刷新与发布 diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 4e7b0a3..599b24e 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/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) diff --git a/xiaobai-datahub/tests/test_atomic_release.py b/xiaobai-datahub/tests/test_atomic_release.py index 4ca2267..71c3dbc 100644 --- a/xiaobai-datahub/tests/test_atomic_release.py +++ b/xiaobai-datahub/tests/test_atomic_release.py @@ -21,14 +21,14 @@ class GroupTransport: def __init__(self) -> None: self.empty: set[str] = set() self.keep_rows: dict[str, int] = {} - self.stocks: list[dict] = [] + 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: + 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) @@ -113,19 +113,48 @@ class ReleaseGroupSwitchTests(unittest.TestCase): 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") + # 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 complete batch is untouched and keeps serving + # the already-published daily 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"]}, - ) + 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] = [] @@ -231,5 +260,56 @@ class ReleaseGroupSwitchTests(unittest.TestCase): 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"])) + + if __name__ == "__main__": unittest.main()