feat(HEL-463): 接入剩余行情数据到 datahub

扩展盘后正式集（涨跌停/人气/龙虎榜/板块日线）与盘中观察 API（报价/指数/分时），网站 bridge 按开关接入并回退旧链路；问天改为按数据依赖跟随开关，不再整栈强制旧路径。

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

.env.example

@@ -7,6 +7,9 @@ TUSHARE_TOKEN=your_tushare_token_here
 
 # Optional xiaobai-datahub client. All DATAHUB_READ_* / DATAHUB_SHADOW_* flags
 # default off in config/datahub.config.json, so the website keeps using Tushare.
+# Extended datasets (HEL-463): LIMIT_EVENTS POPULARITY DRAGON_TIGER SECTOR_DAILY
+# QUOTES INDEX_QUOTES INTRADAY — plus first-batch CALENDAR STOCKS DAILY INDEX_DAILY
+# VALUATION MONEYFLOW AUCTION STATUS.
 DATAHUB_BASE_URL=http://127.0.0.1:8766
 DATAHUB_TOKEN=
 

ARCHITECTURE.md

@@ -54,7 +54,9 @@ background scheduler
   feature repository mixins; do not add feature queries to it.
 - `backend/jobs/` owns job definitions, locks, retries, idempotency, and persisted run state.
   `backend/jobs/service.py` is the application-facing owner of scheduler start/stop, manual
-  refresh submission, and periodic refresh coordination.
+  refresh submission, and periodic refresh coordination. `backend/jobs/refresh.py` owns
+  whether a dashboard payload is a usable refresh result versus a failed job, and whether
+  after-hours official catch-up is due.
 - `backend/llm/` owns model selection, membership/quota checks, fallback, provider transport,
   streaming rules, and call audit. Feature agents only prepare messages and interpret
   feature-specific results.

README.md

@@ -16,7 +16,7 @@
 - **题材库 / 人气热榜 / 龙虎榜**：题材成分、双榜人气、席位与游资档案
 - **智能选股**（会员）：六阶段策略、精选策略库、自然语言编译为受控公式后的确定性筛选与滚动回测；候选需手动加入后才进入五交易日跟踪
 - **问师**（会员）：按选定的游资思维 Skill 单师对话；新增公开角色时在 `游资skills` 下增加含 `SKILL.md` 的目录，并在 `游资skills/mentor_catalog.json` 登记。管理员私有角色放在 `data/private-mentor-skills`（不进 Git / 镜像）
-- **问天**（会员，冻结区，勿改代码）：观势 / 观气 / 观心。卦象、干支、节气与气机由本地程序确定性计算，大模型只负责文字解释
+- **问天**（会员）：观势 / 观气 / 观心。卦象、干支、节气与气机由本地程序确定性计算，大模型只负责文字解释。此前仅冻结过界面视觉方案，现已解冻；问天可纳入后续数据与功能迁移，本阶段不主动重做视觉。
 - **我的复盘**：手工交易日志、每日复盘、提醒中心与复盘助手；不接券商、不自动下单
 
 全局能力：日间 / 夜间主题、股票代码悬停预览日 K 与分时、`Ctrl + K` 全局搜索。图表数据不写入主行情，也不参与情绪、选股或问天计算。
@@ -132,7 +132,7 @@ compose.yaml
 - 本项目是个人研究与复盘工具，全部数据、指标、候选与文字分析均不构成投资建议、证券推荐或买卖要约。
 - 不接券商、不代为下单。交易日志只做手工记录与统计，不代表实际成交。
 - 情绪温度、阶段判定、连板梯队、策略筛选等均为基于公开数据的统计与规则计算，不预测走势，不保证收益。
-- 「问天」属于传统文化视角的观察工具，不具备预测功能，不得作为投资依据；该模块为冻结区，不要改其代码。
+- 「问天」属于传统文化视角的观察工具，不具备预测功能，不得作为投资依据。问天不是永久冻结区：此前只冻结过界面视觉方案，现已解冻，后续数据与功能迁移可以纳入。
 - 行情来自第三方接口，可能延迟、缺失或口径调整；不可用时页面会明确提示，请以交易所与券商正式披露为准。
 - 不要把服务端口直接暴露到公网。不要把 Token、密码、密钥、数据库或 `.env` 提交进 Git。
 - 股市有风险，入市需谨慎。投资决策及其后果由使用者本人承担。

backend/bootstrap/runtime.py

@@ -1,11 +1,23 @@
 from __future__ import annotations
 
 import argparse
+import logging
 from http.server import ThreadingHTTPServer
 from typing import Any
 
 
+def configure_logging() -> None:
+    """让 INFO 级结构化日志（含 datahub 影子对比报告）落到容器日志。"""
+    if logging.getLogger().handlers:
+        return
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+
+
 def main(handler_class: type[Any] | None = None, service: Any | None = None) -> None:
+    configure_logging()
     if handler_class is None or service is None:
         from backend.application import RequestHandler, SERVICE
 

backend/data/datahub/bridge.py

@@ -21,10 +21,14 @@ from backend.data.providers.tushare_client import TushareClient
 
 LOGGER = logging.getLogger("xiaobai.datahub")
 ShadowSink = Callable[[dict[str, Any]], None]
-EMPTY_FAIL_DATASETS = {"stocks", "daily", "index_daily", "valuation", "moneyflow", "auction"}
+EMPTY_FAIL_DATASETS = {
+    "stocks", "daily", "index_daily", "valuation", "moneyflow", "auction",
+    "limit_events", "sector_daily",
+}
 
 
 def looks_like_heaven(module_name: str, filename: str = "") -> bool:
+    """问天调用栈识别（诊断用）。问天按数据集依赖接入，不再整栈强制旧链路。"""
     path = filename.replace("\\", "/")
     return module_name.startswith("backend.features.heaven") or "/features/heaven/" in path
 
@@ -95,7 +99,8 @@ class DatahubBridge:
         legacy_query: Callable[..., list[dict[str, Any]]],
     ) -> list[dict[str, Any]]:
         dataset = API_TO_DATASET.get(api_name)
-        if not dataset or self.heaven_guard():
+        # 问天按实际数据依赖接入：已映射到 hub 的 API 跟随开关；未映射的继续旧链路。
+        if not dataset:
             return legacy_query(api_name, params, fields)
         flags = self.settings.flags(dataset)
         if not flags.read and not flags.shadow:
@@ -106,7 +111,7 @@ class DatahubBridge:
         hub_error: str | None = None
         hub_canonical: list[dict[str, Any]] = []
         try:
-            response = self._fetch_dataset(dataset, params or {})
+            response = self._fetch_dataset(dataset, params or {}, api_name=api_name)
             hub_canonical = self._extract_rows(dataset, response, params or {})
             hub_rows = to_native_rows(dataset, hub_canonical)
             hub_meta = dict(response.meta)
@@ -120,10 +125,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
@@ -132,7 +139,7 @@ class DatahubBridge:
             return project_fields(hub_rows, fields)
         return legacy_query(api_name, params, fields)
 
-    def _fetch_dataset(self, dataset: str, params: dict[str, Any]) -> DatahubResponse:
+    def _fetch_dataset(self, dataset: str, params: dict[str, Any], api_name: str = "") -> DatahubResponse:
         date = yyyymmdd(params.get("trade_date") or params.get("date"))
         start = yyyymmdd(params.get("start_date") or params.get("from") or date)
         end = yyyymmdd(params.get("end_date") or params.get("to") or date)
@@ -149,6 +156,10 @@ class DatahubBridge:
             "valuation": self.client.valuation,
             "moneyflow": self.client.moneyflow,
             "auction": self.client.auction,
+            "limit_events": self.client.limit_events,
+            "popularity": self.client.popularity,
+            "dragon_tiger": self.client.dragon_tiger,
+            "sector_daily": self.client.sectors,
         }
         fetcher = fetchers[dataset]
         query: dict[str, Any] = {}
@@ -163,6 +174,23 @@ class DatahubBridge:
                 query["to"] = end
         if dataset == "daily":
             query["adjust"] = "none"
+        if dataset == "limit_events":
+            limit_type = str(params.get("limit_type") or "").strip().upper()
+            if limit_type:
+                query["limit_type"] = limit_type
+        if dataset == "popularity":
+            if api_name == "ths_hot":
+                query["source"] = "ths"
+            elif api_name == "dc_hot":
+                query["source"] = "dc"
+        if dataset == "sector_daily":
+            family = {
+                "ths_daily": "ths",
+                "dc_index": "dc",
+                "sw_daily": "sw",
+            }.get(api_name, "")
+            if family:
+                query["family"] = family
         return self._paginate(fetcher, query)
 
     def _paginate(self, fetcher: Callable[..., DatahubResponse], params: dict[str, Any]) -> DatahubResponse:
@@ -206,6 +234,10 @@ class DatahubBridge:
             raise DatahubError("STALE", f"{dataset} data is stale")
         if dataset in EMPTY_FAIL_DATASETS and not rows:
             raise DatahubError("EMPTY", f"{dataset} returned no rows")
+        coverage = meta.get("coverage") if isinstance(meta.get("coverage"), dict) else {}
+        if meta.get("incomplete") is True or coverage.get("complete") is False:
+            missing = coverage.get("missing_count")
+            raise DatahubError("INCOMPLETE", f"{dataset} range is incomplete missing={missing}")
 
     def _require_fresh(self, response: DatahubResponse, dataset: str) -> DatahubResponse:
         self._validate_usable(dataset, list(response.data or []) if isinstance(response.data, list) else [], response)

backend/data/datahub/client.py

@@ -60,6 +60,27 @@ class DatahubClient:
     def auction(self, **params: Any) -> DatahubResponse:
         return self.get("/v1/auction", params)
 
+    def limit_events(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/limit-events", params)
+
+    def popularity(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/popularity", params)
+
+    def dragon_tiger(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/dragon-tiger", params)
+
+    def sectors(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/sectors", params)
+
+    def quotes_latest(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/quotes/latest", params)
+
+    def index_quotes(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/indexes/quotes", params)
+
+    def intraday_points(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/intraday/points", params)
+
     def dataset_status(self, date: str) -> DatahubResponse:
         return self.get("/v1/datasets/status", {"date": date})
 

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)

backend/data/datahub/native.py

@@ -17,6 +17,13 @@ API_TO_DATASET = {
     "index_daily": "index_daily",
     "moneyflow": "moneyflow",
     "stk_auction": "auction",
+    "limit_list_d": "limit_events",
+    "ths_hot": "popularity",
+    "dc_hot": "popularity",
+    "hm_detail": "dragon_tiger",
+    "ths_daily": "sector_daily",
+    "dc_index": "sector_daily",
+    "sw_daily": "sector_daily",
 }
 
 SCALE_FIELDS = {
@@ -35,6 +42,16 @@ SCALE_FIELDS = {
         "net_mf_amount": AMOUNT_WAN_YUAN,
     },
     "auction": {"vol": VOLUME_LOT, "float_share": AMOUNT_WAN_YUAN},
+    "limit_events": {
+        "limit_amount": AMOUNT_WAN_YUAN,
+        "float_mv": AMOUNT_WAN_YUAN,
+        "total_mv": AMOUNT_WAN_YUAN,
+    },
+    "dragon_tiger": {
+        "buy_amount": AMOUNT_WAN_YUAN,
+        "sell_amount": AMOUNT_WAN_YUAN,
+        "net_amount": AMOUNT_WAN_YUAN,
+    },
 }
 
 
@@ -67,6 +84,16 @@ def to_native_row(dataset: str, row: dict[str, Any]) -> dict[str, Any]:
             converted[field] = _unscale(converted.get(field), factor)
     if dataset == "stocks":
         converted.pop("updated_at", None)
+    if dataset == "popularity":
+        # keep hub source; callers filter ths/dc themselves when needed
+        if converted.get("ts_name") and not converted.get("name"):
+            converted["name"] = converted.get("ts_name")
+    if dataset == "dragon_tiger":
+        if converted.get("ts_name") and not converted.get("name"):
+            converted["name"] = converted.get("ts_name")
+    if dataset == "sector_daily":
+        if converted.get("pct_change") is not None and converted.get("pct_chg") is None:
+            converted["pct_chg"] = converted.get("pct_change")
     return converted
 
 
@@ -96,6 +123,30 @@ def row_key(dataset: str, row: dict[str, Any]) -> tuple[str, ...]:
         return (str(row.get("ts_code") or "").upper(),)
     if dataset == "status":
         return (str(row.get("dataset") or ""), yyyymmdd(row.get("trade_date")))
+    if dataset == "limit_events":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("limit_type") or ""),
+        )
+    if dataset == "popularity":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("source") or ""),
+        )
+    if dataset == "dragon_tiger":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("hm_name") or ""),
+        )
+    if dataset == "sector_daily":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("family") or ""),
+        )
     return (str(row.get("ts_code") or "").upper(), yyyymmdd(row.get("trade_date")))
 
 

backend/data/datahub/settings.py

@@ -17,6 +17,13 @@ DATASETS = (
     "valuation",
     "moneyflow",
     "auction",
+    "limit_events",
+    "popularity",
+    "dragon_tiger",
+    "sector_daily",
+    "quotes",
+    "index_quotes",
+    "intraday",
     "status",
 )
 
@@ -28,6 +35,13 @@ ENV_DATASET = {
     "valuation": "VALUATION",
     "moneyflow": "MONEYFLOW",
     "auction": "AUCTION",
+    "limit_events": "LIMIT_EVENTS",
+    "popularity": "POPULARITY",
+    "dragon_tiger": "DRAGON_TIGER",
+    "sector_daily": "SECTOR_DAILY",
+    "quotes": "QUOTES",
+    "index_quotes": "INDEX_QUOTES",
+    "intraday": "INTRADAY",
     "status": "STATUS",
 }
 

backend/features/market/service.py

@@ -79,6 +79,8 @@ class MarketServiceMixin:
         if not force:
             snapshot = self.database.get_snapshot(normalized_date)
             if snapshot and str((snapshot.get("meta") or {}).get("source") or "") != "demo":
+                if self._should_retry_incomplete_snapshot(snapshot, normalized_date):
+                    return self.sync_dashboard(normalized_date)
                 snapshot = copy.deepcopy(snapshot)
                 if normalized_date != now.strftime("%Y%m%d"):
                     snapshot.setdefault("meta", {}).update(
@@ -97,6 +99,8 @@ class MarketServiceMixin:
                 "dashboard_request_v1", normalized_date
             )
             if resolved and str((resolved.get("meta") or {}).get("source") or "") != "demo":
+                if self._should_retry_incomplete_snapshot(resolved, normalized_date):
+                    return self.sync_dashboard(normalized_date)
                 resolved = copy.deepcopy(resolved)
                 resolved.setdefault("meta", {})["requested_date"] = self._display_compact_date(
                     normalized_date
@@ -138,6 +142,68 @@ class MarketServiceMixin:
     def _display_compact_date(compact: str) -> str:
         return f"{compact[:4]}-{compact[4:6]}-{compact[6:8]}"
 
+    @staticmethod
+    def _chinese_month_day(value: str) -> str:
+        compact = str(value or "").replace("-", "").replace("/", "")
+        if len(compact) < 8 or not compact[:8].isdigit():
+            return "最近可用交易日"
+        return f"{int(compact[4:6])} 月 {int(compact[6:8])} 日"
+
+    @classmethod
+    def _preparing_display_notice(cls, actual_date: str, requested_date: str) -> str:
+        shown = cls._chinese_month_day(actual_date)
+        requested = str(requested_date or "").replace("-", "")
+        if requested == date.today().strftime("%Y%m%d"):
+            return f"今日数据正在准备，当前展示 {shown}"
+        return f"所选日期数据尚未到齐，当前展示 {shown}"
+
+    @staticmethod
+    def _snapshot_age_seconds(meta: dict[str, Any]) -> float:
+        raw = str(meta.get("updated_at") or "")
+        if not raw:
+            return 10**9
+        try:
+            updated_at = datetime.fromisoformat(raw)
+        except ValueError:
+            return 10**9
+        now = datetime.now().astimezone()
+        if updated_at.tzinfo is None:
+            updated_at = updated_at.replace(tzinfo=now.tzinfo)
+        return (now - updated_at.astimezone(now.tzinfo)).total_seconds()
+
+    def _should_retry_incomplete_snapshot(
+        self, snapshot: dict[str, Any], requested_date: str
+    ) -> bool:
+        if requested_date != date.today().strftime("%Y%m%d"):
+            return False
+        meta = snapshot.get("meta") or {}
+        incomplete = (
+            meta.get("limit_data_source") == "derived"
+            or bool(meta.get("carried_forward"))
+            or str(meta.get("trade_date") or "").replace("-", "") != requested_date
+        )
+        return incomplete and self._snapshot_age_seconds(meta) >= 60
+
+    def _annotate_data_status(self, dashboard: dict[str, Any]) -> dict[str, Any]:
+        meta = dashboard.setdefault("meta", {})
+        notice = str(meta.get("notice") or "")
+        requested = str(meta.get("requested_date") or "").replace("-", "")
+        actual = str(meta.get("trade_date") or "").replace("-", "")
+        if meta.get("limit_data_source") == "derived" and not meta.get("carried_forward"):
+            meta["data_status"] = "partial"
+            meta["display_notice"] = notice or "部分正式数据尚未到齐，当前展示日线推算结果"
+        elif meta.get("carried_forward"):
+            if "非交易日" in notice or "盘前" in notice:
+                meta["data_status"] = "carried"
+                meta["display_notice"] = notice
+            else:
+                meta["data_status"] = "preparing"
+                meta["display_notice"] = self._preparing_display_notice(actual, requested)
+        else:
+            meta["data_status"] = "official"
+            meta.setdefault("display_notice", "")
+        return dashboard
+
     def _carry_dashboard(
         self, snapshot: dict[str, Any], requested_date: str, reason: str
     ) -> dict[str, Any]:
@@ -152,7 +218,7 @@ class MarketServiceMixin:
                 "notice": reason,
             }
         )
