diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 0b737c4..808886c 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/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) diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index f9d67eb..d3b3ff7 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/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") diff --git a/xiaobai-datahub/datahub/steward.py b/xiaobai-datahub/datahub/steward.py index b69c023..907c91a 100644 --- a/xiaobai-datahub/datahub/steward.py +++ b/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": diff --git a/xiaobai-datahub/tests/test_extended_eod.py b/xiaobai-datahub/tests/test_extended_eod.py index e282616..e9cc6ca 100644 --- a/xiaobai-datahub/tests/test_extended_eod.py +++ b/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() diff --git a/xiaobai-datahub/tests/test_steward.py b/xiaobai-datahub/tests/test_steward.py index bed5472..03056d3 100644 --- a/xiaobai-datahub/tests/test_steward.py +++ b/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 = [ {