diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 7f447e9..b8f193b 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -21,6 +21,8 @@ LOGGER = get_logger() HARD_DATASETS = {"daily", "valuation", "index_daily"} SOFT_DATASETS = {"moneyflow", "auction"} +OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS +EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows" STAGING_INSERT = { "daily": ( @@ -97,6 +99,15 @@ EOD_COPY = { } +def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int: + table = DATASET_TABLES[dataset][1] + row = connection.execute( + f"SELECT COUNT(*) AS n FROM {table} WHERE batch_id = ?", + (batch_id,), + ).fetchone() + return int(row["n"] if row is not None else 0) + + class QualityError(RuntimeError): def __init__(self, message: str, report: dict[str, Any]) -> None: super().__init__(message) @@ -176,6 +187,7 @@ class Pipeline: batch_id = self.next_batch_id(dataset, trade_date) max_attempts = attempts or self.settings.max_publish_attempts self._set_batch(batch_id, dataset, trade_date, "scheduled", 0) + rows: list[dict[str, Any]] = [] try: self._set_batch(batch_id, dataset, trade_date, "fetching", 1) rows = retry_call( @@ -189,11 +201,7 @@ class Pipeline: self._set_batch(batch_id, dataset, trade_date, "validating", 1) report = self.validate(dataset, batch_id, trade_date, rows) if report["hard_fail"]: - self._set_batch( - batch_id, dataset, trade_date, "staged", 1, - rows_in=len(rows), rows_out=len(rows), - quality=report, error="; ".join(report["errors"]), - ) + self._reject_batch(batch_id, dataset, trade_date, rows, report) raise QualityError("integrity gate failed", report) self._set_batch(batch_id, dataset, trade_date, "deriving", 1, rows_in=len(rows), rows_out=len(rows), quality=report) self._set_batch(batch_id, dataset, trade_date, "publishing", 1, rows_in=len(rows), rows_out=len(rows), quality=report) @@ -207,7 +215,10 @@ class Pipeline: except RetryError as exc: self._set_batch(batch_id, dataset, trade_date, "failed", max_attempts, error=str(exc), finished=True) raise - except QualityError: + 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, trade_date, rows, exc.report) raise except Exception as exc: self._set_batch(batch_id, dataset, trade_date, "failed", 1, error=str(exc), finished=True) @@ -247,11 +258,13 @@ class Pipeline: null_rate = nulls / row_n if null_rate >= float(quality.get("null_rate_max") or 0.01): errors.append(f"null rate {null_rate:.4f}") - if dataset in SOFT_DATASETS and row_n == 0: - warnings.append("empty soft dataset") - hard_fail = bool(errors) and dataset in HARD_DATASETS.union({"daily", "valuation", "index_daily"}) + empty = row_n == 0 + if empty and dataset in OFFICIAL_DATASETS: + errors.append(EMPTY_BATCH_ERROR) if dataset in SOFT_DATASETS: - hard_fail = bool(dup or bad_date) + hard_fail = bool(dup or bad_date or empty) + else: + hard_fail = bool(errors) and dataset in HARD_DATASETS return { "rows": row_n, "listed": listed_n, @@ -267,6 +280,31 @@ class Pipeline: 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), @@ -364,6 +402,34 @@ class Pipeline: (actor, action, target, detail, isoformat(self.clock())), ) + def _reject_batch( + self, + batch_id: str, + dataset: str, + trade_date: str, + rows: list[dict[str, Any]], + report: dict[str, Any], + ) -> None: + errors = report.get("errors") or [] + LOGGER.warning( + "official batch rejected", + extra={ + "hub": { + "dataset": dataset, + "trade_date": trade_date, + "batch_id": batch_id, + "rows_out": len(rows), + "errors": errors, + "reason": "upstream_empty" if EMPTY_BATCH_ERROR in errors else "integrity_gate", + } + }, + ) + self._set_batch( + batch_id, dataset, trade_date, "staged", 1, + rows_in=len(rows), rows_out=len(rows), + quality=report, error="; ".join(str(item) for item in errors), + ) + def _fetch_dataset(self, dataset: str, trade_date: str) -> list[dict[str, Any]]: if dataset == "daily": raw = self._guarded_fetch("daily", {"trade_date": trade_date}) diff --git a/xiaobai-datahub/tests/test_pipeline.py b/xiaobai-datahub/tests/test_pipeline.py index e435f18..972af7c 100644 --- a/xiaobai-datahub/tests/test_pipeline.py +++ b/xiaobai-datahub/tests/test_pipeline.py @@ -8,11 +8,33 @@ from pathlib import Path from datahub.adapters.tushare import TushareAdapter from datahub.crypto import SecretVault from datahub.db import HubDB -from datahub.pipeline import Pipeline, QualityError +from datahub.pipeline import EMPTY_BATCH_ERROR, Pipeline, QualityError +from datahub.serving import ApiError, V1API from datahub.settings import Settings from datahub.timeutil import SHANGHAI, isoformat from tests.fixtures import TRADE_DATE, fake_transport +DATASET_API = { + "daily": "daily", + "valuation": "daily_basic", + "moneyflow": "moneyflow", + "auction": "stk_auction", + "index_daily": "index_daily", +} + + +def empty_transport_for(*datasets: str): + blocked = {DATASET_API[name] for name in datasets} + if "daily" in datasets: + blocked.add("adj_factor") + + def transport(api_name, params, fields): + if api_name in blocked: + return [] + return fake_transport(api_name, params, fields) + + return transport + def make_pipeline(before_commit=None, clock=None, quality=None) -> tuple[Pipeline, HubDB]: tmp = tempfile.TemporaryDirectory() @@ -144,6 +166,124 @@ class PipelineTests(unittest.TestCase): self.assertEqual(jobs, kept) self.assertEqual(calls, kept) + def test_empty_index_daily_keeps_previous_official_readable(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + first = pipe.run_dataset("index_daily", TRADE_DATE) + self.assertEqual(first["state"], "published") + self.assertGreater(first["rows"], 0) + + pipe.adapter._transport = empty_transport_for("index_daily") + with self.assertRaises(QualityError) as ctx: + pipe.run_dataset("index_daily", TRADE_DATE) + self.assertIn(EMPTY_BATCH_ERROR, ctx.exception.report["errors"]) + + pub = db.fetchone( + "SELECT * FROM publications WHERE dataset='index_daily' AND trade_date=?", + (TRADE_DATE,), + ) + self.assertEqual(pub["active_batch"], first["batch_id"]) + self.assertEqual(pub["state"], "published") + empty_batch = db.fetchone( + "SELECT * FROM batches WHERE batch_id=?", + (f"{TRADE_DATE}-index_daily-002",), + ) + self.assertIsNotNone(empty_batch) + self.assertEqual(empty_batch["state"], "staged") + self.assertEqual(empty_batch["rows_out"], 0) + self.assertNotEqual(empty_batch["state"], "failed") + self.assertIn(EMPTY_BATCH_ERROR, empty_batch["error"] or "") + + api = V1API(pipe.db, pipe, pipe.settings) + payload = api.handle("/v1/indexes/bars", {"date": [TRADE_DATE], "code": ["000001.SH"]}) + self.assertEqual(payload["meta"]["batch_id"], first["batch_id"]) + self.assertEqual(payload["meta"]["tier"], "official") + self.assertEqual(payload["meta"]["state"], "published") + self.assertTrue(payload["data"]) + + def test_empty_index_daily_first_batch_is_not_official(self) -> None: + empty_date = "20260902" + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + pipe.adapter._transport = empty_transport_for("index_daily") + with self.assertRaises(QualityError) as ctx: + pipe.run_dataset("index_daily", empty_date) + self.assertIn(EMPTY_BATCH_ERROR, ctx.exception.report["errors"]) + + batch = db.fetchone("SELECT * FROM batches WHERE batch_id=?", (f"{empty_date}-index_daily-001",)) + self.assertEqual(batch["state"], "staged") + self.assertEqual(batch["rows_out"], 0) + self.assertIsNone( + db.fetchone( + "SELECT * FROM publications WHERE dataset='index_daily' AND trade_date=?", + (empty_date,), + ) + ) + api = V1API(pipe.db, pipe, pipe.settings) + with self.assertRaises(ApiError) as api_ctx: + api.handle("/v1/indexes/bars", {"date": [empty_date], "code": ["000001.SH"]}) + self.assertEqual(api_ctx.exception.code, "DATASET_NOT_PUBLISHED") + + def test_nonempty_index_daily_still_publishes(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + result = pipe.run_dataset("index_daily", TRADE_DATE) + self.assertEqual(result["state"], "published") + self.assertEqual(result["rows"], 4) + pub = db.fetchone( + "SELECT * FROM publications WHERE dataset='index_daily' AND trade_date=?", + (TRADE_DATE,), + ) + self.assertEqual(pub["active_batch"], result["batch_id"]) + rows = db.fetchall("SELECT * FROM eod_index_bars WHERE batch_id=?", (result["batch_id"],)) + self.assertEqual(len(rows), 4) + + def test_empty_batch_guard_covers_all_official_datasets(self) -> None: + for dataset in ("daily", "valuation", "moneyflow", "auction", "index_daily"): + with self.subTest(dataset=dataset): + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + first = pipe.run_dataset(dataset, TRADE_DATE) + pipe.adapter._transport = empty_transport_for(dataset) + with self.assertRaises(QualityError) as ctx: + pipe.run_dataset(dataset, TRADE_DATE) + self.assertTrue(ctx.exception.report["hard_fail"]) + self.assertIn(EMPTY_BATCH_ERROR, ctx.exception.report["errors"]) + pub = db.fetchone( + "SELECT * FROM publications WHERE dataset=? AND trade_date=?", + (dataset, TRADE_DATE), + ) + self.assertEqual(pub["active_batch"], first["batch_id"]) + self.assertEqual(pub["state"], "published") + empty_batch = db.fetchone( + "SELECT * FROM batches WHERE dataset=? AND trade_date=? AND batch_id != ?", + (dataset, TRADE_DATE, first["batch_id"]), + ) + self.assertEqual(empty_batch["state"], "staged") + self.assertEqual(empty_batch["rows_out"], 0) + + def test_publish_entry_rejects_empty_staging_without_moving_pointer(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + first = pipe.run_dataset("moneyflow", TRADE_DATE) + batch_id = pipe.next_batch_id("moneyflow", TRADE_DATE) + pipe._set_batch(batch_id, "moneyflow", TRADE_DATE, "publishing", 1, rows_in=0, rows_out=0) + pipe._stage("moneyflow", batch_id, []) + with self.assertRaises(QualityError) as ctx: + pipe.publish("moneyflow", TRADE_DATE, batch_id, state="published") + self.assertIn(EMPTY_BATCH_ERROR, ctx.exception.report["errors"]) + pub = db.fetchone( + "SELECT * FROM publications WHERE dataset='moneyflow' AND trade_date=?", + (TRADE_DATE,), + ) + self.assertEqual(pub["active_batch"], first["batch_id"]) + self.assertEqual(pub["state"], "published") + history = db.fetchall( + "SELECT batch_id FROM publication_history WHERE dataset='moneyflow' AND trade_date=?", + (TRADE_DATE,), + ) + self.assertEqual({row["batch_id"] for row in history}, {first["batch_id"]}) + if __name__ == "__main__": unittest.main()