diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index 776fd41..b4e36ac 100644 --- a/backend/data/datahub/bridge.py +++ b/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 diff --git a/backend/data/datahub/compare.py b/backend/data/datahub/compare.py index 08b45de..34eeba7 100644 --- a/backend/data/datahub/compare.py +++ b/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) diff --git a/tests/test_datahub_bridge.py b/tests/test_datahub_bridge.py index 5200412..23155df 100644 --- a/tests/test_datahub_bridge.py +++ b/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) diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index f440af9..be86a7a 100644 --- a/xiaobai-datahub/README.md +++ b/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 状态。 diff --git a/xiaobai-datahub/config/hub-quality.config.json b/xiaobai-datahub/config/hub-quality.config.json index d0fafea..93c4514 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/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": [ diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 89bdd0f..4e7b0a3 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/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) diff --git a/xiaobai-datahub/tests/test_atomic_release.py b/xiaobai-datahub/tests/test_atomic_release.py new file mode 100644 index 0000000..4ca2267 --- /dev/null +++ b/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()