fix(HEL-482): 开盘前分时回退最近交易日，并接通中枢失败回旧通道

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

xiaobai-datahub/datahub/adapters/eastmoney.py

@@ -14,6 +14,7 @@ from datahub.numbers import finite_number, round4
 EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
 EASTMONEY_CLIST_URL = "https://push2.eastmoney.com/api/qt/clist/get"
 TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
+HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get"
 BROWSER_UA = (
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
     "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
@@ -166,7 +167,7 @@ class EastmoneyAdapter(MarketAdapter):
             )
         return result
 
-    def fetch_intraday(self, ts_code: str) -> dict[str, Any]:
+    def fetch_intraday(self, ts_code: str, date: str = "") -> dict[str, Any]:
         code = str(ts_code or "").upper()
         if code in INDEX_SECIDS:
             secid = INDEX_SECIDS[code]
@@ -178,25 +179,32 @@ class EastmoneyAdapter(MarketAdapter):
             secid = f"{market}.{symbol}"
             entity = "stock"
             identifier = symbol
-        payload = self._get_json(
-            TRENDS_URL,
-            {
-                "secid": secid,
-                "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
-                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
-                "iscr": "0",
-                "ndays": "1",
-            },
-            referer="https://quote.eastmoney.com/",
-        )
-        data = payload.get("data") or {}
-        points = []
-        for raw in data.get("trends") or []:
-            point = _parse_trend(raw)
-            if point:
-                points.append(point)
+        params = {
+            "secid": secid,
+            "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
+            "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
+            "iscr": "0",
+        }
+        data: dict[str, Any] = {}
+        points: list[dict[str, Any]] = []
+        last_error: Exception | None = None
+        for url, ndays in ((TRENDS_URL, "1"), (TRENDS_URL, "5"), (HIS_TRENDS_URL, "5")):
+            try:
+                payload = self._get_json(
+                    url,
+                    {**params, "ndays": ndays},
+                    referer="https://quote.eastmoney.com/",
+                )
+            except AdapterError as exc:
+                last_error = exc
+                continue
+            data = payload.get("data") or {}
+            parsed = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))]
+            points = _preferred_session(parsed, date)
+            if points:
+                break
         if not points:
