feat(HEL-490): 剩余行情改由数据中枢主线路提供

正式页面以 8766 为主线路，旧接口只作故障备用；compose 钉死全部 DATAHUB_READ_*，避免现网残留 0 造成假完成。

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

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,106 @@ 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"},
+                )
+            self._validate_usable(dataset, list(response.data or []), response)
+            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 +282,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"))
@@ -298,11 +407,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,6 +422,75 @@ 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
@@ -325,5 +504,17 @@ class DatahubAwareTushareClient:
     ) -> list[dict[str, Any]]:
         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)