mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 14:34:15 +08:00
feat: v1.20.0 服务器失败自动 ping 切换(无需手动 easy-tdx ping)
服务器连不上或返回空数据时,自动测速切到延迟最低的可用服务器再试, 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 零新增错误。
This commit is contained in:
@@ -2,6 +2,20 @@
|
|||||||
|
|
||||||
本文件记录 easy-tdx 的版本变更。格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/)。
|
本文件记录 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
|
## [1.19.7] — 2026-07-07
|
||||||
|
|
||||||
**新增「服务器设置」页面:web UI 上测速 + 切换通达信服务器** —— 解决"有些用户获取到的 IP 能连通、有些不能"的问题。不同地区/运营商对通达信各服务器连通性不同,之前用户只能碰运气或手动改 config.json。现在在 web UI 上新增第六个页面「服务器设置」,列出全部 50+ 候选服务器、一键并发测速、点选切换——切换后立即生效(热重连),无需重启服务。
|
**新增「服务器设置」页面:web UI 上测速 + 切换通达信服务器** —— 解决"有些用户获取到的 IP 能连通、有些不能"的问题。不同地区/运营商对通达信各服务器连通性不同,之前用户只能碰运气或手动改 config.json。现在在 web UI 上新增第六个页面「服务器设置」,列出全部 50+ 候选服务器、一键并发测速、点选切换——切换后立即生效(热重连),无需重启服务。
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "easy-tdx"
|
name = "easy-tdx"
|
||||||
version = "1.19.7"
|
version = "1.20.0"
|
||||||
description = "通达信 TCP 协议行情数据客户端,支持在线行情、离线数据读取与写入同步"
|
description = "通达信 TCP 协议行情数据客户端,支持在线行情、离线数据读取与写入同步"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
|
|||||||
+219
-1
@@ -9,13 +9,20 @@
|
|||||||
逐字节重复(仅心跳命令和 logger 名不同)。这里抽出
|
逐字节重复(仅心跳命令和 logger 名不同)。这里抽出
|
||||||
``AsyncHeartbeatMixin`` 收敛这些副本——子类只需实现 ``_heartbeat_cmd()``
|
``AsyncHeartbeatMixin`` 收敛这些副本——子类只需实现 ``_heartbeat_cmd()``
|
||||||
返回一个 awaitable 即可,未来改心跳策略只需改一处。
|
返回一个 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
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
from collections.abc import Awaitable
|
import time
|
||||||
|
from collections.abc import Awaitable, Callable
|
||||||
|
|
||||||
from .exceptions import TdxConnectionError, TdxDecodeError
|
from .exceptions import TdxConnectionError, TdxDecodeError
|
||||||
|
|
||||||
@@ -33,6 +40,217 @@ _HEARTBEAT_RETRYABLE: tuple[type[BaseException], ...] = (
|
|||||||
TdxDecodeError,
|
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:
|
class AsyncHeartbeatMixin:
|
||||||
"""async client 心跳三件套的共享实现(审计复审 L1)。
|
"""async client 心跳三件套的共享实现(审计复审 L1)。
|
||||||
|
|||||||
+153
-26
@@ -22,7 +22,14 @@ from ._df import (
|
|||||||
_merge_txn_datetime,
|
_merge_txn_datetime,
|
||||||
_to_df,
|
_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.block import parse_block_dat
|
||||||
from .codec.financial import parse_financial_dat, parse_financial_file_list
|
from .codec.financial import parse_financial_dat, parse_financial_file_list
|
||||||
from .codec.industry import parse_tdxhy_cfg
|
from .codec.industry import parse_tdxhy_cfg
|
||||||
@@ -55,6 +62,7 @@ from .models.finance import (
|
|||||||
FinancialFileInfo,
|
FinancialFileInfo,
|
||||||
FinancialRecord,
|
FinancialRecord,
|
||||||
)
|
)
|
||||||
|
from .models.quote import SecurityQuote
|
||||||
from .models.security import SecurityInfo
|
from .models.security import SecurityInfo
|
||||||
from .models.stats import FundFlow, HistoricalFundFlow, MarketStat
|
from .models.stats import FundFlow, HistoricalFundFlow, MarketStat
|
||||||
from .models.timeseries import TransactionRecord
|
from .models.timeseries import TransactionRecord
|
||||||
@@ -293,12 +301,24 @@ class TdxClient:
|
|||||||
try:
|
try:
|
||||||
self._execute(GetSecurityCountCmd(Market.SH))
|
self._execute(GetSecurityCountCmd(Market.SH))
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
self._conn.stop_heartbeat()
|
self._reconnect()
|
||||||
self._conn.close()
|
|
||||||
self._conn = TdxConnection(self._host, self._port, self._timeout)
|
def _reconnect(self, host: str | None = None) -> None:
|
||||||
self._conn.connect()
|
"""关闭当前连接并重建(默认连 self._host;传入 host 则切换主机)。
|
||||||
if self._heartbeat_interval > 0:
|
|
||||||
self._conn.start_heartbeat(self._heartbeat_interval)
|
统一收敛所有"重建 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":
|
def __enter__(self) -> "TdxClient":
|
||||||
self.connect()
|
self.connect()
|
||||||
@@ -313,11 +333,18 @@ class TdxClient:
|
|||||||
self.close()
|
self.close()
|
||||||
|
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
# 内部执行:含自动重连
|
# 内部执行:含自动重连 + 跨主机故障转移
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
|
|
||||||
def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
|
两阶段韧性:
|
||||||
|
1. 同主机重试(``_RETRY_DELAYS``,4 次指数退避)——应对瞬时抖动。
|
||||||
|
2. 跨主机故障转移——同主机重试仍失败时,重新测速选延迟最低的另
|
||||||
|
一台服务器再试一轮。服务器连不上时用户无需手动 ``ping``。
|
||||||
|
``auto_reconnect=False`` 时两阶段都不触发,直接抛出原异常。
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
@@ -326,11 +353,22 @@ class TdxClient:
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
time.sleep(delay)
|
time.sleep(delay)
|
||||||
self._conn.close()
|
self._reconnect()
|
||||||
self._conn = TdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
self._conn.connect()
|
return self._conn.execute(cmd)
|
||||||
if self._heartbeat_interval > 0:
|
except TdxConnectionError as e:
|
||||||
self._conn.start_heartbeat(self._heartbeat_interval)
|
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:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
@@ -681,13 +719,18 @@ class TdxClient:
|
|||||||
通达信这三个"统计指数"的计数类字段(涨/跌/平/总数/涨停/跌停家数)
|
通达信这三个"统计指数"的计数类字段(涨/跌/平/总数/涨停/跌停家数)
|
||||||
返回的是真实家数的 1/10,需统一 ×10 还原。成交额/量/市值字段不受影响。
|
返回的是真实家数的 1/10,需统一 ×10 还原。成交额/量/市值字段不受影响。
|
||||||
`suspended_count` 由 `total - up - down - neutral` 推得,用于保证计数守恒。
|
`suspended_count` 由 `total - up - down - neutral` 推得,用于保证计数守恒。
|
||||||
|
|
||||||
|
空数据容错:880005/880001/880006 并非所有服务器都提供,会返回空 quotes。
|
||||||
|
此时不只切换到延迟最低的一台(它可能也不提供),而是按延迟顺序逐台实测,
|
||||||
|
找到第一台返回有效数据的服务器,避免用户手动 ``easy-tdx ping``。
|
||||||
"""
|
"""
|
||||||
# 通达信中 880005 是全市场行情统计,880001 是总市值指数,880006 是涨跌停统计
|
# 通达信中 880005 是全市场行情统计,880001 是总市值指数,880006 是涨跌停统计
|
||||||
quotes = self._execute(
|
_cmd = GetSecurityQuotesCmd(
|
||||||
GetSecurityQuotesCmd(
|
[(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")]
|
||||||
[(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:
|
if not quotes:
|
||||||
raise RuntimeError("无法获取市场统计数据")
|
raise RuntimeError("无法获取市场统计数据")
|
||||||
q = quotes[0]
|
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(
|
def _collect_transaction_records(
|
||||||
self,
|
self,
|
||||||
fetch_page: Callable[[int, int], list[TransactionRecord]],
|
fetch_page: Callable[[int, int], list[TransactionRecord]],
|
||||||
@@ -912,8 +981,29 @@ class AsyncTdxClient(AsyncHeartbeatMixin):
|
|||||||
"""心跳使用的轻量请求(get_security_count,复用 _execute 重连)。"""
|
"""心跳使用的轻量请求(get_security_count,复用 _execute 重连)。"""
|
||||||
return self.get_security_count(Market.SH)
|
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:
|
async def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
|
两阶段韧性与 sync 版对称:先同主机重试(``_RETRY_DELAYS``),再跨主机
|
||||||
|
故障转移(重新测速切到另一台服务器)。整个流程在 ``_execute_lock`` 内
|
||||||
|
串行,避免并发请求触发多次故障转移抖动。``auto_reconnect=False`` 时
|
||||||
|
两阶段都不触发。
|
||||||
|
"""
|
||||||
async with self._execute_lock:
|
async with self._execute_lock:
|
||||||
try:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
@@ -923,9 +1013,22 @@ class AsyncTdxClient(AsyncHeartbeatMixin):
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
await asyncio.sleep(delay)
|
await asyncio.sleep(delay)
|
||||||
await self._conn.close()
|
await self._areconnect()
|
||||||
self._conn = AsyncTdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
await self._conn.connect()
|
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:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
@@ -1214,13 +1317,17 @@ class AsyncTdxClient(AsyncHeartbeatMixin):
|
|||||||
通达信这三个"统计指数"的计数类字段(涨/跌/平/总数/涨停/跌停家数)
|
通达信这三个"统计指数"的计数类字段(涨/跌/平/总数/涨停/跌停家数)
|
||||||
返回的是真实家数的 1/10,需统一 ×10 还原。成交额/量/市值字段不受影响。
|
返回的是真实家数的 1/10,需统一 ×10 还原。成交额/量/市值字段不受影响。
|
||||||
`suspended_count` 由 `total - up - down - neutral` 推得,用于保证计数守恒。
|
`suspended_count` 由 `total - up - down - neutral` 推得,用于保证计数守恒。
|
||||||
|
|
||||||
|
空数据容错:与 sync 版对称——空 quotes 时按延迟顺序逐台实测,找到首台
|
||||||
|
返回有效数据的服务器。
|
||||||
"""
|
"""
|
||||||
# 通达信中 880005 是全市场行情统计,880001 是总市值指数,880006 是涨跌停统计
|
# 通达信中 880005 是全市场行情统计,880001 是总市值指数,880006 是涨跌停统计
|
||||||
quotes = await self._execute(
|
_cmd = GetSecurityQuotesCmd(
|
||||||
GetSecurityQuotesCmd(
|
[(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006")]
|
||||||
[(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:
|
if not quotes:
|
||||||
raise RuntimeError("无法获取市场统计数据")
|
raise RuntimeError("无法获取市场统计数据")
|
||||||
q = quotes[0]
|
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(
|
async def _collect_transaction_records(
|
||||||
self,
|
self,
|
||||||
fetch_page: Callable[[int, int], Awaitable[list[TransactionRecord]]],
|
fetch_page: Callable[[int, int], Awaitable[list[TransactionRecord]]],
|
||||||
|
|||||||
@@ -9,7 +9,12 @@ from types import TracebackType
|
|||||||
from typing import TypeVar
|
from typing import TypeVar
|
||||||
|
|
||||||
from .._df import _apply_bar_time_align_bars, _category_to_minutes
|
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 ..commands.base import BaseCommand
|
||||||
from ..config import get_best_ex_host, get_ex_hosts, save_best_ex_host
|
from ..config import get_best_ex_host, get_ex_hosts, save_best_ex_host
|
||||||
from ..exceptions import TdxConnectionError
|
from ..exceptions import TdxConnectionError
|
||||||
@@ -120,8 +125,25 @@ class ExTdxClient:
|
|||||||
) -> None:
|
) -> None:
|
||||||
self.close()
|
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:
|
def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试(4 次,与 A 股/MAC 统一,审计 #2)。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
|
两阶段韧性与 A 股/MAC 统一(审计 #2):先同主机重试 4 次,再跨主机
|
||||||
|
故障转移(测速切到另一台扩展行情服务器)。
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
@@ -130,9 +152,22 @@ class ExTdxClient:
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
time.sleep(delay)
|
time.sleep(delay)
|
||||||
self._conn.close()
|
self._reconnect()
|
||||||
self._conn = ExTdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
self._conn.connect()
|
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:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
@@ -350,8 +385,22 @@ class AsyncExTdxClient(AsyncHeartbeatMixin):
|
|||||||
"""心跳使用的轻量请求(get_instrument_count,复用 _execute 重连)。"""
|
"""心跳使用的轻量请求(get_instrument_count,复用 _execute 重连)。"""
|
||||||
return self.get_instrument_count()
|
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:
|
async def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试(4 次,与 A 股/MAC 统一,审计 #2)。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。"""
|
||||||
async with self._execute_lock:
|
async with self._execute_lock:
|
||||||
try:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
@@ -361,9 +410,22 @@ class AsyncExTdxClient(AsyncHeartbeatMixin):
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
await asyncio.sleep(delay)
|
await asyncio.sleep(delay)
|
||||||
await self._conn.close()
|
await self._areconnect()
|
||||||
self._conn = AsyncExTdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
await self._conn.connect()
|
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:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
|
|||||||
@@ -15,7 +15,12 @@ from typing import Any, TypeVar
|
|||||||
import pandas as pd
|
import pandas as pd
|
||||||
|
|
||||||
from .._df import _to_df
|
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 ..commands.base import BaseCommand
|
||||||
from ..config import get_best_mac_ex_host, get_mac_ex_hosts, save_best_mac_ex_host
|
from ..config import get_best_mac_ex_host, get_mac_ex_hosts, save_best_mac_ex_host
|
||||||
from ..exceptions import TdxConnectionError
|
from ..exceptions import TdxConnectionError
|
||||||
@@ -148,11 +153,12 @@ class MacExClient:
|
|||||||
self._conn.execute(MacExLoginCmd())
|
self._conn.execute(MacExLoginCmd())
|
||||||
|
|
||||||
def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试(4 次,与 A 股/MAC 统一,审计 #2)。
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
每次重连后必须重新 ``_login()``(MAC 协议扩展行情特有)。登录握手期的
|
每次重连后必须重新 ``_login()``(MAC 协议扩展行情特有)。登录握手期的
|
||||||
``TdxConnectionError`` 与业务请求一样计入退避重试;``TdxCommandError``
|
``TdxConnectionError`` 与业务请求一样计入退避重试;``TdxCommandError``
|
||||||
(登录被拒等确定性失败)不重试,直接抛出。
|
(登录被拒等确定性失败)不重试,直接抛出。跨主机故障转移阶段同样遵循
|
||||||
|
``connect + login`` 纳入重试的语义。
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
@@ -173,6 +179,27 @@ class MacExClient:
|
|||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
last_exc = 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]
|
raise last_exc # type: ignore[misc]
|
||||||
|
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
@@ -585,7 +612,7 @@ class AsyncMacExClient(AsyncHeartbeatMixin):
|
|||||||
await self._conn.execute(MacExLoginCmd())
|
await self._conn.execute(MacExLoginCmd())
|
||||||
|
|
||||||
async def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
async def _execute(self, cmd: "BaseCommand[_T]") -> _T:
|
||||||
"""执行命令;断线时指数退避重试(4 次,与 A 股/MAC 统一,审计 #2)。
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
每次重连后必须重新 ``_login()``(MAC 协议扩展行情特有)。登录握手期的
|
每次重连后必须重新 ``_login()``(MAC 协议扩展行情特有)。登录握手期的
|
||||||
``TdxConnectionError`` 与业务请求一样计入退避重试;``TdxCommandError``
|
``TdxConnectionError`` 与业务请求一样计入退避重试;``TdxCommandError``
|
||||||
@@ -611,6 +638,27 @@ class AsyncMacExClient(AsyncHeartbeatMixin):
|
|||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
last_exc = 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]
|
raise last_exc # type: ignore[misc]
|
||||||
|
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
|
|||||||
+82
-23
@@ -13,7 +13,12 @@ from typing import Any, TypeVar
|
|||||||
import pandas as pd
|
import pandas as pd
|
||||||
|
|
||||||
from .._df import _apply_bar_time_align_df, _period_to_minutes, _to_df
|
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 ..codec.bitmap import Fields, PresetField
|
||||||
from ..commands.base import BaseCommand
|
from ..commands.base import BaseCommand
|
||||||
from ..config import (
|
from ..config import (
|
||||||
@@ -224,12 +229,23 @@ class MacClient:
|
|||||||
try:
|
try:
|
||||||
self._execute(KlineOffsetCmd(0, 1))
|
self._execute(KlineOffsetCmd(0, 1))
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
self._conn.stop_heartbeat()
|
self._reconnect()
|
||||||
self._conn.close()
|
|
||||||
self._conn = TdxConnection(self._host, self._port, self._timeout)
|
def _reconnect(self, host: str | None = None) -> None:
|
||||||
self._conn.connect()
|
"""关闭当前连接并重建(默认连 self._host;传入 host 则切换主机)。
|
||||||
if self._heartbeat_interval > 0:
|
|
||||||
self._conn.start_heartbeat(self._heartbeat_interval)
|
与 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:
|
def __enter__(self) -> MacClient:
|
||||||
self.connect()
|
self.connect()
|
||||||
@@ -244,11 +260,15 @@ class MacClient:
|
|||||||
self.close()
|
self.close()
|
||||||
|
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
# 内部执行:含自动重连
|
# 内部执行:含自动重连 + 跨主机故障转移
|
||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
|
|
||||||
def _execute(self, cmd: BaseCommand[_T]) -> _T:
|
def _execute(self, cmd: BaseCommand[_T]) -> _T:
|
||||||
"""执行命令;断线时指数退避重试。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。
|
||||||
|
|
||||||
|
两阶段韧性与 TdxClient 对称:先同主机重试(``_RETRY_DELAYS``),
|
||||||
|
再跨主机故障转移(重新测速切到另一台 MAC 服务器)。
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
@@ -257,11 +277,24 @@ class MacClient:
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
time.sleep(delay)
|
time.sleep(delay)
|
||||||
self._conn.close()
|
self._reconnect()
|
||||||
self._conn = TdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
self._conn.connect()
|
return self._conn.execute(cmd)
|
||||||
if self._heartbeat_interval > 0:
|
except TdxConnectionError as e:
|
||||||
self._conn.start_heartbeat(self._heartbeat_interval)
|
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:
|
try:
|
||||||
return self._conn.execute(cmd)
|
return self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
@@ -1234,11 +1267,22 @@ class AsyncMacClient(AsyncHeartbeatMixin):
|
|||||||
try:
|
try:
|
||||||
await self._execute(KlineOffsetCmd(0, 1))
|
await self._execute(KlineOffsetCmd(0, 1))
|
||||||
except TdxConnectionError:
|
except TdxConnectionError:
|
||||||
await self._stop_heartbeat()
|
await self._areconnect()
|
||||||
await self._conn.close()
|
|
||||||
self._conn = AsyncTdxConnection(self._host, self._port, self._timeout)
|
async def _areconnect(self, host: str | None = None) -> None:
|
||||||
await self._conn.connect()
|
"""关闭当前连接并重建(默认连 self._host;传入 host 则切换主机)。
|
||||||
self._start_heartbeat()
|
|
||||||
|
与 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:
|
async def __aenter__(self) -> AsyncMacClient:
|
||||||
await self.connect()
|
await self.connect()
|
||||||
@@ -1265,7 +1309,7 @@ class AsyncMacClient(AsyncHeartbeatMixin):
|
|||||||
# ------------------------------------------------------------------ #
|
# ------------------------------------------------------------------ #
|
||||||
|
|
||||||
async def _execute(self, cmd: BaseCommand[_T]) -> _T:
|
async def _execute(self, cmd: BaseCommand[_T]) -> _T:
|
||||||
"""执行命令;断线时指数退避重试。"""
|
"""执行命令;断线时指数退避重试,同主机耗尽则跨主机故障转移。"""
|
||||||
async with self._execute_lock:
|
async with self._execute_lock:
|
||||||
try:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
@@ -1275,9 +1319,24 @@ class AsyncMacClient(AsyncHeartbeatMixin):
|
|||||||
last_exc: TdxConnectionError | None = None
|
last_exc: TdxConnectionError | None = None
|
||||||
for delay in _RETRY_DELAYS:
|
for delay in _RETRY_DELAYS:
|
||||||
await asyncio.sleep(delay)
|
await asyncio.sleep(delay)
|
||||||
await self._conn.close()
|
await self._areconnect()
|
||||||
self._conn = AsyncTdxConnection(self._host, self._port, self._timeout)
|
try:
|
||||||
await self._conn.connect()
|
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:
|
try:
|
||||||
return await self._conn.execute(cmd)
|
return await self._conn.execute(cmd)
|
||||||
except TdxConnectionError as e:
|
except TdxConnectionError as e:
|
||||||
|
|||||||
@@ -10,8 +10,6 @@ from __future__ import annotations
|
|||||||
import pathlib
|
import pathlib
|
||||||
import struct
|
import struct
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
FIXTURES = pathlib.Path(__file__).parent.parent / "fixtures"
|
FIXTURES = pathlib.Path(__file__).parent.parent / "fixtures"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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"]
|
||||||
Reference in New Issue
Block a user