From 0490f67a164fb6a573de603dc35d2499fc13a6c0 Mon Sep 17 00:00:00 2001 From: GitHub Date: Wed, 5 Aug 2026 15:53:49 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(client):=20get=5Fhistory=5Ffund=5Fflow?= =?UTF-8?q?=20=E5=8A=A0=E7=A9=BA=E6=95=B0=E6=8D=AE=E6=95=85=E9=9A=9C?= =?UTF-8?q?=E8=BD=AC=E7=A7=BB=EF=BC=88issue=20#41=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户反馈 get_history_fund_flow(SH, "600519") 返回空 DataFrame，日志显示 "K线响应为空（声称 800 条但首条即解析失败...）"。排查定位：当前 host 对 常见标的也返回 ret_count 撒谎的空 body，但资金流兼容回退路径（直连空 → 拉 K 线 + 历史逐笔重算）未接入 v1.20.4 的空数据故障转移，"服务器回包正常 但内容是假的空" 既非 TdxConnectionError 也不触发换台，用户卡在坏服务器上。 修复：get_history_fund_flow（sync+async）当前 host 直连与 K 线回退均空时， 按延迟顺序逐台实测找首台返回有效数据的服务器（与 get_security_bars/ get_index_bars 同源逻辑）。因资金流获取涉及多命令，无法用单 cmd 复用泛化版 _find_host_returning_data，故内联 _fund_flow_failover：每台候选上跑完整 _fetch_fund_flow_records（原直连+K线回退逻辑抽出成独立方法），返回首台非空 结果。全空返回空 DataFrame（不 raise，区分"真无历史数据"与"服务器缺数据"）。 测试：新增 6 个（4 sync + 2 async）。全套 989 passed；ruff/mypy 改动文件零错误。 --- CHANGELOG.md | 16 +++ pyproject.toml | 2 +- src/easy_tdx/client.py | 108 +++++++++++++++++--- tests/unit/test_failover.py | 191 ++++++++++++++++++++++++++++++++++++ 4 files changed, 304 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fb2bdc4..5a81d2b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ 本文件记录 easy-tdx 的版本变更。格式遵循 [Keep a Changelog](https://keepachangelog.com/zh-CN/)。 +## [1.20.5] — 2026-08-05 + +**资金流空数据故障转移**（Issue #41）—— 用户反馈 `get_history_fund_flow(SH, "600519")` 返回空 DataFrame，日志显示"K线响应为空（声称 800 条但首条即解析失败...）"。排查定位：当前 host 对常见标的也返回 `ret_count` 撒谎的空 body，但资金流这条兼容回退路径（直连空 → 拉 K 线 + 历史逐笔重算）**未接入 v1.20.4 的空数据故障转移**，"服务器回包正常但内容是假的空"既非 `TdxConnectionError` 也不触发换台，用户卡在坏服务器上拿不到数据。本次将资金流路径接入与 K 线同源的空数据故障转移。 + +### 修复 + +- **资金流空数据故障转移**（`src/easy_tdx/client.py`）—— `get_history_fund_flow`（sync+async）当前 host 直连（Category 22）与 K 线回退均空时，按延迟顺序逐台实测找首台返回有效数据的服务器（与 `get_security_bars`/`get_index_bars` 同源逻辑）。因资金流获取涉及多命令（直连 / K 线 + 逐笔），无法用单 cmd 复用泛化版 `_find_host_returning_data`，故内联 `_fund_flow_failover`：每台候选上跑完整 `_fetch_fund_flow_records`，返回首台非空结果。全空返回空 DataFrame（不 raise，区分"真无历史数据"与"服务器缺数据"）。`auto_reconnect=False` 时不触发。 + +### 重构 + +- **提取 `_fetch_fund_flow_records`**（`src/easy_tdx/client.py`）—— 将"直连 + K 线回退"逻辑从 `get_history_fund_flow` 抽出为独立方法（sync+async 对称），便于故障转移在内联 `_try` 中复用。行为不变。 + +### 测试 + +- 新增 6 个测试：`test_failover.py` 的 `TestFundFlowEmptyFailover`（4 个 sync：空数据切台命中 / 全空返回空 df / 首次非空不触发 / `auto_reconnect=False` 不触发）+ `TestAsyncFundFlowEmptyFailover`（2 个 async：空数据切台命中 / 首次非空不触发）。全套 989 passed；ruff/mypy 改动文件零错误。 + ## [1.20.4] — 2026-07-13 **引入服务器健康分引擎 + K线空数据故障转移**（PR #37）—— 彻底解决用户反馈的通达信服务器"跳来跳去"且指数 K 线取不到数据问题。此前代码库零服务器健康记忆（失败的服务器下次又会被低延迟选中），且指数 K 线空数据不触发故障转移（直接返回空 DataFrame）。本次新增进程级健康分引擎 + 泛化空数据转移 + 8 个 client 统一健康分联动。 diff --git a/pyproject.toml b/pyproject.toml index 0093138..06a41aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.20.4" +version = "1.20.5" description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/client.py b/src/easy_tdx/client.py index 9263e6c..df2f022 100644 --- a/src/easy_tdx/client.py +++ b/src/easy_tdx/client.py @@ -880,20 +880,20 @@ class TdxClient: ) return _to_df(_classify_fund_flow(records)) - def get_history_fund_flow( + def _fetch_fund_flow_records( self, market: Market, code: str, start: int, count: int - ) -> pd.DataFrame: - """获取个股历史日线资金流向序列。 + ) -> list[HistoricalFundFlow]: + """在当前 host 上获取资金流记录（直连 + K 线回退）。 - 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 - "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + 优先走 Category 22 直连接口；空则回退为"日 K 线取日期 + 历史逐笔成交重算"。 + 返回空列表代表该 host 既无直连数据也无 K 线数据（或解析失败）。 """ try: direct = self._execute(GetHistoryFundFlowCmd(market, code, start, count)) except Exception: direct = [] if direct: - return _to_df(direct) + return list(direct) bars = self._execute(GetSecurityBarsCmd(market, code, KlineCategory.DAY, start, count)) results: list[HistoricalFundFlow] = [] @@ -910,8 +910,54 @@ class TdxClient: records = self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) + return results + + def get_history_fund_flow( + self, market: Market, code: str, start: int, count: int + ) -> pd.DataFrame: + """获取个股历史日线资金流向序列。 + + 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 + "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + + 空数据故障转移（v1.20.5，Issue #41）：当前 host 直连与 K 线回退均空时， + 按延迟顺序逐台实测找首台返回有效数据的服务器。部分服务器对常见标的也会 + 返回 ret_count 撒谎的空 body（日志"K线响应为空（声称 800 条...）"）， + 此前直接返回空 DataFrame，用户拿不到数据；现复用 K 线故障转移的同源逻辑。 + 注意：真·无历史数据（如新股）所有服务器都返回空，此时换台仍为空，直接 + 返回空 DataFrame 而非 raise——避免把"该股票本就没数据"误报为故障。 + """ + results = self._fetch_fund_flow_records(market, code, start, count) + # 空数据故障转移：与 get_security_bars / get_index_bars 同源逻辑。 + if not results and self._auto_reconnect: + results = self._fund_flow_failover(market, code, start, count) return _to_df(results) + def _fund_flow_failover( + self, market: Market, code: str, start: int, count: int + ) -> list[HistoricalFundFlow]: + """资金流空数据故障转移：逐台实测找首台返回有效数据的服务器。 + + 与 ``_find_host_returning_data`` 区别：资金流获取涉及多命令（直连 / K 线 + + 逐笔），无法用单个 cmd 复用泛化版；这里以内联 ``_try`` 在每台候选上跑完 + 整 ``_fetch_fund_flow_records``，返回首台非空结果。全失败返回空列表。 + """ + bad_host = self._host + ranked = ping_all(get_known_hosts(), self._port, 5.0) + + def _try(host: str) -> bool: + self._reconnect(host) + return bool(self._fetch_fund_flow_records(market, code, start, count)) + + 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 已切到 new_host 并跑过一次，重新取一次拿结果 + return self._fetch_fund_flow_records(market, code, start, count) + # ============================================================ # 异步客户端 @@ -1499,20 +1545,20 @@ class AsyncTdxClient(AsyncHeartbeatMixin): ) return _to_df(_classify_fund_flow(records)) - async def get_history_fund_flow( + async def _fetch_fund_flow_records( self, market: Market, code: str, start: int, count: int - ) -> pd.DataFrame: - """获取个股历史日线资金流向序列。 + ) -> list[HistoricalFundFlow]: + """在当前 host 上获取资金流记录（直连 + K 线回退，async）。 - 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 - "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + 优先走 Category 22 直连接口；空则回退为"日 K 线取日期 + 历史逐笔成交重算"。 + 返回空列表代表该 host 既无直连数据也无 K 线数据（或解析失败）。 """ try: direct = await self._execute(GetHistoryFundFlowCmd(market, code, start, count)) except Exception: direct = [] if direct: - return _to_df(direct) + return list(direct) bars = await self._execute( GetSecurityBarsCmd(market, code, KlineCategory.DAY, start, count) @@ -1531,4 +1577,42 @@ class AsyncTdxClient(AsyncHeartbeatMixin): records = await self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) + return results + + async def get_history_fund_flow( + self, market: Market, code: str, start: int, count: int + ) -> pd.DataFrame: + """获取个股历史日线资金流向序列。 + + 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 + "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + + 空数据故障转移（v1.20.5，Issue #41）：当前 host 直连与 K 线回退均空时， + 按延迟顺序逐台实测找首台返回有效数据的服务器。 + """ + results = await self._fetch_fund_flow_records(market, code, start, count) + if not results and self._auto_reconnect: + results = await self._fund_flow_failover(market, code, start, count) return _to_df(results) + + async def _fund_flow_failover( + self, market: Market, code: str, start: int, count: int + ) -> list[HistoricalFundFlow]: + """资金流空数据故障转移（async）：逐台实测找首台返回有效数据的服务器。 + + 与 ``_find_host_returning_data`` 区别：资金流获取涉及多命令，无法用单个 + cmd 复用泛化版；这里以内联 ``_try`` 在每台候选上跑完整 ``_fetch_fund_flow_records``。 + """ + 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) + return bool(await self._fetch_fund_flow_records(market, code, start, count)) + + 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._fetch_fund_flow_records(market, code, start, count) diff --git a/tests/unit/test_failover.py b/tests/unit/test_failover.py index c756530..eb68b15 100644 --- a/tests/unit/test_failover.py +++ b/tests/unit/test_failover.py @@ -11,6 +11,7 @@ from __future__ import annotations +import asyncio from unittest.mock import MagicMock, patch import pytest @@ -27,6 +28,7 @@ from easy_tdx.commands.security_count import GetSecurityCountCmd from easy_tdx.exceptions import TdxConnectionError from easy_tdx.models.bar import SecurityBar from easy_tdx.models.enums import KlineCategory, Market +from easy_tdx.models.stats import HistoricalFundFlow @pytest.fixture(autouse=True) @@ -588,3 +590,192 @@ class TestBarsEmptyFailover: assert mock_exec.call_count == 4 assert len(df) == 1 + + +# --------------------------------------------------------------------------- # +# get_history_fund_flow 空数据故障转移（Issue #41） +# 部分服务器对常见标的（如 600519）也返回 ret_count 撒谎的空 body，此前直接返回 +# 空 DataFrame；v1.20.5 接入与 K 线同源的空数据故障转移，逐台实测找有效 host。 +# --------------------------------------------------------------------------- # + + +class TestFundFlowEmptyFailover: + """资金流空数据故障转移——验证 get_history_fund_flow 空时逐台实测切 host。""" + + def _make_flow(self) -> HistoricalFundFlow: + """构造一条字段合法的历史资金流，让 _to_df 下游处理走通。""" + return HistoricalFundFlow( + year=2026, + month=7, + day=10, + super_in=1.0, + super_out=0.0, + large_in=2.0, + large_out=0.0, + medium_in=3.0, + medium_out=0.0, + small_in=4.0, + small_out=0.0, + ) + + def test_empty_fund_flow_finds_working_host_and_returns_data(self) -> None: + """当前 host 空 → 逐台实测 → hostB 返回数据，停在该 host。""" + flow = self._make_flow() + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + # _fetch_fund_flow_records 调用序列： + # 1. 首次（bad-host）→ 空 + # 2. 验证 hostA → 空 + # 3. 验证 hostB → 非空（命中） + # 4. 最终再取一次（停在 hostB）→ 非空 + with ( + patch.object( + client, + "_fetch_fund_flow_records", + side_effect=[[], [], [flow], [flow]], + ) as mock_fetch, + patch.object(client, "_reconnect") as mock_reconnect, + patch( + "easy_tdx.client.ping_all", + return_value=[("hostA", 0.01), ("hostB", 0.02)], + ), + ): + df = client.get_history_fund_flow(Market.SH, "600519", 0, 10) + + assert mock_fetch.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_fund_flow_all_candidates_empty_returns_empty_df(self) -> None: + """所有候选都返回空时，返回空 DataFrame（不抛异常）。 + + 与 get_index_bars / get_security_bars 同源逻辑：真·无数据时换台仍为空， + 返回空而非 raise。``_reconnect`` 被 mock 不更新 ``self._host``，故不会 + 再切回 bad-host（与 K 线测试断言一致）。 + """ + client = TdxClient("bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + with ( + patch.object(client, "_fetch_fund_flow_records", 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_history_fund_flow(Market.SH, "600519", 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_fund_flow_does_not_trigger_failover(self) -> None: + """首次即返回数据时，不触发空数据故障转移。""" + flow = self._make_flow() + client = TdxClient("good-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0) + + with ( + patch.object(client, "_fetch_fund_flow_records", return_value=[flow]) as mock_fetch, + patch.object(client, "_fund_flow_failover") as mock_failover, + ): + df = client.get_history_fund_flow(Market.SH, "600519", 0, 10) + + assert mock_fetch.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, "_fetch_fund_flow_records", return_value=[]) as mock_fetch, + patch.object(client, "_fund_flow_failover") as mock_failover, + ): + df = client.get_history_fund_flow(Market.SH, "600519", 0, 10) + + assert mock_fetch.call_count == 1 + mock_failover.assert_not_called() + assert df.empty + + +# --------------------------------------------------------------------------- # +# AsyncTdxClient.get_history_fund_flow 空数据故障转移（Issue #41，async 对称） +# --------------------------------------------------------------------------- # + + +class TestAsyncFundFlowEmptyFailover: + """资金流空数据故障转移——async 版与 sync 语义对称。""" + + def _make_flow(self) -> HistoricalFundFlow: + return HistoricalFundFlow( + year=2026, + month=7, + day=10, + super_in=1.0, + super_out=0.0, + large_in=2.0, + large_out=0.0, + medium_in=3.0, + medium_out=0.0, + small_in=4.0, + small_out=0.0, + ) + + def test_async_empty_fund_flow_finds_working_host(self) -> None: + """当前 host 空 → 逐台实测 → hostB 返回数据。""" + from easy_tdx.client import AsyncTdxClient + + flow = self._make_flow() + + async def main() -> int: + client = AsyncTdxClient( + "bad-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0 + ) + with ( + patch.object( + client, + "_fetch_fund_flow_records", + side_effect=[[], [], [flow], [flow]], + ) as mock_fetch, + patch.object(client, "_areconnect") as mock_reconnect, + patch( + "easy_tdx.client.ping_all", + return_value=[("hostA", 0.01), ("hostB", 0.02)], + ), + ): + df = await client.get_history_fund_flow(Market.SH, "600519", 0, 10) + + assert mock_fetch.call_count == 4 + reconnect_hosts = [c.args[0] for c in mock_reconnect.call_args_list] + assert reconnect_hosts == ["hostA", "hostB"] + return len(df) + + assert asyncio.run(main()) == 1 + + def test_async_non_empty_fund_flow_does_not_trigger_failover(self) -> None: + """首次即返回数据时，不触发空数据故障转移。""" + from easy_tdx.client import AsyncTdxClient + + flow = self._make_flow() + + async def main() -> int: + client = AsyncTdxClient( + "good-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0 + ) + with ( + patch.object( + client, "_fetch_fund_flow_records", return_value=[flow] + ) as mock_fetch, + patch.object(client, "_fund_flow_failover") as mock_failover, + ): + df = await client.get_history_fund_flow(Market.SH, "600519", 0, 10) + + assert mock_fetch.call_count == 1 + mock_failover.assert_not_called() + return len(df) + + assert asyncio.run(main()) == 1 From 60bd0586f982503c5e6ec416fee62ba41b45bf4c Mon Sep 17 00:00:00 2001 From: GitHub Date: Wed, 5 Aug 2026 15:58:00 +0800 Subject: [PATCH 2/2] =?UTF-8?q?style(tests):=20ruff=20format=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20patch.object=20=E5=8F=82=E6=95=B0=E6=8D=A2=E8=A1=8C?= =?UTF-8?q?=EF=BC=88CI=20ruff=20format=20--check=20=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_failover.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/unit/test_failover.py b/tests/unit/test_failover.py index eb68b15..1f1b9b5 100644 --- a/tests/unit/test_failover.py +++ b/tests/unit/test_failover.py @@ -767,9 +767,7 @@ class TestAsyncFundFlowEmptyFailover: "good-host", 7709, 1.0, auto_reconnect=True, heartbeat_interval=0 ) with ( - patch.object( - client, "_fetch_fund_flow_records", return_value=[flow] - ) as mock_fetch, + patch.object(client, "_fetch_fund_flow_records", return_value=[flow]) as mock_fetch, patch.object(client, "_fund_flow_failover") as mock_failover, ): df = await client.get_history_fund_flow(Market.SH, "600519", 0, 10)