fix(HEL-494): retry incomplete sector publications

Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/datahub/pipeline.py

@@ -766,7 +766,7 @@ class Pipeline:
         results: dict[str, Any] = {}
         day = yyyymmdd(trade_date)
         for dataset in datasets:
-            if not force and self.active_batch(dataset, day):
+            if not force and self.extended_publication_complete(dataset, day):
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
@@ -796,6 +796,22 @@ class Pipeline:
                 LOGGER.exception("extended soft publish failed dataset=%s date=%s", dataset, day)
         return results
 
+    def extended_publication_complete(self, dataset: str, trade_date: str) -> bool:
+        """Do not mistake a partially published multi-source dataset for completion."""
+        day = yyyymmdd(trade_date)
+        batch_id = self.active_batch(dataset, day)
+        if not batch_id:
+            return False
+        if dataset != "sector_daily":
+            return True
+        rows = self.db.fetchall(
+            "SELECT DISTINCT family FROM eod_sector_daily "
+            "WHERE trade_date = ? AND batch_id = ?",
+            (day, batch_id),
+        )
+        families = {str(row.get("family") or "") for row in rows}
+        return {"ths", "dc", "sw"} <= families
+
     def run_eod_batch_c(self, trade_date: str, force: bool = False) -> dict[str, Any]:
         return self.run_extended_soft(EOD_C_DATASETS, trade_date, force=force)
 

xiaobai-datahub/datahub/scheduler.py

@@ -206,7 +206,7 @@ class Scheduler:
             LOGGER.warning("eod retry failed for %s", day, exc_info=True)
         ran.append("eod_retry")
         self._settle_eod(day)
-        if "eod_e" in self.jobs and not self.pipeline.active_batch("sector_daily", day):
+        if "eod_e" in self.jobs and not self.pipeline.extended_publication_complete("sector_daily", day):
             try:
                 self.run_job("eod_e", day)
                 ran.append("eod_e")

xiaobai-datahub/datahub/steward.py

@@ -195,6 +195,11 @@ def _try_published(api, api_name: str, dataset: str, params: dict[str, Any], fie
             return None
         raise
     rows = list(payload.get("data") or [])
+    # A published multi-source sector batch can be temporarily incomplete when
+    # one upstream family is late. Let the hub try that family live instead of
+    # returning an authoritative-looking empty result to the website.
+    if dataset == "sector_daily" and query.get("family") and not rows:
+        return None
     if dataset == "stocks":
         rows = _filter_stocks(rows, params)
     if dataset == "calendar":

xiaobai-datahub/tests/test_extended_eod.py

@@ -60,6 +60,22 @@ class ExtendedEodTests(unittest.TestCase):
         names = {item["dataset"] for item in status["data"]}
         self.assertTrue({"limit_events", "popularity", "dragon_tiger", "sector_daily"} <= names)
 
+    def test_partial_sector_families_are_retried_instead_of_skipped(self) -> None:
+        first = self.hub.pipeline.run_eod_batch_e(TRADE_DATE)
+        self.assertEqual(first["sector_daily"]["state"], "published")
+        batch_id = self.hub.pipeline.active_batch("sector_daily", TRADE_DATE)
+        with self.hub.db.write() as connection:
+            connection.execute(
+                "DELETE FROM eod_sector_daily WHERE trade_date = ? AND batch_id = ? AND family = 'sw'",
+                (TRADE_DATE, batch_id),
+            )
+        self.assertFalse(self.hub.pipeline.extended_publication_complete("sector_daily", TRADE_DATE))
+
+        retried = self.hub.pipeline.run_eod_batch_e(TRADE_DATE)
+
+        self.assertEqual(retried["sector_daily"]["state"], "published")
+        self.assertTrue(self.hub.pipeline.extended_publication_complete("sector_daily", TRADE_DATE))
+
 
 if __name__ == "__main__":
     unittest.main()

xiaobai-datahub/tests/test_steward.py

@@ -60,6 +60,26 @@ class StewardQueryTests(unittest.TestCase):
         with self.assertRaises(ApiError):
             self.hub.api.query_api({"api_name": "rt_sw_k", "params": {"ts_code": "801074.SI"}})
 
+    def test_missing_published_sw_family_falls_back_inside_hub(self) -> None:
+        self.hub.pipeline.run_eod_batch_e(TRADE_DATE)
+        batch_id = self.hub.pipeline.active_batch("sector_daily", TRADE_DATE)
+        with self.hub.db.write() as connection:
+            connection.execute(
+                "DELETE FROM eod_sector_daily WHERE trade_date = ? AND batch_id = ? AND family = 'sw'",
+                (TRADE_DATE, batch_id),
+            )
+
+        payload = self.hub.api.query_api(
+            {
+                "api_name": "sw_daily",
+                "params": {"ts_code": "801780.SI", "trade_date": TRADE_DATE},
+                "fields": "ts_code,trade_date,name,pct_change",
+            }
+        )
+
+        self.assertEqual(payload["meta"]["source"], "tushare")
+        self.assertEqual(payload["data"][0]["ts_code"], "801780.SI")
+
     def test_rt_k_uses_free_quotes_not_tushare(self) -> None:
         quotes = [
             {