-            raise AdapterError("No intraday chart data returned")
+            raise AdapterError("No intraday chart data returned") from last_error
         return {
             "entity_type": entity,
             "identifier": identifier,
@@ -227,6 +235,23 @@ class EastmoneyAdapter(MarketAdapter):
             raise AdapterError(f"eastmoney request failed: {exc}") from exc
 
 
+def _preferred_session(points: list[dict[str, Any]], preferred_date: str = "") -> list[dict[str, Any]]:
+    if not points:
+        return []
+    want = ""
+    digits = str(preferred_date or "").replace("-", "")[:8]
+    if len(digits) == 8 and digits.isdigit():
+        want = f"{digits[:4]}-{digits[4:6]}-{digits[6:8]}"
+    if want:
+        matched = [point for point in points if str(point.get("date") or "") == want]
+        if matched:
+            return matched
+    latest = max(str(point.get("date") or "") for point in points)
+    if not latest:
+        return points
+    return [point for point in points if str(point.get("date") or "") == latest]
+
+
 def _parse_trend(raw: Any) -> dict[str, Any] | None:
     text = str(raw or "")
     parts = text.split(",")
@@ -237,11 +262,14 @@ def _parse_trend(raw: Any) -> dict[str, Any] | None:
         when = datetime.strptime(stamp, "%Y-%m-%d %H:%M")
     except ValueError:
         return None
+    close = round4(finite_number(parts[2]))
+    if close <= 0:
+        return None
     return {
         "time": when.strftime("%H:%M"),
         "date": when.strftime("%Y-%m-%d"),
         "open": round4(finite_number(parts[1])),
-        "close": round4(finite_number(parts[2])),
+        "close": close,
         "high": round4(finite_number(parts[3])),
         "low": round4(finite_number(parts[4])),
         "avg_price": round4(finite_number(parts[7] if len(parts) > 7 else parts[2])),

xiaobai-datahub/datahub/realtime_serve.py

@@ -14,6 +14,7 @@ from datahub.adapters.eastmoney import EastmoneyAdapter
 from datahub.adapters.tencent import TencentAdapter
 from datahub.codes import resolve_code
 from datahub.db import HubDB
+from datahub.governance.lkg import LastKnownGood
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 QUOTE_TTL = 60
@@ -108,10 +109,13 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
         return cached
     adapter = EastmoneyAdapter()
     try:
-        payload_data = adapter.fetch_intraday(ts_code)
+        payload_data = adapter.fetch_intraday(ts_code, date)
         source = "eastmoney:trends2"
     except Exception as exc:
-        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
+        recovered = _load_intraday_lkg(db, ts_code, date)
+        if recovered is None:
+            raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
+        return recovered
     payload = _envelope(
         payload_data,
         {
@@ -127,6 +131,47 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
     return payload
 
 
+def _load_intraday_lkg(db: HubDB, ts_code: str, date: str = "") -> dict[str, Any] | None:
+    store = LastKnownGood(db)
+    keys = [f"intraday:{ts_code}:{date or 'today'}"]
+    if date:
+        keys.append(f"intraday:{ts_code}:today")
+    for key in keys:
+        item = store.load(key)
+        payload = _lkg_payload(item)
+        if payload is not None:
+            return payload
+    row = db.fetchone(
+        "SELECT * FROM last_known_good WHERE cache_key LIKE ? ORDER BY stored_at DESC LIMIT 1",
+        (f"intraday:{ts_code}:%",),
+    )
+    if not row:
+        return None
+    try:
+        raw = json.loads(row["payload"])
+    except json.JSONDecodeError:
+        return None
+    return _mark_stale(raw) if isinstance(raw, dict) else None
+
+
+def _lkg_payload(item: dict[str, Any] | None) -> dict[str, Any] | None:
+    if not item:
+        return None
+    payload = item.get("payload")
+    return _mark_stale(payload) if isinstance(payload, dict) else None
+
+
+def _mark_stale(payload: dict[str, Any]) -> dict[str, Any] | None:
+    data = payload.get("data")
+    if not isinstance(data, dict) or not data.get("points"):
+        return None
+    stamped = dict(payload)
+    meta = dict(stamped.get("meta") or {})
+    meta["stale"] = True
+    stamped["meta"] = meta
+    return stamped
+
+
 def _guess_ts_code(code: str) -> str | None:
     raw = str(code or "").strip().upper()
     if "." in raw:

xiaobai-datahub/tests/test_realtime_intraday.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from datahub.adapters.base import AdapterError
+from datahub.adapters.eastmoney import HIS_TRENDS_URL, TRENDS_URL, EastmoneyAdapter
+from datahub.db import HubDB
+from datahub.realtime_serve import fetch_intraday
+
+
+class FakeEastmoney(EastmoneyAdapter):
+    def __init__(self) -> None:
+        super().__init__(timeout=2)
+        self.urls: list[str] = []
+
+    def _get_json(self, url, params, referer):
+        self.urls.append(f"{url}|{params.get('ndays')}")
+        if url == TRENDS_URL:
+            return {"data": {"name": "中国平安", "code": "601318", "preClose": 56.36, "trends": []}}
+        if url == HIS_TRENDS_URL:
+            return {
+                "data": {
+                    "name": "中国平安",
+                    "code": "601318",
+                    "preClose": 55.8,
+                    "trends": [
+                        "2026-09-07 09:30,55.80,55.90,56.00,55.70,100,5580.00,55.900",
+                        "2026-09-07 15:00,56.10,56.20,56.30,56.00,200,11240.00,56.150",
+                        "2026-09-08 09:30,0,0,0,0,0,0.00,0",
+                    ],
+                }
+            }
+        raise AdapterError(f"unexpected url {url}")
+
+
+class EastmoneyIntradayLookbackTests(unittest.TestCase):
+    def test_empty_today_uses_latest_available_session(self):
+        adapter = FakeEastmoney()
+        payload = adapter.fetch_intraday("601318.SH")
+        self.assertEqual(adapter.urls, [f"{TRENDS_URL}|1", f"{TRENDS_URL}|5", f"{HIS_TRENDS_URL}|5"])
+        self.assertEqual(payload["trade_date"], "2026-09-07")
+        self.assertEqual([point["time"] for point in payload["points"]], ["09:30", "15:00"])
+        self.assertEqual(payload["points"][0]["close"], 55.9)
+
+    def test_preferred_date_keeps_that_session(self):
+        adapter = FakeEastmoney()
+        payload = adapter.fetch_intraday("601318.SH", "20260907")
+        self.assertEqual(payload["trade_date"], "2026-09-07")
+        self.assertEqual(len(payload["points"]), 2)
+
+
+class IntradayLkgTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = HubDB(Path(self.tmp.name) / "hub.db")
+
+    def tearDown(self) -> None:
+        self.tmp.cleanup()
+
+    def test_source_failure_returns_last_known_good(self):
+        from datahub.realtime_serve import _envelope, _write_cache
+
+        payload = _envelope(
+            {
+                "entity_type": "stock",
+                "ts_code": "601318.SH",
+                "trade_date": "2026-09-07",
+                "previous_close": 55.8,
+                "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}],
+            },
+            {
+                "tier": "provisional",
+                "trade_date": "20260907",
+                "source": "eastmoney:trends2",
+                "stale": False,
+            },
+        )
+        _write_cache(self.db, "intraday:601318.SH:today", payload, 20, "eastmoney:trends2")
+        self.db.execute(
+            "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?",
+            ("2000-01-01T00:00:00+08:00", "intraday:601318.SH:today"),
+        )
+
+        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
+            mocked.return_value.fetch_intraday.side_effect = AdapterError("down")
+            recovered = fetch_intraday(self.db, "601318.SH")
+        self.assertTrue(recovered["meta"]["stale"])
+        self.assertEqual(recovered["data"]["points"][0]["close"], 55.9)
+
+    def test_source_failure_without_lkg_raises(self):
+        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
+            mocked.return_value.fetch_intraday.side_effect = AdapterError("down")
+            with self.assertRaises(Exception) as ctx:
+                fetch_intraday(self.db, "000001.SZ")
+        self.assertIn("intraday unavailable", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()