diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index 938a92f..df86306 100644 --- a/backend/data/datahub/bridge.py +++ b/backend/data/datahub/bridge.py @@ -21,6 +21,26 @@ from backend.data.providers.tushare_client import TushareClient LOGGER = logging.getLogger("xiaobai.datahub") ShadowSink = Callable[[dict[str, Any]], None] + + +def _usable_intraday_points(rows: list[Any]) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for row in rows: + if not isinstance(row, dict): + continue + try: + close = float(row.get("close") or 0) + except (TypeError, ValueError): + close = 0.0 + if close <= 0: + continue + point = dict(row) + if "average" not in point and point.get("avg_price") is not None: + point["average"] = point.get("avg_price") + points.append(point) + return points + + EMPTY_FAIL_DATASETS = { "stocks", "daily", "index_daily", "valuation", "moneyflow", "auction", "limit_events", "sector_daily", @@ -91,6 +111,34 @@ class DatahubBridge: self._log_failure("status", exc) return None + def try_intraday(self, code: str) -> dict[str, Any] | None: + flags = self.settings.flags("intraday") + if not flags.read: + return None + try: + response = self.client.intraday_points(code=code) + data = response.data + if not isinstance(data, dict): + raise DatahubError("EMPTY", "datahub intraday payload invalid") + points = _usable_intraday_points(data.get("points") or []) + if not points: + raise DatahubError("EMPTY", "datahub intraday empty") + if (response.meta or {}).get("stale"): + raise DatahubError("STALE", "datahub intraday stale") + return { + "entity_type": str(data.get("entity_type") or "stock"), + "identifier": str(data.get("identifier") or code), + "name": str(data.get("name") or ""), + "code": str(data.get("code") or code), + "trade_date": str(data.get("trade_date") or points[-1].get("date") or ""), + "previous_close": float(data.get("previous_close") or 0), + "points": points, + "source": "datahub", + } + except Exception as exc: + self._log_failure("intraday", exc) + return None + def query( self, api_name: str, diff --git a/backend/data/gateway.py b/backend/data/gateway.py index ac577a2..4abdac5 100644 --- a/backend/data/gateway.py +++ b/backend/data/gateway.py @@ -85,12 +85,13 @@ def build_data_gateway( policy = DataSourcePolicy.load() settings = datahub_settings or DatahubSettings.load(credentials=credentials) datahub_client = DatahubClient(settings) + datahub = DatahubBridge(settings, datahub_client) return DataGateway( policy=policy, quality=DataQualityGate.load(policy), tushare_provider=TushareProvider(token_supplier), ifind_provider=IfindProvider(ifind), - chart_data=MarketChartClient(ifind, EastmoneyChartClient()), + chart_data=MarketChartClient(ifind, EastmoneyChartClient(), datahub), realtime_observer=WebRealtimeAggregator(), - datahub=DatahubBridge(settings, datahub_client), + datahub=datahub, ) diff --git a/backend/features/market/charts.py b/backend/features/market/charts.py index 8b40c94..1b1dcbb 100644 --- a/backend/features/market/charts.py +++ b/backend/features/market/charts.py @@ -2,6 +2,7 @@ from __future__ import annotations import http.client import json +import logging import re import time import urllib.error @@ -15,12 +16,15 @@ from typing import Any, ClassVar from backend.bootstrap.config import tushare_code as _stock_market_code from backend.data.providers.ifind_client import IfindError, IfindHttpClient +LOGGER = logging.getLogger("xiaobai.charts") + class ChartDataError(RuntimeError): pass TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get" +HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get" BOARD_LIST_URL = "https://push2delay.eastmoney.com/api/qt/clist/get" BROWSER_USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " @@ -37,14 +41,23 @@ INDEX_SECIDS = { class MarketChartClient: """Prefer iFinD for display charts and retain Eastmoney as a last resort.""" - def __init__(self, ifind: IfindHttpClient, fallback: "EastmoneyChartClient") -> None: + def __init__( + self, + ifind: IfindHttpClient, + fallback: "EastmoneyChartClient", + datahub: Any = None, + ) -> None: self.ifind = ifind self.fallback = fallback + self.datahub = datahub def stock_intraday(self, code: str) -> dict[str, Any]: normalized = str(code or "").strip() if not re.fullmatch(r"\d{6}", normalized): raise ChartDataError("Invalid stock code") + hub_chart = self._datahub_intraday(normalized) + if hub_chart is not None: + return hub_chart ifind_code = _stock_market_code(normalized) try: return self._ifind_intraday(ifind_code, "stock", normalized) @@ -73,11 +86,29 @@ class MarketChartClient: normalized = str(identifier or "").strip().upper() if normalized not in INDEX_SECIDS: raise ChartDataError("Unsupported index") + hub_chart = self._datahub_intraday(normalized) + if hub_chart is not None: + return hub_chart try: return self._ifind_intraday(normalized, "index", normalized) except (IfindError, ChartDataError): return self.fallback.index_intraday(normalized) + def _datahub_intraday(self, code: str) -> dict[str, Any] | None: + if self.datahub is None: + return None + try: + chart = self.datahub.try_intraday(code) + except Exception as exc: + LOGGER.warning("datahub intraday unexpected error: %s", exc) + return None + if not chart: + return None + points = list(chart.get("points") or []) + if not points: + return None + return chart + def board_intraday(self, identifier: str, name: str = "") -> dict[str, Any]: normalized = str(identifier or "").strip().upper() try: @@ -305,21 +336,29 @@ class EastmoneyChartClient: if cached is not None: return cached - payload = self._request_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", - }, - "https://quote.eastmoney.com/", - ) - data = payload.get("data") or {} - points = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))] + 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", + } + last_error: Exception | None = None + data: dict[str, Any] = {} + points: list[dict[str, Any]] = [] + for url, ndays in ((TRENDS_URL, "1"), (TRENDS_URL, "5"), (HIS_TRENDS_URL, "5")): + request_params = {**params, "ndays": ndays} + try: + payload = self._request_json(url, request_params, "https://quote.eastmoney.com/") + except ChartDataError 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 = _latest_session(parsed) + if points: + break if not points: - raise ChartDataError("No intraday chart data returned") + raise ChartDataError("No intraday chart data returned") from last_error result = { "entity_type": entity_type, @@ -433,6 +472,15 @@ class EastmoneyChartClient: raise ChartDataError("Intraday chart request failed") from last_error +def _latest_session(points: list[dict[str, Any]]) -> list[dict[str, Any]]: + if not points: + return [] + 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: fields = str(raw or "").split(",") if len(fields) < 8 or " " not in fields[0]: diff --git a/tests/test_chart_data_provider.py b/tests/test_chart_data_provider.py index db4cc22..a40c956 100644 --- a/tests/test_chart_data_provider.py +++ b/tests/test_chart_data_provider.py @@ -2,7 +2,8 @@ from __future__ import annotations import unittest -from backend.features.market.charts import ChartDataError, EastmoneyChartClient +from backend.data.providers.ifind_client import IfindHttpClient +from backend.features.market.charts import ChartDataError, EastmoneyChartClient, HIS_TRENDS_URL, MarketChartClient, TRENDS_URL from server import DashboardService @@ -72,6 +73,135 @@ class ChartDataProviderTests(unittest.TestCase): self.client.stock_intraday("abc") +class LookbackChartClient(EastmoneyChartClient): + def __init__(self) -> None: + super().__init__(cache_ttl_seconds=20) + self.requests: list[tuple[str, dict[str, str]]] = [] + + def _request_json(self, url, params, referer): + self.requests.append((url, params)) + if url == TRENDS_URL and params.get("ndays") == "1": + return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}} + if url == TRENDS_URL and params.get("ndays") == "5": + return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}} + if url == HIS_TRENDS_URL: + return { + "data": { + "code": "601318", + "name": "中国平安", + "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 ChartDataError("unexpected url") + + +class ChartLookbackTests(unittest.TestCase): + def setUp(self) -> None: + EastmoneyChartClient._cache.clear() + self.client = LookbackChartClient() + + def test_empty_today_falls_back_to_latest_available_session(self): + payload = self.client.stock_intraday("601318") + urls = [url for url, _ in self.client.requests] + self.assertEqual(urls[0], TRENDS_URL) + self.assertEqual(self.client.requests[0][1]["ndays"], "1") + self.assertEqual(urls[1], TRENDS_URL) + self.assertEqual(self.client.requests[1][1]["ndays"], "5") + self.assertEqual(urls[2], HIS_TRENDS_URL) + 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_delay_multiday_can_recover_without_his(self): + class DelayFive(EastmoneyChartClient): + def __init__(self): + super().__init__(cache_ttl_seconds=20) + self.requests = [] + + def _request_json(self, url, params, referer): + self.requests.append((url, params)) + if params.get("ndays") == "1": + return {"data": {"code": "000001", "name": "平安银行", "preClose": 11.7, "trends": []}} + return { + "data": { + "code": "000001", + "name": "平安银行", + "preClose": 11.5, + "trends": [ + "2026-09-07 09:30,11.50,11.60,11.70,11.40,100,1160.00,11.600", + "2026-09-07 15:00,11.70,11.80,11.90,11.60,200,2360.00,11.750", + ], + } + } + + EastmoneyChartClient._cache.clear() + client = DelayFive() + payload = client.stock_intraday("000001") + self.assertEqual(payload["trade_date"], "2026-09-07") + self.assertEqual(len(payload["points"]), 2) + self.assertEqual([url for url, _ in client.requests], [TRENDS_URL, TRENDS_URL]) + + def test_sh_sz_cyb_codes_use_correct_secid(self): + for code, secid in (("601318", "1.601318"), ("000001", "0.000001"), ("300750", "0.300750")): + EastmoneyChartClient._cache.clear() + client = LookbackChartClient() + client.stock_intraday(code) + self.assertEqual(client.requests[0][1]["secid"], secid) + + +class FakeHub: + def __init__(self, chart=None, error=None): + self.chart = chart + self.error = error + self.calls: list[str] = [] + + def try_intraday(self, code): + self.calls.append(code) + if self.error: + raise self.error + return self.chart + + +class DatahubChartFallbackTests(unittest.TestCase): + def setUp(self) -> None: + EastmoneyChartClient._cache.clear() + + def test_datahub_success_skips_old_channel(self): + hub = FakeHub( + { + "entity_type": "stock", + "identifier": "601318", + "name": "中国平安", + "code": "601318", + "trade_date": "2026-09-08", + "previous_close": 56.36, + "points": [{"date": "2026-09-08", "time": "09:30", "close": 56.5, "average": 56.4}], + "source": "datahub", + } + ) + fallback = LookbackChartClient() + client = MarketChartClient(IfindHttpClient(), fallback, hub) + payload = client.stock_intraday("601318") + self.assertEqual(payload["source"], "datahub") + self.assertEqual(hub.calls, ["601318"]) + self.assertEqual(fallback.requests, []) + + def test_datahub_timeout_or_empty_falls_back_to_eastmoney(self): + fallback = LookbackChartClient() + for hub in (FakeHub(chart=None), FakeHub(error=RuntimeError("timeout"))): + EastmoneyChartClient._cache.clear() + client = MarketChartClient(IfindHttpClient(), fallback, hub) + payload = client.stock_intraday("000001") + self.assertEqual(payload["trade_date"], "2026-09-07") + self.assertGreaterEqual(len(payload["points"]), 1) + self.assertTrue(fallback.requests) + + class ChartServiceStub: @staticmethod def _payload(code: str, name: str): diff --git a/tests/test_datahub_bridge.py b/tests/test_datahub_bridge.py index 3949769..9cb16e9 100644 --- a/tests/test_datahub_bridge.py +++ b/tests/test_datahub_bridge.py @@ -349,6 +349,55 @@ class DatahubBridgeTests(unittest.TestCase): self.assertEqual(rows[0]["amount"], 2000.0) self.assertEqual(len(legacy.calls), 1) + def test_try_intraday_respects_switch_and_falls_back_on_bad_payload(self) -> None: + closed = DatahubBridge(flags(), FakeClient(error=DatahubError("INTERNAL", "should not run"))) + self.assertIsNone(closed.try_intraday("601318")) + + empty = DatahubBridge( + flags(intraday=(True, False)), + FakeClient(response=DatahubResponse(data={"points": []}, meta={"stale": False})), + ) + self.assertIsNone(empty.try_intraday("601318")) + + stale = DatahubBridge( + flags(intraday=(True, False)), + FakeClient(response=DatahubResponse( + data={ + "entity_type": "stock", + "code": "601318", + "trade_date": "2026-09-07", + "previous_close": 55.8, + "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9, "avg_price": 55.85}], + }, + meta={"stale": True}, + )), + ) + self.assertIsNone(stale.try_intraday("601318")) + + ok = DatahubBridge( + flags(intraday=(True, False)), + FakeClient(response=DatahubResponse( + data={ + "entity_type": "stock", + "identifier": "601318", + "name": "中国平安", + "code": "601318", + "trade_date": "2026-09-08", + "previous_close": 56.36, + "points": [ + {"date": "2026-09-08", "time": "09:30", "close": 0}, + {"date": "2026-09-08", "time": "09:31", "close": 56.5, "avg_price": 56.4}, + ], + }, + meta={"stale": False}, + )), + ) + chart = ok.try_intraday("601318") + self.assertEqual(chart["source"], "datahub") + self.assertEqual(len(chart["points"]), 1) + self.assertEqual(chart["points"][0]["average"], 56.4) + self.assertEqual(ok.client.paths, ["/v1/intraday/points"]) + def test_features_do_not_import_datahub_client(self) -> None: violations = [] for path in (ROOT / "backend" / "features").rglob("*.py"): diff --git a/xiaobai-datahub/datahub/adapters/eastmoney.py b/xiaobai-datahub/datahub/adapters/eastmoney.py index 32b0bc7..01bf681 100644 --- a/xiaobai-datahub/datahub/adapters/eastmoney.py +++ b/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])), diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index ed883f1..5641105 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/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: diff --git a/xiaobai-datahub/tests/test_realtime_intraday.py b/xiaobai-datahub/tests/test_realtime_intraday.py new file mode 100644 index 0000000..0537507 --- /dev/null +++ b/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()