From d0d65d64b8c54299d588229897a90195305541d4 Mon Sep 17 00:00:00 2001 From: GitHub Date: Wed, 8 Jul 2026 18:07:37 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20v1.20.0=20=E6=9C=8D=E5=8A=A1=E5=99=A8?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E8=87=AA=E5=8A=A8=20ping=20=E5=88=87?= =?UTF-8?q?=E6=8D=A2=EF=BC=88=E6=97=A0=E9=9C=80=E6=89=8B=E5=8A=A8=20easy-t?= =?UTF-8?q?dx=20ping=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 服务器连不上或返回空数据时，自动测速切到延迟最低的可用服务器再试， Python API / CLI / Web API 三入口全部自动生效。 核心改动： - _reconnect.py 新增 select_best_host_sync/async（连接失败 failover） 和 find_working_host_sync/async（空数据逐台实测） - 8 个 client 的 _execute 注入跨主机故障转移（复用 auto_reconnect 开关） - get_market_stat 空数据时按延迟顺序逐台实测找返回数据的服务器 - 新增 _reconnect/_areconnect helper 收敛重建连接副本 - MacClient failover 用 save_best_mac_host（延续 v1.19.4 不污染 best_host） - 顺手修复 test_commands_offline 未使用 import（main CI failure 根因） 测试：925 passed（新增 18 个 failover 测试），ruff/mypy 零新增错误。 --- CHANGELOG.md | 14 + pyproject.toml | 2 +- src/easy_tdx/_reconnect.py | 220 ++++++++++++++- src/easy_tdx/client.py | 179 +++++++++++-- src/easy_tdx/ex/client.py | 80 +++++- src/easy_tdx/ex/mac_client.py | 56 +++- src/easy_tdx/mac/client.py | 105 ++++++-- tests/unit/test_commands_offline.py | 2 - tests/unit/test_failover.py | 398 ++++++++++++++++++++++++++++ 9 files changed, 990 insertions(+), 66 deletions(-) create mode 100644 tests/unit/test_failover.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f528177..21b5a54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ 本文件记录 easy-tdx 的版本变更。格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/)。 +## [1.20.0] — 2026-07-08 + +**服务器失败时自动 ping 切换，无需手动 `easy-tdx ping`** —— 解决普通用户最困惑的痛点：连不上服务器或返回空数据时，之前必须手动跑 `easy-tdx ping` 才能恢复，普通人根本不知道该这么做。现在 Python API / CLI / Web API **三入口全部自动**——服务器连不上或返回空统计指数时，自动测速、切到延迟最低的可用服务器、重试，全程对用户透明。收敛在 `_reconnect.py` 单点注入 8 个 client 的 `_execute`，零冗余、不新增配置开关。 + +### 新增 + +- **跨主机故障转移（连接失败）**（`src/easy_tdx/_reconnect.py`）—— 8 个 client（TdxClient / MacClient / ExTdxClient / MacExClient，各 sync+async）的 `_execute` 在同主机重试耗尽（`_RETRY_DELAYS` 4 次指数退避）后，自动调 `select_best_host_sync/async` 重新测速、切到延迟最低的**另一台**服务器再试一轮。复用 `auto_reconnect` 开关（`False` 时不触发），内置 30s 节流防惊群。 +- **空数据故障转移（`get_market_stat`）**（`src/easy_tdx/_reconnect.py` + `client.py`）—— 880005/880001/880006 统计指数并非所有服务器都提供，返回空 quotes 时触发 `find_working_host_sync/async`：按延迟顺序逐台实测（最多 5 台），找到第一台返回有效数据的服务器。这是 v1.20.0 的核心场景——延迟最低的服务器不一定服务统计指数，必须逐台实测。 +- **统一重建 helper**（`client.py` / `mac/client.py` / `ex/client.py` / `ex/mac_client.py`）—— 新增 `_reconnect`/`_areconnect` 收敛各 client 内"重建连接 + 起心跳"的副本（原 `_execute` / `ensure_connected` 各有一份），消除 4 处重复，保证 failover 与重试逻辑一致。 + +### 修复 + +- **MacClient failover 不污染标准 best_host**（`src/easy_tdx/mac/client.py`）—— MAC 客户端的 failover 用 `save_best_mac_host`（写入独立配置项），而非 `save_best_host`。延续 v1.19.4 的修复（MAC 服务器不再写进标准 best_host），含防回归测试锁定。 + ## [1.19.7] — 2026-07-07 **新增「服务器设置」页面：web UI 上测速 + 切换通达信服务器** —— 解决"有些用户获取到的 IP 能连通、有些不能"的问题。不同地区/运营商对通达信各服务器连通性不同，之前用户只能碰运气或手动改 config.json。现在在 web UI 上新增第六个页面「服务器设置」，列出全部 50+ 候选服务器、一键并发测速、点选切换——切换后立即生效（热重连），无需重启服务。 diff --git a/pyproject.toml b/pyproject.toml index 09b5a64..532d384 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.19.7" +version = "1.20.0" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/_reconnect.py b/src/easy_tdx/_reconnect.py index fba6f4b..730ba16 100644 --- a/src/easy_tdx/_reconnect.py +++ b/src/easy_tdx/_reconnect.py @@ -9,13 +9,20 @@ 逐字节重复（仅心跳命令和 logger 名不同）。这里抽出 ``AsyncHeartbeatMixin`` 收敛这些副本——子类只需实现 ``_heartbeat_cmd()`` 返回一个 awaitable 即可，未来改心跳策略只需改一处。 + +跨主机故障转移（failover）：8 个 client 的 ``_execute`` 在同主机重试耗尽 +（``_RETRY_DELAYS`` 走完仍 ``TdxConnectionError``）后，调用本模块的 +``select_best_host_sync`` / ``select_best_host_async`` 重新测速、切到延迟 +最低的**另一台**服务器再试一轮。这样服务器连不上时无需用户手动 ``ping``， +Python API / CLI / Web API 三入口自动生效（三者最终都汇聚到 ``_execute``）。 """ from __future__ import annotations import asyncio import logging -from collections.abc import Awaitable +import time +from collections.abc import Awaitable, Callable from .exceptions import TdxConnectionError, TdxDecodeError @@ -33,6 +40,217 @@ _HEARTBEAT_RETRYABLE: tuple[type[BaseException], ...] = ( TdxDecodeError, ) +# --------------------------------------------------------------------------- # +# 跨主机故障转移（failover）共享实现 +# --------------------------------------------------------------------------- # +# +# 设计要点： +# 1. 纯函数，不依赖任何 client 状态——8 个 client 各自传入自己的 +# (候选主机列表, 测速函数, 持久化函数, 端口)。便于单测、避免循环依赖。 +# 2. 只返回与 current_host *不同* 的最优主机；若所有候选都不可达或唯一可达 +# 的就是 current_host，返回 None（调用方保持原 host 不变）。 +# 3. 进程级节流：_FAILOVER_PING_THROTTLE_SEC 秒内不重复全量测速——一次失败 +# 可能触发多个并发请求同时进入 failover，节流避免对几十台服务器发起 +# "惊群"式测速。节流窗口内直接返回 None（放弃本次跨主机切换，让外层 +# 同主机重试兜底）。 + +# 同一进程内两次全量测速的最小间隔（秒）。 +_FAILOVER_PING_THROTTLE_SEC: float = 30.0 + +# 上次全量测速完成的时间戳（monotonic）；初始 0 表示"从未测过"。 +_last_failover_ts: float = 0.0 + + +def _throttled() -> bool: + """距上次全量测速是否仍在节流窗口内（True=应跳过本次测速）。""" + global _last_failover_ts + return (time.monotonic() - _last_failover_ts) < _FAILOVER_PING_THROTTLE_SEC + + +def _mark_failover_done() -> None: + """记录"本次全量测速已完成"，开启新一轮节流窗口。""" + global _last_failover_ts + _last_failover_ts = time.monotonic() + + +# 测速函数的统一签名：(hosts, port, timeout) -> [(host, latency_seconds), ...] +PingFn = Callable[..., list[tuple[str, float]]] +# 持久化函数的统一签名：(host) -> None +SaveFn = Callable[[str], None] + + +def select_best_host_sync( + hosts: list[str], + ping_fn: PingFn, + save_fn: SaveFn, + port: int, + ping_timeout: float, + current_host: str, +) -> str | None: + """重新测速并选出优于当前主机的最佳主机（同步）。 + + Args: + hosts: 候选主机列表（如 ``get_known_hosts()``）。 + ping_fn: 测速函数（``ping_all`` / ``ping_mac_all`` / ``ping_ex_all``）， + 签名 ``(hosts, port, timeout) -> [(host, latency), ...]``，已按 + 延迟升序返回，不可达主机不在结果中。 + save_fn: 持久化函数（``save_best_host`` / ``save_best_ex_host`` / + ``save_best_mac_ex_host``），将选中的主机写回 config.json。 + port: 目标端口。 + ping_timeout: 单台测速超时（秒）。 + current_host: 当前正在使用（且已判定不可用）的主机，结果会跳过它。 + + Returns: + 选中的新主机（已 ``save_fn`` 持久化）；若无更优选择或处于节流窗口 + 内，返回 ``None``（调用方保持原 host）。 + """ + if _throttled(): + logging.getLogger(__name__).debug( + "跨主机故障转移：处于 %ss 节流窗口内，跳过本次测速", + _FAILOVER_PING_THROTTLE_SEC, + ) + return None + try: + ranked = ping_fn(hosts, port, ping_timeout) + finally: + # 无论测速是否拿到结果，都视为"完成一次测速"，开启节流窗口， + # 避免失败时被高频重试反复触发。 + _mark_failover_done() + # 跳过当前（已判定不可用）主机，取延迟最低的另一台 + for host, _latency in ranked: + if host != current_host: + save_fn(host) + logging.getLogger(__name__).info("跨主机故障转移：从 %s 切换到 %s", current_host, host) + return host + return None + + +async def select_best_host_async( + hosts: list[str], + ping_fn: PingFn, + save_fn: SaveFn, + port: int, + ping_timeout: float, + current_host: str, +) -> str | None: + """重新测速并选出优于当前主机的最佳主机（异步）。 + + 与 :func:`select_best_host_sync` 语义一致；测速在线程池中执行 + （``ping_fn`` 是阻塞实现，用 ``asyncio.to_thread`` 避免阻塞事件循环）， + 节流与持久化语义不变。 + """ + if _throttled(): + logging.getLogger(__name__).debug( + "跨主机故障转移：处于 %ss 节流窗口内，跳过本次测速", + _FAILOVER_PING_THROTTLE_SEC, + ) + return None + try: + ranked = await asyncio.to_thread(ping_fn, hosts, port, ping_timeout) + finally: + _mark_failover_done() + for host, _latency in ranked: + if host != current_host: + save_fn(host) + logging.getLogger(__name__).info("跨主机故障转移：从 %s 切换到 %s", current_host, host) + return host + return None + + +# 空数据故障转移时最多尝试多少台候选主机（按延迟升序）。统计指数等数据 +# 并非所有服务器都提供，延迟最低的不一定返回数据，故需轮询前几台。 +_WORKING_HOST_MAX_ATTEMPTS = 5 + +# 验证函数签名：(host) -> True 表示该主机可用（如返回非空数据）。 +TryFn = Callable[[str], bool] +AsyncTryFn = Callable[[str], Awaitable[bool]] + + +def find_working_host_sync( + ranked_hosts: list[tuple[str, float]], + try_fn: TryFn, + save_fn: SaveFn, + current_host: str, + max_attempts: int = _WORKING_HOST_MAX_ATTEMPTS, +) -> str | None: + """按延迟顺序逐台测试候选主机，返回第一台"可用"的（同步）。 + + 与 :func:`select_best_host_sync` 的区别：后者只按延迟选一台（用于连接 + 失败的故障转移）；本函数用于"连接成功但数据空"的场景（如 ``get_market_stat`` + 的统计指数 880005/880001/880006 并非所有服务器都提供），需逐台实际查询 + 才能确定哪台返回有效数据。 + + Args: + ranked_hosts: 已按延迟升序排序的 ``[(host, latency), ...]``（来自 + ``ping_fn`` 的返回值）。 + try_fn: 对单台主机的验证函数，返回 ``True`` 表示该主机可用（如返回 + 非空数据）。调用方在其中负责连接、查询、清理。 + save_fn: 持久化函数，选中可用主机后调用。 + current_host: 当前主机（跳过，它已被判定不可用）。 + max_attempts: 最多尝试多少台候选（默认 5），避免极端情况下逐台试探 + 全部候选拖垮响应。 + + Returns: + 第一台可用的主机（已 ``save_fn`` 持久化）；全部不可用则返回 ``None``。 + """ + log = logging.getLogger(__name__) + tried = 0 + for host, _latency in ranked_hosts: + if host == current_host: + continue + if tried >= max_attempts: + break + tried += 1 + try: + if try_fn(host): + save_fn(host) + log.info( + "空数据故障转移：从 %s 切换到 %s（第 %d 台候选可用）", + current_host, + host, + tried, + ) + return host + except Exception: + # 验证单台主机时的任何异常（连接失败、解析错误等）都只跳过该台， + # 继续尝试下一台，不让单台拖垮整个轮询。 + log.debug("空数据故障转移：%s 验证失败，尝试下一台", host, exc_info=True) + return None + + +async def find_working_host_async( + ranked_hosts: list[tuple[str, float]], + try_fn: AsyncTryFn, + save_fn: SaveFn, + current_host: str, + max_attempts: int = _WORKING_HOST_MAX_ATTEMPTS, +) -> str | None: + """按延迟顺序逐台测试候选主机，返回第一台"可用"的（异步）。 + + 与 :func:`find_working_host_sync` 语义一致；``try_fn`` 为 async 函数。 + """ + log = logging.getLogger(__name__) + tried = 0 + for host, _latency in ranked_hosts: + if host == current_host: + continue + if tried >= max_attempts: + break + tried += 1 + try: + if await try_fn(host): + save_fn(host) + log.info( + "空数据故障转移：从 %s 切换到 %s（第 %d 台候选可用）", + current_host, + host, + tried, + ) + return host + except Exception: + log.debug("空数据故障转移：%s 验证失败，尝试下一台", host, exc_info=True) + return None + class AsyncHeartbeatMixin: """async client 心跳三件套的共享实现（审计复审 L1）。 diff --git a/src/easy_tdx/client.py b/src/easy_tdx/client.py index 2b18cf8..23811b5 100644 --- a/src/easy_tdx/client.py +++ b/src/easy_tdx/client.py @@ -22,7 +22,14 @@ from ._df import ( _merge_txn_datetime, _to_df, ) -from ._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin +from ._reconnect import ( + _RETRY_DELAYS, + AsyncHeartbeatMixin, + find_working_host_async, + find_working_host_sync, + select_best_host_async, + select_best_host_sync, +) from .codec.block import parse_block_dat from .codec.financial import parse_financial_dat, parse_financial_file_list from .codec.industry import parse_tdxhy_cfg @@ -55,6 +62,7 @@ from .models.finance import ( FinancialFileInfo, FinancialRecord, ) +from .models.quote import SecurityQuote from .models.security import SecurityInfo from .models.stats import FundFlow, HistoricalFundFlow, MarketStat from .models.timeseries import TransactionRecord @@ -293,12 +301,24 @@ class TdxClient: try: self._execute(GetSecurityCountCmd(Market.SH)) except TdxConnectionError: - self._conn.stop_heartbeat() - self._conn.close() - self._conn = TdxConnection(self._host, self._port, self._timeout) - self._conn.connect() - if self._heartbeat_interval > 0: - self._conn.start_heartbeat(self._heartbeat_interval) + self._reconnect() + + def _reconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + 统一收敛所有"重建 TdxConnection + 起心跳"的副本：``_execute`` 同主机 + 重试、``_execute`` 跨主机故障转移、``ensure_connected``、 + ``get_market_stat`` 空数据重试都走这里，保证 4 处重建逻辑一致。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + self._conn.stop_heartbeat() + self._conn.close() + self._conn = TdxConnection(target, self._port, self._timeout) + self._conn.connect() + if self._heartbeat_interval > 0: + self._conn.start_heartbeat(self._heartbeat_interval) def __enter__(self) -> "TdxClient": self.connect() @@ -313,11 +333,18 @@ class TdxClient: self.close() # ------------------------------------------------------------------ # - # 内部执行：含自动重连 + # 内部执行：含自动重连 + 跨主机故障转移 # ------------------------------------------------------------------ # def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 + + 两阶段韧性： + 1. 同主机重试（``_RETRY_DELAYS``，4 次指数退避）——应对瞬时抖动。 + 2. 跨主机故障转移——同主机重试仍失败时，重新测速选延迟最低的另 + 一台服务器再试一轮。服务器连不上时用户无需手动 ``ping``。 + ``auto_reconnect=False`` 时两阶段都不触发，直接抛出原异常。 + """ try: return self._conn.execute(cmd) except TdxConnectionError: @@ -326,11 +353,22 @@ class TdxClient: last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: time.sleep(delay) - self._conn.close() - self._conn = TdxConnection(self._host, self._port, self._timeout) - self._conn.connect() - if self._heartbeat_interval > 0: - self._conn.start_heartbeat(self._heartbeat_interval) + self._reconnect() + try: + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移——重新测速切到另一台服务器再试一次 + new_host = select_best_host_sync( + get_known_hosts(), + ping_all, + save_best_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + self._reconnect(new_host) try: return self._conn.execute(cmd) except TdxConnectionError as e: @@ -681,13 +719,18 @@ class TdxClient: 通达信这三个"统计指数"的计数类字段（涨/跌/平/总数/涨停/跌停家数） 返回的是真实家数的 1/10，需统一 ×10 还原。成交额/量/市值字段不受影响。 `suspended_count` 由 `total - up - down - neutral` 推得，用于保证计数守恒。 + + 空数据容错：880005/880001/880006 并非所有服务器都提供，会返回空 quotes。 + 此时不只切换到延迟最低的一台（它可能也不提供），而是按延迟顺序逐台实测， + 找到第一台返回有效数据的服务器，避免用户手动 ``easy-tdx ping``。 """ # 通达信中 880005 是全市场行情统计，880001 是总市值指数，880006 是涨跌停统计 - quotes = self._execute( - GetSecurityQuotesCmd( - [(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")] - ) + _cmd = GetSecurityQuotesCmd( + [(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")] ) + quotes = self._execute(_cmd) + if not quotes and self._auto_reconnect: + quotes = self._find_host_returning_quotes(_cmd) if not quotes: raise RuntimeError("无法获取市场统计数据") q = quotes[0] @@ -714,6 +757,32 @@ class TdxClient: ) ) + def _find_host_returning_quotes( + self, cmd: "BaseCommand[list[SecurityQuote]]" + ) -> list[SecurityQuote]: + """空数据故障转移：测速后按延迟顺序逐台实测，返回首台有效数据的 quotes。 + + 专供 ``get_market_stat`` 使用——统计指数并非所有服务器都提供，延迟最低 + 的不一定返回数据，故需逐台实际查询。最多尝试 ``_WORKING_HOST_MAX_ATTEMPTS`` + 台（见 ``_reconnect``）。找到后 client 停在该 host；全失败返回空。 + """ + bad_host = self._host + ranked = ping_all(get_known_hosts(), self._port, 5.0) + + def _try(host: str) -> bool: + # 切换到候选 host 并实测；非空即视为该 host 可用 + self._reconnect(host) + return bool(self._execute(cmd)) + + new_host = find_working_host_sync(ranked, _try, save_best_host, bad_host) + if new_host is None: + # 全部候选都不可用，回退到原 host（保持状态可预测） + if self._host != bad_host: + self._reconnect(bad_host) + return [] + # _try 已把 client 切到 new_host 并执行过 cmd，重新取一次拿结果 + return self._execute(cmd) + def _collect_transaction_records( self, fetch_page: Callable[[int, int], list[TransactionRecord]], @@ -912,8 +981,29 @@ class AsyncTdxClient(AsyncHeartbeatMixin): """心跳使用的轻量请求（get_security_count，复用 _execute 重连）。""" return self.get_security_count(Market.SH) + async def _areconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + async 版的统一重建入口，与 sync ``_reconnect`` 对称，供 ``_execute`` + 同主机重试、跨主机故障转移、``get_market_stat`` 空数据重试复用。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + await self._stop_heartbeat() + await self._conn.close() + self._conn = AsyncTdxConnection(target, self._port, self._timeout) + await self._conn.connect() + self._start_heartbeat() + async def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 + + 两阶段韧性与 sync 版对称：先同主机重试（``_RETRY_DELAYS``），再跨主机 + 故障转移（重新测速切到另一台服务器）。整个流程在 ``_execute_lock`` 内 + 串行，避免并发请求触发多次故障转移抖动。``auto_reconnect=False`` 时 + 两阶段都不触发。 + """ async with self._execute_lock: try: return await self._conn.execute(cmd) @@ -923,9 +1013,22 @@ class AsyncTdxClient(AsyncHeartbeatMixin): last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: await asyncio.sleep(delay) - await self._conn.close() - self._conn = AsyncTdxConnection(self._host, self._port, self._timeout) - await self._conn.connect() + await self._areconnect() + try: + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移 + new_host = await select_best_host_async( + get_known_hosts(), + ping_all, + save_best_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + await self._areconnect(new_host) try: return await self._conn.execute(cmd) except TdxConnectionError as e: @@ -1214,13 +1317,17 @@ class AsyncTdxClient(AsyncHeartbeatMixin): 通达信这三个"统计指数"的计数类字段（涨/跌/平/总数/涨停/跌停家数） 返回的是真实家数的 1/10，需统一 ×10 还原。成交额/量/市值字段不受影响。 `suspended_count` 由 `total - up - down - neutral` 推得，用于保证计数守恒。 + + 空数据容错：与 sync 版对称——空 quotes 时按延迟顺序逐台实测，找到首台 + 返回有效数据的服务器。 """ # 通达信中 880005 是全市场行情统计，880001 是总市值指数，880006 是涨跌停统计 - quotes = await self._execute( - GetSecurityQuotesCmd( - [(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")] - ) + _cmd = GetSecurityQuotesCmd( + [(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")] ) + quotes = await self._execute(_cmd) + if not quotes and self._auto_reconnect: + quotes = await self._find_host_returning_quotes(_cmd) if not quotes: raise RuntimeError("无法获取市场统计数据") q = quotes[0] @@ -1247,6 +1354,26 @@ class AsyncTdxClient(AsyncHeartbeatMixin): ) ) + async def _find_host_returning_quotes( + self, cmd: "BaseCommand[list[SecurityQuote]]" + ) -> list[SecurityQuote]: + """空数据故障转移（async）：与 sync ``_find_host_returning_quotes`` 对称。""" + bad_host = self._host + ranked = await asyncio.to_thread(ping_all, get_known_hosts(), self._port, 5.0) + + async def _try(host: str) -> bool: + await self._areconnect(host) + # mypy 对 async 闭包内泛型参数的推断会宽化为 BaseCommand[object] + # （sync 同模式可正确推断），此处为已知 mypy 限制，非真实类型错误。 + return bool(await self._execute(cmd)) # type: ignore[arg-type] + + new_host = await find_working_host_async(ranked, _try, save_best_host, bad_host) + if new_host is None: + if self._host != bad_host: + await self._areconnect(bad_host) + return [] + return await self._execute(cmd) + async def _collect_transaction_records( self, fetch_page: Callable[[int, int], Awaitable[list[TransactionRecord]]], diff --git a/src/easy_tdx/ex/client.py b/src/easy_tdx/ex/client.py index 1165999..0081bdf 100644 --- a/src/easy_tdx/ex/client.py +++ b/src/easy_tdx/ex/client.py @@ -9,7 +9,12 @@ from types import TracebackType from typing import TypeVar from .._df import _apply_bar_time_align_bars, _category_to_minutes -from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin +from .._reconnect import ( + _RETRY_DELAYS, + AsyncHeartbeatMixin, + select_best_host_async, + select_best_host_sync, +) from ..commands.base import BaseCommand from ..config import get_best_ex_host, get_ex_hosts, save_best_ex_host from ..exceptions import TdxConnectionError @@ -120,8 +125,25 @@ class ExTdxClient: ) -> None: self.close() + def _reconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + 扩展行情无心跳，重建仅 close → new ExTdxConnection → connect。 + 供 ``_execute`` 同主机重试与跨主机故障转移复用。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + self._conn.close() + self._conn = ExTdxConnection(target, self._port, self._timeout) + self._conn.connect() + def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 + + 两阶段韧性与 A 股/MAC 统一（审计 #2）：先同主机重试 4 次，再跨主机 + 故障转移（测速切到另一台扩展行情服务器）。 + """ try: return self._conn.execute(cmd) except TdxConnectionError: @@ -130,9 +152,22 @@ class ExTdxClient: last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: time.sleep(delay) - self._conn.close() - self._conn = ExTdxConnection(self._host, self._port, self._timeout) - self._conn.connect() + self._reconnect() + try: + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移 + new_host = select_best_host_sync( + get_ex_hosts(), + ping_ex_all, + save_best_ex_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + self._reconnect(new_host) try: return self._conn.execute(cmd) except TdxConnectionError as e: @@ -350,8 +385,22 @@ class AsyncExTdxClient(AsyncHeartbeatMixin): """心跳使用的轻量请求（get_instrument_count，复用 _execute 重连）。""" return self.get_instrument_count() + async def _areconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + 供 ``_execute`` 同主机重试与跨主机故障转移复用。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + await self._stop_heartbeat() + await self._conn.close() + self._conn = AsyncExTdxConnection(target, self._port, self._timeout) + await self._conn.connect() + self._start_heartbeat() + async def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。""" async with self._execute_lock: try: return await self._conn.execute(cmd) @@ -361,9 +410,22 @@ class AsyncExTdxClient(AsyncHeartbeatMixin): last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: await asyncio.sleep(delay) - await self._conn.close() - self._conn = AsyncExTdxConnection(self._host, self._port, self._timeout) - await self._conn.connect() + await self._areconnect() + try: + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移 + new_host = await select_best_host_async( + get_ex_hosts(), + ping_ex_all, + save_best_ex_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + await self._areconnect(new_host) try: return await self._conn.execute(cmd) except TdxConnectionError as e: diff --git a/src/easy_tdx/ex/mac_client.py b/src/easy_tdx/ex/mac_client.py index 520dd69..aa96f2e 100644 --- a/src/easy_tdx/ex/mac_client.py +++ b/src/easy_tdx/ex/mac_client.py @@ -15,7 +15,12 @@ from typing import Any, TypeVar import pandas as pd from .._df import _to_df -from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin +from .._reconnect import ( + _RETRY_DELAYS, + AsyncHeartbeatMixin, + select_best_host_async, + select_best_host_sync, +) from ..commands.base import BaseCommand from ..config import get_best_mac_ex_host, get_mac_ex_hosts, save_best_mac_ex_host from ..exceptions import TdxConnectionError @@ -148,11 +153,12 @@ class MacExClient: self._conn.execute(MacExLoginCmd()) def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。 + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 每次重连后必须重新 ``_login()``（MAC 协议扩展行情特有）。登录握手期的 ``TdxConnectionError`` 与业务请求一样计入退避重试；``TdxCommandError`` - （登录被拒等确定性失败）不重试，直接抛出。 + （登录被拒等确定性失败）不重试，直接抛出。跨主机故障转移阶段同样遵循 + ``connect + login`` 纳入重试的语义。 """ try: return self._conn.execute(cmd) @@ -173,6 +179,27 @@ class MacExClient: return self._conn.execute(cmd) except TdxConnectionError as e: last_exc = e + # 第二阶段：跨主机故障转移——测速切到另一台 MAC 扩展行情服务器 + new_host = select_best_host_sync( + get_mac_ex_hosts(), + ping_ex_all, + save_best_mac_ex_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + self._host = new_host + self._conn.close() + self._conn = ExTdxConnection( + self._host, self._port, self._timeout, mac_ex_mode=True + ) + try: + self._conn.connect() + self._login() + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # @@ -585,7 +612,7 @@ class AsyncMacExClient(AsyncHeartbeatMixin): await self._conn.execute(MacExLoginCmd()) async def _execute(self, cmd: "BaseCommand[_T]") -> _T: - """执行命令；断线时指数退避重试（4 次，与 A 股/MAC 统一，审计 #2）。 + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 每次重连后必须重新 ``_login()``（MAC 协议扩展行情特有）。登录握手期的 ``TdxConnectionError`` 与业务请求一样计入退避重试；``TdxCommandError`` @@ -611,6 +638,27 @@ class AsyncMacExClient(AsyncHeartbeatMixin): return await self._conn.execute(cmd) except TdxConnectionError as e: last_exc = e + # 第二阶段：跨主机故障转移 + new_host = await select_best_host_async( + get_mac_ex_hosts(), + ping_ex_all, + save_best_mac_ex_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + self._host = new_host + await self._conn.close() + self._conn = AsyncExTdxConnection( + self._host, self._port, self._timeout, mac_ex_mode=True + ) + try: + await self._conn.connect() + await self._login() + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e raise last_exc # type: ignore[misc] # ------------------------------------------------------------------ # diff --git a/src/easy_tdx/mac/client.py b/src/easy_tdx/mac/client.py index c2e7b4a..f8d3ae2 100644 --- a/src/easy_tdx/mac/client.py +++ b/src/easy_tdx/mac/client.py @@ -13,7 +13,12 @@ from typing import Any, TypeVar import pandas as pd from .._df import _apply_bar_time_align_df, _period_to_minutes, _to_df -from .._reconnect import _RETRY_DELAYS, AsyncHeartbeatMixin +from .._reconnect import ( + _RETRY_DELAYS, + AsyncHeartbeatMixin, + select_best_host_async, + select_best_host_sync, +) from ..codec.bitmap import Fields, PresetField from ..commands.base import BaseCommand from ..config import ( @@ -224,12 +229,23 @@ class MacClient: try: self._execute(KlineOffsetCmd(0, 1)) except TdxConnectionError: - self._conn.stop_heartbeat() - self._conn.close() - self._conn = TdxConnection(self._host, self._port, self._timeout) - self._conn.connect() - if self._heartbeat_interval > 0: - self._conn.start_heartbeat(self._heartbeat_interval) + self._reconnect() + + def _reconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + 与 TdxClient._reconnect 对称：统一收敛 ``_execute`` 同主机重试、 + 跨主机故障转移、``ensure_connected`` 的重建副本。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + self._conn.stop_heartbeat() + self._conn.close() + self._conn = TdxConnection(target, self._port, self._timeout) + self._conn.connect() + if self._heartbeat_interval > 0: + self._conn.start_heartbeat(self._heartbeat_interval) def __enter__(self) -> MacClient: self.connect() @@ -244,11 +260,15 @@ class MacClient: self.close() # ------------------------------------------------------------------ # - # 内部执行：含自动重连 + # 内部执行：含自动重连 + 跨主机故障转移 # ------------------------------------------------------------------ # def _execute(self, cmd: BaseCommand[_T]) -> _T: - """执行命令；断线时指数退避重试。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。 + + 两阶段韧性与 TdxClient 对称：先同主机重试（``_RETRY_DELAYS``）， + 再跨主机故障转移（重新测速切到另一台 MAC 服务器）。 + """ try: return self._conn.execute(cmd) except TdxConnectionError: @@ -257,11 +277,24 @@ class MacClient: last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: time.sleep(delay) - self._conn.close() - self._conn = TdxConnection(self._host, self._port, self._timeout) - self._conn.connect() - if self._heartbeat_interval > 0: - self._conn.start_heartbeat(self._heartbeat_interval) + self._reconnect() + try: + return self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移——测速切到另一台 MAC 服务器再试一次。 + # save_best_mac_host（而非 save_best_host）：MAC 服务器写入独立的 + # best_mac_host 配置项，不污染标准 best_host（v1.19.4 修复的回归）。 + new_host = select_best_host_sync( + get_mac_hosts(), + ping_mac_all, + save_best_mac_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + self._reconnect(new_host) try: return self._conn.execute(cmd) except TdxConnectionError as e: @@ -1234,11 +1267,22 @@ class AsyncMacClient(AsyncHeartbeatMixin): try: await self._execute(KlineOffsetCmd(0, 1)) except TdxConnectionError: - await self._stop_heartbeat() - await self._conn.close() - self._conn = AsyncTdxConnection(self._host, self._port, self._timeout) - await self._conn.connect() - self._start_heartbeat() + await self._areconnect() + + async def _areconnect(self, host: str | None = None) -> None: + """关闭当前连接并重建（默认连 self._host；传入 host 则切换主机）。 + + 与 AsyncTdxClient._areconnect 对称。统一收敛 ``_execute`` 同主机重试、 + 跨主机故障转移、``ensure_connected`` 的重建副本。 + """ + target = host if host is not None else self._host + if host is not None: + self._host = host + await self._stop_heartbeat() + await self._conn.close() + self._conn = AsyncTdxConnection(target, self._port, self._timeout) + await self._conn.connect() + self._start_heartbeat() async def __aenter__(self) -> AsyncMacClient: await self.connect() @@ -1265,7 +1309,7 @@ class AsyncMacClient(AsyncHeartbeatMixin): # ------------------------------------------------------------------ # async def _execute(self, cmd: BaseCommand[_T]) -> _T: - """执行命令；断线时指数退避重试。""" + """执行命令；断线时指数退避重试，同主机耗尽则跨主机故障转移。""" async with self._execute_lock: try: return await self._conn.execute(cmd) @@ -1275,9 +1319,24 @@ class AsyncMacClient(AsyncHeartbeatMixin): last_exc: TdxConnectionError | None = None for delay in _RETRY_DELAYS: await asyncio.sleep(delay) - await self._conn.close() - self._conn = AsyncTdxConnection(self._host, self._port, self._timeout) - await self._conn.connect() + await self._areconnect() + try: + return await self._conn.execute(cmd) + except TdxConnectionError as e: + last_exc = e + # 第二阶段：跨主机故障转移 + # save_best_mac_host：写入独立配置项，不污染标准 best_host + # （v1.19.4 修复的回归：MAC failover 不可用 save_best_host） + new_host = await select_best_host_async( + get_mac_hosts(), + ping_mac_all, + save_best_mac_host, + self._port, + 5.0, + self._host, + ) + if new_host is not None: + await self._areconnect(new_host) try: return await self._conn.execute(cmd) except TdxConnectionError as e: diff --git a/tests/unit/test_commands_offline.py b/tests/unit/test_commands_offline.py index b42f2c7..9e61525 100644 --- a/tests/unit/test_commands_offline.py +++ b/tests/unit/test_commands_offline.py @@ -10,8 +10,6 @@ from __future__ import annotations import pathlib import struct -import pytest - FIXTURES = pathlib.Path(__file__).parent.parent / "fixtures" diff --git a/tests/unit/test_failover.py b/tests/unit/test_failover.py new file mode 100644 index 0000000..bda709e --- /dev/null +++ b/tests/unit/test_failover.py @@ -0,0 +1,398 @@ +"""跨主机故障转移（failover）测试。 + +验证 8 个 client 的 ``_execute`` 在同主机重试耗尽（``_RETRY_DELAYS`` 走完仍 +``TdxConnectionError``）后，会通过 ``select_best_host_sync`` / ``_async`` 重新 +测速、切到延迟最低的另一台服务器再试一轮。同时覆盖： + +- ``select_best_host_sync`` 的节流（30s 窗口内不重复测速）与"跳过当前 host"语义。 +- ``auto_reconnect=False`` 时 failover 不触发。 +- ``get_market_stat`` 空数据时触发 failover 再试。 +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from easy_tdx._reconnect import ( + _FAILOVER_PING_THROTTLE_SEC, + _WORKING_HOST_MAX_ATTEMPTS, + find_working_host_sync, + select_best_host_sync, +) +from easy_tdx.client import TdxClient +from easy_tdx.commands.security_count import GetSecurityCountCmd +from easy_tdx.exceptions import TdxConnectionError +from easy_tdx.models.enums import Market + +# --------------------------------------------------------------------------- # +# select_best_host_sync 单元逻辑 +# --------------------------------------------------------------------------- # + + +class TestSelectBestHostSync: + def setup_method(self) -> None: + # 每个测试前重置节流时间戳，避免上一个测试的节流窗口泄漏 + import easy_tdx._reconnect as r + + r._last_failover_ts = 0.0 + + def test_returns_lowest_latency_host_excluding_current(self) -> None: + """返回延迟最低且与 current_host 不同的主机。""" + ping_fn = MagicMock( + return_value=[("fast", 0.01), ("slow", 0.5)] # 已按延迟升序 + ) + save_fn = MagicMock() + + result = select_best_host_sync(["fast", "slow", "cur"], ping_fn, save_fn, 7709, 1.0, "cur") + assert result == "fast" + ping_fn.assert_called_once_with(["fast", "slow", "cur"], 7709, 1.0) + save_fn.assert_called_once_with("fast") + + def test_skips_current_host_even_if_it_is_fastest(self) -> None: + """当前主机恰好延迟最低时，应跳过它取次优。""" + ping_fn = MagicMock(return_value=[("cur", 0.01), ("other", 0.02)]) + save_fn = MagicMock() + + result = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur") + assert result == "other" + save_fn.assert_called_once_with("other") + + def test_returns_none_when_only_current_reachable(self) -> None: + """只有当前主机可达时返回 None（不切换、不持久化）。""" + ping_fn = MagicMock(return_value=[("cur", 0.01)]) + save_fn = MagicMock() + + result = select_best_host_sync(["cur"], ping_fn, save_fn, 7709, 1.0, "cur") + assert result is None + save_fn.assert_not_called() + + def test_returns_none_when_no_host_reachable(self) -> None: + """所有候选都不可达时返回 None。""" + ping_fn = MagicMock(return_value=[]) + save_fn = MagicMock() + + result = select_best_host_sync(["a", "b"], ping_fn, save_fn, 7709, 1.0, "cur") + assert result is None + save_fn.assert_not_called() + + def test_throttle_skips_ping_within_window(self) -> None: + """节流窗口内（30s）第二次调用直接返回 None，不触发测速。""" + ping_fn = MagicMock(return_value=[("other", 0.01)]) + save_fn = MagicMock() + + # 第一次：正常测速，返回 other + first = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur") + assert first == "other" + assert ping_fn.call_count == 1 + + # 第二次（立即）：应被节流，跳过测速 + second = select_best_host_sync(["cur", "other"], ping_fn, save_fn, 7709, 1.0, "cur") + assert second is None + # 测速调用次数不应增加 + assert ping_fn.call_count == 1 + + def test_throttle_window_is_configurable_constant(self) -> None: + """节流窗口常量存在且为正（防回归：误改成 0 会关闭节流）。""" + assert _FAILOVER_PING_THROTTLE_SEC > 0 + + +# --------------------------------------------------------------------------- # +# find_working_host_sync 单元逻辑（多 host 轮询直到验证通过） +# --------------------------------------------------------------------------- # + + +class TestFindWorkingHostSync: + def test_returns_first_host_passing_validation(self) -> None: + """按延迟顺序逐台测试，返回第一台通过验证的 host。""" + ranked = [("fast", 0.01), ("mid", 0.05), ("slow", 0.5)] + # fast 验证失败，mid 通过 + try_fn = MagicMock(side_effect=[False, True, True]) + save_fn = MagicMock() + + result = find_working_host_sync(ranked, try_fn, save_fn, "cur") + assert result == "mid" + save_fn.assert_called_once_with("mid") + # 只测到通过那台为止（slow 未被测试） + assert try_fn.call_count == 2 + + def test_skips_current_host(self) -> None: + """跳过 current_host，不对其调用验证函数。""" + ranked = [("cur", 0.01), ("other", 0.02)] + try_fn = MagicMock(return_value=True) + save_fn = MagicMock() + + result = find_working_host_sync(ranked, try_fn, save_fn, "cur") + assert result == "other" + # cur 被跳过，只验证了 other + try_fn.assert_called_once_with("other") + + def test_returns_none_when_all_fail_validation(self) -> None: + """所有候选验证都失败时返回 None。""" + ranked = [("a", 0.01), ("b", 0.02)] + try_fn = MagicMock(return_value=False) + save_fn = MagicMock() + + result = find_working_host_sync(ranked, try_fn, save_fn, "cur") + assert result is None + save_fn.assert_not_called() + + def test_respects_max_attempts(self) -> None: + """max_attempts 限制最多测试的候选数。""" + ranked = [("a", 0.01), ("b", 0.02), ("c", 0.03)] + try_fn = MagicMock(return_value=False) + save_fn = MagicMock() + + result = find_working_host_sync(ranked, try_fn, save_fn, "cur", max_attempts=2) + assert result is None + # 只测了前 2 台（受 max_attempts 限制），c 未测 + assert try_fn.call_count == 2 + + def test_validation_exception_skips_host_not_aborts(self) -> None: + """单台验证抛异常只跳过该台，继续尝试下一台。""" + ranked = [("boom", 0.01), ("good", 0.02)] + save_fn = MagicMock() + + def _try(host: str) -> bool: + if host == "boom": + raise RuntimeError("connection refused") + return True + + result = find_working_host_sync(ranked, _try, save_fn, "cur") + assert result == "good" + save_fn.assert_called_once_with("good") + + def test_default_max_attempts_constant(self) -> None: + """默认 max_attempts 常量存在且合理（防回归）。""" + assert _WORKING_HOST_MAX_ATTEMPTS == 5 + + +# --------------------------------------------------------------------------- # +# TdxClient._execute 跨主机故障转移 +# --------------------------------------------------------------------------- # + + +class TestTdxClientFailover: + def setup_method(self) -> None: + import easy_tdx._reconnect as r + + r._last_failover_ts = 0.0 + + def test_failover_switches_host_after_retries_exhausted(self) -> None: + """同主机 4 次重试全失败后，应跨主机切到新 host 并成功。""" + with ( + patch("easy_tdx.client.TdxConnection") as mock_conn_cls, + patch("easy_tdx.client.time.sleep"), + patch("easy_tdx.client.select_best_host_sync", return_value="new-host") as mock_select, + ): + mock_conn = MagicMock() + # 首次 + 4 次重试全失败，第 6 次（failover 后）成功 + mock_conn.execute.side_effect = [ + TdxConnectionError("down"), # 首次 + TdxConnectionError("down"), # 重试1 + TdxConnectionError("down"), # 重试2 + TdxConnectionError("down"), # 重试3 + TdxConnectionError("down"), # 重试4 + 1234, # failover 到新 host 后成功 + ] + mock_conn_cls.return_value = mock_conn + + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + result = client._execute(GetSecurityCountCmd(Market.SH)) + + assert result == 1234 + # failover 被调用，且传入的 current_host 是坏主机 + mock_select.assert_called_once() + args = mock_select.call_args + assert args.args[-1] == "bad-host" # current_host + # client 的 host 已切换到新主机 + assert client._host == "new-host" + + def test_failover_returns_none_keeps_host_and_raises(self) -> None: + """failover 未找到更优 host（返回 None）时，保持原 host 并抛出。""" + with ( + patch("easy_tdx.client.TdxConnection") as mock_conn_cls, + patch("easy_tdx.client.time.sleep"), + patch("easy_tdx.client.select_best_host_sync", return_value=None), + ): + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("always down") + mock_conn_cls.return_value = mock_conn + + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + with pytest.raises(TdxConnectionError): + client._execute(GetSecurityCountCmd(Market.SH)) + # host 未被切换 + assert client._host == "bad-host" + + def test_no_failover_when_auto_reconnect_disabled(self) -> None: + """auto_reconnect=False 时首次失败立即抛出，不进入 failover。""" + with ( + patch("easy_tdx.client.TdxConnection") as mock_conn_cls, + patch("easy_tdx.client.time.sleep"), + patch("easy_tdx.client.select_best_host_sync") as mock_select, + ): + mock_conn = MagicMock() + mock_conn.execute.side_effect = TdxConnectionError("down") + mock_conn_cls.return_value = mock_conn + + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=False, heartbeat_interval=0) + with pytest.raises(TdxConnectionError): + client._execute(GetSecurityCountCmd(Market.SH)) + # failover 完全未被调用 + mock_select.assert_not_called() + + +# --------------------------------------------------------------------------- # +# MacClient 跨主机故障转移（v1.19.4 兼容性：不污染标准 best_host） +# --------------------------------------------------------------------------- # + + +class TestMacClientFailover: + """锁定 v1.19.4 修复：MacClient 的 failover 必须用 save_best_mac_host， + 而非 save_best_host，否则会把 MAC 服务器写进标准 best_host 配置项造成污染。 + + 该 bug 曾在将 failover 改动从旧分支 cherry-pick 到含 v1.19.4 修复的 main 时 + 复现（_execute 的 failover 沿用了旧的 save_best_host）。本测试防止再次倒退。 + """ + + def setup_method(self) -> None: + import easy_tdx._reconnect as r + + r._last_failover_ts = 0.0 + + def test_failover_uses_save_best_mac_host_not_save_best_host(self) -> None: + """MacClient failover 持久化时必须调 save_best_mac_host。""" + from easy_tdx.mac.client import MacClient + from easy_tdx.mac.commands.kline_offset import KlineOffsetCmd + + with ( + patch("easy_tdx.mac.client.TdxConnection") as mock_conn_cls, + patch("easy_tdx.mac.client.time.sleep"), + patch( + "easy_tdx.mac.client.select_best_host_sync", return_value="new-mac-host" + ) as mock_select, + ): + mock_conn = MagicMock() + mock_conn.execute.side_effect = [ + TdxConnectionError("down"), + TdxConnectionError("down"), + TdxConnectionError("down"), + TdxConnectionError("down"), + TdxConnectionError("down"), + 999, # failover 后成功 + ] + mock_conn_cls.return_value = mock_conn + + client = MacClient("bad-mac-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + client._execute(KlineOffsetCmd(0, 1)) + + mock_select.assert_called_once() + # 第 3 个位置参数是 save_fn，必须是 save_best_mac_host（防 v1.19.4 回归） + from easy_tdx.config import save_best_mac_host + + save_fn = mock_select.call_args.args[2] + assert save_fn is save_best_mac_host, ( + "MacClient failover 必须用 save_best_mac_host，" + "否则污染标准 best_host（v1.19.4 修复）" + ) + + +# --------------------------------------------------------------------------- # +# get_market_stat 空数据故障转移 +# --------------------------------------------------------------------------- # + + +class TestMarketStatEmptyFailover: + def setup_method(self) -> None: + import easy_tdx._reconnect as r + + r._last_failover_ts = 0.0 + + def _make_quote(self) -> object: + """构造一个字段合法的统计指数 quote，让 get_market_stat 计算路径走通。""" + from easy_tdx.models.quote import SecurityQuote + + # 880005：price=涨家数/10, open=跌家数/10, low=平/10, high=总数/10 + return SecurityQuote( + market=Market.SH, + code="880005", + price=159.3, # → up=1593 + pre_close=0.0, + open=379.0, # → down=3790 + high=552.8, # → total=5528 + low=13.5, # → neutral=135 + vol=0.0, + cur_vol=0.0, + amount=2.58e12, + s_vol=0.0, + b_vol=0.0, + active1=0, + active2=0, + bid1=0.0, + bid_vol1=0.0, + bid2=0.0, + bid_vol2=0.0, + bid3=0.0, + bid_vol3=0.0, + bid4=0.0, + bid_vol4=0.0, + bid5=0.0, + bid_vol5=0.0, + ask1=0.0, + ask_vol1=0.0, + ask2=0.0, + ask_vol2=0.0, + ask3=0.0, + ask_vol3=0.0, + ask4=0.0, + ask_vol4=0.0, + ask5=0.0, + ask_vol5=0.0, + rise_speed=0.0, + limit_up=None, + limit_down=None, + ) + + def test_empty_quotes_finds_working_host_and_returns_data(self) -> None: + """空 quotes 时按延迟顺序逐台实测，找到返回数据的 host。""" + quote = self._make_quote() + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + # _execute: 首次空（bad-host）→ 验证 hostA 空 → 验证 hostB 非空 → 最终再取一次 + with ( + patch.object(client, "_execute", side_effect=[[], [], [quote], [quote]]) as mock_exec, + patch.object(client, "_reconnect") as mock_reconnect, + patch( + "easy_tdx.client.ping_all", + return_value=[("hostA", 0.01), ("hostB", 0.02)], + ), + ): + df = client.get_market_stat() + + # _execute 调用序列：1 首次 + 2 次 find_working_host 验证(hostA空、hostB非空) + 1 最终取值 + assert mock_exec.call_count == 4 + # _reconnect 切换到 hostA、hostB（逐台实测），最终停在 hostB + reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list] + assert reconnect_hosts == ["hostA", "hostB"] + assert len(df) == 1 + + def test_empty_quotes_all_candidates_empty_raises(self) -> None: + """所有候选都返回空时，抛 RuntimeError。""" + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + with ( + patch.object(client, "_execute", return_value=[]), + patch.object(client, "_reconnect") as mock_reconnect, + patch( + "easy_tdx.client.ping_all", + return_value=[("hostA", 0.01), ("hostB", 0.02)], + ), + ): + with pytest.raises(RuntimeError, match="无法获取市场统计数据"): + client.get_market_stat() + + # find_working_host 逐台实测了 hostA、hostB（_reconnect 被各调一次） + reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list] + assert reconnect_hosts == ["hostA", "hostB"]