From 6b98715a14812f4800abcbc3380174ffa9d389da Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Fri, 10 Jul 2026 19:37:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(client):=20get=5Findex=5Fbars/get=5Fsecurit?= =?UTF-8?q?y=5Fbars=20=E5=8A=A0=E7=A9=BA=E6=95=B0=E6=8D=AE=E6=95=85?= =?UTF-8?q?=E9=9A=9C=E8=BD=AC=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 指数/板块指数（880xxx 等）并非所有服务器都提供，延迟最低的不一定返回数据。 实测约 6/47 台可达主机对所有指数返回空 body 且不报错，from_best_host() 按纯 延迟选主会选中这种"快但空"的主机，导致 get_index_bars 返回空 DataFrame。 get_market_stat 早有 _find_host_returning_quotes 兜底，但 K 线查询缺失同类逻辑。 本提交补齐对称设计： - 新增 _find_host_returning_bars（sync + async）：照搬 _find_host_returning_quotes 模式，复用 find_working_host_sync/_async（最多试 5 台、带持久化、异常吞掉、 全失败回退原 host） - get_security_bars / get_index_bars（sync + async 共 4 处）接入：空数据且 auto_reconnect 开启时触发逐台实测，找首台返回数据的 host - 新增 TestIndexBarsEmptyFailover（5 个测试）：覆盖空数据切 host、全候选空、 非空不触发、auto_reconnect=False 不触发、个股 K 线同链路 验证：23 个 failover 测试全绿，941 个单元测试无回归。 --- src/easy_tdx/client.py | 73 +++++++++++++++++++-- tests/unit/test_failover.py | 123 +++++++++++++++++++++++++++++++++++- 2 files changed, 191 insertions(+), 5 deletions(-) diff --git a/src/easy_tdx/client.py b/src/easy_tdx/client.py index 23811b5..dbcb8e6 100644 --- a/src/easy_tdx/client.py +++ b/src/easy_tdx/client.py @@ -506,7 +506,13 @@ class TdxClient: （= 开始 + 周期时长，与 Tushare/同花顺对齐，上午最后一根标 11:30）。 仅对分钟级周期生效；日线及以上不受影响。 """ - df = _to_df(self._execute(GetSecurityBarsCmd(market, code, category, start, count))) + cmd = GetSecurityBarsCmd(market, code, category, start, count) + bars = self._execute(cmd) + # 空数据故障转移：部分服务器对所有证券返回空 body（不报错），延迟最低的不 + # 一定有数据，故空时按延迟逐台实测找首台返回数据的 host。 + if not bars and self._auto_reconnect: + bars = self._find_host_returning_bars(cmd) + df = _to_df(bars) delta = _category_to_minutes(int(category)) is_intraday = delta is not None df = _apply_bar_time_align_df( @@ -533,7 +539,13 @@ class TdxClient: Args: bar_time: 见 :meth:`get_security_bars`，分钟级周期时间戳可对齐 Tushare 右端点。 """ - df = _to_df(self._execute(GetIndexBarsCmd(market, code, category, start, count))) + cmd = GetIndexBarsCmd(market, code, category, start, count) + bars = self._execute(cmd) + # 空数据故障转移：指数/板块指数（880xxx 等）并非所有服务器都提供，延迟最低 + # 的不一定返回数据，故空时按延迟逐台实测找首台返回数据的 host。 + if not bars and self._auto_reconnect: + bars = self._find_host_returning_bars(cmd) + df = _to_df(bars) delta = _category_to_minutes(int(category)) is_intraday = delta is not None df = _apply_bar_time_align_df( @@ -783,6 +795,31 @@ class TdxClient: # _try 已把 client 切到 new_host 并执行过 cmd，重新取一次拿结果 return self._execute(cmd) + def _find_host_returning_bars(self, cmd: "BaseCommand[list[SecurityBar]]") -> list[SecurityBar]: + """空数据故障转移（K 线类）：与 :meth:`_find_host_returning_quotes` 同模式。 + + 指数/板块指数（880xxx 等）并非所有服务器都提供，延迟最低的不一定返回 + 数据（实测约 1/8 的服务器对所有指数返回空 body 且不报错）。故 K 线查询 + 空数据时按延迟顺序逐台实测，返回首台返回非空 bars 的结果。最多尝试 + ``_WORKING_HOST_MAX_ATTEMPTS`` 台；全失败回退原 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]], @@ -1141,7 +1178,11 @@ class AsyncTdxClient(AsyncHeartbeatMixin): bar_time: str = "start", ) -> pd.DataFrame: """获取 K 线数据。``bar_time`` 见同步版 :meth:`get_security_bars`。""" - df = _to_df(await self._execute(GetSecurityBarsCmd(market, code, category, start, count))) + cmd = GetSecurityBarsCmd(market, code, category, start, count) + bars = await self._execute(cmd) + if not bars and self._auto_reconnect: + bars = await self._find_host_returning_bars(cmd) + df = _to_df(bars) delta = _category_to_minutes(int(category)) is_intraday = delta is not None df = _apply_bar_time_align_df( @@ -1164,7 +1205,11 @@ class AsyncTdxClient(AsyncHeartbeatMixin): bar_time: str = "start", ) -> pd.DataFrame: """获取指数 K 线数据。``bar_time`` 见同步版 :meth:`get_index_bars`。""" - df = _to_df(await self._execute(GetIndexBarsCmd(market, code, category, start, count))) + cmd = GetIndexBarsCmd(market, code, category, start, count) + bars = await self._execute(cmd) + if not bars and self._auto_reconnect: + bars = await self._find_host_returning_bars(cmd) + df = _to_df(bars) delta = _category_to_minutes(int(category)) is_intraday = delta is not None df = _apply_bar_time_align_df( @@ -1374,6 +1419,26 @@ class AsyncTdxClient(AsyncHeartbeatMixin): return [] return await self._execute(cmd) + async def _find_host_returning_bars( + self, cmd: "BaseCommand[list[SecurityBar]]" + ) -> list[SecurityBar]: + """空数据故障转移（K 线类，async）：与 sync ``_find_host_returning_bars`` 对称。""" + 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/tests/unit/test_failover.py b/tests/unit/test_failover.py index bda709e..0b38af3 100644 --- a/tests/unit/test_failover.py +++ b/tests/unit/test_failover.py @@ -24,7 +24,8 @@ from easy_tdx._reconnect import ( 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 +from easy_tdx.models.bar import SecurityBar +from easy_tdx.models.enums import KlineCategory, Market # --------------------------------------------------------------------------- # # select_best_host_sync 单元逻辑 @@ -396,3 +397,123 @@ class TestMarketStatEmptyFailover: # find_working_host 逐台实测了 hostA、hostB（_reconnect 被各调一次） reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list] assert reconnect_hosts == ["hostA", "hostB"] + + +# --------------------------------------------------------------------------- # +# get_index_bars / get_security_bars 空数据故障转移 +# （与 TestMarketStatEmptyFailover 对称：指数/板块指数 880xxx 并非所有服务器都提供） +# --------------------------------------------------------------------------- # + + +class TestIndexBarsEmptyFailover: + """K 线空数据故障转移——验证 get_index_bars/get_security_bars 空时逐台实测切 host。""" + + def setup_method(self) -> None: + import easy_tdx._reconnect as r + + r._last_failover_ts = 0.0 + + def _make_bar(self) -> SecurityBar: + """构造一根字段合法的日 K，让 get_index_bars 下游处理走通。""" + return SecurityBar( + open=10.0, + close=10.5, + high=10.8, + low=9.9, + vol=1000.0, + amount=10500.0, + year=2026, + month=7, + day=10, + hour=15, + minute=0, + ) + + def test_empty_bars_finds_working_host_and_returns_data(self) -> None: + """空 bars 时按延迟顺序逐台实测，找到返回数据的 host。""" + bar = self._make_bar() + 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=[[], [], [bar], [bar]]) 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_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10) + + # _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_bars_all_candidates_empty_returns_empty_df(self) -> None: + """所有候选都返回空时，返回空 DataFrame（不抛异常）。""" + 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)], + ), + ): + df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10) + + # find_working_host 逐台实测了 hostA、hostB（_reconnect 被各调一次） + reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list] + assert reconnect_hosts == ["hostA", "hostB"] + assert df.empty + + def test_non_empty_bars_does_not_trigger_failover(self) -> None: + """首次即返回数据时，不触发空数据故障转移。""" + bar = self._make_bar() + client = TdxClient("good-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + with ( + patch.object(client, "_execute", return_value=[bar]) as mock_exec, + patch.object(client, "_find_host_returning_bars") as mock_failover, + ): + df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10) + + assert mock_exec.call_count == 1 + mock_failover.assert_not_called() + assert len(df) == 1 + + def test_failover_disabled_when_auto_reconnect_off(self) -> None: + """auto_reconnect=False 时空数据不触发故障转移。""" + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=False, heartbeat_interval=0) + + with ( + patch.object(client, "_execute", return_value=[]) as mock_exec, + patch.object(client, "_find_host_returning_bars") as mock_failover, + ): + df = client.get_index_bars(Market.SH, "880008", KlineCategory.DAY, 0, 10) + + assert mock_exec.call_count == 1 + mock_failover.assert_not_called() + assert df.empty + + def test_security_bars_also_triggers_failover(self) -> None: + """get_security_bars（个股 K 线）同样接入空数据故障转移。""" + bar = self._make_bar() + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + with ( + patch.object(client, "_execute", side_effect=[[], [], [bar], [bar]]) as mock_exec, + patch.object(client, "_reconnect"), + patch( + "easy_tdx.client.ping_all", + return_value=[("hostA", 0.01), ("hostB", 0.02)], + ), + ): + df = client.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 10) + + assert mock_exec.call_count == 4 + assert len(df) == 1