fix(HEL-396): 禁止空数据批次冒充正式发布

统一发布入口在有效行数为 0 时不再写成 published，也不推进正式批次指针。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

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})

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()