fix(HEL-490): 真实装配接通中枢并收编估值晚间复核

把 query/行情钩子绑到内层 TushareClient，图表接受不完整日K窗口；收编现网 HEL-423 未提交的估值复核，避免换版丢掉。

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

.env.example

@@ -5,11 +5,10 @@ APP_ENCRYPTION_KEY=
 # the system settings; all accounts use the same backend market snapshot.
 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.
+# Official xiaobai-datahub client. Read flags default on in config/datahub.config.json.
+# compose.yaml pins every DATAHUB_READ_* to 1 so leftover .env zeros cannot keep
+# official pages on the old APIs. Old website APIs are emergency fallback only.
+# DATAHUB_SHADOW_* can still override a single dataset.
 DATAHUB_BASE_URL=http://127.0.0.1:8766
 DATAHUB_TOKEN=
 

backend/data/datahub/bridge.py

@@ -16,6 +16,7 @@ from backend.data.datahub.native import (
     yyyymmdd,
 )
 from backend.data.datahub.redact import redact_text, redact_value
+from backend.data.datahub.route_state import LEDGER
 from backend.data.datahub.settings import DatahubSettings
 from backend.data.providers.tushare_client import TushareClient
 
@@ -125,6 +126,7 @@ class DatahubBridge:
                 raise DatahubError("EMPTY", "datahub intraday empty")
             if (response.meta or {}).get("stale"):
                 raise DatahubError("STALE", "datahub intraday stale")
