From ef13d6feb577a5deaf1958c43925bb29c26b2b50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Tue, 8 Sep 2026 17:38:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(HEL-494):=20=E7=9B=98=E4=B8=AD=E5=8F=8C?= =?UTF-8?q?=E5=85=8D=E8=B4=B9=E6=BA=90=E6=95=85=E9=9A=9C=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E5=B9=B6=E7=A6=81=E6=AD=A2=E9=97=AE=E5=A4=A9=E5=81=870?= =?UTF-8?q?=E8=A6=86=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主源东财失败后自动改走腾讯行情，成功结果写入缓存；两源都失败时返回最近真实快照并标明延迟，不再显示假0。 Co-authored-by: Cursor Co-authored-by: multica-agent --- backend/data/datahub/bridge.py | 21 +- backend/data/providers/tushare_client.py | 2 + backend/data/providers/tushare_dashboard.py | 5 +- backend/data/providers/tushare_industries.py | 182 +++++++++++++++--- backend/data/realtime.py | 14 +- backend/features/heaven/market_context.py | 28 ++- backend/features/heaven/trend.py | 5 +- config/architecture-inventory.json | 32 +-- tests/test_hel494_regressions.py | 68 +++++++ tests/test_realtime_dashboard.py | 10 + xiaobai-datahub/datahub/adapters/tencent.py | 121 +++++++++++- xiaobai-datahub/datahub/realtime_serve.py | 140 +++++++++++--- .../tests/test_realtime_intraday.py | 52 ++++- 13 files changed, 588 insertions(+), 92 deletions(-) diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index 8cc1967..eb96ee0 100644 --- a/backend/data/datahub/bridge.py +++ b/backend/data/datahub/bridge.py @@ -240,9 +240,17 @@ class DatahubBridge: 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")) + stale = bool((response.meta or {}).get("stale")) + delay = int((response.meta or {}).get("staleness_seconds") or 0) + notice = str((response.meta or {}).get("delay_notice") or "") + source = str((response.meta or {}).get("source") or "datahub") + if stale: + for item in rows: + item["delayed"] = True + item["delay_seconds"] = delay + item["delay_notice"] = notice + item["source"] = source + self._record_route(dataset, "datahub", source) return rows except Exception as exc: self._log_failure(dataset, exc) @@ -444,7 +452,7 @@ def _native_quote(row: dict[str, Any]) -> dict[str, Any] | None: 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 { + payload = { "ts_code": ts_code, "name": str(row.get("name") or ts_code).strip(), "pre_close": previous, @@ -458,6 +466,11 @@ def _native_quote(row: dict[str, Any]) -> dict[str, Any] | None: "quote_date": yyyymmdd(row.get("quote_date") or row.get("trade_date")), "source": str(row.get("source") or "datahub"), } + if row.get("delayed"): + payload["delayed"] = True + payload["delay_seconds"] = int(row.get("delay_seconds") or 0) + payload["delay_notice"] = str(row.get("delay_notice") or "") + return payload def _chart_bars(rows: list[Any]) -> list[dict[str, Any]]: diff --git a/backend/data/providers/tushare_client.py b/backend/data/providers/tushare_client.py index 8b8d5b5..ced9a5b 100644 --- a/backend/data/providers/tushare_client.py +++ b/backend/data/providers/tushare_client.py @@ -66,3 +66,5 @@ class TushareClient( _stock_listing_lock: ClassVar[Lock] = Lock() _suspension_cache: ClassVar[dict[str, dict[str, str] | None]] = {} _suspension_lock: ClassVar[Lock] = Lock() + _sw_member_cache: ClassVar[dict[str, Any]] = {} + _sw_member_lock: ClassVar[Lock] = Lock() diff --git a/backend/data/providers/tushare_dashboard.py b/backend/data/providers/tushare_dashboard.py index 70f54d8..2e9130c 100644 --- a/backend/data/providers/tushare_dashboard.py +++ b/backend/data/providers/tushare_dashboard.py @@ -293,7 +293,10 @@ class DashboardMixin: last_error = str(exc) code_list = [item for item in str(codes or "").split(",") if item] try: - quotes = aggregator.tencent_market_quotes(code_list, expected_date=trade_date) + if code_list: + quotes = aggregator.tencent_stock_quotes(code_list, expected_date=trade_date) + else: + quotes = aggregator.tencent_market_quotes(code_list, expected_date=trade_date) except Exception as exc: raise TushareError( f"eastmoney={last_error or 'empty'}；tencent={exc}" diff --git a/backend/data/providers/tushare_industries.py b/backend/data/providers/tushare_industries.py index 67680a8..143d661 100644 --- a/backend/data/providers/tushare_industries.py +++ b/backend/data/providers/tushare_industries.py @@ -1,11 +1,16 @@ from __future__ import annotations +import json from datetime import datetime, timedelta +from pathlib import Path from typing import Any from backend.data.numbers import finite_number as _number from backend.data.providers.tushare_transport import TushareError +_SW_MEMBER_TTL = timedelta(hours=24) +_SW_MEMBER_DIR = Path(__file__).resolve().parents[3] / "data" / "cache" / "sw_members" + class ShenwanIndustryMixin: def sw_stock_industry(self, ts_code: str, trade_date: str) -> dict[str, Any]: @@ -239,23 +244,100 @@ class ShenwanIndustryMixin: sector_code: str, trade_date: str, ) -> list[dict[str, Any]]: - rows = [] - for is_new in ("Y", "N"): - rows.extend( - self.query( - "index_member_all", - {"l2_code": sector_code, "is_new": is_new}, - "l2_code,l2_name,ts_code,name,in_date,out_date,is_new", + cached_rows = self._read_local_sw_members(sector_code) + if cached_rows is not None: + return _active_members(cached_rows, trade_date) + rows: list[dict[str, Any]] = [] + try: + for is_new in ("Y", "N"): + rows.extend( + self.query( + "index_member_all", + {"l2_code": sector_code, "is_new": is_new}, + "l2_code,l2_name,ts_code,name,in_date,out_date,is_new", + ) ) + except TushareError: + stale = self._read_local_sw_members(sector_code, allow_stale=True) or [] + if stale: + return _active_members(stale, trade_date) + raise + reconciled = _reconcile_membership_rows(rows) + self._write_local_sw_members(sector_code, reconciled) + return _active_members(reconciled, trade_date) + + def _read_local_sw_members( + self, + sector_code: str, + allow_stale: bool = False, + ) -> list[dict[str, Any]] | None: + now = datetime.now().astimezone() + cache = getattr(self, "_sw_member_cache", None) + lock = getattr(self, "_sw_member_lock", None) + if isinstance(cache, dict) and lock is not None: + with lock: + packed = cache.get(sector_code) + if isinstance(packed, dict): + loaded_at = packed.get("loaded_at") + rows = packed.get("rows") + fresh = ( + isinstance(loaded_at, datetime) + and now - loaded_at < _SW_MEMBER_TTL + ) + if isinstance(rows, list) and (fresh or allow_stale): + return [dict(item) for item in rows] + path = _sw_member_path(sector_code) + if not path.exists(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + rows = list(payload.get("rows") or []) + updated = str(payload.get("updated_at") or "") + fresh = False + try: + stamped = datetime.fromisoformat(updated) + if stamped.tzinfo is None: + stamped = stamped.replace(tzinfo=now.tzinfo) + fresh = now - stamped.astimezone(now.tzinfo) < _SW_MEMBER_TTL + except ValueError: + fresh = False + if rows and (fresh or allow_stale): + self._remember_sw_members(sector_code, rows) + return rows + return None + + def _write_local_sw_members(self, sector_code: str, rows: list[dict[str, Any]]) -> None: + packed = [dict(item) for item in rows] + self._remember_sw_members(sector_code, packed) + path = _sw_member_path(sector_code) + try: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "sector_code": sector_code, + "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "rows": packed, + }, + ensure_ascii=False, + ), + encoding="utf-8", ) - deduped: dict[str, dict[str, Any]] = {} - for row in _reconcile_membership_rows(rows): - code = str(row.get("ts_code") or "") - if code and _membership_active_on(row, trade_date): - current = deduped.get(code) - if current is None or str(row.get("in_date") or "") > str(current.get("in_date") or ""): - deduped[code] = row - return list(deduped.values()) + except OSError: + pass + + def _remember_sw_members(self, sector_code: str, rows: list[dict[str, Any]]) -> None: + cache = getattr(self, "_sw_member_cache", None) + lock = getattr(self, "_sw_member_lock", None) + if not isinstance(cache, dict) or lock is None: + return + with lock: + cache[sector_code] = { + "loaded_at": datetime.now().astimezone(), + "rows": [dict(item) for item in rows], + } def sw_sector_members(self, sector_code: str, trade_date: str) -> list[dict[str, Any]]: """Return constituents active in a Shenwan L2 industry on the target date.""" @@ -394,8 +476,12 @@ class ShenwanIndustryMixin: codes = [ str(row.get("ts_code") or "") for row in members - if str(row.get("ts_code") or "") in active_codes + if str(row.get("ts_code") or "") ] + if active_codes: + listed = [code for code in codes if code in active_codes] + if listed: + codes = listed quotes, inner_source = self._load_member_realtime_quotes(codes, trade_date) for row in quotes: close = _number(row.get("close")) @@ -439,12 +525,14 @@ class ShenwanIndustryMixin: try: market_rows = self._ensure_realtime_market_cache(trade_date) except TushareError as exc: - inner_precise = False message = str(exc) if "rt_k" in message or "权限" in message: - inner_error = inner_error or "全市场实时行情暂不可用，已避开无权限接口" + market_error = "全市场实时行情暂不可用，已避开无权限接口" else: - inner_error = inner_error or message + market_error = message + if not valid: + inner_precise = False + inner_error = inner_error or market_error capital_map = { str(item.get("ts_code") or ""): item for item in reference.get("capital_rows") or [] @@ -467,9 +555,14 @@ class ShenwanIndustryMixin: average_turnover = sum(sector_turnovers) / len(sector_turnovers) if sector_turnovers else 0 market_turnover = sum(market_turnovers) / len(market_turnovers) if market_turnovers else 0 relative_turnover = average_turnover / market_turnover if market_turnover else 0 - if not relative_turnover: - inner_precise = False - inner_error = inner_error or "Shenwan member relative turnover is unavailable" + delayed = "delayed" in str(inner_source) or any(item.get("delayed") for item in valid) + delay_seconds = max((int(item.get("delay_seconds") or 0) for item in valid), default=0) + delay_notice = "" + if delayed: + delay_notice = next( + (str(item.get("delay_notice") or "") for item in valid if item.get("delay_notice")), + "", + ) or f"主备免费行情均暂不可用，显示最近一次真实快照（延迟 {delay_seconds} 秒）" return { "code": sector_code, "name": str(industry.get("l2_name") or sw_row.get("name") or ""), @@ -511,6 +604,9 @@ class ShenwanIndustryMixin: "precise": inner_precise and outer_precise, "inner_error": inner_error, "outer_error": outer_error, + "delayed": delayed, + "delay_seconds": delay_seconds, + "delay_notice": delay_notice, "schema_version": 6, "methodology": "外显使用已发布 sw_daily 或免费申万实时；内核使用数据中枢/免费实时成分，不调用 rt_sw_k", } @@ -568,7 +664,8 @@ class ShenwanIndustryMixin: if callable(hub_market): filtered = consider(hub_market(trade_date) or [], "datahub") if len(filtered) >= max(1, int(len(wanted) * 0.9)): - return filtered, "datahub" + delayed = any(item.get("delayed") for item in filtered) + return filtered, "datahub_delayed" if delayed else "datahub" hub = getattr(self, "try_quotes", None) if callable(hub): @@ -577,25 +674,38 @@ class ShenwanIndustryMixin: collected.extend(hub(wanted[index:index + _QUOTE_BATCH]) or []) filtered = consider(collected, "datahub") if len(filtered) >= max(1, int(len(wanted) * 0.9)): - return filtered, "datahub" + delayed = any(item.get("delayed") for item in filtered) + return filtered, "datahub_delayed" if delayed else "datahub" aggregator = getattr(self, "realtime_aggregator", None) - loader = getattr(aggregator, "eastmoney_stock_quotes", None) if aggregator else None - if callable(loader): + eastmoney_loader = getattr(aggregator, "eastmoney_stock_quotes", None) if aggregator else None + if callable(eastmoney_loader): try: - filtered = consider(loader(wanted, expected_date=trade_date) or [], "eastmoney_ulist") + filtered = consider(eastmoney_loader(wanted, expected_date=trade_date) or [], "eastmoney_ulist") if len(filtered) >= max(1, int(len(wanted) * 0.9)): return filtered, "eastmoney_ulist" except Exception: pass + tencent_loader = getattr(aggregator, "tencent_stock_quotes", None) if aggregator else None + if callable(tencent_loader): + try: + filtered = consider(tencent_loader(wanted, expected_date=trade_date) or [], "tencent_qt") + if len(filtered) >= max(1, int(len(wanted) * 0.9)): + return filtered, "tencent_qt" + except Exception: + pass + try: - quotes, source = self._free_realtime_quotes(trade_date, "") + quotes, source = self._free_realtime_quotes(trade_date, ",".join(wanted)) consider(quotes, source) except TushareError: pass if best_rows: + delayed = any(item.get("delayed") for item in best_rows) + if delayed and not str(best_source).endswith("_delayed"): + return best_rows, f"{best_source}_delayed" return best_rows, best_source return [], "unavailable" @@ -718,6 +828,22 @@ def _reconcile_membership_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any return list(reconciled.values()) +def _sw_member_path(sector_code: str) -> Path: + safe = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in str(sector_code or "")) + return _SW_MEMBER_DIR / f"{safe or 'unknown'}.json" + + +def _active_members(rows: list[dict[str, Any]], trade_date: str) -> list[dict[str, Any]]: + deduped: dict[str, dict[str, Any]] = {} + for row in rows: + code = str(row.get("ts_code") or "") + if code and _membership_active_on(row, trade_date): + current = deduped.get(code) + if current is None or str(row.get("in_date") or "") > str(current.get("in_date") or ""): + deduped[code] = dict(row) + return list(deduped.values()) + + def _match_sector_row(rows: list[dict[str, Any]], identifier: str) -> dict[str, Any] | None: if not rows: return None diff --git a/backend/data/realtime.py b/backend/data/realtime.py index c935747..dca8d3f 100644 --- a/backend/data/realtime.py +++ b/backend/data/realtime.py @@ -236,6 +236,15 @@ class WebRealtimeAggregator: self, codes: list[str], expected_date: str = "", + ) -> list[dict[str, Any]]: + quotes = self.tencent_stock_quotes(codes, expected_date="", minimum=200) + return self._filter_quotes_by_date(quotes, expected_date) + + def tencent_stock_quotes( + self, + codes: list[str], + expected_date: str = "", + minimum: int | None = None, ) -> list[dict[str, Any]]: symbols: list[str] = [] seen: set[str] = set() @@ -280,10 +289,11 @@ class WebRealtimeAggregator: quotes.extend(rows) elif not status.get("ok"): errors.append(str(status.get("error") or "batch failed")) - if len(quotes) < 200: + floor = minimum if minimum is not None else max(1, int(len(symbols) * 0.5)) + if len(quotes) < floor: detail = f"；{'; '.join(errors[:3])}" if errors else "" raise RealtimeAggregateError( - f"Tencent market snapshot too small: {len(quotes)}{detail}" + f"Tencent quotes too small: {len(quotes)}/{len(symbols)}{detail}" ) return self._filter_quotes_by_date(quotes, expected_date) diff --git a/backend/features/heaven/market_context.py b/backend/features/heaven/market_context.py index 30f68e9..5e81a19 100644 --- a/backend/features/heaven/market_context.py +++ b/backend/features/heaven/market_context.py @@ -299,6 +299,14 @@ class HeavenMarketContextMixin: and not cached.get("realtime") and int(cached.get("schema_version") or 0) >= 6 ) + cached_quotes = int((cached or {}).get("quote_count") or 0) + cached_lkg = bool( + cached + and cached_date == trade_date + and cached.get("taxonomy") == "sw_l2" + and cached_quotes > 0 + and int(cached.get("schema_version") or 0) >= 6 + ) if market_mode != "intraday" and cached_valid: return cached if not self.configured: @@ -311,8 +319,12 @@ class HeavenMarketContextMixin: allow_realtime_close=market_mode == "closed", ) except TushareError as exc: - if cached_valid: - return cached + if cached_lkg: + delayed = dict(cached) + delayed["delayed"] = True + delayed["delay_notice"] = "主备免费行情均暂不可用，显示最近一次真实快照" + delayed["realtime"] = market_mode == "intraday" + return delayed return { "name": "", "code": "", @@ -323,12 +335,16 @@ class HeavenMarketContextMixin: "precise": False, "inner_precise": False, "outer_precise": False, - "coverage": 0, - "member_count": 0, - "quote_count": 0, "error": f"申万二级行业数据获取失败：{exc}", } - if not payload.get("realtime") and payload.get("precise"): + if int(payload.get("quote_count") or 0) > 0: + self.database.save_data_snapshot( + "heaven_sector", + cache_key, + str(payload.get("source") or "tushare"), + payload, + ) + elif not payload.get("realtime") and payload.get("precise"): self.database.save_data_snapshot( "heaven_sector", cache_key, diff --git a/backend/features/heaven/trend.py b/backend/features/heaven/trend.py index bfc46fc..fab2209 100644 --- a/backend/features/heaven/trend.py +++ b/backend/features/heaven/trend.py @@ -243,6 +243,7 @@ class HeavenTrendMixin: "detail": ( f"申万二级 {sector.get('name') or '--'} {sector.get('code') or '--'} " f"成分覆盖 {int(sector.get('quote_count') or 0)}/{int(sector.get('member_count') or 0)}" + + ("；延迟快照" if sector.get("delayed") or sector.get("delay_notice") else "") ), }, { @@ -341,7 +342,9 @@ class HeavenTrendMixin: issues.append("行业外显缺少申万官方行情") if sector and sector_coverage_issue: issues.append(sector_coverage_issue) - if sector.get("realtime") and not sector.get("relative_turnover"): + if sector.get("delay_notice"): + issues.append(str(sector.get("delay_notice"))) + if sector.get("realtime") and not sector.get("relative_turnover") and not sector.get("delayed"): issues.append("行业内核缺少相对全市场换手活跃度") stock = stock or {} diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 1e639c6..a79c671 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -486,6 +486,11 @@ "bytes": 48447, "lines": 665 }, + { + "path": "backend/data/providers/tushare_industries.py", + "bytes": 38757, + "lines": 892 + }, { "path": "backend/features/screener/catalog.py", "bytes": 35571, @@ -498,13 +503,8 @@ }, { "path": "backend/data/providers/tushare_dashboard.py", - "bytes": 34631, - "lines": 812 - }, - { - "path": "backend/data/providers/tushare_industries.py", - "bytes": 33324, - "lines": 766 + "bytes": 34773, + "lines": 815 }, { "path": "database.py", @@ -538,8 +538,8 @@ }, { "path": "backend/features/heaven/trend.py", - "bytes": 16772, - "lines": 370 + "bytes": 17005, + "lines": 373 }, { "path": "backend/features/market/insights_auction_scoring.py", @@ -573,8 +573,8 @@ }, { "path": "backend/features/heaven/market_context.py", - "bytes": 13687, - "lines": 338 + "bytes": 14409, + "lines": 354 }, { "path": "frontend/shared/session.js", @@ -801,16 +801,16 @@ "bytes": 2299, "lines": 57 }, + { + "path": "backend/data/providers/tushare_client.py", + "bytes": 2263, + "lines": 70 + }, { "path": "backend/features/screener/regime.py", "bytes": 2202, "lines": 53 }, - { - "path": "backend/data/providers/tushare_client.py", - "bytes": 2166, - "lines": 68 - }, { "path": "frontend/pages/popularity/page.html", "bytes": 2165, diff --git a/tests/test_hel494_regressions.py b/tests/test_hel494_regressions.py index 7b2ee27..a4f16b1 100644 --- a/tests/test_hel494_regressions.py +++ b/tests/test_hel494_regressions.py @@ -216,6 +216,44 @@ class MemberQuoteCoverageTests(unittest.TestCase): self.assertEqual(source, "datahub") client.try_quotes.assert_not_called() + def test_eastmoney_failure_uses_tencent_member_quotes(self) -> None: + client = TushareClient(token="demo") + wanted = ["000737.SZ", "000630.SZ"] + client.try_market_quotes = MagicMock(return_value=None) + client.try_quotes = MagicMock(return_value=None) + aggregator = MagicMock() + aggregator.eastmoney_stock_quotes.side_effect = RuntimeError("HTTP 503") + aggregator.tencent_stock_quotes.return_value = [ + {"ts_code": "000737.SZ", "close": 12.3, "pre_close": 11.2}, + {"ts_code": "000630.SZ", "close": 4.5, "pre_close": 4.4}, + ] + client.realtime_aggregator = aggregator + client._free_realtime_quotes = MagicMock(side_effect=AssertionError("tencent already won")) + rows, source = client._load_member_realtime_quotes(wanted, "20260908") + self.assertEqual(len(rows), 2) + self.assertEqual(source, "tencent_qt") + aggregator.tencent_stock_quotes.assert_called_once() + + def test_delayed_hub_quotes_are_kept_not_zeroed(self) -> None: + client = TushareClient(token="demo") + delayed = [ + { + "ts_code": "000737.SZ", + "close": 12.3, + "pre_close": 11.2, + "delayed": True, + "delay_seconds": 90, + "delay_notice": "主备免费行情均暂不可用，显示 90 秒前的真实快照", + } + ] + client.try_market_quotes = MagicMock(return_value=delayed) + client.try_quotes = MagicMock() + client.realtime_aggregator = MagicMock() + rows, source = client._load_member_realtime_quotes(["000737.SZ"], "20260908") + self.assertEqual(source, "datahub_delayed") + self.assertEqual(rows[0]["close"], 12.3) + client.try_quotes.assert_not_called() + def test_ignores_non_member_quotes_from_market_snapshot(self) -> None: client = TushareClient(token="demo") client.try_market_quotes = MagicMock( @@ -231,6 +269,36 @@ class MemberQuoteCoverageTests(unittest.TestCase): ) self.assertEqual([row["ts_code"] for row in rows], ["000737.SZ"]) + def test_local_sw_members_survive_tushare_outage(self) -> None: + import tempfile + from pathlib import Path + + from backend.data.providers import tushare_industries as module + + client = TushareClient(token="demo") + stored = [ + { + "ts_code": "000737.SZ", + "name": "北方铜业", + "l2_code": "801074.SI", + "in_date": "20200101", + "out_date": "", + } + ] + with tempfile.TemporaryDirectory() as tmp: + original = module._SW_MEMBER_DIR + module._SW_MEMBER_DIR = Path(tmp) + try: + client._sw_member_cache.clear() + client._write_local_sw_members("801074.SI", stored) + client.query = MagicMock(side_effect=TushareError("index_member_all down")) + members = client._sw_sector_members("801074.SI", "20260908") + finally: + module._SW_MEMBER_DIR = original + client._sw_member_cache.clear() + self.assertEqual([item["ts_code"] for item in members], ["000737.SZ"]) + client.query.assert_not_called() + def test_closed_keeps_daily_inner_when_sw_daily_missing(self) -> None: client = TushareClient(token="demo") client.resolve_trade_context = lambda _date: ("20260908", "20260907") diff --git a/tests/test_realtime_dashboard.py b/tests/test_realtime_dashboard.py index 3295f5f..1f01ea4 100644 --- a/tests/test_realtime_dashboard.py +++ b/tests/test_realtime_dashboard.py @@ -133,6 +133,13 @@ class FakeFreeAggregator: def tencent_market_quotes(self, codes, expected_date=""): return self.eastmoney_market_quotes(expected_date) + def tencent_stock_quotes(self, codes, expected_date="", minimum=None): + wanted = {str(code).strip() for code in (codes or []) if str(code).strip()} + rows = self.tencent_market_quotes(codes, expected_date) + if not wanted: + return rows + return [row for row in rows if str(row.get("ts_code") or "") in wanted] + def eastmoney_indices(self): return [ { @@ -318,6 +325,9 @@ class RealtimeDashboardTests(unittest.TestCase): def tencent_market_quotes(self, codes, expected_date=""): return list(FREE_QUOTES) + def tencent_stock_quotes(self, codes, expected_date="", minimum=None): + return list(FREE_QUOTES) + self.client.query = query self.client.realtime_aggregator = TencentOnlyAggregator() TushareClient._realtime_reference_cache.clear() diff --git a/xiaobai-datahub/datahub/adapters/tencent.py b/xiaobai-datahub/datahub/adapters/tencent.py index b992e99..4981090 100644 --- a/xiaobai-datahub/datahub/adapters/tencent.py +++ b/xiaobai-datahub/datahub/adapters/tencent.py @@ -1,7 +1,6 @@ from __future__ import annotations import time -import urllib.error import urllib.request from datetime import datetime from typing import Any @@ -10,6 +9,8 @@ from datahub.adapters.base import AdapterError, MarketAdapter from datahub.numbers import finite_number, round4 TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006" +TENCENT_QUOTE_URL = "https://qt.gtimg.cn/q=" +TENCENT_QUOTE_BATCH = 80 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" @@ -45,6 +46,16 @@ class TencentAdapter(MarketAdapter): def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]: if dataset in {"indexes_quotes", "index_quotes"}: return self.fetch_indices() + if dataset in {"quotes", "quotes_latest"}: + codes = params.get("codes") or [] + if isinstance(codes, str): + codes = [item.strip() for item in codes.split(",") if item.strip()] + return self.fetch_quotes(list(codes)) + if dataset in {"quotes_market", "market_quotes"}: + codes = params.get("codes") or [] + if isinstance(codes, str): + codes = [item.strip() for item in codes.split(",") if item.strip()] + return self.fetch_quotes(list(codes)) raise AdapterError(f"{self.name} unsupported dataset: {dataset}") def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: @@ -97,3 +108,111 @@ class TencentAdapter(MarketAdapter): if len(result) != 3: raise AdapterError(f"Tencent returned {len(result)}/3 indices") return result + + def fetch_quotes(self, codes: list[str]) -> list[dict[str, Any]]: + symbols: list[str] = [] + seen: set[str] = set() + for raw in codes: + symbol = _tencent_symbol(str(raw or "")) + if not symbol or symbol in seen: + continue + seen.add(symbol) + symbols.append(symbol) + if not symbols: + return [] + result: list[dict[str, Any]] = [] + errors: list[str] = [] + for index in range(0, len(symbols), TENCENT_QUOTE_BATCH): + batch = symbols[index:index + TENCENT_QUOTE_BATCH] + try: + raw = self._get_text(f"{TENCENT_QUOTE_URL}{','.join(batch)}") + except AdapterError as exc: + errors.append(str(exc)) + continue + for line in raw.splitlines(): + quote = _parse_tencent_stock_quote(line) + if quote: + result.append(quote) + if not result: + detail = f"；{'; '.join(errors[:3])}" if errors else "" + raise AdapterError(f"Tencent quotes empty{detail}") + return result + + def _get_text(self, url: str) -> str: + request = urllib.request.Request( + url, + headers={"User-Agent": BROWSER_UA, "Referer": "https://gu.qq.com/"}, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return response.read().decode("gb18030", errors="ignore") + except Exception as exc: + raise AdapterError(f"tencent request failed: {exc}") from exc + + +def _tencent_symbol(code: str) -> str: + raw = str(code or "").strip().upper() + if not raw: + return "" + symbol = raw.split(".")[0] + if not symbol.isdigit() or len(symbol) != 6: + return "" + if raw.endswith(".SH") or symbol.startswith(("5", "6", "9")): + return f"sh{symbol}" + if raw.endswith(".BJ") or symbol.startswith(("4", "8")): + return f"bj{symbol}" + return f"sz{symbol}" + + +def _parse_tencent_stock_quote(line: str) -> dict[str, Any] | None: + if '="' not in line: + return None + prefix, payload = line.split('="', 1) + fields = payload.rsplit('";', 1)[0].split("~") + if len(fields) < 38: + return None + symbol = str(fields[2] or "") + if not symbol.isdigit() or len(symbol) != 6: + return None + close = round4(finite_number(fields[3])) + previous = round4(finite_number(fields[4])) + if not close or not previous or close <= 0 or previous <= 0: + return None + marker = prefix.lower() + if "sh" in marker: + ts_code = f"{symbol}.SH" + elif "bj" in marker: + ts_code = f"{symbol}.BJ" + else: + ts_code = f"{symbol}.SZ" + quote_stamp = "" + quote_date = "" + epoch = 0 + try: + parsed = datetime.strptime(fields[30], "%Y%m%d%H%M%S") + quote_date = parsed.strftime("%Y%m%d") + epoch = int(parsed.timestamp()) + quote_stamp = parsed.astimezone().isoformat(timespec="seconds") + except ValueError: + pass + return { + "ts_code": ts_code, + "name": fields[1] or symbol, + "price": close, + "close": close, + "pct_chg": round4(finite_number(fields[32])), + "change_amount": round4(finite_number(fields[31])), + "open": round4(finite_number(fields[5])), + "high": round4(finite_number(fields[33])), + "low": round4(finite_number(fields[34])), + "pre_close": previous, + "previous_close": previous, + "volume": round4(finite_number(fields[6]) * 100), + "vol": round4(finite_number(fields[6]) * 100), + "amount": round4(finite_number(fields[37]) * 10000), + "quote_date": quote_date, + "quote_time_epoch": epoch, + "quote_time": quote_stamp, + "source": "tencent_qt", + } diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index 17ec21e..537b247 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -11,6 +11,7 @@ import time from datetime import datetime from typing import Any +from datahub.adapters.base import AdapterError from datahub.adapters.eastmoney import EastmoneyAdapter from datahub.adapters.tencent import TencentAdapter from datahub.codes import resolve_code @@ -71,24 +72,32 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]: cached = _read_cache(db, cache_key) if cached is not None: return cached - adapter = EastmoneyAdapter() + errors: list[str] = [] + rows: list[dict[str, Any]] = [] + source = "" try: - rows = adapter.fetch_market_quotes() + rows = EastmoneyAdapter().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", - }, - ) + errors.append(f"eastmoney:{exc}") + try: + listed = _listed_ts_codes(db) + if not listed: + raise AdapterError("no local stock master for tencent market snapshot") + rows = TencentAdapter().fetch_quotes(listed) + if len(rows) < 200: + raise AdapterError(f"Tencent market snapshot too small: {len(rows)}") + source = "tencent:qt" + except Exception as backup_exc: + errors.append(f"tencent:{backup_exc}") + recovered = _load_quotes_lkg(db, cache_key) + if recovered is not None: + return recovered + raise RealtimeApiError( + "SOURCE_UNAVAILABLE", + "market quotes unavailable: " + "；".join(errors), + ) from backup_exc + payload = _quote_payload(rows, source, scope="market") _write_cache(db, cache_key, payload, QUOTE_TTL, source) return payload @@ -110,29 +119,98 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: cached = _read_cache(db, cache_key) if cached is not None: return cached - adapter = EastmoneyAdapter() + minimum = max(1, int(len(resolved) * 0.5)) + errors: list[str] = [] + rows: list[dict[str, Any]] = [] + source = "" try: - rows: list[dict[str, Any]] = [] - for index in range(0, len(resolved), QUOTE_BATCH): - rows.extend(adapter.fetch_quotes(resolved[index:index + QUOTE_BATCH])) + rows = _eastmoney_named_quotes(resolved) + if len(rows) < minimum: + raise AdapterError(f"Eastmoney named quotes too small: {len(rows)}/{len(resolved)}") source = "eastmoney:ulist" except Exception as exc: - raise RealtimeApiError("SOURCE_UNAVAILABLE", f"quotes unavailable: {exc}") from exc - payload = _envelope( - rows, - { - "tier": "provisional", - "trade_date": yyyymmdd(now_shanghai()), - "source": source, - "stale": False, - "staleness_seconds": 0, - "published_at": isoformat(now_shanghai()), - }, - ) + errors.append(f"eastmoney:{exc}") + try: + rows = TencentAdapter().fetch_quotes(resolved) + if len(rows) < minimum: + raise AdapterError(f"Tencent named quotes too small: {len(rows)}/{len(resolved)}") + source = "tencent:qt" + except Exception as backup_exc: + errors.append(f"tencent:{backup_exc}") + recovered = _load_quotes_lkg(db, cache_key) + if recovered is not None: + return recovered + raise RealtimeApiError( + "SOURCE_UNAVAILABLE", + "quotes unavailable: " + "；".join(errors), + ) from backup_exc + payload = _quote_payload(rows, source) _write_cache(db, cache_key, payload, QUOTE_TTL, source) return payload +def _eastmoney_named_quotes(codes: list[str]) -> list[dict[str, Any]]: + adapter = EastmoneyAdapter() + rows: list[dict[str, Any]] = [] + for index in range(0, len(codes), QUOTE_BATCH): + rows.extend(adapter.fetch_quotes(codes[index:index + QUOTE_BATCH])) + return rows + + +def _listed_ts_codes(db: HubDB) -> list[str]: + try: + rows = db.fetchall( + "SELECT ts_code FROM stock_master WHERE list_status = 'L' ORDER BY ts_code" + ) + except Exception: + return [] + return [str(row.get("ts_code") or "") for row in rows if row.get("ts_code")] + + +def _quote_payload( + rows: list[dict[str, Any]], + source: str, + scope: str = "", +) -> dict[str, Any]: + meta: dict[str, Any] = { + "tier": "provisional", + "trade_date": yyyymmdd(now_shanghai()), + "source": source, + "stale": False, + "staleness_seconds": 0, + "published_at": isoformat(now_shanghai()), + "failover": source.startswith("tencent"), + "delay_notice": "", + } + if scope: + meta["scope"] = scope + return _envelope(rows, meta) + + +def _load_quotes_lkg(db: HubDB, cache_key: str) -> dict[str, Any] | None: + store = LastKnownGood(db) + item = store.load(cache_key) + payload = item.get("payload") if item else None + if not isinstance(payload, dict): + return None + data = payload.get("data") + if not isinstance(data, list) or not data: + return None + stamped = dict(payload) + meta = dict(stamped.get("meta") or {}) + stored = str((item or {}).get("stored_at") or "") + try: + age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp())) + except Exception: + age = QUOTE_TTL + meta["stale"] = True + meta["staleness_seconds"] = age + meta["delay_notice"] = f"主备免费行情均暂不可用，显示 {age} 秒前的真实快照" + meta["lkg_source"] = str((item or {}).get("source") or meta.get("source") or "") + stamped["meta"] = meta + return stamped + + def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]: ts_code = resolve_code(db, code) or _guess_ts_code(code) if not ts_code: diff --git a/xiaobai-datahub/tests/test_realtime_intraday.py b/xiaobai-datahub/tests/test_realtime_intraday.py index 9eee3d9..b049562 100644 --- a/xiaobai-datahub/tests/test_realtime_intraday.py +++ b/xiaobai-datahub/tests/test_realtime_intraday.py @@ -232,9 +232,57 @@ class MarketQuotesTests(unittest.TestCase): self.assertEqual(mocked.return_value.fetch_quotes.call_count, 2) self.assertEqual(len(payload["data"]), 70) + def test_named_quotes_fail_over_to_tencent(self) -> None: + named = [ + { + "ts_code": "000737.SZ", + "name": "北方铜业", + "close": 12.3, + "pre_close": 11.2, + } + ] + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_quotes.side_effect = AdapterError("HTTP 503") + tencent.return_value.fetch_quotes.return_value = named + payload = self.api.handle("/v1/quotes/latest", {"codes": ["000737.SZ"]}) + self.assertEqual(payload["data"][0]["ts_code"], "000737.SZ") + self.assertEqual(payload["meta"]["source"], "tencent:qt") + self.assertTrue(payload["meta"]["failover"]) + self.assertFalse(payload["meta"]["stale"]) + + def test_both_quote_sources_return_last_known_good(self) -> None: + from datahub.realtime_serve import _envelope, _write_cache + + cache_key = "quotes:placeholder:1" + payload = _envelope( + [{"ts_code": "000737.SZ", "close": 12.3, "pre_close": 11.2}], + {"source": "eastmoney:ulist", "stale": False, "trade_date": "20260908"}, + ) + _write_cache(self.db, cache_key, payload, 60, "eastmoney:ulist") + self.db.execute( + "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?", + ("2000-01-01T00:00:00+08:00", cache_key), + ) + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent, patch("datahub.realtime_serve.hashlib.sha1") as sha1: + eastmoney.return_value.fetch_quotes.side_effect = AdapterError("HTTP 503") + tencent.return_value.fetch_quotes.side_effect = AdapterError("timeout") + sha1.return_value.hexdigest.return_value = "placeholder" + recovered = self.api.handle("/v1/quotes/latest", {"codes": ["000737.SZ"]}) + self.assertTrue(recovered["meta"]["stale"]) + self.assertIn("真实快照", recovered["meta"]["delay_notice"]) + self.assertEqual(recovered["data"][0]["close"], 12.3) + self.assertNotEqual(recovered["data"][0]["close"], 0) + 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 patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_market_quotes.side_effect = AdapterError("too small") + tencent.return_value.fetch_quotes.side_effect = AdapterError("empty master") with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/quotes/latest", {}) self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")