-        return carried
+        return self._annotate_data_status(carried)
 
     def _realtime_snapshot_due(
         self,
@@ -197,14 +263,14 @@ class MarketServiceMixin:
                 if not self.configured:
                     raise TushareError("公共行情尚未配置")
                 dashboard = self._tushare_client().dashboard(normalized_date)
-
+                meta = dashboard.setdefault("meta", {})
-                if (dashboard.get("meta") or {}).get("limit_data_source") == "derived":
+                meta["source"] = source
-                    raise TushareError(
+                meta["requested_date"] = self._display_compact_date(normalized_date)
-                        str((dashboard.get("meta") or {}).get("notice") or "官方涨跌停数据尚未返回")
+                if meta.get("limit_data_source") == "derived":
+                    meta.setdefault(
+                        "notice",
+                        "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
                     )
-
-                dashboard["meta"]["source"] = source
-                dashboard["meta"]["requested_date"] = self._display_compact_date(normalized_date)
                 dashboard = self._enrich_dashboard_sentiment(dashboard, normalized_date)
                 record_count = self._record_count(dashboard)
                 actual_date = normalize_date(
@@ -233,8 +299,11 @@ class MarketServiceMixin:
             except TushareError as exc:
                 fallback = self.database.get_latest_real_snapshot(normalized_date)
                 if fallback:
+                    actual = str((fallback.get("meta") or {}).get("trade_date") or "")
                     carried = self._carry_dashboard(
-                        fallback, normalized_date, f"最新行情暂不可用，沿用最近收盘快照：{exc}"
+                        fallback,
+                        normalized_date,
+                        self._preparing_display_notice(actual, normalized_date),
                     )
                     self.database.finish_sync(
                         sync_id, "fallback", self._record_count(carried), str(exc), "tushare"
@@ -1160,7 +1229,7 @@ class MarketServiceMixin:
             "storage": "sqlite",
             "cached": cached,
         }
-        return result
+        return self._annotate_data_status(result)
 
     @staticmethod
     def _record_count(dashboard: dict[str, Any]) -> int:

backend/http/handler.py

@@ -109,7 +109,14 @@ class HttpTransportMixin:
             return {}
         if length <= 0 or length > 65536:
             raise ValueError("请求内容为空或过大。")
-        return json.loads(self.rfile.read(length).decode("utf-8"))
+        raw = self.rfile.read(length)
+        try:
+            payload = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            raise ValueError("请求不是合法 JSON。") from None
+        if not isinstance(payload, dict):
+            raise ValueError("请求不是合法 JSON。")
+        return payload
 
     def serve_static(self, request_path: str) -> None:
         relative = unquote(request_path).lstrip("/") or "index.html"

backend/jobs/refresh.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from datetime import datetime, time as dt_time
+
+
+def dashboard_has_usable_data(dashboard: dict[str, object]) -> bool:
+    if not isinstance(dashboard, dict) or dashboard.get("status") == "failed":
+        return False
+    meta = dashboard.get("meta") or {}
+    overview = dashboard.get("overview") or {}
+    if isinstance(meta, dict) and (meta.get("trade_date") or meta.get("carried_forward")):
+        return True
+    return bool(isinstance(overview, dict) and overview)
+
+
+def verified_dashboard_result(dashboard: dict[str, object]) -> dict[str, object]:
+    """Manual refresh and automatic catch-up share this rule.
+
+    Derived limit lists or a previous usable snapshot are not whole-job failures.
+    Only a payload with no displayable market data is recorded as failed.
+    """
+    if dashboard_has_usable_data(dashboard):
+        return dashboard
+    meta = dashboard.get("meta") if isinstance(dashboard, dict) else None
+    notice = ""
+    if isinstance(meta, dict):
+        notice = str(meta.get("notice") or meta.get("display_notice") or "")
+    return {
+        "status": "failed",
+        "error": notice or "未获取到可用行情",
+    }
+
+
+def official_catchup_due(today: str, snapshot: dict[str, object]) -> bool:
+    now = datetime.now().astimezone().time().replace(tzinfo=None)
+    if not (dt_time(15, 5) <= now < dt_time(22, 0)):
+        return False
+    meta = snapshot.get("meta") if isinstance(snapshot.get("meta"), dict) else {}
+    actual = str(meta.get("trade_date") or "").replace("-", "")
+    if (
+        actual == today
+        and meta.get("limit_data_source") != "derived"
+        and not meta.get("carried_forward")
+    ):
+        return False
+    return True

backend/jobs/service.py

@@ -5,16 +5,7 @@ import time
 from datetime import date
 
 from backend.bootstrap.config import normalize_date
-
+from backend.jobs.refresh import official_catchup_due, verified_dashboard_result
-
-def _verified_dashboard_result(dashboard: dict[str, object]) -> dict[str, object]:
-    meta = dashboard.get("meta") or {}
-    if isinstance(meta, dict) and meta.get("carried_forward"):
-        return {
-            "status": "failed",
-            "error": str(meta.get("notice") or "未获取到所选日期的最新行情"),
-        }
-    return dashboard
 
 
 class JobServiceMixin:
@@ -36,7 +27,7 @@ class JobServiceMixin:
         started = self.jobs.submit(
             "market.refresh",
             key,
-            lambda: _verified_dashboard_result(self.sync_dashboard(normalized)),
+            lambda: verified_dashboard_result(self.sync_dashboard(normalized)),
             {"trade_date": normalized, "trigger": "administrator"},
         )
         return {"started": started, "job_key": key if started else ""}
@@ -54,7 +45,15 @@ class JobServiceMixin:
             self.jobs.submit(
                 "market.refresh",
                 f"realtime:{today}:{bucket}",
-                lambda: self.sync_dashboard(today),
+                lambda: verified_dashboard_result(self.sync_dashboard(today)),
                 {"trade_date": today, "trigger": "realtime-poll"},
             )
+        elif official_catchup_due(today, snapshot):
+            bucket = int(time.time() // 300)
+            self.jobs.submit(
+                "market.refresh",
+                f"catchup:{today}:{bucket}",
+                lambda: verified_dashboard_result(self.sync_dashboard(today)),
+                {"trade_date": today, "trigger": "official-catchup"},
+            )
         self._schedule_automatic_screeners(today, snapshot)

config/architecture-inventory.json

@@ -330,6 +330,7 @@
     "system_service": "backend/features/system/service.py",
     "account_bridge": "backend/features/accounts/application.py",
     "job_lifecycle": "backend/jobs/service.py",
+    "job_refresh_status": "backend/jobs/refresh.py",
     "feature_routes": "backend/features/*/routes.py"
   },
   "numeric_normalization": [
@@ -472,8 +473,8 @@
     },
     {
       "path": "frontend/shared/shell.css",
-      "bytes": 63659,
+      "bytes": 63733,
-      "lines": 3763
+      "lines": 3767
     },
     {
       "path": "backend/features/heaven/engine.py",
@@ -560,6 +561,11 @@
       "bytes": 14743,
       "lines": 342
     },
+    {
+      "path": "frontend/shared/dashboard.js",
+      "bytes": 14740,
+      "lines": 316
+    },
     {
       "path": "frontend/shared/admin.js",
       "bytes": 14410,
@@ -575,11 +581,6 @@
       "bytes": 13219,
       "lines": 289
     },
-    {
-      "path": "frontend/shared/dashboard.js",
-      "bytes": 12894,
-      "lines": 274
-    },
     {
       "path": "backend/features/market/insights_auction_data.py",
       "bytes": 12829,
@@ -785,16 +786,16 @@
       "bytes": 2514,
       "lines": 63
     },
+    {
+      "path": "backend/jobs/service.py",
+      "bytes": 2337,
+      "lines": 59
+    },
     {
       "path": "backend/features/mentor/routes.py",
       "bytes": 2299,
       "lines": 57
     },
-    {
-      "path": "backend/jobs/service.py",
-      "bytes": 2219,
-      "lines": 60
-    },
     {
       "path": "backend/features/screener/regime.py",
       "bytes": 2202,
@@ -830,6 +831,11 @@
       "bytes": 1791,
       "lines": 46
     },
+    {
+      "path": "backend/jobs/refresh.py",
+      "bytes": 1728,
+      "lines": 46
+    },
     {
       "path": "backend/features/alerts/routes.py",
       "bytes": 1687,

config/datahub.config.json

@@ -13,6 +13,13 @@
     "valuation": { "read": false, "shadow": false },
     "moneyflow": { "read": false, "shadow": false },
     "auction": { "read": false, "shadow": false },
+    "limit_events": { "read": false, "shadow": false },
+    "popularity": { "read": false, "shadow": false },
+    "dragon_tiger": { "read": false, "shadow": false },
+    "sector_daily": { "read": false, "shadow": false },
+    "quotes": { "read": false, "shadow": false },
+    "index_quotes": { "read": false, "shadow": false },
+    "intraday": { "read": false, "shadow": false },
     "status": { "read": false, "shadow": false }
   }
 }

docs/README.md

@@ -32,4 +32,4 @@
 
 - 旧文档不能删：被替代的旧文档开头要加一行「⚠️ 本文档已过时，仅留档备查，请勿删除」，再写新版。
 - 用中文大白话写，专业词要带通俗解释，让不懂代码的人也能看懂。
-- 「问天」板块是冻结区，任何改动都不许碰；写文档时别误导后来人去改它。
+- 「问天」不是永久冻结区：此前只冻结过界面视觉方案，现已解冻。问天可纳入后续数据与功能迁移，不要再写成“永远不碰”。

docs/任务清单.md

@@ -7,6 +7,7 @@
 | 任务 | 说明 | 状态 |
 |---|---|---|
 | 全站视觉统一改造收尾 | 主线。17 个阶段已完成，正在最终验收、代码合并 | 收尾中 |
+| 行情刷新误报与旧数据提示 | HEL-412：高级接口未到齐不再记整次失败；今日正式数据晚到时提示当前展示日期 | 施工中 |
 | 手机端独立重新设计 | 先出视觉/交互规范和技术架构方案，等老板确认后再施工 | 方案送审中 |
 
 ## 已做完

docs/项目需求.md

@@ -29,11 +29,11 @@
 - **智能工具类（3 个）**：智能选股、问师、问天。
 - **个人类（1 个）**：我的复盘。
 
-其中「问天」是冻结区（见下面的硬规矩）。
+其中「问天」此前只在全站视觉改造阶段冻结过界面方案，现已解冻；问天可以纳入后续数据与功能迁移，但不等于本阶段要重做视觉。
 
 ## 几条硬规矩（不能破坏的边界）
 
-- 「问天」板块是**冻结区**，任何改动都不许碰它。
+- 「问天」板块**不是永久冻结区**：此前冻结的是界面视觉方案，现已解冻。问天现有功能与界面不要破坏；后续数据与功能迁移可以纳入，不主动重做视觉。
 - **不用假数据冒充真行情**；数据缺失就明说“没有/不可用”，不能编。
 - **每个用户自己的数据互相隔离**（自选、复盘、对话、问天历史等），看不到别人的。
 - **计算由程序确定性完成**（情绪周期、智能选股、问天排盘等），AI 大模型（LLM，就是会聊天的那个 AI）只负责解释或编译自然语言条件，不能改计算结果。

frontend/m/js/pages.js

@@ -770,11 +770,33 @@
     scroll.classList.add("m-motion-fade-in");
   }
 
+  function dashboardFreshnessNotice() {
+    const meta = (state.dashboard && state.dashboard.meta) || {};
+    if (meta.display_notice) return String(meta.display_notice);
+    const requested = String(meta.requested_date || "").replace(/-/g, "");
+    const actual = String(meta.trade_date || "").replace(/-/g, "");
+    const compact = actual;
+    const shown = /^\d{8}$/.test(compact)
+      ? (Number(compact.slice(4, 6)) + " 月 " + Number(compact.slice(6, 8)) + " 日")
+      : "";
+    if (meta.data_status === "preparing" || (meta.carried_forward && actual && requested && actual !== requested)) {
+      return shown ? ("今日数据正在准备，当前展示 " + shown) : "今日数据正在准备，当前展示最近可用数据";
+    }
+    if (meta.data_status === "partial" || meta.limit_data_source === "derived") {
+      return meta.notice || "部分正式数据尚未到齐，当前展示日线推算结果";
+    }
+    return "";
+  }
+
   function renderTopArea(key) {
     const page = document.querySelector(".m-page");
     if (!page) return;
     let top = page.querySelector(".m-top");
     let html = buildStrip();
+    const freshness = dashboardFreshnessNotice();
+    if (freshness) {
+      html = '<div class="m-phase-notice"><strong>' + escapeHtml(freshness) + "</strong></div>" + html;
+    }
     if (key === "market/performance") html += performanceConclusion();
     if (!top) {
       top = document.createElement("div");

frontend/shared/dashboard.js

@@ -66,11 +66,13 @@ async function startAdminRefresh() {
     const requestedCompact = requestedDate.replaceAll("-", "");
     const actualCompact = actualDate.replaceAll("-", "");
     const updated = formatTimestamp(meta.updated_at);
-    if (actualCompact !== requestedCompact || meta.carried_forward) {
+    const freshness = dashboardFreshnessMessage(meta);
-      const reason = meta.notice ? `；${meta.notice}` : "";
+    if (freshness || actualCompact !== requestedCompact || meta.carried_forward || meta.limit_data_source === "derived") {
-      setAdminRefreshStatus("warning", `刷新已完成，但没有获取到 ${requestedDate} 的最新行情；当前仍是 ${actualDate || "未知日期"}${reason}`, "triangle-alert");
+      setAdminRefreshStatus("warning", freshness || `部分正式数据尚未到齐，当前展示 ${actualDate || "最近可用数据"}`, "triangle-alert");
-      showToast("刷新完成，但未获取到所选日期的最新行情");
+      setStatus(freshness || "部分正式数据尚未到齐，当前展示最近可用数据");
-    } else if (meta.notice) {
+      return;
+    }
+    if (meta.notice) {
       setAdminRefreshStatus("warning", `已刷新到 ${actualDate}（${updated}），但数据源提示：${meta.notice}`, "triangle-alert");
       showToast(`已刷新到 ${actualDate}，请留意数据源提示`);
     } else {
@@ -105,6 +107,37 @@ async function waitForAdminRefresh(jobKey) {
   throw new Error("刷新等待超时，请稍后重试");
 }
 
+let dashboardCatchupTimer = 0;
+
+function chineseMonthDay(value) {
+  const compact = String(value || "").replaceAll("-", "").replaceAll("/", "");
+  if (!/^\d{8}/.test(compact)) return "";
+  return `${Number(compact.slice(4, 6))} 月 ${Number(compact.slice(6, 8))} 日`;
+}
+
+function dashboardFreshnessMessage(meta = {}) {
+  if (meta.display_notice) return String(meta.display_notice);
+  const requested = String(meta.requested_date || "").replaceAll("-", "");
+  const actual = String(meta.trade_date || "").replaceAll("-", "");
+  const shown = chineseMonthDay(actual);
+  if (meta.data_status === "preparing" || (meta.carried_forward && actual && requested && actual !== requested)) {
+    return shown ? `今日数据正在准备，当前展示 ${shown}` : "今日数据正在准备，当前展示最近可用数据";
+  }
+  if (meta.data_status === "partial" || meta.limit_data_source === "derived") {
+    return meta.notice || "部分正式数据尚未到齐，当前展示日线推算结果";
+  }
+  return "";
+}
+
+function scheduleDashboardCatchup(meta = {}) {
+  window.clearTimeout(dashboardCatchupTimer);
+  const status = String(meta.data_status || "");
+  if (status !== "preparing" && status !== "partial") return;
+  dashboardCatchupTimer = window.setTimeout(() => {
+    loadDashboard(false, true, false);
+  }, 60000);
+}
+
 function applyDashboard(payload, background = false) {
   state.dashboard = payload;
   const selectedDate = payload.meta.requested_date || payload.meta.trade_date;
@@ -112,7 +145,11 @@ function applyDashboard(payload, background = false) {
   document.querySelector("#qiObservationDate").value = selectedDate;
   document.querySelector("#journalDate").value = selectedDate;
   renderDashboard();
-  setStatus(`${dashboardSourceLabel(payload.meta)} · 数据已更新`);
+  const freshness = dashboardFreshnessMessage(payload.meta || {});
+  setStatus(freshness || `${dashboardSourceLabel(payload.meta)} · 数据已更新`);
+  const updatedAt = document.querySelector("#updatedAt");
+  if (updatedAt) updatedAt.dataset.tone = freshness ? "warning" : "ok";
+  scheduleDashboardCatchup(payload.meta || {});
   if (!background) {
     if (state.activeView === "dragonView") loadDragonTiger();
     if (state.activeView === "screenerView") loadScreenerSetup();
@@ -180,7 +217,12 @@ function renderDashboard() {
     }
   }
   updateSentimentGauge(overview.sentiment_score);
-  setText("updatedAt", `${dashboardSourceLabel(meta)} · 更新 ${formatTimestamp(meta.updated_at)}`);
+  const freshness = dashboardFreshnessMessage(meta);
+  setText("updatedAt", freshness
+    ? freshness
+    : `${dashboardSourceLabel(meta)} · 更新 ${formatTimestamp(meta.updated_at)}`);
+  const updatedAt = document.querySelector("#updatedAt");
+  if (updatedAt) updatedAt.dataset.tone = freshness ? "warning" : "ok";
 
   renderLimitTable();
   renderLadderMini(ladders || []);

frontend/shared/shell.css

@@ -921,6 +921,10 @@ body.sidebar-collapsed .app-main {
   text-align: right;
 }
 
+.status-bar #updatedAt[data-tone="warning"] {
+  color: var(--warning);
+}
+
 .status-bar .risk-note {
   display: block;
 

tests/test_admin_refresh_status.py

@@ -1,23 +1,223 @@
 from __future__ import annotations
 
+import copy
+import threading
 import unittest
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
 
-from backend.jobs.service import _verified_dashboard_result
+from backend.features.market.service import MarketServiceMixin
+from backend.jobs.refresh import (
+    dashboard_has_usable_data,
+    official_catchup_due,
+    verified_dashboard_result,
+)
+from backend.data.providers.tushare_transport import TushareError
 
 
 class AdminRefreshStatusTests(unittest.TestCase):
-    def test_carried_snapshot_is_reported_as_failed_job(self):
+    def test_carried_snapshot_is_usable_not_failed_job(self):
-        result = _verified_dashboard_result(
+        result = verified_dashboard_result(
-            {"meta": {"carried_forward": True, "notice": "官方涨跌停数据尚未返回"}}
+            {
+                "meta": {
+                    "trade_date": "2026-09-01",
+                    "requested_date": "2026-09-02",
+                    "carried_forward": True,
+                    "notice": "今日数据正在准备，当前展示 9 月 1 日",
+                    "data_status": "preparing",
+                },
+                "overview": {"limit_up_count": 12},
+            }
         )
 
-        self.assertEqual(result["status"], "failed")
+        self.assertNotEqual(result.get("status"), "failed")
-        self.assertEqual(result["error"], "官方涨跌停数据尚未返回")
+        self.assertEqual(result["meta"]["data_status"], "preparing")
+        self.assertTrue(dashboard_has_usable_data(result))
+
+    def test_derived_limit_snapshot_is_usable_not_failed_job(self):
+        dashboard = {
+            "meta": {
+                "trade_date": "2026-09-02",
+                "limit_data_source": "derived",
+                "notice": "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
+                "data_status": "partial",
+            },
+            "overview": {"limit_up_count": 8},
+        }
+
+        self.assertIs(verified_dashboard_result(dashboard), dashboard)
 
     def test_current_snapshot_is_reported_as_successful_job(self):
         dashboard = {"meta": {"trade_date": "2026-08-28", "carried_forward": False}}
 
-        self.assertIs(_verified_dashboard_result(dashboard), dashboard)
+        self.assertIs(verified_dashboard_result(dashboard), dashboard)
+
+    def test_empty_payload_is_still_failed(self):
+        result = verified_dashboard_result({"meta": {}, "overview": {}})
+        self.assertEqual(result["status"], "failed")
+
+
+class FakeSyncDatabase:
+    def __init__(self, latest=None):
+        self.latest = latest
+        self.saved = []
+        self.finished = []
+
+    def start_sync(self, *_args, **_kwargs):
+        return 1
+
+    def save_snapshot(self, trade_date, source, payload):
+        self.saved.append((trade_date, source, copy.deepcopy(payload)))
+
+    def save_data_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def finish_sync(self, *args, **kwargs):
+        self.finished.append((args, kwargs))
+
+    def get_latest_real_snapshot(self, *_args, **_kwargs):
+        return copy.deepcopy(self.latest)
+
+    def get_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def get_data_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def reason_overrides(self, *_args, **_kwargs):
+        return {}
+
+
+class FakeDerivedClient:
+    def dashboard(self, trade_date: str):
+        return {
+            "meta": {
+                "trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
+                "limit_data_source": "derived",
+                "notice": "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
+                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+            },
+            "overview": {"limit_up_count": 3},
+            "limits": [{"code": "000001"}],
+            "broken": [],
+            "down_limits": [],
+            "yesterday_limits": [],
+        }
+
+
+class FakeMissingDailyClient:
+    def dashboard(self, trade_date: str):
+        raise TushareError(f"No daily data returned for {trade_date}")
+
+
+class SyncHarness(MarketServiceMixin):
+    def __init__(self, client, latest=None):
+        self.configured = True
+        self.sync_lock = threading.Lock()
+        self.database = FakeSyncDatabase(latest)
+        self._client = client
+        self.current_user_id = 1
+
+    def _tushare_client(self):
+        return self._client
+
+    def _enrich_dashboard_sentiment(self, dashboard, _trade_date):
+        return dashboard
+
+    def _apply_reason_overrides(self, dashboard):
+        return dashboard
+
+
+class DashboardFreshnessTests(unittest.TestCase):
+    def test_derived_limits_are_kept_as_partial_success(self):
+        today = date.today().strftime("%Y%m%d")
+        harness = SyncHarness(FakeDerivedClient())
+        payload = harness.sync_dashboard(today)
+        meta = payload["meta"]
+
+        self.assertEqual(meta["limit_data_source"], "derived")
+        self.assertEqual(meta["data_status"], "partial")
+        self.assertFalse(meta.get("carried_forward"))
+        self.assertIn("日线数据推算", meta["display_notice"])
+        self.assertEqual(harness.database.finished[0][0][1], "success")
+        self.assertEqual(verified_dashboard_result(payload), payload)
+
+    def test_missing_official_data_keeps_previous_day_with_preparing_notice(self):
+        today = date.today()
+        previous = (today - timedelta(days=1)).strftime("%Y-%m-%d")
+        latest = {
+            "meta": {"trade_date": previous, "source": "tushare"},
+            "overview": {"limit_up_count": 20},
+        }
+        harness = SyncHarness(FakeMissingDailyClient(), latest)
+        payload = harness.sync_dashboard(today.strftime("%Y%m%d"))
+        meta = payload["meta"]
+
+        self.assertTrue(meta["carried_forward"])
+        self.assertEqual(meta["data_status"], "preparing")
+        self.assertIn("今日数据正在准备，当前展示", meta["display_notice"])
+        self.assertIn("月", meta["display_notice"])
+        self.assertNotIn("No daily data", meta["display_notice"])
+        self.assertNotEqual(verified_dashboard_result(payload).get("status"), "failed")
+
+    def test_weekend_carry_is_not_labeled_as_preparing(self):
+        snapshot = {
+            "meta": {"trade_date": "2026-07-24", "source": "tushare", "updated_at": "2026-07-24T15:00:00+08:00"},
+            "overview": {"limit_up_count": 1},
+        }
+        harness = SyncHarness(FakeMissingDailyClient())
+        carried = harness._carry_dashboard(snapshot, "20260725", "非交易日沿用最近交易日收盘行情")
+        self.assertEqual(carried["meta"]["data_status"], "carried")
+        self.assertIn("非交易日", carried["meta"]["display_notice"])
+
+    def test_stale_derived_snapshot_is_retried(self):
+        today = date.today().strftime("%Y%m%d")
+        old = datetime.now(timezone.utc) - timedelta(minutes=5)
+        snapshot = {
+            "meta": {
+                "source": "tushare",
+                "trade_date": f"{today[:4]}-{today[4:6]}-{today[6:8]}",
+                "limit_data_source": "derived",
+                "updated_at": old.isoformat(),
+            },
+            "overview": {"limit_up_count": 1},
+        }
+        harness = SyncHarness(FakeDerivedClient())
+        harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(snapshot)
+        payload = harness.get_dashboard(today)
+        self.assertEqual(payload["meta"]["data_status"], "partial")
+        self.assertTrue(harness.database.saved)
+
+    def test_official_catchup_skips_complete_today_snapshot(self):
+        today = date.today().strftime("%Y%m%d")
+        iso = f"{today[:4]}-{today[4:6]}-{today[6:8]}"
+        due = official_catchup_due(
+            today,
+            {"meta": {"trade_date": iso, "limit_data_source": "official"}},
+        )
+        derived_due = official_catchup_due(
+            today,
+            {"meta": {"trade_date": iso, "limit_data_source": "derived"}},
+        )
+        now = datetime.now().astimezone().time().replace(tzinfo=None)
+        if datetime.strptime("15:05", "%H:%M").time() <= now < datetime.strptime("22:00", "%H:%M").time():
+            self.assertFalse(due)
+            self.assertTrue(derived_due)
+        else:
+            self.assertFalse(due)
+            self.assertFalse(derived_due)
+
+
+class FrontendRefreshCopyTests(unittest.TestCase):
+    def test_dashboard_script_distinguishes_partial_from_failure(self):
+        script = (Path(__file__).resolve().parents[1] / "frontend" / "shared" / "dashboard.js").read_text(encoding="utf-8")
+        self.assertIn("今日数据正在准备，当前展示", script)
+        self.assertIn("部分正式数据尚未到齐", script)
+        self.assertIn('job.status === "failed"', script)
+        failed_block = script.split("if (job.status === \"failed\")", 1)[1].split("const query", 1)[0]
+        self.assertIn("后台刷新失败", failed_block)
+        success_block = script.split("const freshness = dashboardFreshnessMessage(meta);", 1)[1]
+        self.assertNotIn("后台刷新失败", success_block.split("} else {", 1)[0])
 
 
 if __name__ == "__main__":

tests/test_bootstrap_runtime.py

@@ -0,0 +1,34 @@
+import logging
+import unittest
+
+from backend.bootstrap.runtime import configure_logging
+
+
+class ConfigureLoggingTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._saved_handlers = logging.getLogger().handlers[:]
+        self._saved_level = logging.getLogger().level
+        logging.getLogger().handlers.clear()
+
+    def tearDown(self) -> None:
+        logging.getLogger().handlers[:] = self._saved_handlers
+        logging.getLogger().setLevel(self._saved_level)
+
+    def test_configures_root_logger_at_info(self) -> None:
+        configure_logging()
+        root = logging.getLogger()
+        self.assertTrue(root.handlers)
+        self.assertEqual(root.level, logging.INFO)
+        with self.assertLogs("xiaobai.datahub", level="INFO") as captured:
+            logging.getLogger("xiaobai.datahub").info("datahub shadow %s", {"dataset": "daily"})
+        self.assertIn("datahub shadow", captured.output[0])
+
+    def test_keeps_existing_configuration(self) -> None:
+        handler = logging.NullHandler()
+        logging.getLogger().addHandler(handler)
+        configure_logging()
+        self.assertEqual(logging.getLogger().handlers, [handler])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_datahub_bridge.py

@@ -126,7 +126,7 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertEqual(calendar[0]["is_open"], 1)
         self.assertEqual(calendar_client.paths, [])
 
-    def test_fallback_on_down_401_timeout_empty_unpublished_and_stale(self) -> None:
+    def test_fallback_on_down_401_timeout_empty_unpublished_stale_and_incomplete(self) -> None:
         cases = [
             DatahubError("UNAVAILABLE", "down"),
             DatahubError("UNAUTHORIZED", "401"),
@@ -134,6 +134,7 @@ class DatahubBridgeTests(unittest.TestCase):
             DatahubError("EMPTY", "no rows"),
             DatahubError("DATASET_NOT_PUBLISHED", "not ready"),
             DatahubError("STALE", "old"),
+            DatahubError("INCOMPLETE", "truncated"),
         ]
         for error in cases:
             with self.subTest(error=error.code):
@@ -144,6 +145,16 @@ class DatahubBridgeTests(unittest.TestCase):
                         data=[dict(HUB_DAILY)],
                         meta={"stale": True, "staleness_seconds": 999999},
                     ))
+                elif error.code == "INCOMPLETE":
+                    client = FakeClient(response=DatahubResponse(
+                        data=[dict(HUB_DAILY)],
+                        meta={
+                            "stale": False,
+                            "staleness_seconds": 0,
+                            "incomplete": True,
+                            "coverage": {"complete": False, "missing_count": 80},
+                        },
+                    ))
                 else:
                     client = FakeClient(error=error)
                 legacy = FakeLegacy([LEGACY_DAILY])
@@ -190,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)
@@ -198,7 +290,8 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertEqual(canonical["vol"], 100000.0)
         self.assertEqual(canonical["amount"], 2000000.0)
 
-    def test_heaven_keeps_legacy_even_when_read_flag_is_on(self) -> None:
+    def test_heaven_can_use_hub_when_dataset_flag_is_on(self) -> None:
+        """问天按数据依赖接入：已映射 API 跟随开关，不再整栈强制旧链路。"""
         self.assertTrue(looks_like_heaven("backend.features.heaven.market_context", "backend/features/heaven/market_context.py"))
         self.assertFalse(looks_like_heaven("backend.features.market.service", "backend/features/market/service.py"))
         client = FakeClient()
@@ -209,7 +302,8 @@ class DatahubBridgeTests(unittest.TestCase):
         )
         rows = wrapped.query("daily", {"trade_date": "20240902"}, "amount")
         self.assertEqual(rows[0]["amount"], 2000.0)
-        self.assertEqual(client.paths, [])
+        self.assertEqual(client.paths, ["/v1/bars/daily"])
+        self.assertEqual(legacy.calls, [])
 
     def test_status_flag_does_not_run_when_off_and_falls_back_when_on(self) -> None:
         off = DatahubBridge(flags(), FakeClient(error=DatahubError("UNAVAILABLE", "down")))
@@ -234,6 +328,27 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertIsInstance(client, DatahubAwareTushareClient)
         self.assertFalse(gateway.datahub.settings.any_enabled())
 
+    def test_stock_detail_range_query_is_not_silently_accepted_when_incomplete(self) -> None:
+        source = (ROOT / "backend" / "data" / "providers" / "tushare_stocks.py").read_text(encoding="utf-8")
+        self.assertIn('"daily"', source)
+        self.assertIn("start_date", source)
+        self.assertIn("end_date", source)
+        client = FakeClient(
+            response=DatahubResponse(
+                data=[dict(HUB_DAILY)],
+                meta={"stale": False, "staleness_seconds": 0, "incomplete": True, "coverage": {"complete": False, "missing_count": 89}},
+            )
+        )
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(flags(daily=(True, False)), client))
+        rows = wrapped.query(
+            "daily",
+            {"ts_code": "600000.SH", "start_date": "20240301", "end_date": "20240902"},
+            "ts_code,amount",
+        )
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(len(legacy.calls), 1)
+
     def test_features_do_not_import_datahub_client(self) -> None:
         violations = []
         for path in (ROOT / "backend" / "features").rglob("*.py"):

tools/build_architecture_inventory.py

@@ -97,6 +97,7 @@ def code_hotspots() -> list[dict[str, Any]]:
         "backend/features/system/service.py",
         "backend/features/accounts/application.py",
         "backend/jobs/service.py",
+        "backend/jobs/refresh.py",
         "database.py",
         "backend/features/screener/engine.py",
         "backend/features/screener/catalog.py",
@@ -265,6 +266,7 @@ def build() -> dict[str, Any]:
             "system_service": "backend/features/system/service.py",
             "account_bridge": "backend/features/accounts/application.py",
             "job_lifecycle": "backend/jobs/service.py",
+            "job_refresh_status": "backend/jobs/refresh.py",
             "feature_routes": "backend/features/*/routes.py",
         },
         "numeric_normalization": [

xiaobai-datahub/README.md

@@ -6,11 +6,12 @@
 ## 做什么
 
 - SQLite WAL `datahub.db`，容器名 `xiaobai-datahub`，端口 `8766`
-- Tushare 盘后正式数据：交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction
+- Tushare 盘后正式数据：交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction、limit_list_d、ths_hot/dc_hot、hm_detail、ths_daily/dc_index/sw_daily
+- 盘中观察（provisional）：东财/腾讯指数报价、个股最新价、分时点（`/v1/quotes/latest` `/v1/indexes/quotes` `/v1/intraday/points`）；永不写入 eod_* 正式表
 - 暂存 → 校验 → 整批原子发布 → 可回滚
 - `/v1` 稳定接口（`X-Datahub-Token`）
 - `/admin/` 最小管理后台（总览 / 数据源 / 调度 / 发布 / 数据集 / 审计）
-- 东财/腾讯/同花顺/选股宝/AKShare/iFinD 适配器位已预留，本阶段不拉实时源
+- 同花顺/选股宝/AKShare/iFinD 适配器位仍预留；东财/腾讯已接入盘中观察
 
 ## 单位口径（相对现站）
 
@@ -63,6 +64,66 @@ python -m unittest discover -s tests -v
 
 不调用真实 Tushare；用内存/临时库和假适配器。
 
+## 历史回补
+
+交易日历默认从 `20160101` 拉到今天后 30 天；盘前 `precheck` 与手动回补都走同一 UPSERT，可重复执行。
+
+网站实际使用的指数（上证、深成、创业板、沪深300）按交易日增量发布，默认覆盖 260 个交易日（大于现有 90 天窗口，并覆盖智能选股基准回看）。已发布日期默认跳过。
+
+```bash
+cd xiaobai-datahub
+python -m datahub history-backfill
+# 可选：--calendar-start 20160101 --index-days 260 --force
+```
+
+管理后台也可手动跑 `history_backfill` 任务，或 `POST /admin/api/backfill` 且 `dataset=history`、确认词 `history:full`。
+
+区间接口在 `meta.coverage` / `meta.incomplete` 标明覆盖是否完整；网站只读接入把不完整区间视为不可用并回旧链路。个股日 K 的 90 天区间查询依赖已核实，本阶段不回补全市场历史。
+
+## 估值字段级质量门
+
+`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 状态。
+
+```bash
+cd xiaobai-datahub
+python -m datahub stocks-refresh            # 手动触发；--force 无变化也重发
+```
+
+## 资金流历史回补
+
+网站会沿真实调用链查最近若干交易日的 moneyflow（个股详情任意日期点查 + 智能选股最近 5 个交易日），默认回补最近 60 个交易日（`moneyflow_history_trading_days` 可配，已发布日期自动跳过）。点查未覆盖的历史日期返回 `DATASET_NOT_PUBLISHED` 并附 `available_from` / `available_to`（低于下界时 `reason=history_not_backfilled`），网站据此明确回退旧链路，不会静默拿到半截数据。
+
+```bash
+cd xiaobai-datahub
+python -m datahub moneyflow-backfill        # --trading-days 60 --end-date --force 可选
+```
+
+## 盘后补跑与强制重发
+
+```bash
+cd xiaobai-datahub
+python -m datahub eod-refresh --trade-date 20260904                  # 补不完整的 A/B 边界
+python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
+# --force 按一致性边界整组重发：valuation/daily/moneyflow/auction/stocks → A 组；
+# index_daily → B 组。不可再单独切换某一个正式数据集。
+```
+
+管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
+
 ## 备份
 
 每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：
@@ -74,5 +135,6 @@ python -c "from pathlib import Path; from datahub.db import HubDB; HubDB(Path('d
 ## 安全
 
 - 密钥只以 `configured / 末4位 / 更新时间` 出现在后台，不进日志、不进 `/v1`
+- HTTP 解析失败只记录“请求不是合法 JSON”，不把请求正文、密码或 Token 写入容器日志
 - 回滚、补数需重新输入密码 + 确认词
 - 容器非 root（uid 10002）、read_only、cap_drop ALL

xiaobai-datahub/admin/app.js

@@ -104,11 +104,29 @@ async function render() {
   if (state.page === "overview") {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
+    const eod = data.eod_status || {};
+    const eodLabels = {
+      pending_first_attempt: "等待首次尝试",
+      waiting_upstream: "等待上游",
+      done: "已成功",
+      cutoff_failed: "已截止失败",
+      closed_day: "休市",
+    };
+    const eodExtra = [];
+    if (eod.state === "waiting_upstream") {
+      eodExtra.push(`已试 ${eod.attempts} 次`);
+      if (eod.next_retry_at) eodExtra.push(`下次重试 ${esc(String(eod.next_retry_at).replace("T", " ").slice(11, 16))}`);
+      if (eod.missing_datasets && eod.missing_datasets.length) eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
+    }
+    if (eod.state === "cutoff_failed" && eod.missing_datasets) {
+      eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
+    }
     page.innerHTML = `
       <div class="cards">
         <div class="card"><div class="muted">交易日</div><strong>${esc(data.trade_date)}</strong></div>
         <div class="card"><div class="muted">阶段</div><strong>${esc(data.session_phase)}</strong></div>
         <div class="card"><div class="muted">今日发布</div><strong>${data.publications.length}</strong></div>
+        <div class="card"><div class="muted">盘后补跑</div><strong>${esc(eodLabels[eod.state] || eod.state || "-")}</strong><div class="muted">${eodExtra.join(" · ")}</div></div>
         <div class="card"><div class="muted">异常批次</div><strong class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
       </div>
       <h2>最近调用</h2>
@@ -251,7 +269,7 @@ function renderRelease(data) {
 
 async function dangerous(kind, dataset) {
   const date = ($("rel-date") && $("rel-date").value) || "";
-  const ds = dataset || prompt("数据集（daily / valuation / moneyflow / auction / index_daily / reference）", "daily");
+  const ds = dataset || prompt("数据集（daily/valuation/moneyflow/auction/stocks→A组整批；index_daily→B组；或 reference）", "daily");
   if (!ds) return;
   const password = prompt("二次确认：输入管理密码");
   if (!password) return;

xiaobai-datahub/config/hub-quality.config.json

@@ -11,5 +11,48 @@
   "publication_generations": 3,
   "tushare_rate_per_minute": 300,
   "list_limit_default": 5000,
-  "list_limit_max": 5000
+  "list_limit_max": 5000,
+  "calendar_start": "20160101",
+  "index_history_trading_days": 260,
+  "eod_retry_start": "15:15",
+  "eod_retry_interval_minutes": 30,
+  "eod_retry_cutoff": "23:30",
+  "moneyflow_history_trading_days": 60,
+  "stocks_refresh_times": [
+    "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": [
+        "turnover_rate",
+        "volume_ratio",
+        "total_mv",
+        "circ_mv",
+        "pe_ttm",
+        "pb",
+        "ps_ttm",
+        "dv_ttm"
+      ],
+      "min_nonnull_rate": 0.9,
+      "min_nonnull_rate_by_field": {
+        "pe_ttm": 0.5,
+        "dv_ttm": 0.3
+      },
+      "max_nonnull_drop_vs_prev": 0.15,
+      "max_nonfinite_rate": 0.01
+    }
+  }
 }

xiaobai-datahub/datahub/__main__.py

@@ -0,0 +1,4 @@
+from datahub.cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

xiaobai-datahub/datahub/adapters/__init__.py

@@ -1,13 +1,13 @@
 from datahub.adapters.akshare import ADAPTER as akshare
-from datahub.adapters.eastmoney import ADAPTER as eastmoney
+from datahub.adapters.eastmoney import EastmoneyAdapter
 from datahub.adapters.ifind import ADAPTER as ifind
-from datahub.adapters.tencent import ADAPTER as tencent
+from datahub.adapters.tencent import TencentAdapter
 from datahub.adapters.ths import ADAPTER as ths
 from datahub.adapters.xgb import ADAPTER as xgb
 
 RESERVED = {
-    "eastmoney": eastmoney,
+    "eastmoney": EastmoneyAdapter(),
-    "tencent": tencent,
+    "tencent": TencentAdapter(),
     "ths": ths,
     "xgb": xgb,
     "akshare": akshare,

xiaobai-datahub/datahub/adapters/eastmoney.py

@@ -1,3 +1,250 @@
-from datahub.adapters.base import ReservedAdapter
+from __future__ import annotations
 
-ADAPTER = ReservedAdapter("eastmoney")
+import json
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.base import AdapterError, MarketAdapter
+from datahub.numbers import finite_number, round4
+
+EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
+EASTMONEY_CLIST_URL = "https://push2.eastmoney.com/api/qt/clist/get"
+TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
+BROWSER_UA = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
+)
+INDEX_SECIDS = {
+    "000001.SH": "1.000001",
+    "399001.SZ": "0.399001",
+    "399006.SZ": "0.399006",
+}
+
+
+class EastmoneyAdapter(MarketAdapter):
+    name = "eastmoney"
+
+    def __init__(self, timeout: int = 8) -> None:
+        self.timeout = timeout
+
+    def probe(self) -> dict[str, Any]:
+        started = time.perf_counter()
+        try:
+            rows = self.fetch_indices()
+            state = "ok" if len(rows) == 3 else "empty"
+        except AdapterError as exc:
+            return {
+                "provider": self.name,
+                "configured": True,
+                "state": "error",
+                "message": str(exc),
+                "latency_ms": round((time.perf_counter() - started) * 1000),
+            }
+        return {
+            "provider": self.name,
+            "configured": True,
+            "state": state,
+            "latency_ms": round((time.perf_counter() - started) * 1000),
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset in {"indexes_quotes", "index_quotes"}:
+            return self.fetch_indices()
+        if dataset in {"quotes", "quotes_latest"}:
+            codes = params.get("codes") or []
+            if isinstance(codes, str):
+                codes = [item.strip() for item in codes.split(",") if item.strip()]
+            return self.fetch_quotes(list(codes))
+        raise AdapterError(f"{self.name} unsupported dataset: {dataset}")
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return list(rows)
+
+    def fetch_indices(self) -> list[dict[str, Any]]:
+        payload = self._get_json(
+            EASTMONEY_INDEX_URL,
+            {
+                "secids": "1.000001,0.399001,0.399006",
+                "fltt": "2",
+                "invt": "2",
+                "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f6,f124",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        rows = list((payload.get("data") or {}).get("diff") or [])
+        result = []
+        for row in rows:
+            code = str(row.get("f12") or "")
+            if code not in {"000001", "399001", "399006"}:
+                continue
+            epoch = int(finite_number(row.get("f124")) or 0)
+            ts_code = f"{code}.SH" if code.startswith("0") and code == "000001" else f"{code}.SZ"
+            if code == "000001":
+                ts_code = "000001.SH"
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "code": code,
+                    "name": row.get("f14") or code,
+                    "price": round4(finite_number(row.get("f2"))),
+                    "pct_chg": round4(finite_number(row.get("f3"))),
+                    "change_amount": round4(finite_number(row.get("f4"))),
+                    "open": round4(finite_number(row.get("f17"))),
+                    "high": round4(finite_number(row.get("f15"))),
+                    "low": round4(finite_number(row.get("f16"))),
+                    "previous_close": round4(finite_number(row.get("f18"))),
+                    "amount": round4(finite_number(row.get("f6"))),
+                    "quote_time_epoch": epoch,
+                    "quote_time": (
+                        datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds")
+                        if epoch
+                        else ""
+                    ),
+                    "source": "eastmoney_push2",
+                }
+            )
+        if len(result) != 3:
+            raise AdapterError(f"Eastmoney returned {len(result)}/3 indices")
+        return result
+
+    def fetch_quotes(self, codes: list[str]) -> list[dict[str, Any]]:
+        # Eastmoney clist does not accept arbitrary code lists well; use ulist.np for batches.
+        secids = []
+        for code in codes:
+            ts = str(code or "").upper()
+            symbol = ts.split(".")[0]
+            if ts.endswith(".SH") or symbol.startswith(("5", "6", "9")):
+                secids.append(f"1.{symbol}")
+            else:
+                secids.append(f"0.{symbol}")
+        if not secids:
+            return []
+        payload = self._get_json(
+            EASTMONEY_INDEX_URL,
+            {
+                "secids": ",".join(secids[:60]),
+                "fltt": "2",
+                "invt": "2",
+                "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f5,f6,f8,f124",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        rows = list((payload.get("data") or {}).get("diff") or [])
+        result = []
+        for row in rows:
+            symbol = str(row.get("f12") or "")
+            if not symbol:
+                continue
+            ts_code = f"{symbol}.SH" if symbol.startswith(("5", "6", "9")) else f"{symbol}.SZ"
+            epoch = int(finite_number(row.get("f124")) or 0)
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "name": row.get("f14") or symbol,
+                    "price": round4(finite_number(row.get("f2"))),
+                    "pct_chg": round4(finite_number(row.get("f3"))),
+                    "change_amount": round4(finite_number(row.get("f4"))),
+                    "open": round4(finite_number(row.get("f17"))),
+                    "high": round4(finite_number(row.get("f15"))),
+                    "low": round4(finite_number(row.get("f16"))),
+                    "previous_close": round4(finite_number(row.get("f18"))),
+                    "volume": round4(finite_number(row.get("f5"))),
+                    "amount": round4(finite_number(row.get("f6"))),
+                    "turnover_rate": round4(finite_number(row.get("f8"))),
+                    "quote_time_epoch": epoch,
+                    "quote_time": (
+                        datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds")
+                        if epoch
+                        else ""
+                    ),
+                    "source": "eastmoney_push2",
+                }
+            )
+        return result
+
+    def fetch_intraday(self, ts_code: str) -> dict[str, Any]:
+        code = str(ts_code or "").upper()
+        if code in INDEX_SECIDS:
+            secid = INDEX_SECIDS[code]
+            entity = "index"
+            identifier = code
+        else:
+            symbol = code.split(".")[0]
+            market = "1" if symbol.startswith(("5", "6", "9")) else "0"
+            secid = f"{market}.{symbol}"
+            entity = "stock"
+            identifier = symbol
+        payload = self._get_json(
+            TRENDS_URL,
+            {
+                "secid": secid,
+                "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
+                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
+                "iscr": "0",
+                "ndays": "1",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        data = payload.get("data") or {}
+        points = []
+        for raw in data.get("trends") or []:
+            point = _parse_trend(raw)
+            if point:
+                points.append(point)
+        if not points:
+            raise AdapterError("No intraday chart data returned")
+        return {
+            "entity_type": entity,
+            "identifier": identifier,
+            "ts_code": code if "." in code else f"{identifier}.{'SH' if identifier.startswith(('5','6','9')) else 'SZ'}",
+            "name": str(data.get("name") or ""),
+            "code": str(data.get("code") or identifier),
+            "trade_date": points[-1]["date"],
+            "previous_close": round4(finite_number(data.get("preClose"))),
+            "points": points,
+            "source": "eastmoney_trends2",
+        }
+
+    def _get_json(self, url: str, params: dict[str, str], referer: str) -> dict[str, Any]:
+        request_url = f"{url}?{urllib.parse.urlencode(params)}"
+        request = urllib.request.Request(
+            request_url,
+            headers={
+                "Accept": "application/json,text/plain,*/*",
+                "User-Agent": BROWSER_UA,
+                "Referer": referer,
+            },
+            method="GET",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except Exception as exc:
+            raise AdapterError(f"eastmoney request failed: {exc}") from exc
+
+
+def _parse_trend(raw: Any) -> dict[str, Any] | None:
+    text = str(raw or "")
+    parts = text.split(",")
+    if len(parts) < 8:
+        return None
+    stamp = parts[0]
+    try:
+        when = datetime.strptime(stamp, "%Y-%m-%d %H:%M")
+    except ValueError:
+        return None
+    return {
+        "time": when.strftime("%H:%M"),
+        "date": when.strftime("%Y-%m-%d"),
+        "open": round4(finite_number(parts[1])),
+        "close": round4(finite_number(parts[2])),
+        "high": round4(finite_number(parts[3])),
+        "low": round4(finite_number(parts[4])),
+        "avg_price": round4(finite_number(parts[7] if len(parts) > 7 else parts[2])),
+        "volume": round4(finite_number(parts[5])),
+        "amount": round4(finite_number(parts[6])),
+    }

xiaobai-datahub/datahub/adapters/tencent.py

@@ -1,3 +1,99 @@
-from datahub.adapters.base import ReservedAdapter
+from __future__ import annotations
 
-ADAPTER = ReservedAdapter("tencent")
+import time
+import urllib.error
+import urllib.request
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.base import AdapterError, MarketAdapter
+from datahub.numbers import finite_number, round4
+
+TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006"
+BROWSER_UA = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
+)
+
+
+class TencentAdapter(MarketAdapter):
+    name = "tencent"
+
+    def __init__(self, timeout: int = 8) -> None:
+        self.timeout = timeout
+
+    def probe(self) -> dict[str, Any]:
+        started = time.perf_counter()
+        try:
+            rows = self.fetch_indices()
+            state = "ok" if len(rows) == 3 else "empty"
+        except AdapterError as exc:
+            return {
+                "provider": self.name,
+                "configured": True,
+                "state": "error",
+                "message": str(exc),
+                "latency_ms": round((time.perf_counter() - started) * 1000),
+            }
+        return {
+            "provider": self.name,
+            "configured": True,
+            "state": state,
+            "latency_ms": round((time.perf_counter() - started) * 1000),
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset in {"indexes_quotes", "index_quotes"}:
+            return self.fetch_indices()
+        raise AdapterError(f"{self.name} unsupported dataset: {dataset}")
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return list(rows)
+
+    def fetch_indices(self) -> list[dict[str, Any]]:
+        request = urllib.request.Request(
+            TENCENT_INDEX_URL,
+            headers={"User-Agent": BROWSER_UA, "Referer": "https://gu.qq.com/"},
+            method="GET",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                raw = response.read().decode("gb18030", errors="ignore")
+        except Exception as exc:
+            raise AdapterError(f"tencent request failed: {exc}") from exc
+        result = []
+        for line in raw.splitlines():
+            if '="' not in line:
+                continue
+            fields = line.split('="', 1)[1].rsplit('";', 1)[0].split("~")
+            if len(fields) < 38:
+                continue
+            code = fields[2]
+            if code not in {"000001", "399001", "399006"}:
+                continue
+            try:
+                quote_time = datetime.strptime(fields[30], "%Y%m%d%H%M%S").astimezone()
+            except ValueError as exc:
+                raise AdapterError(f"Tencent invalid quote time for {code}") from exc
+            ts_code = "000001.SH" if code == "000001" else f"{code}.SZ"
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "code": code,
+                    "name": fields[1] or code,
+                    "price": round4(finite_number(fields[3])),
+                    "pct_chg": round4(finite_number(fields[32])),
+                    "change_amount": round4(finite_number(fields[31])),
+                    "open": round4(finite_number(fields[5])),
+                    "high": round4(finite_number(fields[33])),
+                    "low": round4(finite_number(fields[34])),
+                    "previous_close": round4(finite_number(fields[4])),
+                    "amount": round4(finite_number(fields[37]) * 10000),
+                    "quote_time_epoch": int(quote_time.timestamp()),
+                    "quote_time": quote_time.isoformat(timespec="seconds"),
+                    "source": "tencent_qt",
+                }
+            )
+        if len(result) != 3:
+            raise AdapterError(f"Tencent returned {len(result)}/3 indices")
+        return result

xiaobai-datahub/datahub/adapters/tushare.py

@@ -11,8 +11,12 @@ from datahub.normalize import (
     normalize_auction,
     normalize_calendar,
     normalize_daily,
+    normalize_dragon_tiger,
     normalize_index_daily,
+    normalize_limit_event,
     normalize_moneyflow,
+    normalize_popularity,
+    normalize_sector_daily,
     normalize_stock,
     normalize_valuation,
 )
@@ -31,6 +35,21 @@ TUSHARE_FIELDS = {
         "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount"
     ),
     "stk_auction": "ts_code,trade_date,vol,price,amount,pre_close,turnover_rate,volume_ratio,float_share",
+    "limit_list_d": (
+        "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,"
+        "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times,limit_type"
+    ),
+    "ths_hot": "ts_code,ts_name,hot,rank,pct_change,current_price,concept,data_type,trade_date",
+    "dc_hot": "ts_code,ts_name,rank,pct_change,current_price,hot,concept,data_type,trade_date",
+    "hm_detail": "trade_date,ts_code,ts_name,buy_amount,sell_amount,net_amount,hm_name,hm_orgs,tag",
+    "hm_list": "name,desc,orgs",
+    "top_list": "trade_date,ts_code,name,pct_change,reason",
+    "top_inst": "trade_date,ts_code,exalter,buy,buy_rate,sell,sell_rate,net_buy,side,reason",
+    "ths_index": "ts_code,name,count,exchange,list_date,type",
+    "ths_daily": "ts_code,trade_date,open,high,low,close,pre_close,pct_change,vol,turnover_rate",
+    "dc_index": "ts_code,trade_date,name,open,high,low,close,pre_close,pct_change,vol,amount,turnover_rate",
+    "sw_daily": "ts_code,trade_date,name,open,high,low,close,pct_change,vol,amount",
 }
 
 DATASET_API = {
@@ -42,9 +61,15 @@ DATASET_API = {
     "index_daily": "index_daily",
     "moneyflow": "moneyflow",
     "auction": "stk_auction",
+    "limit_events": "limit_list_d",
+    "popularity": "ths_hot",
+    "dragon_tiger": "hm_detail",
+    "sector_daily": "ths_daily",
 }
 
-DEFAULT_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+WEBSITE_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+DEFAULT_INDEX_CODES = WEBSITE_INDEX_CODES
+LIMIT_TYPES = ("U", "D", "Z")
 
 
 class TushareAdapter(MarketAdapter):
@@ -82,6 +107,14 @@ class TushareAdapter(MarketAdapter):
         }
 
     def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset == "limit_events":
+            return self.fetch_limit_events(str(params.get("trade_date") or ""))
+        if dataset == "popularity":
+            return self.fetch_popularity(str(params.get("trade_date") or ""))
+        if dataset == "dragon_tiger":
+            return self.fetch_dragon_tiger(str(params.get("trade_date") or ""))
+        if dataset == "sector_daily":
+            return self.fetch_sector_daily(str(params.get("trade_date") or ""))
         api_name = DATASET_API.get(dataset, dataset)
         fields = TUSHARE_FIELDS.get(api_name, "")
         query_params = dict(params)
@@ -90,10 +123,67 @@ class TushareAdapter(MarketAdapter):
         if api_name == "trade_cal" and "exchange" not in query_params:
             query_params["exchange"] = "SSE"
         if api_name == "index_daily" and "ts_code" not in query_params:
-            # Caller typically loops codes; a missing code would pull nothing useful.
             query_params.setdefault("ts_code", DEFAULT_INDEX_CODES[0])
         return self._query(api_name, query_params, fields)
 
+    def fetch_limit_events(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for limit_type in LIMIT_TYPES:
+            part = self._query(
+                "limit_list_d",
+                {"trade_date": trade_date, "limit_type": limit_type},
+                TUSHARE_FIELDS["limit_list_d"],
+            )
+            for row in part:
+                row = dict(row)
+                row.setdefault("limit_type", limit_type)
+                rows.append(row)
+        return rows
+
+    def fetch_popularity(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for api_name, source in (("ths_hot", "ths"), ("dc_hot", "dc")):
+            for row in self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name]):
+                item = dict(row)
+                item["source"] = source
+                item.setdefault("trade_date", trade_date)
+                rows.append(item)
+        return rows
+
+    def fetch_dragon_tiger(self, trade_date: str) -> list[dict[str, Any]]:
+        details = self._query("hm_detail", {"trade_date": trade_date}, TUSHARE_FIELDS["hm_detail"])
+        top_rows = self._query("top_list", {"trade_date": trade_date}, TUSHARE_FIELDS["top_list"])
+        context = {
+            str(row.get("ts_code") or ""): row
+            for row in top_rows
+            if str(row.get("ts_code") or "")
+        }
+        rows: list[dict[str, Any]] = []
+        for row in details:
+            item = dict(row)
+            stock = context.get(str(item.get("ts_code") or ""), {})
+            if item.get("pct_change") is None and stock.get("pct_change") is not None:
+                item["pct_change"] = stock.get("pct_change")
+            if not item.get("reason") and stock.get("reason"):
+                item["reason"] = stock.get("reason")
+            if not item.get("ts_name") and stock.get("name"):
+                item["ts_name"] = stock.get("name")
+            rows.append(item)
+        return rows
+
+    def fetch_sector_daily(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for api_name, family in (("ths_daily", "ths"), ("dc_index", "dc"), ("sw_daily", "sw")):
+            try:
+                part = self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name])
+            except AdapterError:
+                part = []
+            for row in part:
+                item = dict(row)
+                item["family"] = family
+                rows.append(item)
+        return rows
+
     def fetch_index_daily(self, trade_date: str, codes: tuple[str, ...] = DEFAULT_INDEX_CODES) -> list[dict[str, Any]]:
         rows: list[dict[str, Any]] = []
         for ts_code in codes:
@@ -101,6 +191,17 @@ class TushareAdapter(MarketAdapter):
         return rows
 
     def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        if dataset in {"limit_events", "limit_list_d"}:
+            return [normalize_limit_event(row) for row in rows]
+        if dataset == "popularity":
+            return [normalize_popularity(row, source=str(row.get("source") or "")) for row in rows]
+        if dataset == "dragon_tiger":
+            return [normalize_dragon_tiger(row) for row in rows]
+        if dataset == "sector_daily":
+            return [
+                normalize_sector_daily(row, family=str(row.get("family") or "ths"))
+                for row in rows
+            ]
         mapping = {
             "calendar": normalize_calendar,
             "trade_cal": normalize_calendar,
@@ -146,9 +247,10 @@ class TushareAdapter(MarketAdapter):
             with urllib.request.urlopen(request, timeout=self.timeout) as response:
                 result = json.loads(response.read().decode("utf-8"))
         except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
-            raise AdapterError(f"Tushare request failed: {exc}") from exc
+            raise AdapterError(f"Tushare 请求失败: {exc}") from exc
-        if result.get("code") != 0:
+        if result.get("code") not in (0, "0", None):
-            raise AdapterError(result.get("msg") or "Tushare returned an unknown error")
+            raise AdapterError(str(result.get("msg") or f"Tushare error {result.get('code')}"))
         data = result.get("data") or {}
-        columns = data.get("fields") or []
+        items = data.get("items") or []
-        return [dict(zip(columns, item)) for item in data.get("items") or []]
+        fields_list = data.get("fields") or (fields.split(",") if fields else [])
+        return [dict(zip(fields_list, item)) for item in items]

xiaobai-datahub/datahub/admin_api.py

@@ -6,7 +6,7 @@ from typing import Any
 from datahub.adapters import RESERVED
 from datahub.auth import AuthService
 from datahub.db import HubDB
-from datahub.pipeline import Pipeline
+from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
 from datahub.scheduler import Scheduler
 from datahub.serving import ApiError
 from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
@@ -38,6 +38,7 @@ class AdminAPI:
             "trade_date": today,
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
+            "eod_status": self.scheduler.eod_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -83,11 +84,15 @@ class AdminAPI:
 
     def jobs(self) -> dict[str, Any]:
         runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100")
+        stocks_times = "/".join(self.pipeline.settings.stocks_refresh_times) or "20:00"
         return {
             "jobs": [
                 {"id": "precheck", "at": "08:45", "title": "盘前预检"},
                 {"id": "eod_a", "at": "15:05", "title": "盘后批 A daily/valuation/moneyflow/auction"},
                 {"id": "eod_b", "at": "15:10", "title": "盘后批 B index_daily"},
+                {"id": "eod_retry", "at": "15:15-23:30", "title": "盘后未出数自动重试（每 30 分钟，成功即停）"},
+                {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"},
+                {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},
                 {"id": "backup", "at": "00:40", "title": "SQLite 备份"},
             ],
@@ -130,12 +135,31 @@ class AdminAPI:
         return result
 
     def backfill(self, dataset: str, trade_date: str, password: str, confirm: str, actor: str) -> dict[str, Any]:
-        self._dangerous(password, confirm, f"{dataset}:{trade_date}")
+        day = yyyymmdd(trade_date or now_shanghai())
-        if dataset == "reference":
+        if dataset == "history":
-            result = self.pipeline.ingest_reference(trade_date)
+            self._dangerous(password, confirm, "history:full")
+            result = self.pipeline.backfill_history(day)
         else:
-            result = self.pipeline.run_dataset(dataset, trade_date)
+            self._dangerous(password, confirm, f"{dataset}:{day}")
-        self.pipeline.audit(actor, "backfill", f"{dataset}:{trade_date}", json.dumps({"ok": True}))
+            if dataset == "reference":
+                result = self.pipeline.ingest_reference(day)
+            elif dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET:
+                # Manual same-day republish must rebuild the full A/B boundary.
+                # Gate failures and mid-switch exceptions both surface as
+                # FAILED_PRECONDITION so the admin API never leaks raw
+                # transaction errors to the client.
+                try:
+                    result = self.pipeline.force_republish_boundary(dataset, day)
+                    failures = self.pipeline.eod_failures(result)
+                    if failures:
+                        raise ApiError("FAILED_PRECONDITION", "; ".join(failures))
+                except ApiError:
+                    raise
+                except Exception as exc:
+                    raise ApiError("FAILED_PRECONDITION", str(exc)) from exc
+            else:
+                raise ApiError("INVALID_ARGUMENT", f"unsupported backfill dataset: {dataset}")
+        self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
         return result
 
     def _dangerous(self, password: str, confirm: str, expected: str) -> None:

xiaobai-datahub/datahub/cli.py

@@ -0,0 +1,112 @@
+"""Command-line entry for one-shot datahub operations."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from datahub.hub import build_hub
+from datahub.pipeline import EOD_A_DATASETS, OFFICIAL_DATASETS, STOCKS_DATASET
+from datahub.settings import load_settings
+from datahub.timeutil import yyyymmdd
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="xiaobai-datahub CLI")
+    sub = parser.add_subparsers(dest="command", required=True)
+    history = sub.add_parser("history-backfill", help="回补 2016 年起交易日历和网站所用指数日 K")
+    history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
+    history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
+    history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
+    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已完整发布的一致性边界，仍走质量门禁）")
+    refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
+    refresh.add_argument(
+        "--force", action="store_true",
+        help="强制重发 --dataset 所属的完整一致性边界（A 组或 B 组），生成新批次并保留上一批次可回滚",
+    )
+    refresh.add_argument(
+        "--dataset", default=None,
+        help="配合 --force：指定边界内任一成员（如 valuation→整组 A；index_daily→整组 B）",
+    )
+    stocks_refresh = sub.add_parser("stocks-refresh", help="刷新股票主档并发布正式快照（幂等：无变化则跳过）")
+    stocks_refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
+    stocks_refresh.add_argument("--force", action="store_true", help="即使快照无变化也重新发布")
+    moneyflow_backfill = sub.add_parser(
+        "moneyflow-backfill", help="回补资金流历史（默认覆盖网站所需的最近 N 个交易日，跳过已发布日期）",
+    )
+    moneyflow_backfill.add_argument("--end-date", default=None, help="截止交易日 YYYYMMDD，默认今天")
+    moneyflow_backfill.add_argument("--trading-days", type=int, default=None, help="回补交易日数量，默认配置 moneyflow_history_trading_days")
+    moneyflow_backfill.add_argument("--force", action="store_true", help="覆盖已发布的资金流日期")
+    args = parser.parse_args(argv)
+
+    settings = load_settings()
+    hub = build_hub(settings)
+    if args.command == "history-backfill":
+        result = hub.pipeline.backfill_history(
+            calendar_start=args.calendar_start,
+            index_days=args.index_days,
+            force=args.force,
+        )
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if result.get("ok") else 1
+    if args.command == "eod-refresh":
+        day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
+        if args.force:
+            allowed = set(OFFICIAL_DATASETS) | {STOCKS_DATASET}
+            if not args.dataset:
+                parser.error("--force requires --dataset (e.g. --dataset valuation)")
+            if args.dataset not in allowed:
+                parser.error(f"unknown dataset: {args.dataset}")
+            result = hub.pipeline.force_republish_boundary(args.dataset, day)
+            boundary = "A" if args.dataset in EOD_A_DATASETS or args.dataset == STOCKS_DATASET else "B"
+        else:
+            result = hub.pipeline.run_eod_missing(day)
+            boundary = None
+        hub.pipeline.audit("cli", "eod-refresh", f"eod:{day}", json.dumps(
+            {"force": bool(args.force), "dataset": args.dataset, "boundary": boundary,
+             **{name: item.get("state") for name, item in result.items() if isinstance(item, dict)}},
+            ensure_ascii=False,
+        ))
+        if args.force:
+            failures = hub.pipeline.eod_failures(result)
+            payload = {"trade_date": day, "boundary": boundary, "datasets": result}
+            json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
+            sys.stdout.write("\n")
+            return 0 if not failures else 1
+        missing = hub.pipeline.missing_official_datasets(day)
+        payload = {"trade_date": day, "datasets": result, "missing_after": missing}
+        json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if not missing else 1
+    if args.command == "stocks-refresh":
+        day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
+        result = hub.pipeline.refresh_stocks(day, force=args.force)
+        hub.pipeline.audit("cli", "stocks-refresh", f"stocks:{day}", json.dumps(
+            {"force": bool(args.force), "state": result.get("state"), "batch_id": result.get("batch_id")},
+            ensure_ascii=False,
+        ))
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if result.get("state") != "failed" else 1
+    if args.command == "moneyflow-backfill":
+        result = hub.pipeline.backfill_moneyflow_history(
+            end_date=args.end_date,
+            trading_days=args.trading_days,
+            force=args.force,
+        )
+        hub.pipeline.audit("cli", "moneyflow-backfill", f"moneyflow:{result.get('end')}", json.dumps(
+            {"published": len(result.get("published") or []), "skipped": len(result.get("skipped") or []),
+             "failed": len(result.get("failed") or [])},
+            ensure_ascii=False,
+        ))
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if result.get("ok") else 1
+    parser.error(f"unknown command: {args.command}")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

xiaobai-datahub/datahub/coverage.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+from datahub.db import HubDB
+from datahub.timeutil import iter_yyyymmdd, yyyymmdd
+
+MISSING_SAMPLE_LIMIT = 10
+
+
+def coverage_payload(
+    *,
+    kind: str,
+    start: str,
+    end: str,
+    expected: Iterable[str],
+    available: Iterable[str],
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected_list = sorted({yyyymmdd(item) for item in expected if item})
+    available_set = {yyyymmdd(item) for item in available if item}
+    missing = [item for item in expected_list if item not in available_set]
+    payload: dict[str, Any] = {
+        "kind": kind,
+        "complete": not missing,
+        "requested_from": start,
+        "requested_to": end,
+        "available_from": min(available_set) if available_set else None,
+        "available_to": max(available_set) if available_set else None,
+        "expected_count": len(expected_list),
+        "available_count": len(available_set),
+        "missing_count": len(missing),
+        "missing_sample": missing[:MISSING_SAMPLE_LIMIT],
+    }
+    if extra:
+        payload.update(extra)
+    return payload
+
+
+def calendar_coverage(db: HubDB, start: str, end: str, exchange: str = "SSE") -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected = list(iter_yyyymmdd(start, end))
+    rows = db.fetchall(
+        "SELECT cal_date FROM trade_calendar WHERE exchange = ? AND cal_date >= ? AND cal_date <= ?",
+        (exchange, start, end),
+    )
+    return coverage_payload(
+        kind="calendar",
+        start=start,
+        end=end,
+        expected=expected,
+        available=(row["cal_date"] for row in rows),
+        extra={"exchange": exchange},
+    )
+
+
+def published_range_coverage(
+    db: HubDB,
+    dataset: str,
+    start: str,
+    end: str,
+    ts_code: str = "",
+    table: str = "",
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    calendar = calendar_coverage(db, start, end)
+    open_rows = db.fetchall(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date >= ? AND cal_date <= ?
+        ORDER BY cal_date
+        """,
+        (start, end),
+    )
+    expected_open = [row["cal_date"] for row in open_rows]
+    pubs = db.fetchall(
+        """
+        SELECT trade_date, active_batch FROM publications
+        WHERE dataset = ? AND trade_date >= ? AND trade_date <= ?
+        ORDER BY trade_date
+        """,
+        (dataset, start, end),
+    )
+    published_dates = [row["trade_date"] for row in pubs]
+    available = list(published_dates)
+    extra: dict[str, Any] = {
+        "dataset": dataset,
+        "calendar_complete": calendar["complete"],
+        "calendar_missing_count": calendar["missing_count"],
+    }
+    if ts_code and table and pubs:
+        present_code: list[str] = []
+        for pub in pubs:
+            hit = db.fetchone(
+                f"SELECT 1 AS ok FROM {table} WHERE trade_date = ? AND batch_id = ? AND ts_code = ? LIMIT 1",
+                (pub["trade_date"], pub["active_batch"], ts_code),
+            )
+            if hit:
+                present_code.append(pub["trade_date"])
+        available = present_code
+        extra["code"] = ts_code
+    payload = coverage_payload(
+        kind="published_range",
+        start=start,
+        end=end,
+        expected=expected_open,
+        available=available,
+        extra=extra,
+    )
+    if not calendar["complete"]:
+        payload["complete"] = False
+        payload["calendar_missing_sample"] = calendar["missing_sample"]
+    return payload
+
+
+def point_coverage(trade_date: str, dataset: str = "") -> dict[str, Any]:
+    day = yyyymmdd(trade_date)
+    payload = coverage_payload(
+        kind="point",
+        start=day,
+        end=day,
+        expected=[day],
+        available=[day],
+        extra={"dataset": dataset} if dataset else None,
+    )
+    return payload

xiaobai-datahub/datahub/datasets_ext.py

@@ -0,0 +1,180 @@
+"""Extended EOD datasets beyond the first-batch A/B release groups.
+
+These publish independently (soft): a failure here must not block daily/valuation
+release. Scheduler runs them after the core EOD window.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# Independent soft datasets (not part of A/B atomic groups).
+EXTENDED_SOFT_DATASETS = {
+    "limit_events",
+    "popularity",
+    "dragon_tiger",
+    "sector_daily",
+}
+
+EXTENDED_SCHEMA = """
+CREATE TABLE IF NOT EXISTS eod_limit_events (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL,
+  name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL,
+  limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL,
+  fd_amount REAL, first_time TEXT, last_time TEXT,
+  open_times INTEGER, up_stat TEXT, limit_times INTEGER,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, limit_type, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_limit_events (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL, batch_id TEXT NOT NULL,
+  name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL,
+  limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL,
+  fd_amount REAL, first_time TEXT, last_time TEXT,
+  open_times INTEGER, up_stat TEXT, limit_times INTEGER,
+  PRIMARY KEY (batch_id, ts_code, trade_date, limit_type)
+);
+
+CREATE TABLE IF NOT EXISTS eod_popularity (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL,
+  ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL,
+  hot REAL, concept TEXT, data_type TEXT,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, source, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_popularity (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL, batch_id TEXT NOT NULL,
+  ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL,
+  hot REAL, concept TEXT, data_type TEXT,
+  PRIMARY KEY (batch_id, ts_code, trade_date, source)
+);
+
+CREATE TABLE IF NOT EXISTS eod_dragon_tiger (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL,
+  ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL,
+  hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, hm_name, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_dragon_tiger (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL, batch_id TEXT NOT NULL,
+  ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL,
+  hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT,
+  PRIMARY KEY (batch_id, ts_code, trade_date, hm_name)
+);
+
+CREATE TABLE IF NOT EXISTS eod_sector_daily (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL,
+  name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL,
+  pct_change REAL, vol REAL, turnover_rate REAL, amount REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, family, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_sector_daily (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL, batch_id TEXT NOT NULL,
+  name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL,
+  pct_change REAL, vol REAL, turnover_rate REAL, amount REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date, family)
+);
+
+CREATE TABLE IF NOT EXISTS sector_master (
+  ts_code TEXT PRIMARY KEY,
+  name TEXT,
+  family TEXT NOT NULL,
+  exchange TEXT,
+  list_date TEXT,
+  member_count INTEGER,
+  type TEXT,
+  updated_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_eod_limit_date ON eod_limit_events(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_pop_date ON eod_popularity(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_lhb_date ON eod_dragon_tiger(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_sector_date ON eod_sector_daily(trade_date, family, batch_id);
+"""
+
+EXTENDED_DATASET_TABLES = {
+    "limit_events": ("eod_limit_events", "staging_limit_events"),
+    "popularity": ("eod_popularity", "staging_popularity"),
+    "dragon_tiger": ("eod_dragon_tiger", "staging_dragon_tiger"),
+    "sector_daily": ("eod_sector_daily", "staging_sector_daily"),
+}
+
+EXTENDED_STAGING_INSERT: dict[str, tuple[str, Any]] = {
+    "limit_events": (
+        "INSERT INTO staging_limit_events("
+        "ts_code,trade_date,limit_type,batch_id,name,industry,close,pct_chg,amount,"
+        "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["limit_type"], b,
+            r.get("name"), r.get("industry"), r.get("close"), r.get("pct_chg"), r.get("amount"),
+            r.get("limit_amount"), r.get("float_mv"), r.get("total_mv"), r.get("turnover_ratio"),
+            r.get("fd_amount"), r.get("first_time"), r.get("last_time"),
+            r.get("open_times"), r.get("up_stat"), r.get("limit_times"),
+        ),
+    ),
+    "popularity": (
+        "INSERT INTO staging_popularity("
+        "ts_code,trade_date,source,batch_id,ts_name,rank,pct_change,current_price,hot,concept,data_type) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["source"], b,
+            r.get("ts_name"), r.get("rank"), r.get("pct_change"), r.get("current_price"),
+            r.get("hot"), r.get("concept"), r.get("data_type"),
+        ),
+    ),
+    "dragon_tiger": (
+        "INSERT INTO staging_dragon_tiger("
+        "ts_code,trade_date,hm_name,batch_id,ts_name,buy_amount,sell_amount,net_amount,"
+        "hm_orgs,tag,pct_change,reason) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["hm_name"], b,
+            r.get("ts_name"), r.get("buy_amount"), r.get("sell_amount"), r.get("net_amount"),
+            r.get("hm_orgs"), r.get("tag"), r.get("pct_change"), r.get("reason"),
+        ),
+    ),
+    "sector_daily": (
+        "INSERT INTO staging_sector_daily("
+        "ts_code,trade_date,family,batch_id,name,open,high,low,close,pre_close,"
+        "pct_change,vol,turnover_rate,amount) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["family"], b,
+            r.get("name"), r.get("open"), r.get("high"), r.get("low"), r.get("close"),
+            r.get("pre_close"), r.get("pct_change"), r.get("vol"), r.get("turnover_rate"),
+            r.get("amount"),
+        ),
+    ),
+}
+
+EXTENDED_EOD_COPY = {
+    "limit_events": (
+        "INSERT OR REPLACE INTO eod_limit_events "
+        "SELECT ts_code,trade_date,limit_type,name,industry,close,pct_chg,amount,"
+        "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times,batch_id "
+        "FROM staging_limit_events WHERE batch_id = ?"
+    ),
+    "popularity": (
+        "INSERT OR REPLACE INTO eod_popularity "
+        "SELECT ts_code,trade_date,source,ts_name,rank,pct_change,current_price,hot,concept,data_type,batch_id "
+        "FROM staging_popularity WHERE batch_id = ?"
+    ),
+    "dragon_tiger": (
+        "INSERT OR REPLACE INTO eod_dragon_tiger "
+        "SELECT ts_code,trade_date,hm_name,ts_name,buy_amount,sell_amount,net_amount,"
+        "hm_orgs,tag,pct_change,reason,batch_id "
+        "FROM staging_dragon_tiger WHERE batch_id = ?"
+    ),
+    "sector_daily": (
+        "INSERT OR REPLACE INTO eod_sector_daily "
+        "SELECT ts_code,trade_date,family,name,open,high,low,close,pre_close,"
+        "pct_change,vol,turnover_rate,amount,batch_id "
+        "FROM staging_sector_daily WHERE batch_id = ?"
+    ),
+}

xiaobai-datahub/datahub/db.py

@@ -7,9 +7,10 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
+from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA
 from datahub.timeutil import isoformat
 
-SCHEMA = """
+_BASE_SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_migrations (
   version INTEGER PRIMARY KEY,
   applied_at TEXT NOT NULL
@@ -114,6 +115,21 @@ CREATE TABLE IF NOT EXISTS eod_index_bars (
   PRIMARY KEY (ts_code, trade_date, batch_id)
 ) WITHOUT ROWID;
 
+CREATE TABLE IF NOT EXISTS eod_stocks (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  symbol TEXT, name TEXT, area TEXT, industry TEXT, market TEXT,
+  list_status TEXT, list_date TEXT,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_stocks (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  symbol TEXT, name TEXT, area TEXT, industry TEXT, market TEXT,
+  list_status TEXT, list_date TEXT,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
 CREATE TABLE IF NOT EXISTS staging_bars (
   ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
   open REAL, high REAL, low REAL, close REAL, pct_chg REAL,
@@ -212,6 +228,17 @@ CREATE TABLE IF NOT EXISTS job_runs (
   detail TEXT
 );
 
+CREATE TABLE IF NOT EXISTS eod_progress (
+  trade_date TEXT PRIMARY KEY,
+  state TEXT NOT NULL,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at TEXT,
+  next_retry_at TEXT,
+  finished_at TEXT,
+  detail TEXT,
+  updated_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS audit_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,
   actor TEXT NOT NULL,
@@ -256,12 +283,16 @@ CREATE INDEX IF NOT EXISTS idx_eod_bars_date ON eod_bars(trade_date, batch_id);
 CREATE INDEX IF NOT EXISTS idx_calendar_open ON trade_calendar(is_open, cal_date);
 """
 
+SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA
+
 DATASET_TABLES = {
     "daily": ("eod_bars", "staging_bars"),
     "valuation": ("eod_valuation", "staging_valuation"),
     "moneyflow": ("eod_moneyflow", "staging_moneyflow"),
     "auction": ("eod_auction", "staging_auction"),
     "index_daily": ("eod_index_bars", "staging_index_bars"),
+    "stocks": ("eod_stocks", "staging_stocks"),
+    **EXTENDED_DATASET_TABLES,
 }
 
 

xiaobai-datahub/datahub/httpapp.py

@@ -190,7 +190,15 @@ class HubRequestHandler(BaseHTTPRequestHandler):
             return {}
         if length <= 0 or length > 65536:
             raise ValueError("请求内容为空或过大")
-        return json.loads(self.rfile.read(length).decode("utf-8"))
+        raw = self.rfile.read(length)
+        try:
+            payload = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            LOGGER.warning("invalid json request body")
+            raise ValueError("请求不是合法 JSON") from None
+        if not isinstance(payload, dict):
+            raise ValueError("请求不是合法 JSON")
+        return payload
 
     def _cookie_value(self, name: str) -> str:
         cookie = SimpleCookie()

xiaobai-datahub/datahub/logutil.py

@@ -2,7 +2,9 @@ from __future__ import annotations
 
 import json
 import logging
+import re
 import sys
+import traceback
 from typing import Any
 
 from datahub.timeutil import isoformat
@@ -11,6 +13,13 @@ _SECRET_KEYS = (
     "token", "password", "secret", "key", "authorization", "credential",
     "tushare_token", "datahub_token", "encryption_key", "cookie",
 )
+_SECRET_JSON = re.compile(
+    r'(?i)("(?:' + "|".join(re.escape(key) for key in _SECRET_KEYS) + r')"\s*:\s*")([^"\\]*(?:\\.[^"\\]*)*)(")'
+)
+
+
+def redact_log_text(text: str) -> str:
+    return _SECRET_JSON.sub(r"\1***\3", str(text))
 
 
 def _redact(value: Any, key: str = "") -> Any:
@@ -21,22 +30,39 @@ def _redact(value: Any, key: str = "") -> Any:
         return {str(item_key): _redact(item_value, str(item_key)) for item_key, item_value in value.items()}
     if isinstance(value, list):
         return [_redact(item) for item in value]
+    if isinstance(value, str):
+        return redact_log_text(value)
     return value
 
 
+def _safe_exc_text(exc_info: tuple[Any, Any, Any]) -> str:
+    exc = exc_info[1]
+    if isinstance(exc, json.JSONDecodeError):
+        return f"JSONDecodeError: invalid json at position {exc.pos}"
+    cause = getattr(exc, "__cause__", None)
+    if isinstance(cause, json.JSONDecodeError):
+        return f"{type(exc).__name__}: invalid json in request"
+    text = "".join(traceback.format_exception(*exc_info))
+    if isinstance(cause, json.JSONDecodeError) and cause.doc:
+        text = text.replace(cause.doc, "")
+    if isinstance(exc, json.JSONDecodeError) and exc.doc:
+        text = text.replace(exc.doc, "")
+    return redact_log_text(text)
+
+
 class JsonFormatter(logging.Formatter):
     def format(self, record: logging.LogRecord) -> str:
         payload: dict[str, Any] = {
             "ts": isoformat(),
             "level": record.levelname,
             "logger": record.name,
-            "message": record.getMessage(),
+            "message": redact_log_text(record.getMessage()),
         }
         extra = getattr(record, "hub", None)
         if isinstance(extra, dict):
             payload.update(_redact(extra))
         if record.exc_info:
-            payload["exc"] = self.formatException(record.exc_info)
+            payload["exc"] = _safe_exc_text(record.exc_info)
         return json.dumps(payload, ensure_ascii=False, default=str)
 
 

xiaobai-datahub/datahub/normalize.py

@@ -156,6 +156,95 @@ def normalize_stock(row: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def normalize_limit_event(row: dict[str, Any]) -> dict[str, Any]:
+    """limit_list_d. float_mv/total_mv/limit_amount are 万元 → yuan; amount/fd_amount already yuan."""
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "limit_type": str(row.get("limit_type") or "").strip().upper() or "U",
+        "name": str(row.get("name") or "").strip() or None,
+        "industry": str(row.get("industry") or "").strip() or None,
+        "close": round4(finite_number(row.get("close"))),
+        "pct_chg": round4(finite_number(row.get("pct_chg"))),
+        "amount": round4(finite_number(row.get("amount"))),
+        "limit_amount": round4(_scale(row.get("limit_amount"), AMOUNT_WAN_YUAN)),
+        "float_mv": round4(_scale(row.get("float_mv"), AMOUNT_WAN_YUAN)),
+        "total_mv": round4(_scale(row.get("total_mv"), AMOUNT_WAN_YUAN)),
+        "turnover_ratio": round4(finite_number(row.get("turnover_ratio"))),
+        "fd_amount": round4(finite_number(row.get("fd_amount"))),
+        "first_time": str(row.get("first_time") or "").strip() or None,
+        "last_time": str(row.get("last_time") or "").strip() or None,
+        "open_times": _optional_int(row.get("open_times")),
+        "up_stat": str(row.get("up_stat") or "").strip() or None,
+        "limit_times": _optional_int(row.get("limit_times")),
+    }
+
+
+def normalize_popularity(row: dict[str, Any], source: str = "") -> dict[str, Any]:
+    src = str(source or row.get("source") or "").strip().lower() or "ths"
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "source": src,
+        "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None,
+        "rank": _optional_int(row.get("rank")),
+        "pct_change": round4(
+            finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg"))
+        ),
+        "current_price": round4(finite_number(row.get("current_price") or row.get("price"))),
+        "hot": round4(finite_number(row.get("hot"))),
+        "concept": str(row.get("concept") or "").strip() or None,
+        "data_type": str(row.get("data_type") or "").strip() or None,
+    }
+
+
+def normalize_dragon_tiger(row: dict[str, Any]) -> dict[str, Any]:
+    """hm_detail amounts are 万元 → yuan."""
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "hm_name": str(row.get("hm_name") or "未命名游资").strip() or "未命名游资",
+        "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None,
+        "buy_amount": round4(_scale(row.get("buy_amount"), AMOUNT_WAN_YUAN)),
+        "sell_amount": round4(_scale(row.get("sell_amount"), AMOUNT_WAN_YUAN)),
+        "net_amount": round4(_scale(row.get("net_amount"), AMOUNT_WAN_YUAN)),
+        "hm_orgs": str(row.get("hm_orgs") or "").strip() or None,
+        "tag": str(row.get("tag") or "").strip() or None,
+        "pct_change": round4(finite_number(row.get("pct_change"))),
+        "reason": str(row.get("reason") or "").strip() or None,
+    }
+
+
+def normalize_sector_daily(row: dict[str, Any], family: str = "ths") -> dict[str, Any]:
+    fam = str(family or row.get("family") or "ths").strip().lower()
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "family": fam,
+        "name": str(row.get("name") or "").strip() or None,
+        "open": round4(finite_number(row.get("open"))),
+        "high": round4(finite_number(row.get("high"))),
+        "low": round4(finite_number(row.get("low"))),
+        "close": round4(finite_number(row.get("close"))),
+        "pre_close": round4(finite_number(row.get("pre_close"))),
+        "pct_change": round4(
+            finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg"))
+        ),
+        "vol": round4(finite_number(row.get("vol"))),
+        "turnover_rate": round4(finite_number(row.get("turnover_rate"))),
+        "amount": round4(finite_number(row.get("amount"))),
+    }
+
+
+def _optional_int(value: Any) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(float(value))
+    except (TypeError, ValueError):
+        return None
+
+
 def apply_qfq(price: float | None, factor: float | None, latest_factor: float | None) -> float | None:
     if price is None:
         return None
@@ -184,6 +273,11 @@ NORMALIZERS = {
     "calendar": normalize_calendar,
     "stock_basic": normalize_stock,
     "stocks": normalize_stock,
+    "limit_events": normalize_limit_event,
+    "limit_list_d": normalize_limit_event,
+    "popularity": normalize_popularity,
+    "dragon_tiger": normalize_dragon_tiger,
+    "sector_daily": normalize_sector_daily,
 }
 
 

xiaobai-datahub/datahub/realtime_serve.py

@@ -0,0 +1,188 @@
+"""Provisional (盘中观察) serving: quotes, index quotes, intraday points.
+
+Free sources only. Never writes official eod_* tables. Uses rt_cache + LKG.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.eastmoney import EastmoneyAdapter
+from datahub.adapters.tencent import TencentAdapter
+from datahub.codes import resolve_code
+from datahub.db import HubDB
+from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
+
+QUOTE_TTL = 60
+INDEX_TTL = 60
+INTRADAY_TTL = 20
+
+
+class RealtimeApiError(RuntimeError):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+def _envelope(data: Any, meta: dict[str, Any]) -> dict[str, Any]:
+    from datahub import SCHEMA_VERSION
+
+    return {"schema_version": SCHEMA_VERSION, "data": data, "meta": meta}
+
+
+def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
+    cache_key = "indexes:quotes"
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    eastmoney = EastmoneyAdapter()
+    try:
+        rows = eastmoney.fetch_indices()
+        source = "eastmoney:ulist"
+    except Exception:
+        rows = TencentAdapter().fetch_indices()
+        source = "tencent:qt"
+    if len(rows) < 3:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete")
+    payload = _envelope(
+        rows,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, INDEX_TTL, source)
+    return payload
+
+
+def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
+    if not codes:
+        raise RealtimeApiError("INVALID_ARGUMENT", "codes is required")
+    resolved: list[str] = []
+    for code in codes[:60]:
+        item = resolve_code(db, code) or _guess_ts_code(code)
+        if item:
+            resolved.append(item)
+    if not resolved:
+        raise RealtimeApiError("INVALID_ARGUMENT", "no resolvable codes")
+    cache_key = "quotes:" + ",".join(sorted(resolved))
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    adapter = EastmoneyAdapter()
+    try:
+        rows = adapter.fetch_quotes(resolved)
+        source = "eastmoney:clist"
+    except Exception as exc:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"quotes unavailable: {exc}") from exc
+    payload = _envelope(
+        rows,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, QUOTE_TTL, source)
+    return payload
+
+
+def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
+    ts_code = resolve_code(db, code) or _guess_ts_code(code)
+    if not ts_code:
+        raise RealtimeApiError("INVALID_ARGUMENT", f"ambiguous code: {code}")
+    cache_key = f"intraday:{ts_code}:{date or 'today'}"
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    adapter = EastmoneyAdapter()
+    try:
+        payload_data = adapter.fetch_intraday(ts_code)
+        source = "eastmoney:trends2"
+    except Exception as exc:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
+    payload = _envelope(
+        payload_data,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(payload_data.get("trade_date") or date or now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, INTRADAY_TTL, source)
+    return payload
+
+
+def _guess_ts_code(code: str) -> str | None:
+    raw = str(code or "").strip().upper()
+    if "." in raw:
+        return raw
+    if len(raw) == 6 and raw.isdigit():
+        if raw.startswith(("5", "6", "9")):
+            return f"{raw}.SH"
+        return f"{raw}.SZ"
+    return None
+
+
+def _read_cache(db: HubDB, cache_key: str) -> dict[str, Any] | None:
+    row = db.fetchone("SELECT * FROM rt_cache WHERE cache_key = ?", (cache_key,))
+    if not row:
+        return None
+    expires = str(row.get("expires_at") or "")
+    now = isoformat(now_shanghai())
+    if expires and expires < now:
+        return None
+    try:
+        payload = json.loads(row["payload"])
+    except json.JSONDecodeError:
+        return None
+    if isinstance(payload, dict) and isinstance(payload.get("meta"), dict):
+        stored = str(row.get("stored_at") or "")
+        try:
+            age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp()))
+        except Exception:
+            age = 0
+        payload["meta"]["staleness_seconds"] = age
+        payload["meta"]["stale"] = age > QUOTE_TTL
+    return payload
+
+
+def _write_cache(db: HubDB, cache_key: str, payload: dict[str, Any], ttl: int, source: str) -> None:
+    from datetime import timedelta
+
+    now = now_shanghai()
+    stored = isoformat(now)
+    expires = isoformat(now + timedelta(seconds=ttl))
+    db.execute(
+        """
+        INSERT INTO rt_cache(cache_key, payload, source, stored_at, expires_at)
+        VALUES (?,?,?,?,?)
+        ON CONFLICT(cache_key) DO UPDATE SET
+          payload=excluded.payload, source=excluded.source,
+          stored_at=excluded.stored_at, expires_at=excluded.expires_at
+        """,
+        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored, expires),
+    )
+    db.execute(
+        """
+        INSERT INTO last_known_good(cache_key, payload, source, stored_at)
+        VALUES (?,?,?,?)
+        ON CONFLICT(cache_key) DO UPDATE SET
+          payload=excluded.payload, source=excluded.source, stored_at=excluded.stored_at
+        """,
+        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored),
+    )

xiaobai-datahub/datahub/scheduler.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import threading
 from collections.abc import Callable
-from datetime import datetime, time
+from datetime import datetime, time, timedelta
 from typing import Any
 
 from datahub.db import HubDB
@@ -14,6 +14,8 @@ LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+
 
 def is_open_day(db: HubDB, day: str) -> bool:
     row = db.fetchone(
@@ -25,8 +27,19 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def _hhmm(value: str) -> time:
+    return datetime.strptime(value, "%H:%M").time()
+
+
 class Scheduler:
-    """Calendar-driven in-process scheduler. Non-trading days skip EOD fetches."""
+    """Calendar-driven in-process scheduler. Non-trading days skip EOD fetches.
+
+    EOD datasets that failed to publish (e.g. upstream not ready at 15:05)
+    are retried automatically every ``eod_retry_interval_minutes`` between
+    ``eod_retry_start`` and ``eod_retry_cutoff``. Progress is persisted in
+    ``eod_progress`` so a container restart catches up instead of waiting
+    for the next day, and completed days are never re-fetched.
+    """
 
     def __init__(self, db: HubDB, pipeline: Pipeline, jobs: dict[str, JobFn] | None = None) -> None:
         self.db = db
@@ -35,12 +48,20 @@ class Scheduler:
             "precheck": self._precheck,
             "eod_a": self._eod_a,
             "eod_b": self._eod_b,
+            "eod_c": self._eod_c,
+            "eod_d": self._eod_d,
+            "eod_e": self._eod_e,
+            "eod_f": self._eod_f,
+            "eod_retry": self._eod_retry,
+            "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
+            "history_backfill": self._history_backfill,
         }
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
         self._fired: set[tuple[str, str, str]] = set()
+        self._eod_lock = threading.Lock()
 
     def start(self, interval_seconds: float = 30.0) -> None:
         if self._thread and self._thread.is_alive():
@@ -62,17 +83,23 @@ class Scheduler:
             self._thread.join(timeout)
 
     def tick(self, clock: datetime | None = None) -> list[str]:
-        now = clock or now_shanghai()
+        now = now_shanghai(clock)
         day = yyyymmdd(now)
-        current = now.timetz() if False else now.time()
+        current = now.time()
         ran: list[str] = []
         plan = [
             ("precheck", time(8, 45)),
             ("eod_a", time(15, 5)),
             ("eod_b", time(15, 10)),
+            ("eod_c", time(16, 40)),
+            ("eod_d", time(16, 45)),
+            ("eod_e", time(18, 5)),
+            ("eod_f", time(22, 40)),
             ("cleanup", time(0, 30)),
             ("backup", time(0, 40)),
         ]
+        for refresh_at in self.pipeline.settings.stocks_refresh_times:
+            plan.append(("stocks_refresh", _hhmm(refresh_at)))
         open_day = is_open_day(self.db, day)
         for job_id, at in plan:
             if current < at:
@@ -80,18 +107,187 @@ class Scheduler:
             key = (job_id, day, at.strftime("%H%M"))
             if key in self._fired:
                 continue
-            if job_id in {"eod_a", "eod_b"} and not open_day:
+            if job_id in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"} and not open_day:
                 self._fired.add(key)
                 continue
             self._fired.add(key)
-            self.run_job(job_id, day)
+            if job_id in {"eod_a", "eod_b"}:
+                # Record the attempt before running: even a crash must not
+                # hide that today's first EOD try already happened.
+                self._record_eod_attempt(day, now)
+            try:
+                self.run_job(job_id, day)
+            except Exception:
+                if job_id not in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"}:
+                    raise
+                # Keep the tick alive; evening retries take over.
+                LOGGER.exception("scheduled job %s failed for %s", job_id, day)
             ran.append(job_id)
+            if job_id in {"eod_a", "eod_b"}:
+                self._settle_eod(day)
+        ran.extend(self._eod_retry_tick(now, day, open_day))
         return ran
 
+    # ------------------------------------------------------------------
+    # EOD retry window
+    # ------------------------------------------------------------------
+
+    def _eod_retry_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day:
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.eod_retry_start)
+        cutoff = _hhmm(settings.eod_retry_cutoff)
+        interval = timedelta(minutes=settings.eod_retry_interval_minutes)
+        missing = self.pipeline.missing_official_datasets(day)
+        row = self.eod_progress(day)
+
+        if not missing:
+            if row is None or row["state"] != "done":
+                self._save_eod_progress(day, state="done", finished_at=isoformat(now))
+            return []
+        if current < start:
+            return []
+        if row and row["state"] == "cutoff_failed":
+            return []
+        if current >= cutoff:
+            detail = "截止时间已到，缺失数据集: " + ",".join(missing)
+            self._save_eod_progress(day, state="cutoff_failed", finished_at=isoformat(now), detail=detail)
+            with self.db.write() as connection:
+                connection.execute(
+                    "INSERT INTO job_runs(job_id, state, started_at, finished_at, error, attempt, detail)"
+                    " VALUES ('eod_retry','failed',?,?,?,?,?)",
+                    (isoformat(now), isoformat(now), detail, int((row or {}).get("attempts") or 0), "eod cutoff reached"),
+                )
+            LOGGER.warning(
+                "eod retry window closed without data",
+                extra={"hub": {"trade_date": day, "missing": missing, "reason": "eod_cutoff"}},
+            )
+            return []
+        last = None
+        if row and row["last_attempt_at"]:
+            try:
+                last = datetime.fromisoformat(str(row["last_attempt_at"]))
+            except ValueError:
+                last = None
+        if last is not None and now_shanghai(last).replace(tzinfo=None) + interval > now.replace(tzinfo=None):
+            return []
+        if "eod_retry" not in self.jobs:
+            return []
+        self._record_eod_attempt(day, now)
+        ran = []
+        try:
+            self.run_job("eod_retry", day)
+        except Exception:
+            # job_runs already carries the failure; the window keeps retrying.
+            LOGGER.warning("eod retry failed for %s", day, exc_info=True)
+        ran.append("eod_retry")
+        self._settle_eod(day)
+        return ran
+
+    def _settle_eod(self, day: str) -> None:
+        """Flip the day to done as soon as every official dataset is published."""
+        if not self.pipeline.missing_official_datasets(day):
+            row = self.eod_progress(day)
+            if row is None or row["state"] != "done":
+                self._save_eod_progress(day, state="done", finished_at=isoformat())
+
+    def eod_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,))
+
+    def eod_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """Human/admin facing view: 等待上游 / 下次重试 / 已成功 / 已截止失败."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        now = now_shanghai(clock)
+        row = self.eod_progress(day)
+        open_day = is_open_day(self.db, day)
+        missing = self.pipeline.missing_official_datasets(day)
+        if row and row["state"] == "done":
+            state = "done"
+        elif not open_day:
+            state = "closed_day"
+        elif not missing:
+            state = "done"
+        elif row and row["state"] == "cutoff_failed":
+            state = "cutoff_failed"
+        elif now.time() < _hhmm("15:05"):
+            state = "pending_first_attempt"
+        else:
+            state = "waiting_upstream"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "missing_datasets": missing,
+            "attempts": int((row or {}).get("attempts") or 0),
+            "last_attempt_at": (row or {}).get("last_attempt_at"),
+            "next_retry_at": (row or {}).get("next_retry_at") if state == "waiting_upstream" else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "detail": (row or {}).get("detail"),
+        }
+
+    def _record_eod_attempt(self, day: str, now: datetime) -> None:
+        row = self.eod_progress(day)
+        attempts = int((row or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.eod_retry_interval_minutes
+        self._save_eod_progress(
+            day,
+            state="waiting_upstream",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+
+    def _save_eod_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM eod_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({"trade_date": day, "state": "waiting_upstream", "attempts": 0})
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO eod_progress({','.join(columns)}) VALUES ({placeholders})",
+                    tuple(payload[name] for name in columns),
+                )
+            else:
+                assignments = ", ".join(f"{name} = ?" for name in fields)
+                connection.execute(
+                    f"UPDATE eod_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
+    # ------------------------------------------------------------------
+    # Job execution
+    # ------------------------------------------------------------------
+
     def run_job(self, job_id: str, trade_date: str) -> dict[str, Any]:
         fn = self.jobs.get(job_id)
         if fn is None:
             raise KeyError(job_id)
+        if job_id in EOD_JOB_IDS:
+            if not self._eod_lock.acquire(blocking=False):
+                return {
+                    "job_id": job_id,
+                    "state": "skipped",
+                    "detail": "another EOD job is already running",
+                }
+            try:
+                return self._run_job(fn, job_id, trade_date)
+            finally:
+                self._eod_lock.release()
+        return self._run_job(fn, job_id, trade_date)
+
+    def _run_job(self, fn: JobFn, job_id: str, trade_date: str) -> dict[str, Any]:
         started = isoformat()
         run_id = None
         with self.db.write() as connection:
@@ -102,6 +298,10 @@ class Scheduler:
             run_id = cur.lastrowid
         try:
             result = fn(trade_date) or {}
+            if isinstance(result, dict):
+                failures = self.pipeline.eod_failures(result) if job_id in EOD_JOB_IDS else []
+                if failures:
+                    raise RuntimeError("; ".join(failures))
             with self.db.write() as connection:
                 connection.execute(
                     "UPDATE job_runs SET state=?, finished_at=?, rows_out=?, detail=? WHERE id=?",
@@ -125,6 +325,27 @@ class Scheduler:
     def _eod_b(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_batch_b(trade_date)
 
+    def _eod_c(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_c(trade_date)
+
+    def _eod_d(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_d(trade_date)
+
+    def _eod_e(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_e(trade_date)
+
+    def _eod_f(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_f(trade_date)
+
+    def _eod_retry(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_missing(trade_date)
+
+    def _stocks_refresh(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.refresh_stocks(trade_date)
+
+    def _history_backfill(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.backfill_history(trade_date)
+
     def _cleanup(self, trade_date: str) -> dict[str, Any]:
         result = self.pipeline.cleanup()
         if now_shanghai().weekday() == 6:

xiaobai-datahub/datahub/serving.py

@@ -6,6 +6,7 @@ from urllib.parse import parse_qs
 
 from datahub import SCHEMA_VERSION
 from datahub.codes import resolve_code
+from datahub.coverage import calendar_coverage, point_coverage, published_range_coverage
 from datahub.db import HubDB
 from datahub.normalize import qfq_bar
 from datahub.numbers import finite_number
@@ -72,6 +73,20 @@ class V1API:
             return self.moneyflow(q)
         if path == "/v1/auction":
             return self.auction(q)
+        if path == "/v1/limit-events":
+            return self.limit_events(q)
+        if path == "/v1/popularity":
+            return self.popularity(q)
+        if path == "/v1/dragon-tiger":
+            return self.dragon_tiger(q)
+        if path == "/v1/sectors":
+            return self.sectors(q)
+        if path == "/v1/quotes/latest":
+            return self.quotes_latest(q)
+        if path == "/v1/indexes/quotes":
+            return self.index_quotes(q)
+        if path == "/v1/intraday/points":
+            return self.intraday_points(q)
         if path == "/v1/datasets/status":
             return self.dataset_status(q.get("date") or "")
         if path == "/v1/batches":
@@ -129,10 +144,34 @@ class V1API:
             }
             for row in rows
         ]
-        return envelope(items, self._official_meta("calendar", end if items else start, source="tushare:trade_cal"))
+        meta = self._official_meta("calendar", end if items else start, source="tushare:trade_cal")
+        return envelope(items, attach_coverage(meta, calendar_coverage(self.db, start, end)))
 
     def stocks(self, updated_since: str, q: dict[str, str]) -> dict[str, Any]:
         limit, offset = self._page(q)
+        today = yyyymmdd(now_shanghai())
+        batch_id, snapshot = self.pipeline.published_stock_snapshot(today)
+        if batch_id:
+            # Formal view: the latest published stock snapshot, with batch
+            # metadata. Filters are applied in-memory on the snapshot.
+            pub = self.pipeline.latest_stocks_publication(today) or {}
+            rows = snapshot
+            if updated_since:
+                rows = []
+            rows = rows[offset: offset + limit]
+            return envelope(
+                rows,
+                {
+                    "tier": "official",
+                    "trade_date": pub.get("trade_date"),
+                    "published_at": pub.get("published_at"),
+                    "source": "tushare:stock_basic",
+                    "batch_id": batch_id,
+                    "stale": False,
+                    "staleness_seconds": 0,
+                    "state": pub.get("state"),
+                },
+            )
         if updated_since:
             rows = self.db.fetchall(
                 "SELECT * FROM stock_master WHERE updated_at >= ? ORDER BY ts_code LIMIT ? OFFSET ?",
@@ -172,9 +211,75 @@ class V1API:
     def auction(self, q: dict[str, str]) -> dict[str, Any]:
         return self._published_rows(dataset="auction", table="eod_auction", q=q, source="tushare:stk_auction")
 
+    def limit_events(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="limit_events",
+            table="eod_limit_events",
+            q=q,
+            source="tushare:limit_list_d",
+            extra_filters={"limit_type": q.get("limit_type") or ""},
+        )
+
+    def popularity(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="popularity",
+            table="eod_popularity",
+            q=q,
+            source="tushare:ths_hot+dc_hot",
+            extra_filters={"source": q.get("source") or ""},
+        )
+
+    def dragon_tiger(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="dragon_tiger",
+            table="eod_dragon_tiger",
+            q=q,
+            source="tushare:hm_detail",
+        )
+
+    def sectors(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="sector_daily",
+            table="eod_sector_daily",
+            q=q,
+            source="tushare:ths_daily+dc_index+sw_daily",
+            extra_filters={"family": q.get("family") or ""},
+        )
+
+    def quotes_latest(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_quotes
+
+        codes = [item.strip() for item in str(q.get("codes") or "").split(",") if item.strip()]
+        try:
+            return fetch_quotes(self.db, codes)
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
+    def index_quotes(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_index_quotes
+
+        try:
+            return fetch_index_quotes(self.db)
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
+    def intraday_points(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_intraday
+
+        code = str(q.get("code") or "").strip()
+        if not code:
+            raise ApiError("INVALID_ARGUMENT", "code is required")
+        try:
+            return fetch_intraday(self.db, code, yyyymmdd(q.get("date") or ""))
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
     def dataset_status(self, date: str) -> dict[str, Any]:
         trade_date = yyyymmdd(date or now_shanghai())
-        datasets = ("daily", "valuation", "moneyflow", "auction", "index_daily")
+        datasets = (
+            "daily", "valuation", "moneyflow", "auction", "index_daily", "stocks",
+            "limit_events", "popularity", "dragon_tiger", "sector_daily",
+        )
         items = []
         for dataset in datasets:
             pub = self.db.fetchone(
@@ -219,6 +324,7 @@ class V1API:
         source: str,
         adjust: str = "none",
         default_code: str = "",
+        extra_filters: dict[str, str] | None = None,
     ) -> dict[str, Any]:
         trade_date = q.get("date") or q.get("trade_date") or ""
         code = q.get("code") or default_code
@@ -239,6 +345,7 @@ class V1API:
             if resolved is None:
                 raise ApiError("INVALID_ARGUMENT", f"ambiguous code: {code}")
             ts_code = resolved
+        filters = {key: value for key, value in (extra_filters or {}).items() if value}
         # For a range, use per-date published batch. Single-date is the common path.
         if start == end:
             pub = self.db.fetchone(
@@ -249,7 +356,7 @@ class V1API:
                 raise ApiError(
                     "DATASET_NOT_PUBLISHED",
                     f"{dataset} {start} 尚未发布",
-                    extra={"expected_at": "15:05+08:00"},
+                    extra=self._unpublished_extra(dataset, start),
                 )
             limit, offset = self._page(q)
             sql = f"SELECT * FROM {table} WHERE trade_date = ? AND batch_id = ?"
@@ -257,6 +364,9 @@ class V1API:
             if ts_code:
                 sql += " AND ts_code = ?"
                 params.append(ts_code)
+            for key, value in filters.items():
+                sql += f" AND {key} = ?"
+                params.append(value)
             sql += " ORDER BY ts_code LIMIT ? OFFSET ?"
             params.extend([limit, offset])
             rows = [dict(row) for row in self.db.fetchall(sql, tuple(params))]
@@ -272,14 +382,18 @@ class V1API:
                 "staleness_seconds": 0,
                 "state": pub["state"],
             }
-            return envelope(rows, meta)
+            return envelope(rows, attach_coverage(meta, point_coverage(start, dataset)))
         # multi-day: walk published dates
         pubs = self.db.fetchall(
             "SELECT * FROM publications WHERE dataset = ? AND trade_date >= ? AND trade_date <= ? ORDER BY trade_date",
             (dataset, start, end),
         )
         if not pubs:
-            raise ApiError("DATASET_NOT_PUBLISHED", f"{dataset} {start}-{end} 尚未发布")
+            raise ApiError(
+                "DATASET_NOT_PUBLISHED",
+                f"{dataset} {start}-{end} 尚未发布",
+                extra=self._unpublished_extra(dataset, end),
+            )
         rows: list[dict[str, Any]] = []
         limit, offset = self._page(q)
         for pub in pubs:
@@ -288,23 +402,37 @@ class V1API:
             if ts_code:
                 sql += " AND ts_code = ?"
                 params.append(ts_code)
+            for key, value in filters.items():
+                sql += f" AND {key} = ?"
+                params.append(value)
             sql += " ORDER BY ts_code"
             rows.extend(self.db.fetchall(sql, tuple(params)))
         sliced = rows[offset: offset + limit]
         if adjust == "qfq" and dataset == "daily":
             sliced = self._apply_qfq(sliced)
         last = pubs[-1]
+        coverage = published_range_coverage(
+            self.db,
+            dataset,
+            start,
+            end,
+            ts_code=ts_code,
+            table=table,
+        )
         return envelope(
             sliced,
-            {
+            attach_coverage(
-                "tier": "official",
+                {
-                "trade_date": last["trade_date"],
+                    "tier": "official",
-                "published_at": last["published_at"],
+                    "trade_date": last["trade_date"],
-                "source": source,
+                    "published_at": last["published_at"],
-                "batch_id": last["active_batch"],
+                    "source": source,
-                "stale": False,
+                    "batch_id": last["active_batch"],
-                "staleness_seconds": 0,
+                    "stale": False,
-            },
+                    "staleness_seconds": 0,
+                },
+                coverage,
+            ),
         )
 
     def _apply_qfq(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -337,6 +465,20 @@ class V1API:
         offset = max(0, offset)
         return limit, offset
 
+    def _unpublished_extra(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Identifiable coverage info: is this a history gap or today-not-yet?"""
+        extra: dict[str, Any] = {"expected_at": "15:05+08:00"}
+        row = self.db.fetchone(
+            "SELECT MIN(trade_date) AS a, MAX(trade_date) AS b FROM publications WHERE dataset = ?",
+            (dataset,),
+        )
+        if row and row.get("a"):
+            extra["available_from"] = row["a"]
+            extra["available_to"] = row["b"]
+            if str(trade_date) < str(row["a"]):
+                extra["reason"] = "history_not_backfilled"
+        return extra
+
     def _official_meta(self, dataset: str, trade_date: str, source: str) -> dict[str, Any]:
         pub = self.db.fetchone(
             "SELECT * FROM publications WHERE dataset = ? AND trade_date = ?",
@@ -359,6 +501,13 @@ def add_default(days: int) -> str:
     return (now_shanghai() + timedelta(days=days)).strftime("%Y%m%d")
 
 
+def attach_coverage(meta: dict[str, Any], coverage: dict[str, Any]) -> dict[str, Any]:
+    merged = dict(meta)
+    merged["coverage"] = coverage
+    merged["incomplete"] = not bool(coverage.get("complete"))
+    return merged
+
+
 def parse_query(raw: str) -> dict[str, list[str]]:
     return parse_qs(raw, keep_blank_values=True)
 

xiaobai-datahub/datahub/settings.py

@@ -48,6 +48,37 @@ class Settings:
     def list_limit_max(self) -> int:
         return int(self.quality.get("list_limit_max") or 5000)
 
+    @property
+    def calendar_start(self) -> str:
+        return str(self.quality.get("calendar_start") or "20160101")
+
+    @property
+    def index_history_trading_days(self) -> int:
+        return int(self.quality.get("index_history_trading_days") or 260)
+
+    @property
+    def moneyflow_history_trading_days(self) -> int:
+        return int(self.quality.get("moneyflow_history_trading_days") or 60)
+
+    @property
+    def stocks_refresh_times(self) -> tuple[str, ...]:
+        raw = self.quality.get("stocks_refresh_times") or ["20:00", "23:10"]
+        if isinstance(raw, str):
+            raw = [raw]
+        return tuple(str(item) for item in raw)
+
+    @property
+    def eod_retry_start(self) -> str:
+        return str(self.quality.get("eod_retry_start") or "15:15")
+
+    @property
+    def eod_retry_interval_minutes(self) -> int:
+        return int(self.quality.get("eod_retry_interval_minutes") or 30)
+
+    @property
+    def eod_retry_cutoff(self) -> str:
+        return str(self.quality.get("eod_retry_cutoff") or "23:30")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/datahub/timeutil.py

@@ -58,6 +58,16 @@ def add_days(trade_date: str, days: int) -> str:
     return (parse_trade_date(trade_date) + timedelta(days=days)).strftime("%Y%m%d")
 
 
+def iter_yyyymmdd(start: str, end: str):
+    cursor = parse_trade_date(start)
+    last = parse_trade_date(end)
+    if cursor > last:
+        return
+    while cursor <= last:
+        yield cursor.strftime("%Y%m%d")
+        cursor += timedelta(days=1)
+
+
 def utc_timestamp(value: Any) -> str:
     if isinstance(value, datetime):
         return isoformat(value)

xiaobai-datahub/tests/fixtures.py

@@ -25,7 +25,7 @@ RAW = {
     ],
     "daily_basic": [
         {"ts_code": "600000.SH", "trade_date": "20240902", "turnover_rate": 1.2, "volume_ratio": 0.8, "total_mv": 1000.0, "circ_mv": 800.0, "pe_ttm": 5.1, "pb": 0.6, "ps_ttm": 1.1, "dv_ttm": 4.0},
-        {"ts_code": "000001.SZ", "trade_date": "20240902", "turnover_rate": 2.2, "volume_ratio": 1.1, "total_mv": 2000.0, "circ_mv": 1500.0, "pe_ttm": 6.2, "pb": 0.7, "ps_ttm": 1.2, "dv_ttm": 3.0},
+        {"ts_code": "000001.SZ", "trade_date": "20240902", "turnover_rate": 2.2, "volume_ratio": 1.1, "total_mv": 2000.0, "circ_mv": 1500.0, "pe_ttm": 6.2, "pb": 0.7, "ps_ttm": 1.2, "dv_ttm": None},
     ],
     "adj_factor": [
         {"ts_code": "600000.SH", "trade_date": "20240902", "adj_factor": 1.1},
@@ -45,15 +45,54 @@ RAW = {
         {"ts_code": "600000.SH", "trade_date": "20240902", "vol": 100, "price": 10.15, "amount": 1500000, "pre_close": 10.00, "turnover_rate": 0.1, "volume_ratio": 1.2, "float_share": 2000},
         {"ts_code": "000001.SZ", "trade_date": "20240902", "vol": 80, "price": 11.05, "amount": 1200000, "pre_close": 11.10, "turnover_rate": 0.2, "volume_ratio": 0.9, "float_share": 1800},
     ],
+    "limit_list_d": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "industry": "银行", "name": "浦发银行", "close": 10.2, "pct_chg": 9.95, "amount": 1e8, "limit_amount": 5000, "float_mv": 800, "total_mv": 1000, "turnover_ratio": 5.0, "fd_amount": 2e7, "first_time": "09:30:01", "last_time": "14:55:00", "open_times": 0, "up_stat": "1/1", "limit_times": 1, "limit_type": "U"},
+    ],
+    "ths_hot": [
+        {"ts_code": "600000.SH", "ts_name": "浦发银行", "hot": 90.0, "rank": 1, "pct_change": 1.2, "current_price": 10.2, "concept": "银行", "data_type": "热股", "trade_date": "20240902"},
+    ],
+    "dc_hot": [
+        {"ts_code": "600000.SH", "ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2, "hot": 80.0, "concept": "银行", "data_type": "A股市场", "trade_date": "20240902"},
+    ],
+    "hm_detail": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "ts_name": "浦发银行", "buy_amount": 1000, "sell_amount": 200, "net_amount": 800, "hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
+    ],
+    "top_list": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "name": "浦发银行", "pct_change": 9.95, "reason": "涨幅偏离值达7%"},
+    ],
+    "ths_daily": [
+        {"ts_code": "885811.TI", "trade_date": "20240902", "open": 1000, "high": 1010, "low": 990, "close": 1005, "pre_close": 995, "pct_change": 1.0, "vol": 100, "turnover_rate": 1.2},
+    ],
+    "dc_index": [
+        {"ts_code": "BK0475", "trade_date": "20240902", "name": "银行", "open": 100, "high": 101, "low": 99, "close": 100.5, "pre_close": 99.5, "pct_change": 1.0, "vol": 10, "amount": 1e8, "turnover_rate": 0.5},
+    ],
+    "sw_daily": [
+        {"ts_code": "801780.SI", "trade_date": "20240902", "name": "银行", "open": 2000, "high": 2010, "low": 1990, "close": 2005, "pct_change": 0.8, "vol": 50, "amount": 2e8},
+    ],
 }
 
 
 def fake_transport(api_name: str, params: dict, fields: str):
     if api_name == "index_daily":
         code = params.get("ts_code")
-        return [row for row in RAW["index_daily"] if row["ts_code"] == code]
+        rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+        trade_date = str(params.get("trade_date") or "")
+        start = str(params.get("start_date") or "")
+        end = str(params.get("end_date") or "")
+        if trade_date:
+            rows = [row for row in rows if row["trade_date"] == trade_date]
+        if start:
+            rows = [row for row in rows if row["trade_date"] >= start]
+        if end:
+            rows = [row for row in rows if row["trade_date"] <= end]
+        return rows
     if api_name == "trade_cal":
         start = str(params.get("start_date") or "")
         end = str(params.get("end_date") or "99999999")
         return [row for row in RAW["trade_cal"] if start <= row["cal_date"] <= end]
-    return list(RAW.get(api_name) or [])
+    rows = list(RAW.get(api_name) or [])
+    if api_name == "limit_list_d":
+        limit_type = str(params.get("limit_type") or "")
+        if limit_type:
+            rows = [row for row in rows if str(row.get("limit_type") or "") == limit_type]
+    return rows

xiaobai-datahub/tests/test_admin.py

@@ -1,17 +1,21 @@
 from __future__ import annotations
 
+import io
 import json
+import logging
 import tempfile
 import threading
 import unittest
 from http.server import ThreadingHTTPServer
 from pathlib import Path
+from urllib.error import HTTPError
 from urllib.request import Request, urlopen
 
 from datahub.adapters.tushare import TushareAdapter
 from datahub.crypto import SecretVault
 from datahub.httpapp import make_handler
 from datahub.hub import Hub
+from datahub.logutil import JsonFormatter
 from datahub.settings import Settings
 from tests.fixtures import fake_transport
 
@@ -91,6 +95,53 @@ class AdminTests(unittest.TestCase):
             )
         self.assertEqual(ctx.exception.code, 401)
 
+    def test_invalid_json_does_not_log_request_body_secrets(self) -> None:
+        secret = "SuperSecretPass1!"
+        token = "hub-token-should-not-leak"
+        raw = json.dumps({"password": secret, "token": token, "username": "hub_admin"}) + "{not-json"
+        stream = io.StringIO()
+        logger = logging.getLogger("datahub")
+        handler = logging.StreamHandler(stream)
+        handler.setFormatter(JsonFormatter())
+        logger.addHandler(handler)
+        previous_level = logger.level
+        logger.setLevel(logging.DEBUG)
+        try:
+            req = Request(
+                self.base + "/admin/api/login",
+                data=raw.encode("utf-8"),
+                headers={"Content-Type": "application/json"},
+                method="POST",
+            )
+            with self.assertRaises(HTTPError) as ctx:
+                urlopen(req, timeout=5)
+            body = ctx.exception.read().decode("utf-8")
+            self.assertEqual(ctx.exception.code, 400)
+            self.assertNotIn(secret, body)
+            self.assertNotIn(token, body)
+            blob = stream.getvalue() + body
+            self.assertNotIn(secret, blob)
+            self.assertNotIn(token, blob)
+            self.assertNotIn(raw, blob)
+        finally:
+            logger.removeHandler(handler)
+            logger.setLevel(previous_level)
+
+    def test_json_formatter_drops_decode_error_document(self) -> None:
+        secret = "ParseSecretTokenXYZ"
+        formatter = JsonFormatter()
+        logger = logging.getLogger("datahub.test")
+        record = logger.makeRecord(
+            "datahub.test", logging.ERROR, __file__, 1, "parse failed", (), None
+        )
+        try:
+            json.loads('{"password": "%s"}{' % secret)
+        except json.JSONDecodeError as exc:
+            record.exc_info = (type(exc), exc, exc.__traceback__)
+        blob = formatter.format(record)
+        self.assertNotIn(secret, blob)
+        self.assertIn("invalid json", blob)
+
 
 if __name__ == "__main__":
     unittest.main()

xiaobai-datahub/tests/test_api.py

@@ -107,6 +107,9 @@ class ApiContractTests(unittest.TestCase):
             self.assertIn("data", body)
             self.assertIn("meta", body)
             self.assertIn("tier", body["meta"])
+            if "calendar" in path or "bars" in path or "indexes" in path or "valuation" in path or "moneyflow" in path or "auction" in path:
+                self.assertIn("coverage", body["meta"])
+                self.assertIn("incomplete", body["meta"])
 
     def test_qfq_matches_formula(self) -> None:
         _, none = self._get(f"/v1/bars/daily?date={TRADE_DATE}&code=600000.SH&adjust=none", token=self.token)

xiaobai-datahub/tests/test_atomic_release.py

@@ -0,0 +1,408 @@
+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] | None = None
+        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 is not None:
+            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)
+        # incomplete A-group restages daily with the others; valuation fails → no A switch
+        self.assertEqual(results["daily"]["state"], "failed")
+        self.assertEqual(results["valuation"]["state"], "failed")
+        # the already-published daily batch is untouched and keeps serving
+        self.assertEqual(self.pipe.active_batch("daily", TRADE_DATE), first["batch_id"])
+        pubs = publications_map(self.db, TRADE_DATE)
+        self.assertEqual(pubs["daily"], first["batch_id"])
+        self.assertNotIn("valuation", pubs)
+        self.assertNotIn("moneyflow", pubs)
+        self.assertNotIn("auction", pubs)
+        # B-group is an independent boundary and may still publish
+        self.assertEqual(results["index_daily"]["state"], "published")
+        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_partial_group_retry_does_not_mix_batches(self) -> None:
+        """Already-published A members must be restaged with missing ones."""
+        first_daily = self.pipe.run_dataset("daily", TRADE_DATE)
+        first_moneyflow = self.pipe.run_dataset("moneyflow", TRADE_DATE)
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # A-group switched as one boundary; B-group (index) also published
+        for name in (*GROUP_A, "stocks"):
+            self.assertEqual(results[name]["state"], "published", name)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        pubs = self.db.fetchall(
+            "SELECT dataset, active_batch, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        # old partial batches replaced — no cross-batch mix of the first wave
+        self.assertNotEqual(by_ds["daily"]["active_batch"], first_daily["batch_id"])
+        self.assertNotEqual(by_ds["moneyflow"]["active_batch"], first_moneyflow["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in (*GROUP_A, "stocks")}
+        self.assertEqual(len(a_times), 1)
+        # serving resolves the new complete A-group batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        daily = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(daily["meta"]["batch_id"], results["daily"]["batch_id"])
+        self.assertEqual(daily["meta"]["batch_id"], by_ds["daily"]["active_batch"])
+
+    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)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        detail = str(audit["detail"])
+        self.assertIn("killed mid-switch", detail)
+        self.assertIn("failed", detail)
+
+    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)
+
+
+class StocksRefreshAtomicTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.transport.stocks = [
+            {"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"},
+        ]
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(first["state"], "published")
+        self.first_batch = first["batch_id"]
+
+    def test_refresh_keeps_master_when_snapshot_publish_fails(self) -> None:
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "金钛股份", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+
+        def explode() -> None:
+            raise RuntimeError("snapshot switch killed")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("snapshot switch killed", str(audit["detail"]))
+
+    def test_refresh_keeps_master_when_quality_gate_rejects(self) -> None:
+        self.transport.stocks = []  # empty → hard fail before publish
+        with self.assertRaises(Exception):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+
+
+class ForceBoundaryEntryTests(unittest.TestCase):
+    """CLI force / admin backfill must rebuild the full A/B boundary."""
+
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.first = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+
+    def test_force_republish_valuation_rebuilds_whole_a_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+            self.assertEqual(after[name], results[name]["batch_id"], name)
+        # B-group left alone
+        self.assertEqual(after["index_daily"], before["index_daily"])
+        pubs = self.db.fetchall(
+            "SELECT dataset, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        a_times = {row["published_at"] for row in pubs if row["dataset"] in {*GROUP_A, "stocks"}}
+        self.assertEqual(len(a_times), 1)
+
+    def test_force_republish_index_rebuilds_only_b_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("index_daily", TRADE_DATE)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertNotEqual(after["index_daily"], before["index_daily"])
+        for name in GROUP_A:
+            self.assertEqual(after[name], before[name], name)
+
+    def test_admin_backfill_official_dataset_uses_boundary(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+        result = admin.backfill("moneyflow", TRADE_DATE, "StartPass1", f"moneyflow:{TRADE_DATE}", "tester")
+        self.assertEqual(result["moneyflow"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+        # bad password / wrong confirm still rejected
+        with self.assertRaises(ApiError):
+            admin.backfill("daily", TRADE_DATE, "wrong", f"daily:{TRADE_DATE}", "tester")
+
+    def test_admin_backfill_switch_crash_is_failed_precondition(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(ApiError) as ctx:
+            admin.backfill("valuation", TRADE_DATE, "StartPass1", f"valuation:{TRADE_DATE}", "tester")
+        self.assertEqual(ctx.exception.code, "FAILED_PRECONDITION")
+        self.assertIn("killed mid-switch", ctx.exception.message)
+        # previous complete A/B versions keep serving
+        self.assertEqual(publications_map(self.db, TRADE_DATE), before)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("killed mid-switch", str(audit["detail"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_eod_retry.py

@@ -0,0 +1,241 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime
+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.scheduler import Scheduler
+from datahub.settings import Settings
+from datahub.timeutil import SHANGHAI
+from tests.fixtures import fake_transport
+
+OFFICIAL = {"daily", "valuation", "moneyflow", "auction", "index_daily"}
+
+
+class DelayedTransport:
+    """Upstream that only returns rows for dates it has "published" yet."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, ready_dates: set[str]) -> None:
+        self.ready = set(ready_dates)
+        self.calls: list[str] = []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        if api_name in self.DATE_APIS:
+            trade_date = str(params.get("trade_date") or "")
+            if trade_date and trade_date not in self.ready:
+                return []
+        return fake_transport(api_name, params, fields)
+
+
+def clock_at(day: str, hh: int, mm: int) -> datetime:
+    return datetime(int(day[:4]), int(day[4:6]), int(day[6:8]), hh, mm, tzinfo=SHANGHAI)
+
+
+class EodRetryTests(unittest.TestCase):
+    def _make(self, ready_dates: set[str]):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        db = HubDB(Path(tmp.name) / "hub.db")
+        transport = DelayedTransport(ready_dates)
+        adapter = TushareAdapter("x", transport=transport)
+        settings = Settings(
+            encryption_key=SecretVault.generate_key(),
+            db_path=db.path,
+            backup_dir=Path(tmp.name) / "backups",
+        )
+        pipe = Pipeline(db, adapter, settings)
+        pipe.ingest_reference("20240902")
+        sched = Scheduler(db, pipe)
+        return db, transport, pipe, sched
+
+    def _job_runs(self, db: HubDB, job_id: str) -> list[dict]:
+        return db.fetchall("SELECT * FROM job_runs WHERE job_id = ? ORDER BY id", (job_id,))
+
+    def _batches(self, db: HubDB, day: str) -> list[dict]:
+        placeholders = ",".join("?" for _ in OFFICIAL)
+        return db.fetchall(
+            f"SELECT * FROM batches WHERE trade_date = ? AND dataset IN ({placeholders})",
+            (day, *sorted(OFFICIAL)),
+        )
+
+    @staticmethod
+    def _batch_ids(db: HubDB, day: str) -> set[str]:
+        return {str(row["batch_id"]) for row in db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (day,))}
+
+    @staticmethod
+    def _eod_calls(transport: DelayedTransport) -> list[str]:
+        return [name for name in transport.calls if name in DelayedTransport.DATE_APIS]
+
+    def _published(self, db: HubDB, day: str) -> set[str]:
+        placeholders = ",".join("?" for _ in OFFICIAL)
+        rows = db.fetchall(
+            f"SELECT dataset FROM publications WHERE trade_date = ? AND dataset IN ({placeholders})",
+            (day, *sorted(OFFICIAL)),
+        )
+        return {str(row["dataset"]) for row in rows}
+
+    def test_first_empty_then_retry_succeeds(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make(set())
+
+        sched.tick(clock_at(day, 15, 5))   # eod_a: upstream empty -> failed
+        sched.tick(clock_at(day, 15, 10))  # eod_b: upstream empty -> failed
+        self.assertEqual(self._published(db, day), set())  # quality gate held
+
+        sched.tick(clock_at(day, 15, 20))  # inside window, but <30min since 15:10
+        self.assertEqual(self._job_runs(db, "eod_retry"), [])
+        status = sched.eod_status(day, clock=clock_at(day, 15, 20))
+        self.assertEqual(status["state"], "waiting_upstream")
+        self.assertTrue(status["next_retry_at"])
+        self.assertEqual(status["missing_datasets"], sorted(OFFICIAL))
+
+        sched.tick(clock_at(day, 15, 40))  # retry #1, still empty
+        runs = self._job_runs(db, "eod_retry")
+        self.assertEqual(len(runs), 1)
+        self.assertEqual(runs[0]["state"], "failed")
+        self.assertEqual(self._published(db, day), set())
+
+        transport.ready.add(day)
+        sched.tick(clock_at(day, 16, 10))  # retry #2 succeeds
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        self.assertEqual(sched.eod_status(day, clock=clock_at(day, 16, 10))["state"], "done")
+        progress = db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,))
+        self.assertEqual(progress["state"], "done")
+        self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
+
+        # success stops all further same-day requests
+        batches_before = len(self._batches(db, day))
+        eod_calls_before = len(self._eod_calls(transport))
+        sched.tick(clock_at(day, 17, 0))
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        self.assertEqual(len(self._eod_calls(transport)), eod_calls_before)
+
+    def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make(set())
+        sched.tick(clock_at(day, 15, 5))
+        sched.tick(clock_at(day, 15, 10))
+        sched.tick(clock_at(day, 15, 40))
+        sched.tick(clock_at(day, 16, 10))
+        sched.tick(clock_at(day, 23, 29))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 3)
+
+        sched.tick(clock_at(day, 23, 35))  # past cutoff 23:30
+        status = sched.eod_status(day, clock=clock_at(day, 23, 35))
+        self.assertEqual(status["state"], "cutoff_failed")
+        cutoff_runs = [r for r in self._job_runs(db, "eod_retry") if "截止" in str(r["error"])]
+        self.assertEqual(len(cutoff_runs), 1)
+        self.assertEqual(self._published(db, day), set())
+
+        attempts = db.fetchone("SELECT attempts FROM eod_progress WHERE trade_date = ?", (day,))["attempts"]
+        sched.tick(clock_at(day, 23, 59))
+        self.assertEqual(
+            db.fetchone("SELECT attempts FROM eod_progress WHERE trade_date = ?", (day,))["attempts"],
+            attempts,
+        )
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 4)  # 3 retries + 1 cutoff record
+        self.assertEqual(self._published(db, day), set())
+
+    def test_restart_catches_up_without_overwriting(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))   # eod_a publishes 4 datasets
+        sched.tick(clock_at(day, 15, 10))  # eod_b publishes index
+        self.assertEqual(self._published(db, day), OFFICIAL)
+
+        def official_batches() -> list[str]:
+            placeholders = ",".join("?" for _ in OFFICIAL)
+            return [
+                str(row["batch_id"])
+                for row in db.fetchall(
+                    f"SELECT batch_id FROM batches WHERE trade_date = ? AND dataset IN ({placeholders})",
+                    (day, *sorted(OFFICIAL)),
+                )
+            ]
+
+        active = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+        active_map = {row["dataset"]: row["active_batch"] for row in active if row["dataset"] in OFFICIAL}
+        batches_before = set(official_batches())
+        calls_before = self._eod_calls(transport)
+
+        # container restart: fresh scheduler, missed-time catch-up fires eod_a/eod_b
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(day, 21, 0))
+        self.assertIn("eod_a", ran)
+        self.assertIn("eod_b", ran)
+        self.assertNotIn("eod_retry", ran)
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+        self.assertEqual(
+            {row["dataset"]: row["active_batch"] for row in after if row["dataset"] in OFFICIAL},
+            active_map,
+        )
+        self.assertEqual(set(official_batches()), batches_before)  # no duplicate batches
+        self.assertEqual(self._eod_calls(transport), calls_before)  # no duplicate upstream EOD calls
+        self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 21, 0))["state"], "done")
+
+    def test_restart_with_partial_publish_only_fetches_missing(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))  # eod_a publishes 4; container "crashes" before eod_b
+        self.assertEqual(self._published(db, day), {"daily", "valuation", "moneyflow", "auction"})
+        batches_before = self._batch_ids(db, day)
+
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(day, 15, 20))  # restart: eod_b catch-up, eod_a all skipped
+        self.assertIn("eod_b", ran)
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        new_ids = self._batch_ids(db, day) - batches_before
+        new_datasets = {str(b["dataset"]) for b in self._batches(db, day) if str(b["batch_id"]) in new_ids}
+        self.assertEqual(new_datasets, {"index_daily"})
+        self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 15, 20))["state"], "done")
+
+    def test_closed_day_skips_all_eod_work(self) -> None:
+        day = "20240907"  # closed in fixture calendar
+        db, transport, pipe, sched = self._make(set())
+        for hh, mm in ((15, 5), (15, 10), (15, 40), (16, 10), (20, 0), (23, 40)):
+            ran = sched.tick(clock_at(day, hh, mm))
+            self.assertNotIn("eod_retry", ran)
+        eod_runs = db.fetchall("SELECT * FROM job_runs WHERE job_id LIKE 'eod%'")
+        self.assertEqual(eod_runs, [])
+        self.assertIsNone(db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,)))
+        self.assertEqual(self._published(db, day), set())
+        self.assertEqual(sched.eod_status(day, clock=clock_at(day, 20, 0))["state"], "closed_day")
+
+    def test_duplicate_and_concurrent_execution_are_safe(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))
+        sched.tick(clock_at(day, 15, 10))
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        batches_before = len(self._batches(db, day))
+        calls_before = len(transport.calls)
+
+        out = sched.run_job("eod_retry", day)  # manual duplicate run
+        self.assertEqual(out["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        self.assertEqual(len(transport.calls), calls_before)
+
+        sched._eod_lock.acquire()  # simulate an in-flight EOD job
+        try:
+            busy = sched.run_job("eod_retry", day)
+            self.assertEqual(busy["state"], "skipped")
+            busy_a = sched.run_job("eod_a", day)
+            self.assertEqual(busy_a["state"], "skipped")
+        finally:
+            sched._eod_lock.release()
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_extended_eod.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.hub import Hub
+from datahub.settings import Settings
+from tests.fixtures import TRADE_DATE, fake_transport
+
+
+class ExtendedEodTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        key = SecretVault.generate_key()
+        settings = Settings(
+            host="127.0.0.1",
+            port=0,
+            encryption_key=key,
+            api_token="k" * 32,
+            admin_password="StartPass1",
+            tushare_token="tushare-secret",
+            db_path=Path(self.tmp.name) / "hub.db",
+            backup_dir=Path(self.tmp.name) / "backups",
+            scheduler_enabled=False,
+            quality={"daily_row_ratio": 0.5, "null_rate_max": 0.5, "list_limit_default": 5000, "list_limit_max": 5000},
+        )
+        adapter = TushareAdapter("tushare-secret", transport=fake_transport)
+        self.hub = Hub(settings, adapter=adapter)
+        self.hub.pipeline.ingest_reference(TRADE_DATE)
+        for dataset in ("daily", "valuation", "moneyflow", "auction", "index_daily"):
+            self.hub.pipeline.run_dataset(dataset, TRADE_DATE)
+
+    def tearDown(self) -> None:
+        self.hub.stop()
+        self.tmp.cleanup()
+
+    def test_extended_soft_datasets_publish_and_serve(self) -> None:
+        results = self.hub.pipeline.run_extended_soft(
+            ("limit_events", "popularity", "dragon_tiger", "sector_daily"),
+            TRADE_DATE,
+        )
+        for name in ("limit_events", "popularity", "dragon_tiger", "sector_daily"):
+            self.assertEqual(results[name]["state"], "published", results[name])
+        api = self.hub.api
+        limits = api.handle("/v1/limit-events", {"date": [TRADE_DATE]})
+        self.assertGreaterEqual(len(limits["data"]), 1)
+        self.assertEqual(limits["meta"]["tier"], "official")
+        pop = api.handle("/v1/popularity", {"date": [TRADE_DATE], "source": ["ths"]})
+        self.assertEqual(pop["data"][0]["source"], "ths")
+        lhb = api.handle("/v1/dragon-tiger", {"date": [TRADE_DATE]})
+        self.assertEqual(lhb["data"][0]["hm_name"], "测试游资")
+        # hub stores 万元→元
+        self.assertEqual(lhb["data"][0]["buy_amount"], 10_000_000.0)
+        sectors = api.handle("/v1/sectors", {"date": [TRADE_DATE], "family": ["ths"]})
+        self.assertEqual(sectors["data"][0]["family"], "ths")
+        status = api.handle("/v1/datasets/status", {"date": [TRADE_DATE]})
+        names = {item["dataset"] for item in status["data"]}
+        self.assertTrue({"limit_events", "popularity", "dragon_tiger", "sector_daily"} <= names)
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_history_backfill.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import unittest
+from datetime import date, timedelta
+
+from datahub.coverage import calendar_coverage, point_coverage, published_range_coverage
+from datahub.serving import V1API
+from tests.fixtures import TRADE_DATE, fake_transport
+from tests.test_pipeline import make_pipeline
+
+
+def history_transport(open_dates: list[str], extra_closed: list[str] | None = None):
+    open_set = set(open_dates)
+    start = date(int(open_dates[0][:4]), int(open_dates[0][4:6]), int(open_dates[0][6:8]))
+    end = date(int(open_dates[-1][:4]), int(open_dates[-1][4:6]), int(open_dates[-1][6:8]))
+    calendar = []
+    cursor = start
+    while cursor <= end:
+        compact = cursor.strftime("%Y%m%d")
+        calendar.append(
+            {
+                "exchange": "SSE",
+                "cal_date": compact,
+                "is_open": 1 if compact in open_set else 0,
+                "pretrade_date": compact,
+            }
+        )
+        cursor += timedelta(days=1)
+    for day in extra_closed or []:
+        calendar.append(
+            {"exchange": "SSE", "cal_date": day, "is_open": 0, "pretrade_date": open_dates[0]}
+        )
+    index_codes = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+    index_rows = []
+    for ts_code in index_codes:
+        for day in open_dates:
+            index_rows.append(
+                {
+                    "ts_code": ts_code,
+                    "trade_date": day,
+                    "open": 100,
+                    "high": 101,
+                    "low": 99,
+                    "close": 100.5,
+                    "pct_chg": 0.1,
+                    "vol": 10.0,
+                    "amount": 20.0,
+                }
+            )
+
+    def transport(api_name, params, fields):
+        if api_name == "trade_cal":
+            start = str(params.get("start_date") or "")
+            end = str(params.get("end_date") or "99999999")
+            return [row for row in calendar if start <= row["cal_date"] <= end]
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in index_rows if row["ts_code"] == code]
+            trade_date = str(params.get("trade_date") or "")
+            start = str(params.get("start_date") or "")
+            end = str(params.get("end_date") or "")
+            if trade_date:
+                rows = [row for row in rows if row["trade_date"] == trade_date]
+            if start:
+                rows = [row for row in rows if row["trade_date"] >= start]
+            if end:
+                rows = [row for row in rows if row["trade_date"] <= end]
+            return rows
+        return fake_transport(api_name, params, fields)
+
+    return transport
+
+
+def consecutive_open_days(end: str, count: int) -> list[str]:
+    cursor = date(int(end[:4]), int(end[4:6]), int(end[6:8]))
+    days: list[str] = []
+    while len(days) < count:
+        if cursor.weekday() < 5:
+            days.append(cursor.strftime("%Y%m%d"))
+        cursor -= timedelta(days=1)
+    return sorted(days)
+
+
+class CoverageApiTests(unittest.TestCase):
+    def test_calendar_marks_holes_incomplete(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/calendar", {"from": ["20240901"], "to": ["20240907"]})
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+        self.assertGreater(payload["meta"]["coverage"]["missing_count"], 0)
+        self.assertIn("20240901", payload["meta"]["coverage"]["missing_sample"])
+
+    def test_calendar_complete_when_every_day_present(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/calendar", {"from": ["20240902"], "to": ["20240903"]})
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertTrue(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["expected_count"], 2)
+        self.assertEqual(len(payload["data"]), 2)
+
+    def test_index_range_incomplete_without_history(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("index_daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/indexes/bars",
+            {"from": ["20240902"], "to": ["20240903"], "code": ["000001.SH"]},
+        )
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["available_count"], 1)
+        self.assertIn("20240903", payload["meta"]["coverage"]["missing_sample"])
+
+    def test_index_point_query_stays_complete(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("index_daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/indexes/bars", {"date": [TRADE_DATE], "code": ["000001.SH"]})
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertTrue(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["kind"], "point")
+
+    def test_daily_range_incomplete_without_stock_history(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/bars/daily",
+            {"from": ["20240902"], "to": ["20240903"], "code": ["600000.SH"]},
+        )
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+
+
+class HistoryBackfillTests(unittest.TestCase):
+    def test_index_history_is_idempotent_and_covers_requested_days(self) -> None:
+        open_dates = consecutive_open_days(TRADE_DATE, 5)
+        pipe, db = make_pipeline(quality={"index_history_trading_days": 5, "calendar_start": open_dates[0]})
+        pipe.adapter._transport = history_transport(open_dates)
+        first = pipe.backfill_history(TRADE_DATE, index_days=5)
+        self.assertTrue(first["ok"])
+        self.assertEqual(first["calendar"]["calendar_from"], open_dates[0])
+        self.assertEqual(first["index_daily"]["requested_days"], 5)
+        self.assertEqual(len(first["index_daily"]["published"]), 5)
+        self.assertEqual(first["index_daily"]["skipped"], [])
+        pubs = db.fetchall("SELECT trade_date FROM publications WHERE dataset='index_daily'")
+        self.assertEqual(sorted(row["trade_date"] for row in pubs), open_dates)
+
+        second = pipe.backfill_index_history(TRADE_DATE, trading_days=5)
+        self.assertTrue(second["ok"])
+        self.assertEqual(second["published"], [])
+        self.assertEqual(second["skipped"], open_dates)
+
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/indexes/bars",
+            {"from": [open_dates[0]], "to": [open_dates[-1]], "code": ["000001.SH"]},
+        )
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertEqual(payload["meta"]["coverage"]["available_count"], 5)
+        self.assertEqual(len(payload["data"]), 5)
+
+    def test_index_history_retries_failed_dates_without_dropping_success(self) -> None:
+        open_dates = consecutive_open_days(TRADE_DATE, 3)
+        base = history_transport(open_dates)
+
+        def missing_cyb(api_name, params, fields):
+            if api_name == "index_daily" and params.get("ts_code") == "399006.SZ":
+                raise RuntimeError("upstream down")
+            return base(api_name, params, fields)
+
+        pipe, db = make_pipeline(quality={"index_history_trading_days": 3, "max_publish_attempts": 1})
+        pipe.adapter._transport = missing_cyb
+        first = pipe.backfill_history(TRADE_DATE, calendar_start=open_dates[0], index_days=3)
+        self.assertFalse(first["ok"])
+        self.assertTrue(any(item.get("ts_code") == "399006.SZ" for item in first["index_daily"]["failed"]))
+        published_first = {
+            row["trade_date"]
+            for row in db.fetchall("SELECT trade_date FROM publications WHERE dataset='index_daily'")
+        }
+        self.assertEqual(published_first, set(open_dates))
+
+        pipe.adapter._transport = base
+        retry = pipe.backfill_index_history(TRADE_DATE, trading_days=3)
+        self.assertTrue(retry["ok"])
+        self.assertEqual(len(retry["published"]), 3)
+        for day in open_dates:
+            rows = db.fetchall(
+                """
+                SELECT DISTINCT ts_code FROM eod_index_bars
+                WHERE trade_date = ? AND batch_id = (
+                    SELECT active_batch FROM publications
+                    WHERE dataset='index_daily' AND trade_date = ?
+                )
+                """,
+                (day, day),
+            )
+            self.assertEqual({row["ts_code"] for row in rows}, {"000001.SH", "399001.SZ", "399006.SZ", "000300.SH"})
+
+    def test_prepared_rows_skip_upstream_fetch(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        calls = {"n": 0}
+        original = pipe.adapter._transport
+
+        def counting(api_name, params, fields):
+            calls["n"] += 1
+            return original(api_name, params, fields)
+
+        pipe.adapter._transport = counting
+        rows = pipe.adapter.normalize("index_daily", original("index_daily", {"ts_code": "000001.SH", "trade_date": TRADE_DATE}, ""))
+        before = calls["n"]
+        result = pipe.run_dataset("index_daily", TRADE_DATE, prepared_rows=rows)
+        self.assertEqual(result["rows"], 1)
+        self.assertEqual(calls["n"], before)
+
+    def test_coverage_helpers_point_and_calendar(self) -> None:
+        pipe, db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        point = point_coverage(TRADE_DATE, "index_daily")
+        self.assertTrue(point["complete"])
+        cal = calendar_coverage(db, "20240902", "20240903")
+        self.assertTrue(cal["complete"])
+        pub = published_range_coverage(db, "index_daily", "20240902", "20240903")
+        self.assertFalse(pub["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_layout.py

@@ -19,11 +19,17 @@ class LayoutTests(unittest.TestCase):
     def test_reserved_adapters_present(self) -> None:
         from datahub.adapters import RESERVED
 
-        for name in ("eastmoney", "tencent", "ths", "xgb", "akshare", "ifind"):
+        for name in ("ths", "xgb", "akshare", "ifind"):
             self.assertIn(name, RESERVED)
             probe = RESERVED[name].probe()
             self.assertEqual(probe["state"], "reserved")
             self.assertFalse(probe["configured"])
+        for name in ("eastmoney", "tencent"):
+            self.assertIn(name, RESERVED)
+            probe = RESERVED[name].probe()
+            # Live free adapters: probe may be ok/error/empty depending on network.
+            self.assertIn(probe["state"], {"ok", "empty", "error"})
+            self.assertTrue(probe["configured"])
 
 
 if __name__ == "__main__":

xiaobai-datahub/tests/test_moneyflow_backfill.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import unittest
+from datetime import date, timedelta
+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.serving import ApiError, V1API
+from datahub.settings import Settings
+from tests.fixtures import fake_transport
+
+OPEN_DATES = ["20240826", "20240827", "20240828", "20240829", "20240830", "20240902", "20240903"]
+EMPTY_UPSTREAM = {"20240828"}  # one date the upstream cannot serve
+
+
+def build_calendar(open_dates: list[str], span_days: int = 16) -> list[dict]:
+    start = date(int(open_dates[0][:4]), int(open_dates[0][4:6]), int(open_dates[0][6:8]))
+    rows = []
+    open_set = set(open_dates)
+    for offset in range(span_days):
+        cursor = start + timedelta(days=offset)
+        compact = cursor.strftime("%Y%m%d")
+        rows.append(
+            {"exchange": "SSE", "cal_date": compact, "is_open": 1 if compact in open_set else 0, "pretrade_date": compact}
+        )
+    return rows
+
+
+def moneyflow_rows(day: str) -> list[dict]:
+    return [
+        {
+            "ts_code": "600000.SH", "trade_date": day,
+            "buy_sm_amount": 10 + int(day[-2:]), "sell_sm_amount": 8, "buy_md_amount": 20, "sell_md_amount": 15,
+            "buy_lg_amount": 30, "sell_lg_amount": 25, "buy_elg_amount": 40, "sell_elg_amount": 35, "net_mf_amount": 17,
+        },
+        {
+            "ts_code": "000001.SZ", "trade_date": day,
+            "buy_sm_amount": 11, "sell_sm_amount": 9, "buy_md_amount": 21, "sell_md_amount": 16,
+            "buy_lg_amount": 31, "sell_lg_amount": 26, "buy_elg_amount": 41, "sell_elg_amount": 36, "net_mf_amount": 18,
+        },
+    ]
+
+
+class MoneyflowHistoryTransport:
+    def __init__(self) -> None:
+        self.calendar = build_calendar(OPEN_DATES)
+        self.moneyflow_fetches: list[str] = []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        if api_name == "trade_cal":
+            start = str(params.get("start_date") or "")
+            end = str(params.get("end_date") or "99999999")
+            return [row for row in self.calendar if start <= row["cal_date"] <= end]
+        if api_name == "moneyflow":
+            day = str(params.get("trade_date") or "")
+            self.moneyflow_fetches.append(day)
+            if day in EMPTY_UPSTREAM:
+                return []
+            return moneyflow_rows(day)
+        return fake_transport(api_name, params, fields)
+
+
+class MoneyflowBackfillTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = MoneyflowHistoryTransport()
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.db = HubDB(Path(tmp.name) / "hub.db")
+        adapter = TushareAdapter("x", transport=self.transport)
+        settings = Settings(
+            encryption_key=SecretVault.generate_key(),
+            db_path=self.db.path,
+            backup_dir=Path(tmp.name) / "backups",
+            quality={"max_publish_attempts": 2, "publication_generations": 3},
+        )
+        self.pipe = Pipeline(self.db, adapter, settings)
+        self.pipe.ingest_reference("20240903")
+        self.api = V1API(self.db, self.pipe, settings)
+
+    def test_backfill_publishes_window_and_reports_failures(self) -> None:
+        result = self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        published = [item["trade_date"] for item in result["published"]]
+        self.assertEqual(published, ["20240829", "20240830", "20240902", "20240903"])
+        self.assertEqual(result["failed"][0]["trade_date"], "20240828")
+        self.assertFalse(result["ok"])
+        rows = self.db.fetchall("SELECT * FROM eod_moneyflow WHERE trade_date='20240830'")
+        self.assertEqual(len(rows), 2)
+
+    def test_backfill_is_idempotent(self) -> None:
+        self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        fetches_after_first = list(self.transport.moneyflow_fetches)
+        second = self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        # only the still-missing date is re-fetched; published dates are skipped
+        self.assertEqual(self.transport.moneyflow_fetches[len(fetches_after_first):], ["20240828"])
+        self.assertEqual(len(second["skipped"]), 4)
+
+    def test_point_query_on_backfilled_date_serves_data(self) -> None:
+        self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        payload = self.api.handle("/v1/moneyflow", {"date": ["20240830"]})
+        self.assertEqual(len(payload["data"]), 2)
+        self.assertEqual(payload["data"][0]["net_mf_amount"], 180000.0)
+
+    def test_unpublished_point_below_window_is_identifiable(self) -> None:
+        self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        with self.assertRaises(ApiError) as ctx:
+            self.api.handle("/v1/moneyflow", {"date": ["20240801"]})
+        extra = ctx.exception.extra
+        self.assertEqual(extra["available_from"], "20240829")  # window starts at the first published date
+        self.assertEqual(extra["available_to"], "20240903")
+        self.assertEqual(extra["reason"], "history_not_backfilled")
+        self.assertEqual(extra["expected_at"], "15:05+08:00")
+
+    def test_range_query_flags_missing_dates(self) -> None:
+        self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5)
+        payload = self.api.handle("/v1/moneyflow", {"from": ["20240828"], "to": ["20240903"]})
+        coverage = payload["meta"]["coverage"]
+        self.assertFalse(coverage["complete"])
+        self.assertEqual(coverage["missing_count"], 1)
+        self.assertEqual(coverage["missing_sample"], ["20240828"])
+        self.assertTrue(payload["meta"]["incomplete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_pipeline.py

@@ -69,6 +69,7 @@ class PipelineTests(unittest.TestCase):
         pipe, db = make_pipeline()
         ref = pipe.ingest_reference(TRADE_DATE)
         self.assertEqual(ref["stocks"], 2)
+        self.assertEqual(ref["calendar_from"], "20160101")
         result = pipe.run_dataset("daily", TRADE_DATE)
         self.assertEqual(result["state"], "published")
         self.assertEqual(result["rows"], 2)

xiaobai-datahub/tests/test_quality_gates.py

@@ -0,0 +1,262 @@
+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, QualityError
+from datahub.scheduler import Scheduler
+from datahub.settings import Settings
+from tests.fixtures import TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+
+FIELD_GATES = {
+    "valuation": {
+        "fields": [
+            "turnover_rate", "volume_ratio", "total_mv", "circ_mv",
+            "pe_ttm", "pb", "ps_ttm", "dv_ttm",
+        ],
+        "min_nonnull_rate": 0.9,
+        "min_nonnull_rate_by_field": {"pe_ttm": 0.5, "dv_ttm": 0.3},
+        "max_nonnull_drop_vs_prev": 0.15,
+        "max_nonfinite_rate": 0.01,
+    },
+}
+
+
+class ValuationTransport:
+    """fake_transport with switchable daily_basic degradation modes."""
+
+    def __init__(self) -> None:
+        self.mode = "ok"
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        rows = fake_transport(api_name, params, fields)
+        if api_name != "daily_basic":
+            return rows
+        trade_date = str(params.get("trade_date") or "")
+        if trade_date:
+            rows = [{**row, "trade_date": trade_date} for row in rows]
+        if self.mode == "ok":
+            return rows
+        patched = []
+        for row in rows:
+            item = dict(row)
+            if self.mode == "fields_all_null":
+                item["volume_ratio"] = None
+                item["dv_ttm"] = None
+            elif self.mode == "vr_all_null":
+                item["volume_ratio"] = None
+            elif self.mode == "dv_all_null":
+                item["dv_ttm"] = None
+            elif self.mode == "nonfinite":
+                item["volume_ratio"] = float("inf")
+            patched.append(item)
+        return patched
+
+
+def make_pipe(transport, quality_extra=None, clock=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": 3,
+        "publication_generations": 3,
+        "job_run_retain_days": 90,
+        "staging_retain_days": 14,
+        "field_gates": FIELD_GATES,
+    }
+    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, clock=clock)
+    pipe._tmp = tmp
+    return pipe, db
+
+
+class ValuationFieldGateTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = ValuationTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+
+    def _active(self) -> str | None:
+        row = self.db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )
+        return str(row["active_batch"]) if row else None
+
+    def test_normal_batch_with_legit_dv_nulls_passes(self) -> None:
+        result = self.pipe.run_dataset("valuation", TRADE_DATE)
+        self.assertEqual(result["state"], "published")
+        fields = result["quality"]["fields"]
+        # fixture: 1 of 2 stocks has null dv_ttm → 0.5 non-null ≥ 0.3 floor
+        self.assertEqual(fields["dv_ttm"]["nonnull_rate"], 0.5)
+        self.assertEqual(fields["volume_ratio"]["nonnull_rate"], 1.0)
+        self.assertFalse(result["quality"]["errors"])
+
+    def test_all_null_fields_rejected_and_prev_batch_kept(self) -> None:
+        first = self.pipe.run_dataset("valuation", TRADE_DATE)
+        self.transport.mode = "fields_all_null"
+        with self.assertRaises(QualityError) as ctx:
+            self.pipe.run_dataset("valuation", TRADE_DATE)
+        errors = "; ".join(ctx.exception.report["errors"])
+        self.assertIn("field gate: valuation.volume_ratio non-null rate 0.0000 < 0.9", errors)
+        self.assertIn("field gate: valuation.dv_ttm non-null rate 0.0000 < 0.3", errors)
+        # previous good publication stays active
+        self.assertEqual(self._active(), first["batch_id"])
+        # rejected batch left staged with readable error + field stats
+        rejected = self.db.fetchone(
+            "SELECT * FROM batches WHERE state='staged' AND dataset='valuation' ORDER BY started_at DESC",
+        )
+        self.assertIsNotNone(rejected)
+        self.assertIn("field gate: valuation.volume_ratio", str(rejected["error"]))
+        import json
+
+        quality = json.loads(rejected["quality_json"])
+        self.assertEqual(quality["fields"]["volume_ratio"]["nonnull"], 0)
+        self.assertEqual(quality["fields"]["dv_ttm"]["nonnull"], 0)
+
+    def test_volume_ratio_all_null_alone_rejected(self) -> None:
+        self.pipe.run_dataset("valuation", TRADE_DATE)
+        self.transport.mode = "vr_all_null"
+        with self.assertRaises(QualityError):
+            self.pipe.run_dataset("valuation", TRADE_DATE)
+        self.assertEqual(
+            self.db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            "20240902-valuation-001",
+        )
+
+    def test_dv_ttm_all_null_rejected_by_floor_and_collapse(self) -> None:
+        prev_day = "20240830"
+        prev = self.pipe.run_dataset("valuation", prev_day)  # prev dv nonnull 0.5
+        self.transport.mode = "dv_all_null"
+        with self.assertRaises(QualityError) as ctx:
+            self.pipe.run_dataset("valuation", TRADE_DATE)
+        errors = "; ".join(ctx.exception.report["errors"])
+        self.assertIn("field gate: valuation.dv_ttm non-null rate 0.0000 < 0.3", errors)
+        self.assertIn(f"dropped > 0.15 vs prev batch {prev['batch_id']}", errors)
+
+    def test_nonfinite_values_rejected(self) -> None:
+        self.pipe.run_dataset("valuation", TRADE_DATE)
+        rows = self.pipe.adapter.normalize(
+            "valuation", self.pipe._guarded_fetch("valuation", {"trade_date": TRADE_DATE})
+        )
+        for row in rows:
+            row["volume_ratio"] = float("inf")
+        with self.assertRaises(QualityError) as ctx:
+            self.pipe.run_dataset("valuation", TRADE_DATE, prepared_rows=rows)
+        errors = "; ".join(ctx.exception.report["errors"])
+        self.assertIn("field gate: valuation.volume_ratio non-finite rate 1.0000 > 0.01", errors)
+
+    def test_gate_off_when_not_configured(self) -> None:
+        pipe, _db = make_pipe(ValuationTransport(), quality_extra={"field_gates": {}})
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.adapter._transport.mode = "fields_all_null"
+        result = pipe.run_dataset("valuation", TRADE_DATE)
+        self.assertEqual(result["state"], "published")  # legacy behavior when unconfigured
+
+    def test_gate_applies_to_any_configured_dataset(self) -> None:
+        gates = {"daily": {"fields": ["volume"], "min_nonnull_rate": 0.9, "max_nonfinite_rate": 0.01}}
+        pipe, _db = make_pipe(ValuationTransport(), quality_extra={"field_gates": gates})
+        pipe.ingest_reference(TRADE_DATE)
+
+        def null_volume(api_name, params, fields):
+            if api_name != "daily":
+                return fake_transport(api_name, params, fields)
+            rows = fake_transport(api_name, params, fields)
+            for row in rows:
+                row["vol"] = None
+            return rows
+
+        pipe.adapter._transport = null_volume
+        with self.assertRaises(QualityError) as ctx:
+            pipe.run_dataset("daily", TRADE_DATE)
+        errors = "; ".join(ctx.exception.report["errors"])
+        self.assertIn("field gate: daily.volume non-null rate 0.0000 < 0.9", errors)
+
+
+class GateRetryInterplayTests(unittest.TestCase):
+    def test_rejected_valuation_stays_missing_and_retry_publishes_later(self) -> None:
+        transport = ValuationTransport()
+        transport.mode = "fields_all_null"
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        db = HubDB(Path(tmp.name) / "hub.db")
+        adapter = TushareAdapter("x", transport=transport)
+        settings = Settings(
+            encryption_key=SecretVault.generate_key(),
+            db_path=db.path,
+            backup_dir=Path(tmp.name) / "backups",
+            quality={"field_gates": FIELD_GATES, "max_publish_attempts": 2},
+        )
+        pipe = Pipeline(db, adapter, settings)
+        pipe.ingest_reference("20240902")
+        sched = Scheduler(db, pipe)
+
+        sched.tick(clock_at("20240902", 15, 5))   # valuation rejected by field gate
+        sched.tick(clock_at("20240902", 15, 10))
+        self.assertIn("valuation", pipe.missing_official_datasets("20240902"))
+        self.assertEqual(
+            pipe.active_batch("valuation", "20240902"),
+            None,
+        )
+
+        transport.mode = "ok"
+        sched.tick(clock_at("20240902", 15, 45))  # retry passes the gate
+        self.assertNotIn("valuation", pipe.missing_official_datasets("20240902"))
+        rows = db.fetchall("SELECT * FROM eod_valuation WHERE trade_date='20240902'")
+        self.assertTrue(rows)
+        self.assertTrue(all(row["volume_ratio"] is not None for row in rows))
+
+
+class ForceRepublishTests(unittest.TestCase):
+    def test_force_boundary_republish_keeps_prev_for_rollback(self) -> None:
+        transport = ValuationTransport()
+        pipe, db = make_pipe(transport)
+        pipe.ingest_reference(TRADE_DATE)
+        first = pipe.run_eod_batch_a(TRADE_DATE)
+        first_val = first["valuation"]["batch_id"]
+        first_daily = first["daily"]["batch_id"]
+        transport.mode = "vr_all_null"
+        blocked = pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual(blocked["valuation"]["state"], "failed")
+        self.assertEqual(pipe.active_batch("valuation", TRADE_DATE), first_val)
+        self.assertEqual(pipe.active_batch("daily", TRADE_DATE), first_daily)
+        transport.mode = "ok"
+        second = pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual(second["valuation"]["state"], "published")
+        self.assertNotEqual(second["valuation"]["batch_id"], first_val)
+        self.assertNotEqual(second["daily"]["batch_id"], first_daily)
+        pubs = db.fetchall(
+            "SELECT dataset, active_batch, prev_batch, published_at FROM publications WHERE trade_date=?",
+            (TRADE_DATE,),
+        )
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        a_times = {by_ds[name]["published_at"] for name in ("daily", "valuation", "moneyflow", "auction", "stocks")}
+        self.assertEqual(len(a_times), 1)
+        self.assertEqual(by_ds["valuation"]["active_batch"], second["valuation"]["batch_id"])
+        self.assertEqual(by_ds["valuation"]["prev_batch"], first_val)
+        rolled = pipe.rollback("valuation", TRADE_DATE, actor="cli")
+        self.assertEqual(rolled["active_batch"], first_val)
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_stocks_refresh.py

@@ -0,0 +1,168 @@
+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.scheduler import Scheduler
+from datahub.serving import V1API
+from datahub.settings import Settings
+from tests.fixtures import TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+
+
+class StockMasterTransport:
+    """fake_transport with a mutable stock_basic list (new listings / renames)."""
+
+    def __init__(self) -> None:
+        self.stocks = [
+            {"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"},
+        ]
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        if api_name == "stock_basic":
+            return [dict(row) for row in self.stocks]
+        return fake_transport(api_name, params, fields)
+
+    def rename_and_add(self) -> None:
+        for row in self.stocks:
+            if row["ts_code"] == "920071.BJ":
+                row["name"] = "金钛股份"  # N-prefix removed the day after listing
+        self.stocks.append(
+            {"ts_code": "920289.BJ", "symbol": "920289", "name": "N华汇", "area": "广东", "industry": "专用机械", "market": "北交所", "list_status": "L", "list_date": "20240902"}
+        )
+
+
+def make_pipe(transport):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    adapter = TushareAdapter("test-token", transport=transport)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        admin_password="admin-pass",
+        tushare_token="test-token",
+        db_path=db.path,
+        quality={"max_publish_attempts": 3, "publication_generations": 3},
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    pipe._tmp = tmp
+    return pipe, db
+
+
+class StocksRefreshTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = StockMasterTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+
+    def _stocks_api(self) -> dict:
+        return V1API(self.db, self.pipe, self.pipe.settings).handle("/v1/stocks", {})
+
+    def test_first_refresh_publishes_snapshot_with_meta(self) -> None:
+        result = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(result["state"], "published")
+        self.assertEqual(result["rows"], 2)
+        self.assertTrue(result["batch_id"].startswith("20240902-stocks-"))
+        payload = self._stocks_api()
+        self.assertEqual(payload["meta"]["batch_id"], result["batch_id"])
+        self.assertIsNotNone(payload["meta"]["published_at"])
+        self.assertEqual(len(payload["data"]), 2)
+        names = {row["ts_code"]: row["name"] for row in payload["data"]}
+        self.assertEqual(names["920071.BJ"], "N金钛")
+        self.assertNotIn("batch_id", payload["data"][0])
+
+    def test_new_listing_and_rename_publish_new_batch(self) -> None:
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.transport.rename_and_add()
+        second = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(second["state"], "published")
+        self.assertNotEqual(second["batch_id"], first["batch_id"])
+        payload = self._stocks_api()
+        names = {row["ts_code"]: row["name"] for row in payload["data"]}
+        self.assertEqual(names["920071.BJ"], "金钛股份")
+        self.assertIn("920289.BJ", names)
+        self.assertEqual(names["920289.BJ"], "N华汇")
+        # stock_master is refreshed too (code resolution stays current)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code='920289.BJ'")
+        self.assertEqual(master["name"], "N华汇")
+
+    def test_unchanged_refresh_is_idempotent(self) -> None:
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        again = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(again["state"], "skipped")
+        self.assertEqual(again["reason"], "unchanged")
+        self.assertEqual(again["batch_id"], first["batch_id"])
+        count = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM batches WHERE dataset='stocks' AND trade_date=?",
+            (TRADE_DATE,),
+        )["n"]
+        self.assertEqual(count, 1)
+
+    def test_force_republishes_even_unchanged(self) -> None:
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        forced = self.pipe.refresh_stocks(TRADE_DATE, force=True)
+        self.assertEqual(forced["state"], "published")
+        self.assertNotEqual(forced["batch_id"], first["batch_id"])
+
+    def test_snapshot_pinned_until_next_publish(self) -> None:
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.transport.rename_and_add()
+        # upstream changed but no refresh ran: published snapshot is untouched
+        _, snapshot = self.pipe.published_stock_snapshot(TRADE_DATE)
+        names = {row["ts_code"]: row["name"] for row in snapshot}
+        self.assertEqual(names["920071.BJ"], "N金钛")
+        self.assertNotIn("920289.BJ", names)
+        self.assertEqual(len(snapshot), 2)
+
+    def test_dataset_status_includes_stocks(self) -> None:
+        result = self.pipe.refresh_stocks(TRADE_DATE)
+        payload = V1API(self.db, self.pipe, self.pipe.settings).handle(
+            "/v1/datasets/status", {"date": [TRADE_DATE]}
+        )
+        by_name = {item["dataset"]: item for item in payload["data"]}
+        self.assertIn("stocks", by_name)
+        self.assertEqual(by_name["stocks"]["batch_id"], result["batch_id"])
+        self.assertIsNotNone(by_name["stocks"]["published_at"])
+
+
+class StocksRefreshSchedulingTests(unittest.TestCase):
+    def _make(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        db = HubDB(Path(tmp.name) / "hub.db")
+        transport = StockMasterTransport()
+        adapter = TushareAdapter("x", transport=transport)
+        settings = Settings(
+            encryption_key=SecretVault.generate_key(),
+            db_path=db.path,
+            backup_dir=Path(tmp.name) / "backups",
+            quality={"stocks_refresh_times": ["20:00", "23:10"]},
+        )
+        pipe = Pipeline(db, adapter, settings)
+        pipe.ingest_reference(TRADE_DATE)
+        return db, pipe, Scheduler(db, pipe)
+
+    def test_scheduled_refresh_runs_on_open_day(self) -> None:
+        db, pipe, sched = self._make()
+        ran = sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertIn("stocks_refresh", ran)
+        ran = sched.tick(clock_at(TRADE_DATE, 23, 10))
+        self.assertIn("stocks_refresh", ran)  # second slot catches late renames
+        self.assertIsNotNone(pipe.active_batch("stocks", TRADE_DATE))
+
+    def test_no_refresh_on_closed_day(self) -> None:
+        db, _pipe, sched = self._make()
+        sched.tick(clock_at("20240907", 20, 30))  # fixture: Saturday closed
+        runs = db.fetchall("SELECT * FROM job_runs WHERE job_id='stocks_refresh'")
+        self.assertEqual(runs, [])
+
+
+if __name__ == "__main__":
+    unittest.main()