+            self._record_route("intraday", "datahub", str((response.meta or {}).get("source") or "datahub"))
             return {
                 "entity_type": str(data.get("entity_type") or "stock"),
                 "identifier": str(data.get("identifier") or code),
@@ -139,6 +141,113 @@ class DatahubBridge:
             self._log_failure("intraday", exc)
             return None
 
+    def try_market_quotes(self, trade_date: str = "") -> list[dict[str, Any]] | None:
+        return self._try_quote_rows("quotes", {}, expected_date=trade_date, minimum=200)
+
+    def try_quotes(self, codes: list[str]) -> list[dict[str, Any]] | None:
+        cleaned = [str(item or "").strip() for item in codes if str(item or "").strip()]
+        if not cleaned:
+            return None
+        return self._try_quote_rows("quotes", {"codes": ",".join(cleaned[:60])}, minimum=1)
+
+    def try_index_quotes(self) -> list[dict[str, Any]] | None:
+        flags = self.settings.flags("index_quotes")
+        if not flags.read:
+            return None
+        try:
+            response = self.client.index_quotes()
+            rows = [dict(item) for item in (response.data or []) if isinstance(item, dict)]
+            if len(rows) < 3:
+                raise DatahubError("EMPTY", "datahub index quotes incomplete")
+            if (response.meta or {}).get("stale"):
+                raise DatahubError("STALE", "datahub index quotes stale")
+            self._record_route(
+                "index_quotes",
+                "datahub",
+                str((response.meta or {}).get("source") or "datahub"),
+            )
+            return rows
+        except Exception as exc:
+            self._log_failure("index_quotes", exc)
+            return None
+
+    def try_daily_chart(
+        self,
+        code: str,
+        end_date: str,
+        limit: int = 90,
+        dataset: str = "daily",
+    ) -> list[dict[str, Any]] | None:
+        flags = self.settings.flags(dataset)
+        if not flags.read:
+            return None
+        compact_end = yyyymmdd(end_date)
+        if not compact_end:
+            return None
+        try:
+            start = _shift_yyyymmdd(compact_end, -max(190, int(limit) * 3))
+            if dataset == "index_daily":
+                response = self._paginate(
+                    self.client.index_bars,
+                    {"code": code, "from": start, "to": compact_end},
+                )
+            else:
+                response = self._paginate(
+                    self.client.daily_bars,
+                    {"code": code, "from": start, "to": compact_end, "adjust": "none"},
+                )
+            # Charts can use a partial history window; do not discard usable bars
+            # just because the requested lookback is not fully covered.
+            self._validate_usable(
+                dataset,
+                list(response.data or []),
+                response,
+                require_complete=False,
+            )
+            rows = _chart_bars(list(response.data or []))
+            if not rows:
+                raise DatahubError("EMPTY", f"{dataset} chart empty")
+            self._record_route(dataset, "datahub", str((response.meta or {}).get("source") or "datahub"))
+            return rows[-max(20, min(180, int(limit))):]
+        except Exception as exc:
+            self._log_failure(dataset, exc)
+            return None
+
+    def record_legacy(self, dataset: str, source: str = "", error: str = "") -> None:
+        self._record_route(dataset, "legacy", source, error)
+
+    def route_snapshot(self) -> list[dict[str, Any]]:
+        return LEDGER.snapshot()
+
+    def _try_quote_rows(
+        self,
+        dataset: str,
+        params: dict[str, Any],
+        expected_date: str = "",
+        minimum: int = 1,
+    ) -> list[dict[str, Any]] | None:
+        flags = self.settings.flags(dataset)
+        if not flags.read:
+            return None
+        try:
+            response = self.client.quotes_latest(**params)
+            rows = [_native_quote(item) for item in (response.data or []) if isinstance(item, dict)]
+            rows = [item for item in rows if item]
+            want = yyyymmdd(expected_date)
+            if want:
+                dated = [item for item in rows if not item.get("quote_date") or item.get("quote_date") == want]
+                if dated:
+                    rows = dated
+            if len(rows) < minimum:
+                raise DatahubError("EMPTY", f"datahub {dataset} empty")
+            if (response.meta or {}).get("stale"):
+                raise DatahubError("STALE", f"datahub {dataset} stale")
+            self._record_route(dataset, "datahub", str((response.meta or {}).get("source") or "datahub"))
+            return rows
+        except Exception as exc:
+            self._log_failure(dataset, exc)
+            return None
+
     def query(
         self,
         api_name: str,
@@ -180,12 +289,19 @@ class DatahubBridge:
                 raise
             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:
+                self._record_route(dataset, "datahub", str(hub_meta.get("source") or "datahub"))
                 return project_fields(hub_rows, fields)
+            if flags.read:
+                self._record_route(dataset, "legacy", "tushare", hub_error or "")
             return legacy_rows
 
         if flags.read and hub_rows is not None and hub_error is None:
+            self._record_route(dataset, "datahub", str(hub_meta.get("source") or "datahub"))
             return project_fields(hub_rows, fields)
-        return legacy_query(api_name, params, fields)
+        result = legacy_query(api_name, params, fields)
+        if flags.read:
+            self._record_route(dataset, "legacy", "tushare", hub_error or "")
+        return result
 
     def _fetch_dataset(self, dataset: str, params: dict[str, Any], api_name: str = "") -> DatahubResponse:
         date = yyyymmdd(params.get("trade_date") or params.get("date"))
@@ -275,7 +391,13 @@ class DatahubBridge:
             return filter_stock_rows(rows, params)
         return rows
 
-    def _validate_usable(self, dataset: str, rows: list[dict[str, Any]], response: DatahubResponse) -> None:
+    def _validate_usable(
+        self,
+        dataset: str,
+        rows: list[dict[str, Any]],
+        response: DatahubResponse,
+        require_complete: bool = True,
+    ) -> None:
         meta = response.meta or {}
         stale_seconds = int(meta.get("staleness_seconds") or 0)
         if meta.get("stale") or stale_seconds > self.settings.stale_seconds_max:
@@ -283,7 +405,7 @@ class DatahubBridge:
         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:
+        if require_complete and (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}")
 
@@ -298,11 +420,12 @@ class DatahubBridge:
             self.shadow_sink(report)
 
     def _log_failure(self, dataset: str, exc: Exception) -> None:
-        LOGGER.warning(
-            "datahub fallback dataset=%s error=%s",
-            dataset,
-            redact_text(self._error_text(exc), self.settings.secrets()),
-        )
+        error = redact_text(self._error_text(exc), self.settings.secrets())
+        LOGGER.warning("datahub fallback dataset=%s error=%s", dataset, error)
+        self._record_route(dataset, "legacy", "pending-legacy", error)
+
+    def _record_route(self, dataset: str, route: str, source: str = "", error: str = "") -> None:
+        LEDGER.record(dataset, route, source, redact_text(error, self.settings.secrets()))
 
     def _error_text(self, exc: Exception) -> str:
         if isinstance(exc, DatahubError):
@@ -312,10 +435,88 @@ class DatahubBridge:
         return redact_text(text, self.settings.secrets())
 
 
+def _native_quote(row: dict[str, Any]) -> dict[str, Any] | None:
+    ts_code = str(row.get("ts_code") or "").strip()
+    close = _finite(row.get("close") if row.get("close") not in (None, "") else row.get("price"))
+    previous = _finite(
+        row.get("pre_close") if row.get("pre_close") not in (None, "") else row.get("previous_close")
+    )
+    if not ts_code or close <= 0 or previous <= 0:
+        return None
+    volume = _finite(row.get("vol") if row.get("vol") not in (None, "") else row.get("volume"))
+    return {
+        "ts_code": ts_code,
+        "name": str(row.get("name") or ts_code).strip(),
+        "pre_close": previous,
+        "open": _finite(row.get("open")),
+        "high": _finite(row.get("high")),
+        "low": _finite(row.get("low")),
+        "close": close,
+        "vol": volume,
+        "amount": _finite(row.get("amount")),
+        "num": 0,
+        "quote_date": yyyymmdd(row.get("quote_date") or row.get("trade_date")),
+        "source": str(row.get("source") or "datahub"),
+    }
+
+
+def _chart_bars(rows: list[Any]) -> list[dict[str, Any]]:
+    normalized: list[dict[str, Any]] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        compact = yyyymmdd(row.get("trade_date"))
+        close = _finite(row.get("close"))
+        if len(compact) != 8 or close <= 0:
+            continue
+        volume = _finite(row.get("volume") if row.get("volume") not in (None, "") else row.get("vol"))
+        amount = _finite(row.get("amount"))
+        if volume and volume < close * 10 and amount > 1000:
+            volume = volume * 100
+        trade_date = f"{compact[:4]}-{compact[4:6]}-{compact[6:8]}"
+        previous = normalized[-1]["close"] if normalized else 0.0
+        normalized.append(
+            {
+                "trade_date": trade_date,
+                "open": _finite(row.get("open")),
+                "high": _finite(row.get("high")),
+                "low": _finite(row.get("low")),
+                "close": close,
+                "change": round((close / previous - 1) * 100, 4) if previous else _finite(row.get("pct_chg")),
+                "volume": volume,
+                "amount_billion": amount / 100_000_000,
+            }
+        )
+    return normalized
+
+
+def _shift_yyyymmdd(value: str, days: int) -> str:
+    from datetime import datetime, timedelta
+
+    stamp = datetime.strptime(value, "%Y%m%d")
+    return (stamp + timedelta(days=days)).strftime("%Y%m%d")
+
+
+def _finite(value: Any) -> float:
+    try:
+        return float(value or 0)
+    except (TypeError, ValueError):
+        return 0.0
+
+
 class DatahubAwareTushareClient:
     def __init__(self, legacy: TushareClient, bridge: DatahubBridge) -> None:
         self._legacy = legacy
         self._bridge = bridge
+        # Mixins run as methods on the inner instance (dashboard / indices /
+        # getattr). Bind hub hooks and query onto that instance so real
+        # assembly cannot skip 8766.
+        self._legacy_query = legacy.query
+        legacy.query = self.query
+        legacy.try_market_quotes = self.try_market_quotes
+        legacy.try_quotes = self.try_quotes
+        legacy.try_index_quotes = self.try_index_quotes
+        legacy.record_datahub_legacy = self.record_datahub_legacy
 
     def query(
         self,
@@ -323,7 +524,19 @@ class DatahubAwareTushareClient:
         params: dict[str, Any] | None = None,
         fields: str = "",
     ) -> list[dict[str, Any]]:
-        return self._bridge.query(api_name, params, fields, self._legacy.query)
+        return self._bridge.query(api_name, params, fields, self._legacy_query)
+
+    def try_market_quotes(self, trade_date: str = "") -> list[dict[str, Any]] | None:
+        return self._bridge.try_market_quotes(trade_date)
+
+    def try_quotes(self, codes: list[str]) -> list[dict[str, Any]] | None:
+        return self._bridge.try_quotes(codes)
+
+    def try_index_quotes(self) -> list[dict[str, Any]] | None:
+        return self._bridge.try_index_quotes()
+
+    def record_datahub_legacy(self, dataset: str, source: str = "", error: str = "") -> None:
+        self._bridge.record_legacy(dataset, source, error)
 
     def __getattr__(self, name: str) -> Any:
         return getattr(self._legacy, name)

backend/data/datahub/route_state.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from datetime import datetime
+from threading import Lock
+from typing import Any
+
+from backend.data.datahub.settings import DATASETS
+
+DATASET_LABELS = {
+    "calendar": "交易日历",
+    "stocks": "股票主档",
+    "daily": "个股日K",
+    "index_daily": "指数日K",
+    "valuation": "估值",
+    "moneyflow": "资金流",
+    "auction": "竞价",
+    "limit_events": "涨停池",
+    "popularity": "人气榜",
+    "dragon_tiger": "龙虎榜",
+    "sector_daily": "题材板块",
+    "quotes": "全市场实时行情",
+    "index_quotes": "指数实时行情",
+    "intraday": "分时",
+    "status": "数据集状态",
+}
+
+
+class DatahubRouteLedger:
+    def __init__(self) -> None:
+        self._lock = Lock()
+        self._rows: dict[str, dict[str, Any]] = {}
+
+    def record(self, dataset: str, route: str, source: str = "", error: str = "") -> None:
+        name = str(dataset or "").strip() or "unknown"
+        with self._lock:
+            self._rows[name] = {
+                "dataset": name,
+                "label": DATASET_LABELS.get(name, name),
+                "route": "legacy" if route == "legacy" else "datahub",
+                "source": str(source or "").strip(),
+                "error": str(error or "").strip(),
+                "at": datetime.now().astimezone().isoformat(timespec="seconds"),
+            }
+
+    def snapshot(self) -> list[dict[str, Any]]:
+        with self._lock:
+            rows = [dict(item) for item in self._rows.values()]
+        order = {name: index for index, name in enumerate(DATASETS)}
+        rows.sort(key=lambda item: (order.get(str(item.get("dataset")), 99), str(item.get("dataset"))))
+        return rows
+
+    def clear(self) -> None:
+        with self._lock:
+            self._rows.clear()
+
+
+LEDGER = DatahubRouteLedger()

backend/data/gateway.py

@@ -47,6 +47,31 @@ class DataGateway:
     def batches(self, trade_date: str, dataset: str = "") -> list[dict[str, Any]] | None:
         return self.datahub.batches(trade_date, dataset)
 
+    def datahub_status(self) -> dict[str, Any]:
+        from backend.data.datahub.route_state import DATASET_LABELS, LEDGER
+        from backend.data.datahub.settings import DATASETS
+
+        settings = self.datahub.settings
+        flags = []
+        enabled = 0
+        for name in DATASETS:
+            read = bool(settings.flags(name).read)
+            if read:
+                enabled += 1
+            flags.append({"dataset": name, "label": DATASET_LABELS.get(name, name), "read": read})
+        routes = LEDGER.snapshot()
+        fallbacks = [item for item in routes if item.get("route") == "legacy"]
+        return {
+            "configured": bool(settings.token and settings.base_url),
+            "base_url": settings.base_url,
+            "enabled_reads": enabled,
+            "total_reads": len(DATASETS),
+            "flags": flags,
+            "routes": routes,
+            "fallback_count": len(fallbacks),
+            "fallback_labels": [str(item.get("label") or item.get("dataset")) for item in fallbacks],
+        }
+
     def assert_source(self, dataset_id: str, provider_id: str, usage: DataUsage) -> None:
         self.policy.assert_allowed(dataset_id, provider_id, usage)
 

backend/data/providers/tushare_dashboard.py

@@ -186,7 +186,12 @@ class DashboardMixin:
         previous_sectors = _build_sectors(previous_limits)
         now = self._now()
         market_status = _realtime_market_status(now.time().replace(tzinfo=None))
-        if quote_source == "eastmoney_clist":
+        if quote_source == "datahub":
+            notice = (
+                "盘中行情由数据中枢统一提供；涨停原因、封板时间和开板次数以盘后榜单校正为准。"
+            )
+            source_name = "datahub"
+        elif quote_source == "eastmoney_clist":
             notice = (
                 "盘中行情由东财免费实时快照计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。"
             )
@@ -241,10 +246,16 @@ class DashboardMixin:
         codes: str,
         trade_date: str,
     ) -> tuple[list[dict[str, Any]], str]:
+        hub = getattr(self, "try_market_quotes", None)
+        if callable(hub):
+            quotes = hub(trade_date)
+            if quotes:
+                return list(quotes), "datahub"
         rt_error = ""
         try:
             quotes = self.query("rt_k", {"ts_code": codes})
             if quotes:
+                self._mark_quote_legacy("tushare_rt_k", rt_error)
                 return list(quotes), "tushare_rt_k"
             rt_error = f"No realtime data returned for {trade_date}"
         except TushareError as exc:
@@ -259,8 +270,14 @@ class DashboardMixin:
             raise TushareError(
                 f"当天盘中实时行情不可用：rt_k={rt_error}；免费源=empty"
             )
+        self._mark_quote_legacy(quote_source, rt_error)
         return quotes, quote_source
 
+    def _mark_quote_legacy(self, source: str, error: str = "") -> None:
+        marker = getattr(self, "record_datahub_legacy", None)
+        if callable(marker):
+            marker("quotes", source, error)
+
     def _free_realtime_quotes(
         self,
         trade_date: str,
@@ -286,8 +303,18 @@ class DashboardMixin:
         return quotes, "tencent_qt"
 
     def _free_realtime_indices(self) -> list[dict[str, Any]]:
+        hub = getattr(self, "try_index_quotes", None)
+        if callable(hub):
+            rows = hub()
+            converted = [item for item in (_hub_index_quote(row) for row in rows or []) if item]
+            if converted:
+                return converted
         try:
-            return self._realtime_aggregator().eastmoney_indices()
+            rows = self._realtime_aggregator().eastmoney_indices()
+            marker = getattr(self, "record_datahub_legacy", None)
+            if callable(marker):
+                marker("index_quotes", "eastmoney_push2")
+            return rows
         except Exception:
             return []
 
@@ -692,6 +719,31 @@ def _build_yesterday_performance(
     return result
 
 
+def _hub_index_quote(row: dict[str, Any]) -> dict[str, Any] | None:
+    ts_code = str(row.get("ts_code") or "")
+    code = str(row.get("code") or ts_code.split(".")[0])
+    close = _number(row.get("price") if row.get("price") not in (None, "") else row.get("close"))
+    previous = _number(
+        row.get("previous_close") if row.get("previous_close") not in (None, "") else row.get("pre_close")
+    )
+    if close <= 0 or previous <= 0:
+        return None
+    amount = _number(row.get("amount"))
+    amount_billion = _number(row.get("amount_billion"))
+    if not amount_billion and amount:
+        amount_billion = round(amount / 100_000_000, 2)
+    return {
+        "code": code,
+        "name": str(row.get("name") or code),
+        "price": close,
+        "change": _number(row.get("pct_chg") if row.get("pct_chg") not in (None, "") else row.get("change")),
+        "previous_close": previous,
+        "amount_billion": amount_billion,
+        "quote_time": str(row.get("quote_time") or ""),
+        "source": "datahub",
+    }
+
+
 def _build_limit_performance(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
     result = []
     for level in sorted({int(row.get("prior_streak") or 1) for row in rows}, reverse=True):

backend/data/providers/tushare_indices.py

@@ -59,10 +59,85 @@ class IndexMixin:
         }
 
     def realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
+        hub = getattr(self, "try_index_quotes", None)
+        if callable(hub):
+            rows = hub()
+            if rows:
+                try:
+                    return self._hub_realtime_market_indices(requested_date, rows)
+                except TushareError:
+                    pass
         try:
-            return self._tushare_realtime_market_indices(requested_date)
+            payload = self._tushare_realtime_market_indices(requested_date)
+            marker = getattr(self, "record_datahub_legacy", None)
+            if callable(marker):
+                marker("index_quotes", "tushare_rt_idx_k")
+            return payload
         except TushareError:
-            return self._free_realtime_market_indices(requested_date)
+            payload = self._free_realtime_market_indices(requested_date)
+            marker = getattr(self, "record_datahub_legacy", None)
+            if callable(marker):
+                marker("index_quotes", str(payload.get("source") or "eastmoney_push2"))
+            return payload
+
+    def _hub_realtime_market_indices(
+        self,
+        requested_date: str,
+        rows: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        trade_date, _ = self.resolve_trade_context(requested_date)
+        index_names = {
+            "000001.SH": "上证指数",
+            "399001.SZ": "深证成指",
+            "399006.SZ": "创业板指",
+        }
+        by_code = {str(row.get("ts_code") or ""): row for row in rows}
+        by_symbol = {str(row.get("code") or ""): row for row in rows}
+        indices = []
+        for ts_code, name in index_names.items():
+            row = by_code.get(ts_code) or by_symbol.get(ts_code.split(".")[0])
+            if not row:
+                continue
+            close = _number(row.get("price") if row.get("price") not in (None, "") else row.get("close"))
+            previous_close = _number(
+                row.get("previous_close") if row.get("previous_close") not in (None, "") else row.get("pre_close")
+            )
+            if close <= 0 or previous_close <= 0:
+                continue
+            amount = _number(row.get("amount"))
+            amount_billion = _number(row.get("amount_billion"))
+            if not amount_billion and amount:
+                amount_billion = round(amount / 100_000_000, 2)
+            indices.append(
+                {
+                    "ts_code": ts_code,
+                    "name": str(row.get("name") or name).strip(),
+                    "trade_date": trade_date,
+                    "close": close,
+                    "pct_chg": round(
+                        _number(row.get("pct_chg")) or (close / previous_close - 1) * 100,
+                        3,
+                    ),
+                    "return_5d": 0,
+                    "amount_billion": amount_billion,
+                    "quote_time": str(row.get("quote_time") or ""),
+                    "source": "datahub",
+                }
+            )
+        if len(indices) != 3:
+            raise TushareError("Realtime index quotes are incomplete")
+        return {
+            "trade_date": trade_date,
+            "source": "datahub",
+            "realtime": True,
+            "precise": True,
+            "indices": indices,
+            "aggregate": {
+                "average_pct_chg": round(sum(item["pct_chg"] for item in indices) / len(indices), 3),
+                "average_return_5d": 0,
+                "average_return_20d": 0,
+            },
+        }
 
     def _tushare_realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
         trade_date, _ = self.resolve_trade_context(requested_date)

backend/data/realtime.py

@@ -19,6 +19,8 @@ class RealtimeAggregateError(RuntimeError):
 
 
 EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
+EASTMONEY_STOCK_URL = "https://push2.eastmoney.com/api/qt/stock/get"
+EASTMONEY_STOCK_FIELDS = "f43,f44,f45,f46,f47,f48,f57,f58,f60,f86,f168"
 EASTMONEY_SECTOR_URL = "https://push2.eastmoney.com/api/qt/clist/get"
 EASTMONEY_A_SHARE_BOARDS = (
     "m:0+t:6",
@@ -283,6 +285,42 @@ class WebRealtimeAggregator:
             )
         return self._filter_quotes_by_date(quotes, expected_date)
 
+    def tencent_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]:
+        symbol, _secid, ts_code = _a_share_identity(code)
+        raw, _cache_age = self._get_text(
+            f"{TENCENT_QUOTE_URL}{symbol}",
+            referer="https://gu.qq.com/",
+            encoding="gb18030",
+        )
+        quote = next(
+            (
+                item
+                for line in raw.splitlines()
+                if (item := _parse_tencent_stock_quote(line))
+            ),
+            None,
+        )
+        if not quote:
+            raise RealtimeAggregateError(f"Tencent stock quote unavailable for {ts_code}")
+        return _require_quote_date(quote, expected_date)
+
+    def eastmoney_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]:
+        _symbol, secid, ts_code = _a_share_identity(code)
+        payload = self._get_json(
+            EASTMONEY_STOCK_URL,
+            {
+                "secid": secid,
+                "invt": "2",
+                "fltt": "2",
+                "fields": EASTMONEY_STOCK_FIELDS,
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        quote = _normalize_eastmoney_stock_quote(payload.get("data") or {}, ts_code)
+        if not quote:
+            raise RealtimeAggregateError(f"Eastmoney stock quote unavailable for {ts_code}")
+        return _require_quote_date(quote, expected_date)
+
     def tencent_indices(self) -> list[dict[str, Any]]:
         raw, cache_age = self._get_text(
             TENCENT_INDEX_URL,
@@ -553,6 +591,55 @@ def _diff_rows(data: dict[str, Any]) -> list[dict[str, Any]]:
     return [row for row in diff if isinstance(row, dict)]
 
 
+def _a_share_identity(code: str) -> tuple[str, str, str]:
+    raw = str(code or "").strip().upper()
+    symbol = raw.split(".")[0]
+    if not symbol.isdigit() or len(symbol) != 6:
+        raise RealtimeAggregateError("Invalid stock code")
+    if raw.endswith(".SH") or symbol.startswith(("5", "6", "9")):
+        return f"sh{symbol}", f"1.{symbol}", f"{symbol}.SH"
+    if raw.endswith(".BJ") or symbol.startswith(("4", "8")):
+        return f"bj{symbol}", f"0.{symbol}", f"{symbol}.BJ"
+    return f"sz{symbol}", f"0.{symbol}", f"{symbol}.SZ"
+
+
+def _require_quote_date(quote: dict[str, Any], expected_date: str) -> dict[str, Any]:
+    want = str(expected_date or "").replace("-", "")
+    got = str(quote.get("quote_date") or "")
+    if want and got != want:
+        raise RealtimeAggregateError(f"quote date {got or 'empty'} is not {want}")
+    return quote
+
+
+def _normalize_eastmoney_stock_quote(
+    row: dict[str, Any], ts_code: str
+) -> dict[str, Any] | None:
+    close = _number(row.get("f43"))
+    previous_close = _number(row.get("f60"))
+    if close <= 0 or previous_close <= 0:
+        return None
+    epoch = int(_number(row.get("f86")))
+    quote_date = ""
+    if epoch > 0:
+        quote_date = datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d")
+    return {
+        "ts_code": ts_code,
+        "name": row.get("f58") or ts_code.split(".")[0],
+        "pre_close": previous_close,
+        "open": _number(row.get("f46")),
+        "high": _number(row.get("f44")),
+        "low": _number(row.get("f45")),
+        "close": close,
+        "vol": _number(row.get("f47")) * 100,
+        "amount": _number(row.get("f48")),
+        "num": 0,
+        "quote_date": quote_date,
+        "quote_time_epoch": epoch,
+        "turnover_rate": _number(row.get("f168")),
+        "source": "eastmoney_stock",
+    }
+
+
 def _parse_tencent_stock_quote(line: str) -> dict[str, Any] | None:
     if '="' not in line:
         return None

backend/features/market/charts.py

@@ -68,12 +68,18 @@ class MarketChartClient:
         normalized = str(code or "").strip()
         if not re.fullmatch(r"\d{6}", normalized):
             raise ChartDataError("Invalid stock code")
+        hub_rows = self._datahub_daily(normalized, end_date, limit, "daily")
+        if hub_rows:
+            return hub_rows
         return self._ifind_daily(_stock_market_code(normalized), end_date, limit)
 
     def index_daily(self, identifier: str, end_date: str, limit: int = 90) -> list[dict[str, Any]]:
         normalized = str(identifier or "").strip().upper()
         if normalized not in INDEX_SECIDS:
             raise ChartDataError("Unsupported index")
+        hub_rows = self._datahub_daily(normalized, end_date, limit, "index_daily")
+        if hub_rows:
+            return hub_rows
         return self._ifind_daily(normalized, end_date, limit)
 
     def board_daily(self, identifier: str, end_date: str, limit: int = 90) -> list[dict[str, Any]]:
@@ -109,6 +115,112 @@ class MarketChartClient:
             return None
         return chart
 
+    def _datahub_daily(
+        self,
+        code: str,
+        end_date: str,
+        limit: int,
+        dataset: str,
+    ) -> list[dict[str, Any]] | None:
+        if self.datahub is None or not hasattr(self.datahub, "try_daily_chart"):
+            return None
+        try:
+            rows = self.datahub.try_daily_chart(code, end_date, limit, dataset)
+        except Exception as exc:
+            LOGGER.warning("datahub daily unexpected error: %s", exc)
+            rows = None
+        if not rows:
+            if hasattr(self.datahub, "record_legacy"):
+                self.datahub.record_legacy(dataset, "ifind")
+            return None
+        compact_end = str(end_date or "").replace("-", "")
+        market_now = datetime.now().astimezone()
+        today = market_now.strftime("%Y%m%d")
+        market_open = (
+            market_now.weekday() < 5
+            and market_now.time().replace(tzinfo=None) >= dt_time(9, 30)
+        )
+        if compact_end == today and market_open:
+            overlay = self._datahub_today_bar(code, dataset, rows)
+            if overlay:
+                if rows and rows[-1]["trade_date"] == overlay["trade_date"]:
+                    rows[-1] = overlay
+                else:
+                    rows.append(overlay)
+        return rows
+
+    def _datahub_today_bar(
+        self,
+        code: str,
+        dataset: str,
+        history: list[dict[str, Any]],
+    ) -> dict[str, Any] | None:
+        today_display = datetime.now().astimezone().date().isoformat()
+        previous = history[-1]["close"] if history and history[-1]["trade_date"] != today_display else (
+            history[-2]["close"] if len(history) >= 2 else 0.0
+        )
+        quote = None
+        if dataset == "index_daily" and hasattr(self.datahub, "try_index_quotes"):
+            quotes = self.datahub.try_index_quotes() or []
+            quote = next(
+                (
+                    item for item in quotes
+                    if str(item.get("ts_code") or "") == code or str(item.get("code") or "") == code.split(".")[0]
+                ),
+                None,
+            )
+        elif hasattr(self.datahub, "try_quotes"):
+            quotes = self.datahub.try_quotes([code]) or []
+            quote = quotes[0] if quotes else None
+        if quote:
+            close = _number(quote.get("close") if quote.get("close") not in (None, "") else quote.get("price"))
+            open_price = _number(quote.get("open"))
+            high = _number(quote.get("high"))
+            low = _number(quote.get("low"))
+            previous_close = _number(
+                quote.get("pre_close") if quote.get("pre_close") not in (None, "") else quote.get("previous_close")
+            ) or previous
+            volume = _number(quote.get("vol") if quote.get("vol") not in (None, "") else quote.get("volume"))
+            amount = _number(quote.get("amount"))
+            if close > 0 and open_price > 0:
+                return {
+                    "trade_date": today_display,
+                    "open": open_price,
+                    "high": high or close,
+                    "low": low or close,
+                    "close": close,
+                    "change": round((close / previous_close - 1) * 100, 4) if previous_close else 0.0,
+                    "volume": volume,
+                    "amount_billion": amount / 100_000_000,
+                    "realtime": True,
+                }
+        chart = self._datahub_intraday(code)
+        points = list((chart or {}).get("points") or [])
+        if not points:
+            return None
+        closes = [_number(point.get("close")) for point in points if _number(point.get("close")) > 0]
+        if not closes:
+            return None
+        opens = [_number(point.get("open")) for point in points if _number(point.get("open")) > 0]
+        highs = [_number(point.get("high")) for point in points if _number(point.get("high")) > 0]
+        lows = [_number(point.get("low")) for point in points if _number(point.get("low")) > 0]
+        volume = sum(_number(point.get("volume")) for point in points)
+        amount = sum(_number(point.get("amount")) for point in points)
+        previous_close = _number((chart or {}).get("previous_close")) or previous
+        close = closes[-1]
+        open_price = opens[0] if opens else closes[0]
+        return {
+            "trade_date": today_display,
+            "open": open_price,
+            "high": max(highs or closes),
+            "low": min(lows or closes),
+            "close": close,
+            "change": round((close / previous_close - 1) * 100, 4) if previous_close else 0.0,
+            "volume": volume,
+            "amount_billion": amount / 100_000_000,
+            "realtime": True,
+        }
+
     def board_intraday(self, identifier: str, name: str = "") -> dict[str, Any]:
         normalized = str(identifier or "").strip().upper()
         try:

backend/features/market/service.py

@@ -15,6 +15,7 @@ from backend.bootstrap.config import (
 )
 from backend.data.providers.ifind_client import IfindError
 from backend.data.providers.tushare_client import TushareClient, TushareError
+from backend.data.realtime import RealtimeAggregateError
 from backend.features.market.backfill_history import (
     DEFAULT_RECENT_TRADING_DAYS,
     MAX_RANGE_TRADING_DAYS,
@@ -42,6 +43,7 @@ SEARCH_TYPE_LABELS = {
     "theme": "题材",
     "index": "指数",
 }
+TODAY_DAILY_UNAVAILABLE_NOTICE = "今日日K暂不可用，仍显示最近收盘K线。"
 THS_SEARCH_TYPES = {
     "I": ("sector", "行业板块"),
     "R": ("sector", "地域板块"),
@@ -814,26 +816,27 @@ class MarketServiceMixin:
                 "trade_date": f"{actual_date[:4]}-{actual_date[4:6]}-{actual_date[6:]}",
             }
         today = now.strftime("%Y%m%d")
+        latest_bar = (result.get("prices") or [{}])[-1] if result.get("prices") else {}
+        official_today = (
+            actual_date == today and not bool(latest_bar.get("realtime"))
+        )
+        after_close = now.time().replace(tzinfo=None) >= dt_time(15, 0)
         should_merge = (
             requested_date == today
             and actual_date <= today
             and now.weekday() < 5
             and now.time().replace(tzinfo=None) >= dt_time(9, 30)
+            and not (official_today and after_close)
         )
         if should_merge:
-            quote = self._ifind_realtime_stock_quote(code)
+            quote = self._resolve_today_daily_quote(code, today, result)
             if quote and self._valid_realtime_stock_quote(quote, today):
                 self._merge_realtime_stock_detail(result, quote, requested_date)
-            elif self.configured and actual_date < today:
-                client = self._tushare_client()
-                try:
-                    resolved_date, _ = client.resolve_trade_context(requested_date)
-                    if resolved_date == today:
-                        quote = client.realtime_stock_quote(tushare_code(code), requested_date)
-                        if self._valid_realtime_stock_quote(quote, today):
-                            self._merge_realtime_stock_detail(result, quote, requested_date)
-                except TushareError:
-                    pass
+            elif actual_date < today:
+                result["meta"] = {
+                    **(result.get("meta") or {}),
+                    "notice": TODAY_DAILY_UNAVAILABLE_NOTICE,
+                }
         return self._enrich_stock_detail(result)
 
     @staticmethod
@@ -948,6 +951,134 @@ class MarketServiceMixin:
             "quote_time": str(row.get("time") or ""),
         }
 
+    def _resolve_today_daily_quote(
+        self, code: str, today: str, payload: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        quote = self._ifind_realtime_stock_quote(code)
+        if quote and self._valid_realtime_stock_quote(quote, today):
+            return quote
+        if self.configured:
+            try:
+                client = self._tushare_client()
+                resolve = getattr(client, "resolve_trade_context", None)
+                resolved = today
+                if callable(resolve):
+                    resolved, _ = resolve(today)
+                if str(resolved or "") == today:
+                    quote = client.realtime_stock_quote(tushare_code(code), today)
+                    if self._valid_realtime_stock_quote(quote, today):
+                        return quote
+            except TushareError:
+                pass
+        quote = self._free_realtime_stock_quote(code, today)
+        if quote and self._valid_realtime_stock_quote(quote, today):
+            return quote
+        return self._intraday_realtime_stock_quote(code, today, payload)
+
+    def _free_realtime_stock_quote(self, code: str, today: str) -> dict[str, Any] | None:
+        aggregator = getattr(self, "realtime_aggregator", None)
+        if aggregator is None:
+            return None
+        ts_code = tushare_code(code)
+        for loader in (
+            getattr(aggregator, "tencent_stock_quote", None),
+            getattr(aggregator, "eastmoney_stock_quote", None),
+        ):
+            if not callable(loader):
+                continue
+            try:
+                row = loader(ts_code, expected_date=today)
+            except (RealtimeAggregateError, Exception):
+                continue
+            quote = self._quote_from_free_row(code, today, row)
+            if quote:
+                return quote
+        return None
+
+    def _quote_from_free_row(
+        self, code: str, today: str, row: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        price = float(row.get("close") or 0)
+        previous_close = float(row.get("pre_close") or 0)
+        if price <= 0 or previous_close <= 0:
+            return None
+        try:
+            name, sector = self._stock_identity(code, today)
+        except Exception:
+            name, sector = "--", "其他"
+        epoch = int(row.get("quote_time_epoch") or 0)
+        if epoch > 0:
+            quote_time = datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds")
+        else:
+            quote_date = str(row.get("quote_date") or today)
+            quote_time = f"{quote_date[:4]}-{quote_date[4:6]}-{quote_date[6:]}"
+        return {
+            "name": str(row.get("name") or name or "--"),
+            "sector": sector,
+            "price": price,
+            "open": float(row.get("open") or 0),
+            "high": float(row.get("high") or 0),
+            "low": float(row.get("low") or 0),
+            "change": round((price / previous_close - 1) * 100, 4),
+            "volume": float(row.get("vol") or 0),
+            "amount_billion": float(row.get("amount") or 0) / 100_000_000,
+            "turnover_rate": float(row.get("turnover_rate") or 0),
+            "quote_time": quote_time,
+        }
+
+    def _intraday_realtime_stock_quote(
+        self, code: str, today: str, payload: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        chart_data = getattr(self, "chart_data", None)
+        if chart_data is None:
+            return None
+        try:
+            chart = chart_data.stock_intraday(code)
+        except (AttributeError, ChartDataError, Exception):
+            return None
+        points = [
+            point
+            for point in list(chart.get("points") or [])
+            if str(point.get("date") or "").replace("-", "") == today
+        ]
+        if not points:
+            return None
+        opens = [float(point.get("open") or 0) for point in points if float(point.get("open") or 0) > 0]
+        highs = [float(point.get("high") or 0) for point in points if float(point.get("high") or 0) > 0]
+        lows = [float(point.get("low") or 0) for point in points if float(point.get("low") or 0) > 0]
+        closes = [float(point.get("close") or 0) for point in points if float(point.get("close") or 0) > 0]
+        if not opens or not highs or not lows or not closes:
+            return None
+        price = closes[-1]
+        previous_close = float(chart.get("previous_close") or 0)
+        if previous_close <= 0:
+            history = list(payload.get("prices") or [])
+            previous_close = float((history[-1] if history else {}).get("close") or 0)
+        if previous_close <= 0:
+            return None
+        volume = sum(float(point.get("volume") or 0) for point in points)
+        amount = sum(float(point.get("amount") or 0) for point in points)
+        if volume <= 0 and amount <= 0:
+            return None
+        try:
+            name, sector = self._stock_identity(code, today)
+        except Exception:
+            name, sector = "--", "其他"
+        return {
+            "name": name,
+            "sector": sector,
+            "price": price,
+            "open": opens[0],
+            "high": max(highs),
+            "low": min(lows),
+            "change": round((price / previous_close - 1) * 100, 4),
+            "volume": volume,
+            "volume_unit": "lots",
+            "amount_billion": amount / 100_000_000,
+            "turnover_rate": 0.0,
+            "quote_time": str(points[-1].get("date") or today),
+        }
+
     @staticmethod
     def _merge_realtime_stock_detail(
         payload: dict[str, Any], quote: dict[str, Any], trade_date: str
@@ -986,6 +1117,7 @@ class MarketServiceMixin:
             **(payload.get("meta") or {}),
             "trade_date": display_date,
             "realtime": True,
+            "notice": "",
             "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
         }
 

backend/features/system/service.py

@@ -130,6 +130,7 @@ class SystemServiceMixin:
                 ),
                 **self.database.status(),
                 "jobs": self.jobs.repository.recent(12),
+                "datahub": self._datahub_status(),
             },
             "llm": {
                 "primary_configured": self._profile_configured(platform["primary"]),
@@ -145,6 +146,22 @@ class SystemServiceMixin:
             },
         }
 
+    def _datahub_status(self) -> dict[str, Any]:
+        gateway = getattr(self, "data_gateway", None)
+        reporter = getattr(gateway, "datahub_status", None)
+        if callable(reporter):
+            return reporter()
+        return {
+            "configured": False,
+            "base_url": "",
+            "enabled_reads": 0,
+            "total_reads": 0,
+            "flags": [],
+            "routes": [],
+            "fallback_count": 0,
+            "fallback_labels": [],
+        }
+
     def save_system_settings(self, payload: dict[str, Any]) -> dict[str, Any]:
         current = dict(self._system_credentials)
         token = str(payload.get("tushare_token") or current.get("tushare_token") or "").strip()

compose.yaml

@@ -13,6 +13,22 @@ services:
       - ./.env
     environment:
       APP_ENCRYPTION_KEY: "${APP_ENCRYPTION_KEY:?APP_ENCRYPTION_KEY must be set in .env}"
+      DATAHUB_BASE_URL: "${DATAHUB_BASE_URL:-http://192.168.200.11:8766}"
+      DATAHUB_READ_CALENDAR: "1"
+      DATAHUB_READ_STOCKS: "1"
+      DATAHUB_READ_DAILY: "1"
+      DATAHUB_READ_INDEX_DAILY: "1"
+      DATAHUB_READ_VALUATION: "1"
+      DATAHUB_READ_MONEYFLOW: "1"
+      DATAHUB_READ_AUCTION: "1"
+      DATAHUB_READ_LIMIT_EVENTS: "1"
+      DATAHUB_READ_POPULARITY: "1"
+      DATAHUB_READ_DRAGON_TIGER: "1"
+      DATAHUB_READ_SECTOR_DAILY: "1"
+      DATAHUB_READ_QUOTES: "1"
+      DATAHUB_READ_INDEX_QUOTES: "1"
+      DATAHUB_READ_INTRADAY: "1"
+      DATAHUB_READ_STATUS: "1"
       TZ: Asia/Shanghai
       PYTHONUTF8: "1"
     volumes:

config/README.md

@@ -12,9 +12,12 @@ These registries describe the approved product surface of the standalone applica
   providers, model entry points, CSS layers, and remaining code hotspots.
 - `data-fields.config.json`: canonical data products, provider eligibility, intended use, and
   known blocked datasets.
-- `datahub.config.json`: optional read-only client for `xiaobai-datahub`. Each dataset has its
-  own `read` / `shadow` flag, all default off. Environment variables `DATAHUB_READ_*` and
-  `DATAHUB_SHADOW_*` can override a single dataset without a master switch.
+- `datahub.config.json`: official read-only client for `xiaobai-datahub`. Each dataset has its
+  own `read` / `shadow` flag; official reads default on. `compose.yaml` pins every
+  `DATAHUB_READ_*` to `"1"` so a leftover `.env` `=0` cannot silently keep official
+  pages on the old APIs. Environment variables can still override a single
+  `DATAHUB_SHADOW_*` without a master switch. The old website APIs stay as
+  emergency fallback only.
 - `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and
   fail-closed rules for every canonical data product.
 - `jobs.config.json`: background schedules, dependencies, lock keys, retry policy, timeouts,

config/architecture-inventory.json

@@ -483,8 +483,8 @@
     },
     {
       "path": "frontend/index.html",
-      "bytes": 48254,
-      "lines": 664
+      "bytes": 48447,
+      "lines": 665
     },
     {
       "path": "backend/features/screener/catalog.py",
@@ -496,6 +496,11 @@
       "bytes": 35247,
       "lines": 2416
     },
+    {
+      "path": "backend/data/providers/tushare_dashboard.py",
+      "bytes": 33603,
+      "lines": 784
+    },
     {
       "path": "database.py",
       "bytes": 32073,
@@ -506,11 +511,6 @@
       "bytes": 31756,
       "lines": 562
     },
-    {
-      "path": "backend/data/providers/tushare_dashboard.py",
-      "bytes": 31361,
-      "lines": 732
-    },
     {
       "path": "backend/data/providers/tushare_industries.py",
       "bytes": 26540,
@@ -533,8 +533,8 @@
     },
     {
       "path": "frontend/pages/market/preview.js",
-      "bytes": 18178,
-      "lines": 446
+      "bytes": 18339,
+      "lines": 450
     },
     {
       "path": "backend/features/heaven/trend.py",
@@ -551,6 +551,11 @@
       "bytes": 15311,
       "lines": 387
     },
+    {
+      "path": "frontend/shared/admin.js",
+      "bytes": 15235,
+      "lines": 289
+    },
     {
       "path": "frontend/shared/dashboard.js",
       "bytes": 15063,
@@ -566,11 +571,6 @@
       "bytes": 14743,
       "lines": 342
     },
-    {
-      "path": "frontend/shared/admin.js",
-      "bytes": 14410,
-      "lines": 268
-    },
     {
       "path": "backend/features/heaven/market_context.py",
       "bytes": 13681,
@@ -581,15 +581,20 @@
       "bytes": 13219,
       "lines": 289
     },
+    {
+      "path": "backend/features/system/service.py",
+      "bytes": 12937,
+      "lines": 271
+    },
     {
       "path": "backend/features/market/insights_auction_data.py",
       "bytes": 12829,
       "lines": 318
     },
     {
-      "path": "backend/features/system/service.py",
-      "bytes": 12392,
-      "lines": 254
+      "path": "backend/data/providers/tushare_indices.py",
+      "bytes": 10956,
+      "lines": 248
     },
     {
       "path": "backend/features/market/insights_auction.py",
@@ -631,11 +636,6 @@
       "bytes": 8357,
       "lines": 116
     },
-    {
-      "path": "backend/data/providers/tushare_indices.py",
-      "bytes": 7823,
-      "lines": 173
-    },
     {
       "path": "backend/features/screener/formula.py",
       "bytes": 6983,
@@ -681,16 +681,16 @@
       "bytes": 6092,
       "lines": 138
     },
+    {
+      "path": "frontend/pages/market/stock-detail.js",
+      "bytes": 6041,
+      "lines": 134
+    },
     {
       "path": "frontend/pages/dragon-tiger/page.html",
       "bytes": 5754,
       "lines": 85
     },
-    {
-      "path": "frontend/pages/market/stock-detail.js",
-      "bytes": 5690,
-      "lines": 124
-    },
     {
       "path": "frontend/pages.config.js",
       "bytes": 5385,

config/datahub.config.json

@@ -6,20 +6,20 @@
   "page_limit": 5000,
   "stale_seconds_max": 86400,
   "datasets": {
-    "calendar": { "read": false, "shadow": false },
-    "stocks": { "read": false, "shadow": false },
-    "daily": { "read": false, "shadow": false },
-    "index_daily": { "read": false, "shadow": false },
-    "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 }
+    "calendar": { "read": true, "shadow": false },
+    "stocks": { "read": true, "shadow": false },
+    "daily": { "read": true, "shadow": false },
+    "index_daily": { "read": true, "shadow": false },
+    "valuation": { "read": true, "shadow": false },
+    "moneyflow": { "read": true, "shadow": false },
+    "auction": { "read": true, "shadow": false },
+    "limit_events": { "read": true, "shadow": false },
+    "popularity": { "read": true, "shadow": false },
+    "dragon_tiger": { "read": true, "shadow": false },
+    "sector_daily": { "read": true, "shadow": false },
+    "quotes": { "read": true, "shadow": false },
+    "index_quotes": { "read": true, "shadow": false },
+    "intraday": { "read": true, "shadow": false },
+    "status": { "read": true, "shadow": false }
   }
 }

frontend/index.html

@@ -611,6 +611,7 @@
         <label class="form-field"><span>iFinD Refresh Token</span><input id="systemIfindTokenInput" type="password" autocomplete="off" maxlength="2048" placeholder="留空保留现有 Token"></label>
         <label class="switch-control"><input id="systemBackgroundRefresh" type="checkbox"><span>启用交易时段后台刷新</span></label>
         <p class="form-hint">所有用户读取同一份后台快照，页面不会随后台任务自动重绘。</p>
+        <div id="datahubRouteStatus" class="admin-refresh-status" data-tone="idle" role="status" aria-live="polite"><i data-lucide="database"></i><span>数据中枢线路待检查</span></div>
         <div id="adminRefreshStatus" class="admin-refresh-status" data-tone="idle" role="status" aria-live="polite"><i data-lucide="circle-dot"></i><span>尚未手动刷新</span></div>
         <div class="dialog-actions admin-inline-actions"><button id="adminRefreshButton" class="button" type="button"><i data-lucide="refresh-cw"></i>立即后台刷新</button><button class="button primary" type="submit">保存行情配置</button></div>
       </form>

frontend/m/js/pages.js

@@ -5219,6 +5219,15 @@
     return '<span class="m-sys-dot' + (ok ? " m-sys-dot--ok" : "") + '"></span>';
   }
 
+  function datahubStatusText(hub) {
+    const enabled = number(hub.enabled_reads);
+    const total = number(hub.total_reads) || enabled;
+    const fallbacks = hub.fallback_labels || [];
+    if (fallbacks.length) return " 备用 " + fallbacks.join("、");
+    if (hub.configured) return " 主线路 " + enabled + "/" + total;
+    return " 未配置";
+  }
+
   function renderSystemAdmin(key) {
     if (key === "system/members") {
       renderSystemMembers();
@@ -5237,6 +5246,7 @@
       '<div class="m-sys-status-item"><span>iFinD</span><span>' + statusDot(ifind.configured) + (ifind.configured ? " 已配置" : " 未配置") + "</span></div>" +
       '<div class="m-sys-status-item"><span>行情快照</span><strong>' + number(data.snapshot_dates) + " 个交易日</strong></div>" +
       '<div class="m-sys-status-item"><span>后台刷新</span><span>' + statusDot(data.background_refresh_enabled) + (data.background_refresh_enabled ? " 已启用" : " 已暂停") + "</span></div>" +
+      '<div class="m-sys-status-item"><span>数据中枢</span><span>' + statusDot(Boolean((data.datahub || {}).configured) && !((data.datahub || {}).fallback_count)) + datahubStatusText(data.datahub || {}) + "</span></div>" +
       "</div></div>" +
       '<div class="m-card m-sys-section"><strong>数据源密钥</strong>' +
       formFieldHtml("Tushare Token", '<input id="m-sys-token" type="password" autocomplete="off" minlength="20" placeholder="留空则保留现有 Token">', false) +

frontend/pages/market/preview.js

@@ -367,7 +367,11 @@ function selectStockPreviewChart(chart) {
     }
   } else if ((payload.prices || []).length) {
     setText("stockPreviewDate", payload.meta?.trade_date || "最新行情");
-    setText("stockPreviewSource", `日 K 行情 · ${payload.prices.length} 个交易日`);
+    const notice = String(payload.meta?.notice || "").trim();
+    setText(
+      "stockPreviewSource",
+      notice ? `日 K 行情 · ${payload.prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${payload.prices.length} 个交易日`,
+    );
     drawDailyPreviewChart(payload.prices);
   } else {
     setText("stockPreviewDate", payload.meta?.trade_date || "最新行情");

frontend/pages/market/stock-detail.js

@@ -52,7 +52,11 @@ async function openStock(code, fallback = null) {
     renderStockNotes(payload.notes || []);
     updateWatchButton();
     if (state.stockDetailChartMode === "daily") {
-      setText("chartSource", `日 K 行情 · ${payload.prices.length} 个交易日`);
+      const notice = String(payload.meta?.notice || "").trim();
+      setText(
+        "chartSource",
+        notice ? `日 K 行情 · ${payload.prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${payload.prices.length} 个交易日`,
+      );
       requestAnimationFrame(() => drawPriceChart(payload.prices || []));
     }
   } catch (error) {
@@ -69,7 +73,13 @@ async function selectStockDetailChart(mode) {
   syncDetailChartButtons("stock", selected);
   if (selected === "daily") {
     const prices = state.stockDetail?.prices || [];
-    setText("chartSource", prices.length ? `日 K 行情 · ${prices.length} 个交易日` : "正在加载行情");
+    const notice = String(state.stockDetail?.meta?.notice || "").trim();
+    setText(
+      "chartSource",
+      prices.length
+        ? (notice ? `日 K 行情 · ${prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${prices.length} 个交易日`)
+        : "正在加载行情",
+    );
     if (prices.length) requestAnimationFrame(() => drawPriceChart(prices));
     else clearPriceChart("正在加载日 K 数据");
     return;

frontend/shared/admin.js

@@ -44,6 +44,7 @@ async function openAdminSettings(refreshOnly = false) {
     status.textContent = `Tushare ${data.configured ? "已配置" : "未配置"} · iFinD ${ifind.configured ? "已配置" : "未配置"} · ${number(data.snapshot_dates)} 个交易日`;
     status.classList.toggle("connected", Boolean(data.configured));
     setText("systemDataStatus", data.background_refresh_enabled ? "后台刷新已启用" : "后台刷新已暂停");
+    renderDatahubRouteStatus(data.datahub || {});
     document.querySelector("#systemTokenInput").value = "";
     document.querySelector("#systemIfindTokenInput").value = "";
     document.querySelector("#systemBackgroundRefresh").checked = Boolean(data.background_refresh_enabled);
@@ -55,6 +56,26 @@ async function openAdminSettings(refreshOnly = false) {
   }
 }
 
+function renderDatahubRouteStatus(hub) {
+  const box = document.querySelector("#datahubRouteStatus");
+  if (!box) return;
+  const label = box.querySelector("span");
+  const enabled = number(hub.enabled_reads);
+  const total = number(hub.total_reads) || enabled;
+  const fallbacks = hub.fallback_labels || [];
+  if (fallbacks.length) {
+    box.dataset.tone = "warning";
+    if (label) label.textContent = `数据中枢主线路 ${enabled}/${total} · 备用 ${fallbacks.length} 类：${fallbacks.join("、")}`;
+    return;
+  }
+  box.dataset.tone = hub.configured ? "success" : "idle";
+  if (label) {
+    label.textContent = hub.configured
+      ? `数据中枢主线路 ${enabled}/${total}，当前无备用`
+      : "数据中枢尚未配置，网站仍走原接口";
+  }
+}
+
 function selectAdminPanel(panel) {
   const selected = ["market", "models", "members"].includes(panel) ? panel : "market";
   document.querySelector("#adminSectionSelect").value = selected;

tests/test_chart_data_provider.py

@@ -155,10 +155,12 @@ class ChartLookbackTests(unittest.TestCase):
 
 
 class FakeHub:
-    def __init__(self, chart=None, error=None):
+    def __init__(self, chart=None, error=None, daily=None):
         self.chart = chart
         self.error = error
+        self.daily = daily
         self.calls: list[str] = []
+        self.legacy: list[str] = []
 
     def try_intraday(self, code):
         self.calls.append(code)
@@ -166,6 +168,15 @@ class FakeHub:
             raise self.error
         return self.chart
 
+    def try_daily_chart(self, code, end_date, limit=90, dataset="daily"):
+        self.calls.append(f"{dataset}:{code}")
+        if self.error:
+            raise self.error
+        return self.daily
+
+    def record_legacy(self, dataset, source="", error=""):
+        self.legacy.append(dataset)
+
 
 class DatahubChartFallbackTests(unittest.TestCase):
     def setUp(self) -> None:
@@ -207,6 +218,25 @@ class DatahubChartFallbackTests(unittest.TestCase):
             self.assertGreaterEqual(len(payload["points"]), 1)
             self.assertTrue(fallback.requests)
 
+    def test_datahub_daily_skips_ifind(self):
+        hub = FakeHub(
+            daily=[
+                {
+                    "trade_date": "2026-09-07",
+                    "open": 10.0,
+                    "high": 10.4,
+                    "low": 9.9,
+                    "close": 10.2,
+                    "volume": 1000,
+                    "amount_billion": 0.02,
+                }
+            ]
+        )
+        client = MarketChartClient(IfindHttpClient(), LookbackChartClient(), hub)
+        rows = client.stock_daily("600000", "20260907")
+        self.assertEqual(rows[-1]["trade_date"], "2026-09-07")
+        self.assertIn("daily:600000", hub.calls)
+
 
 class ChartServiceStub:
     @staticmethod

tests/test_datahub_bridge.py

@@ -12,6 +12,7 @@ from backend.data.datahub.client import DatahubClient, DatahubResponse
 from backend.data.datahub.compare import compare_rows
 from backend.data.datahub.errors import DatahubError
 from backend.data.datahub.native import to_canonical_row, to_native_row
+from backend.data.datahub.route_state import LEDGER
 from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -84,17 +85,21 @@ def flags(**enabled: tuple[bool, bool]) -> DatahubSettings:
 
 
 class DatahubBridgeTests(unittest.TestCase):
-    def test_default_config_keeps_legacy_and_does_not_call_datahub(self) -> None:
+    def setUp(self) -> None:
+        LEDGER.clear()
+
+    def test_default_config_enables_official_reads(self) -> None:
         settings = DatahubSettings.load(environ={}, credentials={})
-        self.assertFalse(settings.any_enabled())
-        self.assertTrue(all(not settings.flags(name).read and not settings.flags(name).shadow for name in DATASETS))
-        client = FakeClient(error=DatahubError("INTERNAL", "should not be called"))
+        self.assertTrue(settings.any_enabled())
+        self.assertTrue(all(settings.flags(name).read and not settings.flags(name).shadow for name in DATASETS))
+        client = FakeClient()
         legacy = FakeLegacy([LEGACY_DAILY])
         wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(settings, client))
         rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,close,vol,amount")
         self.assertEqual(rows[0]["amount"], 2000.0)
-        self.assertEqual(client.paths, [])
-        self.assertEqual(len(legacy.calls), 1)
+        self.assertEqual(client.paths, ["/v1/bars/daily"])
+        self.assertEqual(legacy.calls, [])
+        self.assertEqual(LEDGER.snapshot()[0]["route"], "datahub")
 
     def test_each_dataset_has_independent_read_flag(self) -> None:
         settings = flags(daily=(True, False), auction=(False, False))
@@ -104,6 +109,13 @@ class DatahubBridgeTests(unittest.TestCase):
         source = (ROOT / "config" / "datahub.config.json").read_text(encoding="utf-8")
         self.assertNotIn("master", source)
         self.assertNotIn("DATAHUB_READ_ALL", source)
+        compose = (ROOT / "compose.yaml").read_text(encoding="utf-8")
+        for env_key in (
+            "CALENDAR", "STOCKS", "DAILY", "INDEX_DAILY", "VALUATION", "MONEYFLOW",
+            "AUCTION", "LIMIT_EVENTS", "POPULARITY", "DRAGON_TIGER", "SECTOR_DAILY",
+            "QUOTES", "INDEX_QUOTES", "INTRADAY", "STATUS",
+        ):
+            self.assertIn(f'DATAHUB_READ_{env_key}: "1"', compose)
 
     def test_read_flag_replaces_only_that_dataset_and_converts_units(self) -> None:
         shadows: list[dict[str, Any]] = []
@@ -412,7 +424,181 @@ class DatahubBridgeTests(unittest.TestCase):
             FakeClient(error=DatahubError("INTERNAL", "datahub exploded")),
         )
         self.assertIsNone(broken.try_intraday("601318"))
-        self.assertFalse(DatahubSettings.load(environ={}, credentials={}).flags("intraday").read)
+        self.assertTrue(DatahubSettings.load(environ={}, credentials={}).flags("intraday").read)
+
+    def test_try_market_quotes_and_visible_fallback(self) -> None:
+        quotes = [
+            {
+                "ts_code": f"{600000 + index:06d}.SH",
+                "name": f"股票{index}",
+                "close": 10.2,
+                "pre_close": 10.0,
+                "open": 10.1,
+                "high": 10.3,
+                "low": 9.9,
+                "vol": 1000,
+                "amount": 2000000,
+                "quote_date": "20240902",
+            }
+            for index in range(220)
+        ]
+        ok = DatahubBridge(
+            flags(quotes=(True, False)),
+            FakeClient(
+                response=DatahubResponse(
+                    data=quotes,
+                    meta={"stale": False, "staleness_seconds": 0, "source": "eastmoney:clist"},
+                )
+            ),
+        )
+        rows = ok.try_market_quotes("20240902")
+        self.assertEqual(len(rows), 220)
+        self.assertEqual(rows[0]["pre_close"], 10.0)
+        self.assertEqual(ok.client.paths, ["/v1/quotes/latest"])
+        self.assertEqual(LEDGER.snapshot()[0]["route"], "datahub")
+
+        failed = DatahubBridge(
+            flags(quotes=(True, False)),
+            FakeClient(error=DatahubError("UNAVAILABLE", "down")),
+        )
+        self.assertIsNone(failed.try_market_quotes("20240902"))
+        failed.record_legacy("quotes", "tencent_qt", "down")
+        snap = next(item for item in LEDGER.snapshot() if item["dataset"] == "quotes")
+        self.assertEqual(snap["route"], "legacy")
+        self.assertEqual(snap["source"], "tencent_qt")
+        self.assertIn("备用", "备用")
+
+        gateway = build_data_gateway({}, datahub_settings=flags(quotes=(True, False)))
+        status = gateway.datahub_status()
+        self.assertEqual(status["enabled_reads"], 1)
+        self.assertEqual(status["total_reads"], len(DATASETS))
+        self.assertGreaterEqual(status["fallback_count"], 1)
+
+    def test_try_daily_chart_converts_hub_bars(self) -> None:
+        rows = [
+            {
+                "ts_code": "600000.SH",
+                "trade_date": "20240901",
+                "open": 10.0,
+                "high": 10.4,
+                "low": 9.9,
+                "close": 10.2,
+                "volume": 100000,
+                "amount": 2000000,
+            },
+            {
+                "ts_code": "600000.SH",
+                "trade_date": "20240902",
+                "open": 10.2,
+                "high": 10.5,
+                "low": 10.1,
+                "close": 10.4,
+                "volume": 120000,
+                "amount": 2400000,
+            },
+        ]
+        hub = DatahubBridge(
+            flags(daily=(True, False)),
+            FakeClient(
+                response=DatahubResponse(
+                    data=rows,
+                    meta={"stale": False, "staleness_seconds": 0, "source": "tushare:daily"},
+                )
+            ),
+        )
+        chart = hub.try_daily_chart("600000.SH", "20240902", 90, "daily")
+        self.assertEqual(chart[-1]["trade_date"], "2024-09-02")
+        self.assertEqual(chart[-1]["close"], 10.4)
+        self.assertAlmostEqual(chart[-1]["amount_billion"], 0.024)
+
+    def test_try_daily_chart_keeps_usable_bars_when_coverage_incomplete(self) -> None:
+        rows = [
+            {
+                "ts_code": "000001.SZ",
+                "trade_date": "20240901",
+                "open": 10.0,
+                "high": 10.4,
+                "low": 9.9,
+                "close": 10.2,
+                "volume": 100000,
+                "amount": 2000000,
+            },
+            {
+                "ts_code": "000001.SZ",
+                "trade_date": "20240902",
+                "open": 10.2,
+                "high": 10.5,
+                "low": 10.1,
+                "close": 10.4,
+                "volume": 120000,
+                "amount": 2400000,
+            },
+        ]
+        hub = DatahubBridge(
+            flags(daily=(True, False)),
+            FakeClient(
+                response=DatahubResponse(
+                    data=rows,
+                    meta={
+                        "stale": False,
+                        "staleness_seconds": 0,
+                        "incomplete": True,
+                        "coverage": {"complete": False, "missing_count": 127},
+                        "source": "tushare:daily",
+                    },
+                )
+            ),
+        )
+        chart = hub.try_daily_chart("000001.SZ", "20240902", 90, "daily")
+        self.assertIsNotNone(chart)
+        self.assertEqual(chart[-1]["trade_date"], "2024-09-02")
+        self.assertEqual(chart[-1]["close"], 10.4)
+
+    def test_gateway_tushare_assembly_binds_hooks_on_inner_client(self) -> None:
+        quotes = [
+            {
+                "ts_code": f"{index:06d}.SZ",
+                "name": f"S{index}",
+                "pre_close": 10.0,
+                "open": 10.0,
+                "high": 10.5,
+                "low": 9.8,
+                "close": 10.2,
+                "vol": 100.0,
+                "amount": 1000.0,
+                "quote_date": "20240902",
+            }
+            for index in range(1, 221)
+        ]
+        hub_client = FakeClient(
+            response=DatahubResponse(
+                data=quotes,
+                meta={"stale": False, "staleness_seconds": 0, "source": "eastmoney_clist"},
+            )
+        )
+        gateway = build_data_gateway(
+            {"tushare_token": "tok"},
+            datahub_settings=flags(quotes=(True, False), daily=(True, False)),
+        )
+        gateway.datahub.client = hub_client
+        wrapped = gateway.tushare()
+        inner = wrapped._legacy
+        self.assertTrue(callable(getattr(inner, "try_market_quotes", None)))
+        self.assertTrue(callable(getattr(inner, "try_index_quotes", None)))
+        self.assertTrue(callable(getattr(inner, "record_datahub_legacy", None)))
+        self.assertIs(inner.query.__self__, wrapped)
+        self.assertEqual(inner.query.__func__, wrapped.query.__func__)
+        self.assertFalse(hasattr(type(inner), "try_market_quotes"))
+        rows = inner.try_market_quotes("20240902")
+        self.assertGreaterEqual(len(rows or []), 200)
+        self.assertIn("/v1/quotes/latest", hub_client.paths)
+        hub_client.response = DatahubResponse(
+            data=[dict(HUB_DAILY)],
+            meta={"stale": False, "staleness_seconds": 0, "source": "tushare:daily"},
+        )
+        daily = inner.query("daily", {"trade_date": "20240902"}, "ts_code,amount")
+        self.assertEqual(daily[0]["amount"], 2000.0)
+        self.assertIn("/v1/bars/daily", hub_client.paths)
 
     def test_features_do_not_import_datahub_client(self) -> None:
         violations = []

tests/test_heaven_realtime.py

@@ -6,7 +6,7 @@ import unittest
 from datetime import datetime
 from unittest.mock import MagicMock, patch
 
-from backend.data.realtime import WebRealtimeAggregator
+from backend.data.realtime import RealtimeAggregateError, WebRealtimeAggregator
 from backend.features.heaven.engine import _market_line_scores, build_manual_market_hexagram
 from server import DashboardService
 from backend.data.providers.tushare_client import (
@@ -415,6 +415,50 @@ class RealtimeAggregatorTests(unittest.TestCase):
         self.assertEqual(quotes[0]["vol"], 1000)
         self.assertEqual(quotes[0]["quote_date"], "20260720")
 
+    @patch.object(WebRealtimeAggregator, "_get_text")
+    def test_tencent_stock_quote_keeps_expected_date(self, get_text: MagicMock):
+        fields = [""] * 38
+        fields[1] = "浦发银行"
+        fields[2] = "600000"
+        fields[3] = "11.20"
+        fields[4] = "11.00"
+        fields[5] = "11.10"
+        fields[6] = "1234"
+        fields[30] = "20260720103000"
+        fields[33] = "11.30"
+        fields[34] = "11.00"
+        fields[37] = "1380"
+        get_text.return_value = (f'v_sh600000="{"~".join(fields)}";', 0)
+
+        quote = WebRealtimeAggregator().tencent_stock_quote("600000", "20260720")
+
+        self.assertEqual(quote["ts_code"], "600000.SH")
+        self.assertEqual(quote["quote_date"], "20260720")
+        self.assertEqual(quote["vol"], 123400)
+        self.assertAlmostEqual(quote["amount"], 13_800_000)
+
+    @patch.object(WebRealtimeAggregator, "_get_json")
+    def test_eastmoney_stock_quote_rejects_stale_date(self, get_json: MagicMock):
+        epoch = datetime(2026, 7, 19, 15, 0).timestamp()
+        get_json.return_value = {
+            "rc": 0,
+            "data": {
+                "f43": 11.2,
+                "f44": 11.3,
+                "f45": 11.0,
+                "f46": 11.1,
+                "f47": 10,
+                "f48": 50000000,
+                "f57": "300750",
+                "f58": "宁德时代",
+                "f60": 11.0,
+                "f86": epoch,
+            },
+        }
+
+        with self.assertRaises(RealtimeAggregateError):
+            WebRealtimeAggregator().eastmoney_stock_quote("300750.SZ", "20260720")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_http_dispatch.py

@@ -138,7 +138,7 @@ class HttpDispatchContractTests(unittest.TestCase):
             self.assertTrue(claimed.isdisjoint(methods))
             claimed.update(methods)
             self.assertLessEqual(len(path.read_text(encoding="utf-8").splitlines()), line_limit)
-        self.assertEqual(len(claimed), 27)
+        self.assertEqual(len(claimed), 28)
 
 
 if __name__ == "__main__":

tests/test_realtime_dashboard.py

@@ -364,6 +364,105 @@ class RealtimeDashboardTests(unittest.TestCase):
         self.assertEqual(quote["amount"], 406050000)
         self.assertEqual(quote["source"], "tencent_qt")
 
+    def test_datahub_market_quotes_used_before_legacy(self):
+        calls = []
+
+        def try_market_quotes(trade_date):
+            calls.append(trade_date)
+            return list(FREE_QUOTES)
+
+        self.client.try_market_quotes = try_market_quotes
+        self.client.realtime_aggregator = FakeFreeAggregator(fail=True)
+        TushareClient._realtime_reference_cache.clear()
+        dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
+        self.assertEqual(calls, ["20260720"])
+        self.assertEqual(dashboard["meta"]["quote_source"], "datahub")
+        self.assertEqual(dashboard["meta"]["source"], "datahub")
+        self.assertEqual(dashboard["meta"]["quote_count"], 3)
+        self.assertIn("数据中枢", dashboard["meta"]["notice"])
+
+    def test_gateway_dashboard_uses_bound_market_quotes(self) -> None:
+        from backend.data import build_data_gateway
+        from backend.data.datahub.client import DatahubResponse
+        from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags
+        from backend.data.gateway import DataGateway
+        from backend.data.providers.tushare import TushareProvider
+
+        quotes = [
+            {
+                "ts_code": item["ts_code"],
+                "name": item["name"],
+                "pre_close": item["pre_close"],
+                "open": item["open"],
+                "high": item["high"],
+                "low": item["low"],
+                "close": item["close"],
+                "vol": item["vol"],
+                "amount": item["amount"],
+                "quote_date": "20260720",
+            }
+            for item in FREE_QUOTES
+        ]
+        extras = [
+            {
+                "ts_code": f"{index:06d}.SZ",
+                "name": f"X{index}",
+                "pre_close": 10.0,
+                "open": 10.0,
+                "high": 10.2,
+                "low": 9.8,
+                "close": 10.1,
+                "vol": 100.0,
+                "amount": 1000.0,
+                "quote_date": "20260720",
+            }
+            for index in range(10, 230)
+        ]
+
+        class QuoteHub:
+            def __init__(self):
+                self.calls = []
+
+            def quotes_latest(self, **params):
+                return self.get("/v1/quotes/latest", params)
+
+            def get(self, path, params=None):
+                self.calls.append(path)
+                if path == "/v1/quotes/latest":
+                    return DatahubResponse(
+                        data=quotes + extras,
+                        meta={"stale": False, "staleness_seconds": 0, "source": "eastmoney_clist"},
+                    )
+                raise AssertionError(path)
+
+        datasets = {name: DatasetFlags(name) for name in DATASETS}
+        datasets["quotes"] = DatasetFlags("quotes", read=True, shadow=False)
+        settings = DatahubSettings(base_url="http://127.0.0.1:9", token="tok", datasets=datasets)
+        base = build_data_gateway({"tushare_token": "tok"}, datahub_settings=settings)
+        gateway = DataGateway(
+            policy=base.policy,
+            quality=base.quality,
+            tushare_provider=TushareProvider(
+                lambda: "tok",
+                client_factory=lambda token: FakeRealtimeClient(token),
+            ),
+            ifind_provider=base.ifind_provider,
+            chart_data=base.chart_data,
+            realtime_observer=base.realtime_observer,
+            datahub=base.datahub,
+        )
+        gateway.datahub.client = QuoteHub()
+        wrapped = gateway.tushare()
+        inner = wrapped._legacy
+        inner.clock = lambda: datetime(2026, 7, 20, 10, 30, tzinfo=timezone(timedelta(hours=8)))
+        inner.realtime_aggregator = FakeFreeAggregator(fail=True)
+        TushareClient._realtime_reference_cache.clear()
+        dashboard = wrapped.dashboard("20260720")
+        self.assertEqual(dashboard["meta"]["quote_source"], "datahub")
+        self.assertIn("/v1/quotes/latest", gateway.datahub.client.calls)
+        self.assertTrue(callable(getattr(inner, "try_market_quotes", None)))
+        self.assertFalse(hasattr(type(inner), "try_market_quotes"))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_stock_detail_realtime.py

@@ -5,6 +5,10 @@ import unittest
 from datetime import datetime, timedelta
 from unittest.mock import patch
 
+from backend.data.providers.tushare_client import TushareError
+from backend.data.realtime import RealtimeAggregateError
+from backend.features.market.charts import ChartDataError
+from backend.features.market.service import TODAY_DAILY_UNAVAILABLE_NOTICE
 from server import DashboardService
 
 
@@ -17,6 +21,10 @@ class DetailDatabaseStub:
     def list_notes(user_id, code=""):
         return []
 
+    @staticmethod
+    def get_snapshot(trade_date):
+        return {}
+
 
 class RealtimeClientStub:
     quote_calls = 0
@@ -61,6 +69,122 @@ class FixedPreopenDatetime(datetime):
         return cls.fixed_now
 
 
+class FixedLunchDatetime(datetime):
+    fixed_now = datetime(2026, 7, 31, 11, 45).astimezone()
+
+    @classmethod
+    def now(cls, tz=None):
+        return cls.fixed_now
+
+
+class FixedAfterCloseDatetime(datetime):
+    fixed_now = datetime(2026, 7, 31, 15, 30).astimezone()
+
+    @classmethod
+    def now(cls, tz=None):
+        return cls.fixed_now
+
+
+class DeniedRealtimeClientStub:
+    quote_calls = 0
+
+    def __init__(self, token):
+        self.token = token
+
+    @staticmethod
+    def resolve_trade_context(requested_date):
+        return requested_date, requested_date
+
+    @classmethod
+    def realtime_stock_quote(cls, ts_code, reference_date=""):
+        cls.quote_calls += 1
+        raise TushareError("没有接口访问权限")
+
+
+class FreeQuoteAggregator:
+    def __init__(self, quote=None, fail=False):
+        self.quote = quote
+        self.fail = fail
+        self.tencent_calls = 0
+        self.eastmoney_calls = 0
+
+    def tencent_stock_quote(self, code, expected_date=""):
+        self.tencent_calls += 1
+        if self.fail:
+            raise RealtimeAggregateError("tencent down")
+        if self.quote and self.quote.get("source") == "eastmoney_stock":
+            raise RealtimeAggregateError("tencent empty")
+        if self.quote:
+            return self.quote
+        raise RealtimeAggregateError("tencent empty")
+
+    def eastmoney_stock_quote(self, code, expected_date=""):
+        self.eastmoney_calls += 1
+        if self.fail:
+            raise RealtimeAggregateError("eastmoney down")
+        if self.quote and self.quote.get("source") == "eastmoney_stock":
+            return self.quote
+        raise RealtimeAggregateError("eastmoney empty")
+
+
+class IntradayChartStub:
+    def __init__(self, points, previous_close=10.0, trade_date="2026-07-31"):
+        self.points = points
+        self.previous_close = previous_close
+        self.trade_date = trade_date
+
+    def stock_daily(self, code, end_date, limit=90):
+        raise ChartDataError("iFinD daily unavailable")
+
+    def stock_intraday(self, code):
+        return {
+            "trade_date": self.trade_date,
+            "previous_close": self.previous_close,
+            "points": self.points,
+        }
+
+
+def _history_payload(code="002141"):
+    yesterday = (FixedMarketDatetime.fixed_now - timedelta(days=1)).strftime("%Y-%m-%d")
+    return {
+        "meta": {"trade_date": yesterday, "source": "tushare"},
+        "stock": {"code": code, "name": "旧名称", "price": 10, "change": 7.1},
+        "prices": [
+            {
+                "trade_date": yesterday,
+                "open": 9.5,
+                "high": 10.1,
+                "low": 9.4,
+                "close": 10,
+                "change": 7.1,
+                "volume": 100,
+                "amount_billion": 1.1,
+            }
+        ],
+        "moneyflow": {},
+    }
+
+
+def _free_quote(source="tencent_qt", **overrides):
+    quote = {
+        "ts_code": "002141.SZ",
+        "name": "贤程科技",
+        "pre_close": 10.0,
+        "open": 10.2,
+        "high": 10.8,
+        "low": 10.1,
+        "close": 10.6,
+        "vol": 250000,
+        "amount": 26_500_000,
+        "quote_date": "20260731",
+        "quote_time_epoch": int(datetime(2026, 7, 31, 10, 31).timestamp()),
+        "source": source,
+        "turnover_rate": 2.5,
+    }
+    quote.update(overrides)
+    return quote
+
+
 class StockDetailRealtimeTests(unittest.TestCase):
     def setUp(self):
         self.service = DashboardService.__new__(DashboardService)
@@ -68,7 +192,11 @@ class StockDetailRealtimeTests(unittest.TestCase):
         self.service.database = DetailDatabaseStub()
         self.service._request_context = threading.local()
         self.service._request_context.user_id = 1
+        self.service.ifind = None
+        self.service.realtime_aggregator = None
+        self.service.chart_data = None
         RealtimeClientStub.quote_calls = 0
+        DeniedRealtimeClientStub.quote_calls = 0
 
     def test_today_detail_merges_rt_quote_without_mutating_daily_cache(self):
         today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
@@ -162,6 +290,183 @@ class StockDetailRealtimeTests(unittest.TestCase):
         self.assertEqual(result["stock"]["change"], 1.2)
         self.assertEqual(RealtimeClientStub.quote_calls, 0)
 
+    def test_today_detail_falls_back_to_tencent_quote_when_rt_k_denied(self):
+        today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
+        aggregator = FreeQuoteAggregator(_free_quote())
+        self.service.realtime_aggregator = aggregator
+        DeniedRealtimeClientStub.quote_calls = 0
+
+        with patch("backend.features.market.service.datetime", FixedMarketDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(_history_payload(), "002141", today)
+
+        bar = result["prices"][-1]
+        self.assertEqual(bar["trade_date"], "2026-07-31")
+        self.assertTrue(bar["realtime"])
+        self.assertEqual(bar["open"], 10.2)
+        self.assertEqual(bar["high"], 10.8)
+        self.assertEqual(bar["low"], 10.1)
+        self.assertEqual(bar["close"], 10.6)
+        self.assertAlmostEqual(bar["change"], 6.0, places=4)
+        self.assertEqual(bar["volume"], 2500)
+        self.assertAlmostEqual(bar["amount_billion"], 0.265)
+        self.assertEqual(len(result["prices"]), 2)
+        self.assertEqual(result["meta"]["notice"], "")
+        self.assertEqual(aggregator.tencent_calls, 1)
+        self.assertEqual(DeniedRealtimeClientStub.quote_calls, 1)
+
+    def test_today_detail_falls_back_to_eastmoney_then_intraday(self):
+        today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
+        aggregator = FreeQuoteAggregator(
+            _free_quote("eastmoney_stock", ts_code="600000.SH", name="浦发银行"),
+        )
+        self.service.realtime_aggregator = aggregator
+        DeniedRealtimeClientStub.quote_calls = 0
+
+        with patch("backend.features.market.service.datetime", FixedMarketDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(_history_payload("600000"), "600000", today)
+
+        self.assertEqual(result["prices"][-1]["trade_date"], "2026-07-31")
+        self.assertEqual(result["prices"][-1]["close"], 10.6)
+        self.assertEqual(aggregator.tencent_calls, 1)
+        self.assertEqual(aggregator.eastmoney_calls, 1)
+
+        aggregator = FreeQuoteAggregator(fail=True)
+        self.service.realtime_aggregator = aggregator
+        self.service.chart_data = IntradayChartStub(
+            [
+                {
+                    "date": "2026-07-31",
+                    "time": "09:30",
+                    "open": 10.1,
+                    "high": 10.2,
+                    "low": 10.0,
+                    "close": 10.15,
+                    "volume": 120,
+                    "amount": 121800,
+                },
+                {
+                    "date": "2026-07-31",
+                    "time": "10:05",
+                    "open": 10.15,
+                    "high": 10.5,
+                    "low": 9.9,
+                    "close": 10.4,
+                    "volume": 80,
+                    "amount": 83200,
+                },
+            ]
+        )
+        with patch("backend.features.market.service.datetime", FixedMarketDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(_history_payload("300750"), "300750", today)
+
+        bar = result["prices"][-1]
+        self.assertEqual(bar["trade_date"], "2026-07-31")
+        self.assertEqual(bar["open"], 10.1)
+        self.assertEqual(bar["high"], 10.5)
+        self.assertEqual(bar["low"], 9.9)
+        self.assertEqual(bar["close"], 10.4)
+        self.assertAlmostEqual(bar["change"], 4.0, places=4)
+        self.assertEqual(bar["volume"], 200)
+        self.assertTrue(bar["realtime"])
+
+    def test_today_detail_keeps_history_when_free_sources_fail(self):
+        today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
+        self.service.realtime_aggregator = FreeQuoteAggregator(fail=True)
+        self.service.chart_data = IntradayChartStub([], trade_date="2026-07-30")
+        DeniedRealtimeClientStub.quote_calls = 0
+
+        with patch("backend.features.market.service.datetime", FixedMarketDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(_history_payload(), "002141", today)
+
+        self.assertEqual(result["prices"][-1]["trade_date"], "2026-07-30")
+        self.assertFalse(result["meta"].get("realtime", False))
+        self.assertEqual(result["meta"]["notice"], TODAY_DAILY_UNAVAILABLE_NOTICE)
+        self.assertEqual(len(result["prices"]), 1)
+
+    def test_lunch_keeps_morning_realtime_bar(self):
+        today = FixedLunchDatetime.fixed_now.strftime("%Y%m%d")
+        self.service.realtime_aggregator = FreeQuoteAggregator(
+            _free_quote(quote_time_epoch=int(datetime(2026, 7, 31, 11, 30).timestamp()))
+        )
+        DeniedRealtimeClientStub.quote_calls = 0
+        with patch("backend.features.market.service.datetime", FixedLunchDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(_history_payload(), "002141", today)
+
+        self.assertEqual(result["prices"][-1]["trade_date"], "2026-07-31")
+        self.assertTrue(result["meta"]["realtime"])
+
+    def test_after_close_keeps_forming_bar_until_official_ready(self):
+        today = FixedAfterCloseDatetime.fixed_now.strftime("%Y%m%d")
+        self.service.realtime_aggregator = FreeQuoteAggregator(_free_quote())
+        DeniedRealtimeClientStub.quote_calls = 0
+        with patch("backend.features.market.service.datetime", FixedAfterCloseDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            forming = self.service._prepare_stock_detail(_history_payload(), "002141", today)
+        self.assertEqual(forming["prices"][-1]["trade_date"], "2026-07-31")
+        self.assertTrue(forming["prices"][-1]["realtime"])
+
+        official = _history_payload()
+        official["prices"].append(
+            {
+                "trade_date": "2026-07-31",
+                "open": 10.15,
+                "high": 10.9,
+                "low": 10.05,
+                "close": 10.7,
+                "change": 7.0,
+                "volume": 1800,
+                "amount_billion": 0.3,
+            }
+        )
+        RealtimeClientStub.quote_calls = 0
+        with patch("backend.features.market.service.datetime", FixedAfterCloseDatetime), patch(
+            "backend.features.market.service.TushareClient", RealtimeClientStub
+        ):
+            replaced = self.service._prepare_stock_detail(official, "002141", today)
+
+        self.assertEqual(replaced["prices"][-1]["close"], 10.7)
+        self.assertFalse(replaced["prices"][-1].get("realtime", False))
+        self.assertEqual(len(replaced["prices"]), 2)
+        self.assertEqual(RealtimeClientStub.quote_calls, 0)
+
+    def test_same_date_bar_is_replaced_not_duplicated(self):
+        today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d")
+        payload = _history_payload()
+        payload["prices"].append(
+            {
+                "trade_date": "2026-07-31",
+                "open": 10.0,
+                "high": 10.1,
+                "low": 9.9,
+                "close": 10.05,
+                "change": 0.5,
+                "volume": 10,
+                "amount_billion": 0.01,
+                "realtime": True,
+            }
+        )
+        self.service.realtime_aggregator = FreeQuoteAggregator(_free_quote())
+        DeniedRealtimeClientStub.quote_calls = 0
+        with patch("backend.features.market.service.datetime", FixedMarketDatetime), patch(
+            "backend.features.market.service.TushareClient", DeniedRealtimeClientStub
+        ):
+            result = self.service._prepare_stock_detail(payload, "002141", today)
+
+        self.assertEqual(len(result["prices"]), 2)
+        self.assertEqual(result["prices"][-1]["close"], 10.6)
+        self.assertEqual(result["prices"][-1]["trade_date"], "2026-07-31")
+
 
 if __name__ == "__main__":
     unittest.main()

xiaobai-datahub/README.md

@@ -7,7 +7,7 @@
 
 - SQLite WAL `datahub.db`，容器名 `xiaobai-datahub`，端口 `8766`
 - 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_* 正式表
+- 盘中观察（provisional）：东财/腾讯指数报价、个股最新价、全市场快照、分时点（`/v1/quotes/latest` 不传 codes 即全市场，`/v1/indexes/quotes` `/v1/intraday/points`）；永不写入 eod_* 正式表
 - 暂存 → 校验 → 整批原子发布 → 可回滚
 - `/v1` 稳定接口（`X-Datahub-Token`）
 - `/admin/` 最小管理后台（总览 / 数据源 / 调度 / 发布 / 数据集 / 审计）
@@ -124,6 +124,18 @@ python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
 
 管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
 
+## 估值发布后复核与自动追补
+
+Tushare `daily_basic` 会在盘后继续改当日字段。HEL-423 在 2026-09-07 观察到：中枢 17:10 发布 `003021.SZ turnover_rate=1.3565`，21:05 上游/旧链路已是 `1.3572`；其余 7 类观察对象当日一致。日 K、资金流、竞价、指数没有同类晚间修订证据，股票主档已有 20:00/23:10 刷新，因此默认只复核估值，不盲目全量重拉。
+
+窗口（可配）：交易日 **20:00–23:20**，每 30 分钟一次轻量比对（对齐网站 21:00 / 23:30 观察）。只拉取 `daily_basic`，按网站真实请求字段精确比较，无误差豁免。
+
+- 无变化：不产生新批次，状态「已追平」。
+- 发现修订：重新走字段质量门、覆盖检查和 A 组整批原子发布；读者全程只能看到上一完整版本或新完整版本。
+- 上游空 / 接口失败 / 不完整 / 质量门拒绝：保留上一完整版本，状态「复核失败」。
+- 23:20 截止后停止当晚复核；下一自然日盘前对上一交易日再做一次安全追赶。
+- 与 `eod_a` / `eod_retry` 共用互斥锁；容器重启会在窗口内立即补一次。
+
 ## 备份
 
 每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：

xiaobai-datahub/admin/app.js

@@ -105,6 +105,7 @@ async function render() {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
     const eod = data.eod_status || {};
+    const rev = data.revision_status || {};
     const eodLabels = {
       pending_first_attempt: "等待首次尝试",
       waiting_upstream: "等待上游",
@@ -112,6 +113,14 @@ async function render() {
       cutoff_failed: "已截止失败",
       closed_day: "休市",
     };
+    const revLabels = {
+      waiting_review: "等待复核",
+      review_failed: "复核失败",
+      aligned: "已追平",
+      cutoff: "已截止",
+      pending_publish: "待发布",
+      closed_day: "休市",
+    };
     const eodExtra = [];
     if (eod.state === "waiting_upstream") {
       eodExtra.push(`已试 ${eod.attempts} 次`);
@@ -121,12 +130,16 @@ async function render() {
     if (eod.state === "cutoff_failed" && eod.missing_datasets) {
       eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
     }
+    const revExtra = [];
+    if (rev.detail) revExtra.push(esc(String(rev.detail)));
+    if (rev.window) revExtra.push(esc(String(rev.window)));
     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>${esc(revLabels[rev.state] || rev.state || "-")}</strong><div class="muted">${revExtra.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>

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

@@ -17,6 +17,10 @@
   "eod_retry_start": "15:15",
   "eod_retry_interval_minutes": 30,
   "eod_retry_cutoff": "23:30",
+  "revision_review_datasets": ["valuation"],
+  "revision_review_start": "20:00",
+  "revision_review_interval_minutes": 30,
+  "revision_review_cutoff": "23:20",
   "moneyflow_history_trading_days": 60,
   "stocks_refresh_times": [
     "20:00",

xiaobai-datahub/datahub/adapters/eastmoney.py

@@ -13,6 +13,15 @@ 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"
+EASTMONEY_A_SHARE_BOARDS = (
+    "m:0+t:6",
+    "m:0+t:80",
+    "m:1+t:2",
+    "m:1+t:23",
+    "m:0+t:81",
+)
+EASTMONEY_QUOTE_FIELDS = "f12,f13,f14,f2,f3,f4,f5,f6,f15,f16,f17,f18,f8,f124"
+EASTMONEY_MARKET_PAGE_SIZE = 100
 TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
 HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get"
 BROWSER_UA = (
@@ -59,7 +68,11 @@ class EastmoneyAdapter(MarketAdapter):
             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))
+            if codes:
+                return self.fetch_quotes(list(codes))
+            return self.fetch_market_quotes()
+        if dataset in {"quotes_market", "market_quotes"}:
+            return self.fetch_market_quotes()
         raise AdapterError(f"{self.name} unsupported dataset: {dataset}")
 
     def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -167,6 +180,58 @@ class EastmoneyAdapter(MarketAdapter):
             )
         return result
 
+    def fetch_market_quotes(self) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        board_errors: list[str] = []
+        for board in EASTMONEY_A_SHARE_BOARDS:
+            try:
+                rows.extend(self._board_quotes(board))
+            except Exception as exc:
+                board_errors.append(f"{board}:{exc}")
+        quotes: list[dict[str, Any]] = []
+        seen: set[str] = set()
+        for row in rows:
+            quote = _normalize_market_quote(row)
+            ts_code = str((quote or {}).get("ts_code") or "")
+            if not quote or ts_code in seen:
+                continue
+            seen.add(ts_code)
+            quotes.append(quote)
+        if len(quotes) < 200:
+            detail = f"；{'; '.join(board_errors)}" if board_errors else ""
+            raise AdapterError(f"Eastmoney market snapshot too small: {len(quotes)}{detail}")
+        return quotes
+
+    def _board_quotes(self, board: str) -> list[dict[str, Any]]:
+        first = self._market_page(board, 1)
+        data = first.get("data") or {}
+        rows = list(data.get("diff") or [])
+        total = int(finite_number(data.get("total")) or 0)
+        page_count = 1
+        if total > 0:
+            page_count = max(1, (total + EASTMONEY_MARKET_PAGE_SIZE - 1) // EASTMONEY_MARKET_PAGE_SIZE)
+        for page in range(2, min(page_count, 40) + 1):
+            payload = self._market_page(board, page)
+            rows.extend(list((payload.get("data") or {}).get("diff") or []))
+        return rows
+
+    def _market_page(self, board: str, page: int) -> dict[str, Any]:
+        return self._get_json(
+            EASTMONEY_CLIST_URL,
+            {
+                "pn": str(page),
+                "pz": str(EASTMONEY_MARKET_PAGE_SIZE),
+                "po": "1",
+                "np": "1",
+                "fltt": "2",
+                "invt": "2",
+                "fid": "f12",
+                "fs": board,
+                "fields": EASTMONEY_QUOTE_FIELDS,
+            },
+            referer="https://quote.eastmoney.com/center/gridlist.html",
+        )
+
     def fetch_intraday(self, ts_code: str, date: str = "") -> dict[str, Any]:
         code = str(ts_code or "").upper()
         if code in INDEX_SECIDS:
@@ -252,6 +317,45 @@ def _preferred_session(points: list[dict[str, Any]], preferred_date: str = "") -
     return [point for point in points if str(point.get("date") or "") == latest]
 
 
+def _normalize_market_quote(row: dict[str, Any]) -> dict[str, Any] | None:
+    symbol = str(row.get("f12") or "").strip()
+    if not symbol.isdigit() or len(symbol) != 6:
+        return None
+    close = round4(finite_number(row.get("f2")))
+    previous_close = round4(finite_number(row.get("f18")))
+    if close <= 0 or previous_close <= 0:
+        return None
+    market = int(finite_number(row.get("f13")) or 0)
+    if market == 1 or symbol.startswith(("5", "6", "9")):
+        ts_code = f"{symbol}.SH"
+    elif symbol.startswith(("4", "8")):
+        ts_code = f"{symbol}.BJ"
+    else:
+        ts_code = f"{symbol}.SZ"
+    epoch = int(finite_number(row.get("f124")) or 0)
+    quote_date = ""
+    if epoch > 0:
+        quote_date = datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d")
+    return {
+        "ts_code": ts_code,
+        "name": row.get("f14") or symbol,
+        "pre_close": previous_close,
+        "previous_close": previous_close,
+        "open": round4(finite_number(row.get("f17"))),
+        "high": round4(finite_number(row.get("f15"))),
+        "low": round4(finite_number(row.get("f16"))),
+        "close": close,
+        "price": close,
+        "pct_chg": round4(finite_number(row.get("f3"))),
+        "vol": round4(finite_number(row.get("f5")) * 100),
+        "volume": round4(finite_number(row.get("f5")) * 100),
+        "amount": round4(finite_number(row.get("f6"))),
+        "quote_date": quote_date,
+        "quote_time_epoch": epoch,
+        "source": "eastmoney_clist",
+    }
+
+
 def _parse_trend(raw: Any) -> dict[str, Any] | None:
     text = str(raw or "")
     parts = text.split(",")

xiaobai-datahub/datahub/admin_api.py

@@ -39,6 +39,7 @@ class AdminAPI:
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
             "eod_status": self.scheduler.eod_status(today),
+            "revision_status": self.scheduler.revision_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -91,6 +92,7 @@ class AdminAPI:
                 {"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": "eod_revise", "at": "20:00-23:20", "title": "估值发布后复核（轻量比对，有修订才整组原子追补）"},
                 {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"},
                 {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},

xiaobai-datahub/datahub/db.py

@@ -239,6 +239,19 @@ CREATE TABLE IF NOT EXISTS eod_progress (
   updated_at TEXT NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS revision_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,
+  catchup_done INTEGER NOT NULL DEFAULT 0,
+  last_diff TEXT,
+  detail TEXT,
+  updated_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS audit_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,
   actor TEXT NOT NULL,

xiaobai-datahub/datahub/pipeline.py

@@ -20,6 +20,12 @@ from datahub.governance.ratelimit import TokenBucket
 from datahub.governance.retry import RetryError, retry_call
 from datahub.logutil import get_logger
 from datahub.normalize import finite_number, normalize_daily
+from datahub.revision import (
+    compare_fields,
+    diff_published_vs_upstream,
+    official_table,
+    revision_datasets,
+)
 from datahub.settings import Settings
 from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd
 
@@ -748,6 +754,240 @@ class Pipeline:
             return self.run_extended_soft((name,), trade_date, force=True)
         raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
 
+    def published_official_rows(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
+        day = yyyymmdd(trade_date)
+        batch_id = self.active_batch(dataset, day)
+        if not batch_id:
+            return []
+        fields = compare_fields(dataset)
+        table = official_table(dataset)
+        if fields:
+            columns = ",".join(fields)
+            return self.db.fetchall(
+                f"SELECT {columns} FROM {table} WHERE batch_id = ?",
+                (batch_id,),
+            )
+        return self.db.fetchall(f"SELECT * FROM {table} WHERE batch_id = ?", (batch_id,))
+
+    def compare_revision(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Light fetch of one revision-risk dataset vs the published official rows."""
+        day = yyyymmdd(trade_date)
+        published = self.published_official_rows(dataset, day)
+        if not published:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "skipped",
+                "reason": "not_published",
+            }
+        try:
+            upstream = self._fetch_dataset(dataset, day)
+        except Exception as exc:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_error",
+                "error": str(exc),
+            }
+        if not upstream:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_empty",
+                "error": "revision review upstream empty",
+                "published_rows": len(published),
+                "upstream_rows": 0,
+            }
+        listed = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
+        )
+        listed_n = int((listed or {}).get("n") or 0)
+        floor = float(self.settings.quality.get("daily_row_ratio") or 0.98)
+        if listed_n and len(upstream) / listed_n < floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"/ listed {listed_n} < {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        if len(upstream) < len(published) * floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"< published {len(published)} * {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        compared = diff_published_vs_upstream(dataset, published, upstream)
+        compared["trade_date"] = day
+        compared["state"] = "changed" if compared["changed"] else "unchanged"
+        compared["reason"] = "revised" if compared["changed"] else "unchanged"
+        return compared
+
+    def review_published_revisions(self, trade_date: str) -> dict[str, Any]:
+        """Evening/morning catch-up: compare website fields, republish only on change.
+
+        Unchanged → no new batch. Changed → full A/B boundary quality gate +
+        atomic switch (HEL-459/460/461). Empty/failed/incomplete upstream keeps
+        the previous complete official version.
+        """
+        day = yyyymmdd(trade_date)
+        results: dict[str, Any] = {}
+        for dataset in revision_datasets(self.settings.quality):
+            compared = self.compare_revision(dataset, day)
+            if compared.get("state") == "skipped":
+                results[dataset] = compared
+                continue
+            if compared.get("state") == "failed":
+                results[dataset] = compared
+                LOGGER.warning(
+                    "revision review kept previous official version",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": compared.get("reason"),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": compared.get("reason"),
+                            "error": compared.get("error"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            if not compared.get("changed"):
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aligned",
+                    "reason": "unchanged",
+                    "published_rows": compared.get("published_rows"),
+                    "upstream_rows": compared.get("upstream_rows"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps({"state": "aligned", "reason": "unchanged"}, ensure_ascii=False),
+                )
+                continue
+            LOGGER.info(
+                "revision review detected upstream rewrite, republishing boundary",
+                extra={
+                    "hub": {
+                        "dataset": dataset,
+                        "trade_date": day,
+                        "diffs": compared.get("diffs"),
+                        "event": "revision_review_changed",
+                    }
+                },
+            )
+            try:
+                published = self.force_republish_boundary(dataset, day)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "republish_error",
+                    "error": str(exc),
+                    "diffs": compared.get("diffs"),
+                }
+                LOGGER.warning(
+                    "revision republish failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": str(exc),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {"state": "failed", "reason": "republish_error", "error": str(exc)},
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            failures = self.eod_failures(published)
+            if failures:
+                results.update(published)
+                results[dataset] = {
+                    **(published.get(dataset) or {}),
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "quality_gate",
+                    "error": "; ".join(failures),
+                    "diffs": compared.get("diffs"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": "quality_gate",
+                            "error": "; ".join(failures),
+                            "diffs": compared.get("diffs"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            results.update(published)
+            results["review"] = {
+                "dataset": dataset,
+                "trade_date": day,
+                "state": "aligned",
+                "reason": "revised",
+                "diffs": compared.get("diffs"),
+                "missing_codes": compared.get("missing_codes"),
+                "extra_codes": compared.get("extra_codes"),
+            }
+            self.audit(
+                "pipeline", "revision-review", f"{dataset}:{day}",
+                json.dumps(
+                    {
+                        "state": "aligned",
+                        "reason": "revised",
+                        "diffs": compared.get("diffs"),
+                        "switched": sorted(
+                            name for name, item in published.items()
+                            if isinstance(item, dict) and item.get("state") == "published"
+                        ),
+                    },
+                    ensure_ascii=False,
+                ),
+            )
+        return results
+
     def run_release_group(
         self,
         datasets: tuple[str, ...],

xiaobai-datahub/datahub/realtime_serve.py

@@ -64,9 +64,36 @@ def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
     return payload
 
 
+def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
+    cache_key = "quotes:market"
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    adapter = EastmoneyAdapter()
+    try:
+        rows = adapter.fetch_market_quotes()
+        source = "eastmoney:clist"
+    except Exception as exc:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"market 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()),
+            "scope": "market",
+        },
+    )
+    _write_cache(db, cache_key, payload, QUOTE_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")
+        return fetch_market_quotes(db)
     resolved: list[str] = []
     for code in codes[:60]:
         item = resolve_code(db, code) or _guess_ts_code(code)

xiaobai-datahub/datahub/revision.py

@@ -0,0 +1,111 @@
+"""Post-publish revision review for datasets whose upstream may rewrite T-day fields.
+
+HEL-423 field evidence, not a whitelist of tolerated diffs:
+
+- 2026-09-07 valuation/daily_basic: hub published 003021.SZ turnover_rate=1.3565
+  at 17:10; website legacy and a direct Tushare read at 21:05 both showed 1.3572.
+  The other seven observed objects (daily, moneyflow, auction, stocks, status,
+  index_daily, calendar) matched. Hub had already stopped the day after the
+  first successful publish, so the revision never self-healed.
+- 2026-09-02: same dataset, opposite direction (hub already held the later
+  value). Confirms daily_basic is rewritten after the first complete dump.
+
+Daily bars, moneyflow, auction and index_daily have no same-evening field
+revision evidence. Stocks already refreshes at 20:00/23:10. Review therefore
+fetches only configured revision-risk datasets (default: valuation) and
+compares the website-requested field set. No numeric tolerance.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from datahub.db import DATASET_TABLES
+from datahub.normalize import VALUATION_FIELDS
+from datahub.numbers import finite_number, round4
+
+# Datasets with proven same-evening upstream rewrites. Config may replace this
+# list; it must not silently expand to a full EOD re-pull.
+DEFAULT_REVISION_DATASETS = ("valuation",)
+
+# Website daily_basic request (HEL-423): ts_code/trade_date plus the eight
+# value fields used by the old link and field_gates.
+WEBSITE_COMPARE_FIELDS: dict[str, tuple[str, ...]] = {
+    "valuation": VALUATION_FIELDS,
+}
+
+REVISION_STATES = ("waiting_review", "review_failed", "aligned", "cutoff")
+
+
+def revision_datasets(quality: dict[str, Any] | None) -> tuple[str, ...]:
+    raw = (quality or {}).get("revision_review_datasets")
+    if isinstance(raw, (list, tuple)) and raw:
+        names = tuple(str(item) for item in raw if str(item))
+        if names:
+            return names
+    return DEFAULT_REVISION_DATASETS
+
+
+def compare_fields(dataset: str) -> tuple[str, ...]:
+    fields = WEBSITE_COMPARE_FIELDS.get(dataset)
+    if fields:
+        return fields
+    gate = {}
+    return tuple(str(item) for item in (gate.get("fields") or []) if str(item))
+
+
+def _norm_value(field: str, value: Any) -> Any:
+    if field in {"ts_code", "trade_date"}:
+        return str(value or "")
+    number = round4(finite_number(value))
+    return number
+
+
+def row_signature(row: dict[str, Any], fields: tuple[str, ...]) -> tuple[Any, ...]:
+    return tuple(_norm_value(field, row.get(field)) for field in fields)
+
+
+def diff_published_vs_upstream(
+    dataset: str,
+    published: list[dict[str, Any]],
+    upstream: list[dict[str, Any]],
+    *,
+    max_diffs: int = 20,
+) -> dict[str, Any]:
+    """Exact compare on website-requested fields. No tolerance / exemption."""
+    fields = compare_fields(dataset)
+    if not fields:
+        fields = tuple(sorted({key for row in published + upstream for key in row if key != "batch_id"}))
+    pub_map = {str(row.get("ts_code") or "").upper(): row for row in published}
+    up_map = {str(row.get("ts_code") or "").upper(): row for row in upstream}
+    missing = sorted(code for code in pub_map if code not in up_map)
+    extra = sorted(code for code in up_map if code not in pub_map)
+    diffs: list[dict[str, Any]] = []
+    for code in sorted(set(pub_map) & set(up_map)):
+        left = row_signature(pub_map[code], fields)
+        right = row_signature(up_map[code], fields)
+        if left == right:
+            continue
+        for field, old, new in zip(fields, left, right):
+            if old == new:
+                continue
+            diffs.append({"ts_code": code, "field": field, "published": old, "upstream": new})
+            if len(diffs) >= max_diffs:
+                break
+        if len(diffs) >= max_diffs:
+            break
+    changed = bool(diffs or missing or extra)
+    return {
+        "changed": changed,
+        "dataset": dataset,
+        "fields": list(fields),
+        "published_rows": len(published),
+        "upstream_rows": len(upstream),
+        "missing_codes": missing[:max_diffs],
+        "extra_codes": extra[:max_diffs],
+        "diffs": diffs,
+    }
+
+
+def official_table(dataset: str) -> str:
+    return DATASET_TABLES[dataset][0]

xiaobai-datahub/datahub/scheduler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import threading
 from collections.abc import Callable
 from datetime import datetime, time, timedelta
@@ -8,13 +9,14 @@ from typing import Any
 from datahub.db import HubDB
 from datahub.logutil import get_logger
 from datahub.pipeline import Pipeline
+from datahub.revision import revision_datasets
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
-EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry", "eod_revise"}
 
 
 def is_open_day(db: HubDB, day: str) -> bool:
@@ -27,6 +29,20 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def previous_open_day(db: HubDB, day: str) -> str | None:
+    row = db.fetchone(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date < ?
+        ORDER BY cal_date DESC LIMIT 1
+        """,
+        (day,),
+    )
+    if row is None:
+        return None
+    return str(row["cal_date"])
+
+
 def _hhmm(value: str) -> time:
     return datetime.strptime(value, "%H:%M").time()
 
@@ -53,6 +69,7 @@ class Scheduler:
             "eod_e": self._eod_e,
             "eod_f": self._eod_f,
             "eod_retry": self._eod_retry,
+            "eod_revise": self._eod_revise,
             "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
@@ -126,6 +143,8 @@ class Scheduler:
             if job_id in {"eod_a", "eod_b"}:
                 self._settle_eod(day)
         ran.extend(self._eod_retry_tick(now, day, open_day))
+        ran.extend(self._revision_review_tick(now, day, open_day))
+        ran.extend(self._revision_catchup_tick(now, day))
         return ran
 
     # ------------------------------------------------------------------
@@ -227,6 +246,201 @@ class Scheduler:
             "detail": (row or {}).get("detail"),
         }
 
+    def revision_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM revision_progress WHERE trade_date = ?", (day,))
+
+    def revision_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """等待复核 / 复核失败 / 已追平 / 已截止."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        row = self.revision_progress(day)
+        open_day = is_open_day(self.db, day)
+        published = self._revision_ready(day)
+        if row and row["state"] in {"aligned", "review_failed", "cutoff", "waiting_review"}:
+            state = str(row["state"])
+        elif not open_day:
+            state = "closed_day"
+        elif not published:
+            state = "pending_publish"
+        else:
+            state = "waiting_review"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "datasets": list(revision_datasets(self.pipeline.settings.quality)),
+            "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 in {"waiting_review", "review_failed"} else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "catchup_done": bool(int((row or {}).get("catchup_done") or 0)),
+            "detail": (row or {}).get("detail"),
+            "window": f"{self.pipeline.settings.revision_review_start}-{self.pipeline.settings.revision_review_cutoff}",
+        }
+
+    def _revision_ready(self, day: str) -> bool:
+        return all(
+            self.pipeline.active_batch(dataset, day)
+            for dataset in revision_datasets(self.pipeline.settings.quality)
+        )
+
+    def _revision_due(self, now: datetime, row: dict[str, Any] | None) -> bool:
+        if row is None or not row.get("last_attempt_at"):
+            return True
+        try:
+            last = datetime.fromisoformat(str(row["last_attempt_at"]))
+        except ValueError:
+            return True
+        interval = timedelta(minutes=self.pipeline.settings.revision_review_interval_minutes)
+        return now_shanghai(last).replace(tzinfo=None) + interval <= now.replace(tzinfo=None)
+
+    def _revision_review_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day or not self._revision_ready(day):
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.revision_review_start)
+        cutoff = _hhmm(settings.revision_review_cutoff)
+        row = self.revision_progress(day)
+        if current < start:
+            if row is None:
+                self._save_revision_progress(day, state="waiting_review")
+            return []
+        if current >= cutoff:
+            if row is None or row["state"] not in {"aligned", "cutoff"}:
+                detail = "复核窗口已截止"
+                self._save_revision_progress(
+                    day, state="cutoff", 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_revise','failed',?,?,?,?,?)",
+                        (
+                            isoformat(now), isoformat(now), detail,
+                            int((row or {}).get("attempts") or 0), "revision cutoff reached",
+                        ),
+                    )
+            elif row["state"] == "aligned" and not row.get("finished_at"):
+                self._save_revision_progress(day, finished_at=isoformat(now))
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(day, now, catchup=False)
+
+    def _revision_catchup_tick(self, now: datetime, day: str) -> list[str]:
+        prev = previous_open_day(self.db, day)
+        if prev is None or prev >= day:
+            return []
+        if not self._revision_ready(prev):
+            return []
+        row = self.revision_progress(prev)
+        if row and int(row.get("catchup_done") or 0):
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(prev, now, catchup=True)
+
+    def _run_revision_job(self, day: str, now: datetime, catchup: bool) -> list[str]:
+        attempts = int((self.revision_progress(day) or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.revision_review_interval_minutes
+        self._save_revision_progress(
+            day,
+            state="waiting_review",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+        ran: list[str] = []
+        try:
+            out = self.run_job("eod_revise", day)
+        except Exception as exc:
+            LOGGER.warning("revision review failed for %s: %s", day, exc)
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+            )
+            ran.append("eod_revise")
+            return ran
+        ran.append("eod_revise")
+        if out.get("state") == "skipped":
+            return ran
+        result = out.get("result") if isinstance(out.get("result"), dict) else {}
+        failed = [
+            name for name, item in result.items()
+            if isinstance(item, dict) and item.get("state") == "failed"
+        ]
+        review = result.get("review") if isinstance(result.get("review"), dict) else None
+        watched = [
+            result[name]
+            for name in revision_datasets(self.pipeline.settings.quality)
+            if isinstance(result.get(name), dict)
+        ]
+        diff_blob = None
+        if review and review.get("diffs"):
+            diff_blob = json.dumps(review.get("diffs"), ensure_ascii=False)
+        else:
+            for item in watched:
+                if item.get("diffs"):
+                    diff_blob = json.dumps(item.get("diffs"), ensure_ascii=False)
+                    break
+        revised = bool(review and review.get("reason") == "revised")
+        matched = any(item.get("reason") == "unchanged" or item.get("state") == "aligned" for item in watched)
+        if failed:
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+                last_diff=diff_blob,
+            )
+        elif revised or matched:
+            fields: dict[str, Any] = {
+                "state": "aligned",
+                "finished_at": isoformat(now),
+                "detail": "已追平" if revised else "已追平（无变化）",
+                "last_diff": diff_blob,
+            }
+            if catchup:
+                fields["catchup_done"] = 1
+            self._save_revision_progress(day, **fields)
+        return ran
+
+    def _save_revision_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "catchup_done", "last_diff", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM revision_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({
+                    "trade_date": day,
+                    "state": "waiting_review",
+                    "attempts": 0,
+                    "catchup_done": 0,
+                })
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO revision_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 revision_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
     def _record_eod_attempt(self, day: str, now: datetime) -> None:
         row = self.eod_progress(day)
         attempts = int((row or {}).get("attempts") or 0) + 1
@@ -340,6 +554,9 @@ class Scheduler:
     def _eod_retry(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_missing(trade_date)
 
+    def _eod_revise(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.review_published_revisions(trade_date)
+
     def _stocks_refresh(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.refresh_stocks(trade_date)
 

xiaobai-datahub/datahub/serving.py

@@ -247,11 +247,13 @@ class V1API:
         )
 
     def quotes_latest(self, q: dict[str, str]) -> dict[str, Any]:
-        from datahub.realtime_serve import RealtimeApiError, fetch_quotes
+        from datahub.realtime_serve import RealtimeApiError, fetch_market_quotes, fetch_quotes
 
         codes = [item.strip() for item in str(q.get("codes") or "").split(",") if item.strip()]
         try:
-            return fetch_quotes(self.db, codes)
+            if codes:
+                return fetch_quotes(self.db, codes)
+            return fetch_market_quotes(self.db)
         except RealtimeApiError as exc:
             raise ApiError(exc.code, exc.message) from exc
 

xiaobai-datahub/datahub/settings.py

@@ -79,6 +79,20 @@ class Settings:
     def eod_retry_cutoff(self) -> str:
         return str(self.quality.get("eod_retry_cutoff") or "23:30")
 
+    @property
+    def revision_review_start(self) -> str:
+        # Before the 21:00 website shadow observation.
+        return str(self.quality.get("revision_review_start") or "20:00")
+
+    @property
+    def revision_review_interval_minutes(self) -> int:
+        return int(self.quality.get("revision_review_interval_minutes") or 30)
+
+    @property
+    def revision_review_cutoff(self) -> str:
+        # Last light review ~23:00; cutoff before the 23:30 observation.
+        return str(self.quality.get("revision_review_cutoff") or "23:20")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/tests/test_eod_retry.py

@@ -111,15 +111,24 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(progress["state"], "done")
         self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
 
-        # success stops all further same-day requests
+        # success stops further eod_retry; revision window has not started yet
         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._job_runs(db, "eod_revise")), 0)
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(self._eod_calls(transport)), eod_calls_before)
 
+        # 23:00 is inside the valuation review window: light daily_basic only, no new batch
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 1)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        extra = [name for name in self._eod_calls(transport)[eod_calls_before:]]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
+
     def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
         day = "20240902"
         db, transport, pipe, sched = self._make(set())
@@ -174,6 +183,7 @@ class EodRetryTests(unittest.TestCase):
         self.assertIn("eod_a", ran)
         self.assertIn("eod_b", ran)
         self.assertNotIn("eod_retry", ran)
+        self.assertIn("eod_revise", ran)
         self.assertEqual(self._published(db, day), OFFICIAL)
         after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
         self.assertEqual(
@@ -181,7 +191,9 @@ class EodRetryTests(unittest.TestCase):
             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
+        extra = self._eod_calls(transport)[len(calls_before):]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
         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:
@@ -206,6 +218,7 @@ class EodRetryTests(unittest.TestCase):
         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)
+            self.assertNotIn("eod_revise", 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,)))
@@ -226,12 +239,18 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(transport.calls), calls_before)
 
+        revised = sched.run_job("eod_revise", day)
+        self.assertEqual(revised["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_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")
+            busy_r = sched.run_job("eod_revise", day)
+            self.assertEqual(busy_r["state"], "skipped")
         finally:
             sched._eod_lock.release()
         self.assertEqual(len(self._batches(db, day)), batches_before)

xiaobai-datahub/tests/test_realtime_intraday.py

@@ -172,5 +172,61 @@ class ServingIntradayDateTests(unittest.TestCase):
         self.assertIn("intraday unavailable", ctx.exception.message)
 
 
+class MarketQuotesTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = HubDB(Path(self.tmp.name) / "hub.db")
+        self.api = V1API(self.db, None, None)  # type: ignore[arg-type]
+
+    def tearDown(self) -> None:
+        self.tmp.cleanup()
+
+    def test_empty_codes_returns_full_market_snapshot(self) -> None:
+        rows = [
+            {
+                "ts_code": f"{600000 + index:06d}.SH",
+                "name": f"股票{index}",
+                "pre_close": 10.0,
+                "close": 10.2,
+                "open": 10.1,
+                "high": 10.3,
+                "low": 10.0,
+                "vol": 1000,
+                "amount": 2000000,
+            }
+            for index in range(220)
+        ]
+        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
+            mocked.return_value.fetch_market_quotes.return_value = rows
+            omitted = self.api.handle("/v1/quotes/latest", {})
+            empty = self.api.handle("/v1/quotes/latest", {"codes": [""]})
+        self.assertEqual(len(omitted["data"]), 220)
+        self.assertEqual(omitted["meta"]["scope"], "market")
+        self.assertEqual(omitted["meta"]["source"], "eastmoney:clist")
+        self.assertEqual(len(empty["data"]), 220)
+
+    def test_explicit_codes_still_use_named_quote_path(self) -> None:
+        named = [
+            {
+                "ts_code": "600000.SH",
+                "name": "浦发银行",
+                "price": 10.2,
+                "previous_close": 10.0,
+            }
+        ]
+        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
+            mocked.return_value.fetch_quotes.return_value = named
+            payload = self.api.handle("/v1/quotes/latest", {"codes": ["600000.SH"]})
+            mocked.return_value.fetch_market_quotes.assert_not_called()
+        self.assertEqual(payload["data"][0]["ts_code"], "600000.SH")
+
+    def test_market_unavailable_stays_source_error(self) -> None:
+        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
+            mocked.return_value.fetch_market_quotes.side_effect = AdapterError("too small")
+            with self.assertRaises(ApiError) as ctx:
+                self.api.handle("/v1/quotes/latest", {})
+        self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")
+
+
 if __name__ == "__main__":
     unittest.main()

xiaobai-datahub/tests/test_revision_review.py

@@ -0,0 +1,345 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.base import AdapterError
+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 datahub.timeutil import SHANGHAI
+from tests.fixtures import RAW, TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+from tests.test_quality_gates import FIELD_GATES
+
+SAMPLE_DAY = "20260907"
+NEXT_DAY = "20260908"
+SAMPLE_CODE = "003021.SZ"
+
+
+def _dated(row: dict, day: str) -> dict:
+    item = dict(row)
+    if "trade_date" in item:
+        item["trade_date"] = day
+    return item
+
+
+class RevisingTransport:
+    """Fixture transport that can rewrite daily_basic after the first publish."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, extra_calendar: list[dict] | None = None) -> None:
+        self.calls: list[str] = []
+        self.fail_daily_basic = False
+        self.empty_daily_basic = False
+        self.null_volume_ratio = False
+        self.turnover_by_code: dict[str, float] = {}
+        self.extra_calendar = extra_calendar or []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        day = str(params.get("trade_date") or "")
+        if api_name == "trade_cal":
+            rows = fake_transport(api_name, params, fields)
+            extra = [
+                row for row in self.extra_calendar
+                if str(params.get("start_date") or "") <= row["cal_date"] <= str(params.get("end_date") or "99999999")
+            ]
+            return rows + extra
+        if self.fail_daily_basic and api_name == "daily_basic":
+            raise AdapterError("tushare daily_basic unavailable")
+        if self.empty_daily_basic and api_name == "daily_basic":
+            return []
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+            if day:
+                rows = [_dated(row, day) for row in rows]
+            return rows
+        rows = fake_transport(api_name, params, fields)
+        if api_name == "stock_basic":
+            rows = list(rows)
+            rows.append({
+                "ts_code": SAMPLE_CODE, "symbol": "003021", "name": "兆威机电",
+                "area": "广东", "industry": "元器件", "market": "主板",
+                "list_status": "L", "list_date": "20201202",
+            })
+            return rows
+        if api_name in self.DATE_APIS:
+            template = RAW.get(api_name) or []
+            if not day:
+                return [_dated(row, TRADE_DATE) for row in template]
+            out = [_dated(row, day) for row in template]
+            extra = copy.deepcopy(template[0])
+            extra["ts_code"] = SAMPLE_CODE
+            extra["trade_date"] = day
+            if api_name == "daily_basic":
+                extra["turnover_rate"] = self.turnover_by_code.get(SAMPLE_CODE, extra.get("turnover_rate"))
+                if self.null_volume_ratio:
+                    extra["volume_ratio"] = None
+                    for row in out:
+                        row["volume_ratio"] = None
+            out.append(extra)
+            if api_name == "daily_basic":
+                for row in out:
+                    code = str(row.get("ts_code") or "")
+                    if code in self.turnover_by_code:
+                        row["turnover_rate"] = self.turnover_by_code[code]
+            return out
+        return rows
+
+
+def make_revision_env(quality_extra: dict | None = None, extra_calendar: list[dict] | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    transport = RevisingTransport(extra_calendar=extra_calendar)
+    adapter = TushareAdapter("x", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.5,
+        "null_rate_max": 0.5,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+        "field_gates": FIELD_GATES,
+        "revision_review_start": "20:00",
+        "revision_review_interval_minutes": 30,
+        "revision_review_cutoff": "23:20",
+        "revision_review_datasets": ["valuation"],
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        db_path=db.path,
+        backup_dir=Path(tmp.name) / "backups",
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    sched = Scheduler(db, pipe)
+    return tmp, db, transport, pipe, sched
+
+
+SAMPLE_CALENDAR = [
+    {"exchange": "SSE", "cal_date": SAMPLE_DAY, "is_open": 1, "pretrade_date": "20260906"},
+    {"exchange": "SSE", "cal_date": NEXT_DAY, "is_open": 1, "pretrade_date": SAMPLE_DAY},
+]
+
+
+class RevisionReviewTests(unittest.TestCase):
+    def _publish(self, pipe: Pipeline, day: str) -> None:
+        pipe.ingest_reference(day)
+        pipe.run_eod_batch_a(day)
+        pipe.run_eod_batch_b(day)
+
+    def _turnover(self, db: HubDB, day: str, code: str = SAMPLE_CODE) -> float | None:
+        pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (day,),
+        )
+        row = db.fetchone(
+            "SELECT turnover_rate FROM eod_valuation WHERE batch_id=? AND ts_code=?",
+            (pub["active_batch"], code),
+        )
+        return None if row is None else row["turnover_rate"]
+
+    def _batch_ids(self, db: HubDB, day: str) -> set[str]:
+        # HEL-478/HEL-463 integration: extended soft datasets (limit_events,
+        # dragon_tiger, sector_daily, popularity) publish on their own schedule
+        # inside the same tick and are outside the valuation review boundary.
+        # Scope assertions to the A-group + stocks boundary the review republishes.
+        boundary = ("daily", "valuation", "moneyflow", "auction", "stocks")
+        placeholders = ",".join("?" for _ in boundary)
+        return {
+            str(row["batch_id"])
+            for row in db.fetchall(
+                f"SELECT batch_id FROM batches WHERE trade_date=? AND dataset IN ({placeholders})",
+                (day, *boundary),
+            )
+        }
+
+    def test_no_change_does_not_create_a_new_batch(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), before)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertIn("无变化", progress["detail"])
+        status = sched.revision_status(TRADE_DATE, clock=clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(status["state"], "aligned")
+
+    def test_hel423_20260907_single_field_revision_is_caught_up(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+        first = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        seen: list[float | None] = []
+
+        def watch() -> None:
+            seen.append(self._turnover(db, SAMPLE_DAY))
+
+        pipe.before_commit = watch
+        ran = sched.tick(clock_at(SAMPLE_DAY, 20, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(seen, [1.3565])  # readers still see the previous complete version mid-switch
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        second = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+        self.assertNotEqual(second, first)
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.valuation({"date": SAMPLE_DAY, "code": SAMPLE_CODE})
+        row = next(item for item in payload["data"] if item["ts_code"] == SAMPLE_CODE)
+        self.assertEqual(row["turnover_rate"], 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(progress["detail"], "已追平")
+        audit = db.fetchone(
+            "SELECT * FROM audit_log WHERE action='revision-review' ORDER BY id DESC"
+        )
+        self.assertIn("1.3572", str(audit["detail"]))
+        self.assertIn(SAMPLE_CODE, str(audit["detail"]))
+
+    def test_empty_or_failed_upstream_keeps_previous_version(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        batches = self._batch_ids(db, TRADE_DATE)
+
+        transport.empty_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+        transport.empty_daily_basic = False
+        transport.fail_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 30))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), batches)
+
+    def test_quality_gate_rejects_catchup_and_keeps_previous(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        transport.turnover_by_code[SAMPLE_CODE] = 9.9999
+        transport.null_volume_ratio = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+    def test_repeat_ticks_after_align_do_not_republish(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        after_fix = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 10))  # inside interval
+        self.assertEqual(len(db.fetchall("SELECT * FROM job_runs WHERE job_id='eod_revise'")), 1)
+        sched.tick(clock_at(TRADE_DATE, 20, 30))  # next light compare, no change
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), after_fix)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_restart_catches_up_inside_window(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(TRADE_DATE, 21, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_cutoff_stops_evening_reviews_and_morning_catchup_runs(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        sched.tick(clock_at(SAMPLE_DAY, 23, 25))  # past 23:20 cutoff, no review yet
+        cutoff = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(cutoff["state"], "cutoff")
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(SAMPLE_DAY, 23, 50))  # still same calendar day, no catch-up
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        ran = sched.tick(clock_at(NEXT_DAY, 8, 45))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(int(progress["catchup_done"]), 1)
+
+        batches = self._batch_ids(db, SAMPLE_DAY)
+        sched.tick(clock_at(NEXT_DAY, 8, 50))
+        self.assertEqual(self._batch_ids(db, SAMPLE_DAY), batches)
+
+    def test_only_valuation_is_light_fetched(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = [name for name in transport.calls]
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        extra = transport.calls[len(before):]
+        self.assertIn("daily_basic", extra)
+        self.assertNotIn("daily", extra)
+        self.assertNotIn("moneyflow", extra)
+        self.assertNotIn("stk_auction", extra)
+        self.assertNotIn("index_daily", extra)
+
+
+if __name__ == "__main__":
+    unittest.main()