From 574ffdd2a4d3d9e55b419d443d29f6e330e8ee95 Mon Sep 17 00:00:00 2001 From: GitHub Date: Wed, 26 Aug 2026 15:00:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(client):=20=E5=8E=86=E5=8F=B2=E8=B5=84?= =?UTF-8?q?=E9=87=91=E6=B5=81=E5=BD=93=E6=97=A5=E8=A1=8C=E5=85=A8=E9=9B=B6?= =?UTF-8?q?=20+=20=E4=B8=BB=E5=8A=9B=E5=87=80=E9=A2=9D=E5=88=97=E7=BC=BA?= =?UTF-8?q?=E5=A4=B1=EF=BC=88issue=20#52=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三个根因（全部实测核实）： 1. Category 22 直连接口为虚构协议——52 台已知服务器中 46 台可达的 全部仅回 2 字节空包，从未成功过；移除死代码与臆造解析格式。 2. 历史逐笔接口当日数据要收盘清算后才有，日 K 盘中已含当日 bar， 故 start=0 的最新一行恒为全 0——当日 bar 改走当日实时逐笔接口。 3. main_net_inflow 此前仅为 dataclass property，asdict 静默丢弃， 返回 DataFrame 无主力净额列——新增 _fund_flow_df_with_net 物化 （history 紧随 date 列、当日快照放首列）。 sync + async 双客户端同步修改；更新示例与三份文档；重写/新增回归 测试（当日实时逐笔路径、主力净额列断言）。 --- docs/api_reference.md | 10 +- docs/field_mapping.md | 4 +- docs/protocol-unknown-fields.md | 13 ++- examples/08_fund_flow/history_fund_flow.py | 24 +++-- src/easy_tdx/client.py | 112 +++++++++++++-------- src/easy_tdx/commands/fund_flow.py | 79 --------------- tests/unit/test_a_share_extensions.py | 85 +++++++++++----- tests/unit/test_protocol_fixes.py | 58 ----------- 8 files changed, 158 insertions(+), 227 deletions(-) delete mode 100644 src/easy_tdx/commands/fund_flow.py diff --git a/docs/api_reference.md b/docs/api_reference.md index fd4d24f..412f478 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -329,10 +329,10 @@ c.get_block_info(filename: str) -> list[TdxBlock] ### get_fund_flow ```python -c.get_fund_flow(market: Market, code: str) -> FundFlow +c.get_fund_flow(market: Market, code: str) -> pd.DataFrame ``` -获取个股当日资金流向（基于 L1 逐笔数据统计）。 +获取个股当日资金流向（基于 L1 逐笔数据统计）。返回含 `main_net_inflow`（主力净流入）列。 **资金分级**： | 级别 | 单笔成交额 | @@ -346,10 +346,12 @@ c.get_fund_flow(market: Market, code: str) -> FundFlow ```python c.get_history_fund_flow(market: Market, code: str, - start: int, count: int) -> list[HistoricalFundFlow] + start: int, count: int) -> pd.DataFrame ``` -获取历史日线资金流向序列。优先走直连接口，若服务器不支持则自动回退为逐笔成交重算。 +获取历史日线资金流向序列，由"日K线取日期 + 逐笔成交重算"实现（标准服务器 +无资金流专用指令，Issue #52）。当日 bar 盘中取当日实时逐笔。返回列含 +`main_net_inflow`（主力净流入，单位元）。 --- diff --git a/docs/field_mapping.md b/docs/field_mapping.md index 31ebc99..42aa4c3 100644 --- a/docs/field_mapping.md +++ b/docs/field_mapping.md @@ -260,6 +260,7 @@ | 字段名 | 中文 | 类型 | 说明 | |--------|------|------|------| +| `main_net_inflow` | 主力净流入 | `float` | 列：超大+大净额（Issue #52 起物化为列） | | `super_in` | 超大单流入 | `float` | 单笔 >100 万 | | `super_out` | 超大单流出 | `float` | | | `large_in` | 大单流入 | `float` | 20~100 万 | @@ -268,7 +269,6 @@ | `medium_out` | 中单流出 | `float` | | | `small_in` | 小单流入 | `float` | ≤4 万 | | `small_out` | 小单流出 | `float` | | -| `main_net_inflow` | 主力净流入 | `float` | 属性：超大+大 | | `total_net_inflow` | 全单净流入 | `float` | 属性：全部 | --- @@ -277,7 +277,7 @@ 来源：`get_history_fund_flow()` -字段同 FundFlow，额外包含： +字段同 FundFlow（`main_net_inflow` 同样物化为列，紧随 `date` 之后），额外包含： | 字段名 | 中文 | 类型 | |--------|------|------| diff --git a/docs/protocol-unknown-fields.md b/docs/protocol-unknown-fields.md index 83b80be..61a60f7 100644 --- a/docs/protocol-unknown-fields.md +++ b/docs/protocol-unknown-fields.md @@ -231,12 +231,15 @@ gotdx 同样命名为 `Unknown`，pytdx 直接 `pos += 9`。 fixture 中 87 条记录全部为 0x00（87/87）。无其他可能值样本，推测为保留/对齐字节。 gotdx 命名为 `Unknown`，pytdx 注释 `# noused`。 -### 4.3 fund_flow 9字节响应头部（已确认：同 xdxr_info 格式） +### 4.3 fund_flow（Category 22）——已证伪并移除（Issue #52） -**位置**：`commands/fund_flow.py:45` +**位置**：`commands/fund_flow.py`（已删除） -与 xdxr_info 相同的 9 字节头部格式：prefix(2) + market(1) + code(6)。 -后接 uint16 num（记录数量），然后是 36 字节/条的固定记录。 +曾有实现假设 0x052D + category=22 为"历史资金流向"指令，并按 +"9 字节头 + uint16 num + 36 字节/条"解析。**2026-08-26 实测**（Issue #52）： +全部 46 台可达标准行情服务器对该请求仅回 2 字节 body（`0000` 或 `2003`）， +即 0 条记录 / ret_count 撒谎空包，从未在任何环境返回过有效数据。 +响应格式属臆造，该命令已删除；资金流改为"日K取日期 + 逐笔成交重算"。 gotdx 未实现此命令。 @@ -374,7 +377,7 @@ xdxr_info 9字节头部（fixture 验证）： xdxr_info 每条1字节 padding： - 87 条记录全部为 0x00，保留/对齐字节 -fund_flow 9字节头部：同 xdxr_info 格式（prefix + market + code） +fund_flow 9字节头部：~~同 xdxr_info 格式~~ 2026-08-26 证伪（见 4.3，命令已删除） block 384字节头部：gotdx 命名"头信息, 忽略"，所有实现跳过 diff --git a/examples/08_fund_flow/history_fund_flow.py b/examples/08_fund_flow/history_fund_flow.py index 779f883..32407fa 100644 --- a/examples/08_fund_flow/history_fund_flow.py +++ b/examples/08_fund_flow/history_fund_flow.py @@ -2,18 +2,20 @@ 使用 TdxClient 标准协议客户端，调用 get_history_fund_flow() 获取个股历史每日资金流向。 返回 HistoricalFundFlow DataFrame，每行代表一个交易日的资金流向数据。 -优先走 Category 22 直连接口；若服务器返回空，自动回退为日K线+逐笔重算。 +实现为"日K线取日期 + 逐笔成交重算"（通达信标准服务器无资金流专用指令， +Category 22 实测全部服务器仅回空包，见 Issue #52）。 DataFrame 列说明: - date str 交易日期（datetime） - super_in float 超大单流入（元） - super_out float 超大单流出（元） - large_in float 大单流入（元） - large_out float 大单流出（元） - medium_in float 中单流入（元） - medium_out float 中单流出（元） - small_in float 小单流入（元） - small_out float 小单流出（元） + date str 交易日期（datetime） + main_net_inflow float 主力净流入 = (超大单+大单)流入 - 流出（元） + super_in float 超大单流入（元） + super_out float 超大单流出（元） + large_in float 大单流入（元） + large_out float 大单流出（元） + medium_in float 中单流入（元） + medium_out float 中单流出（元） + small_in float 小单流入（元） + small_out float 小单流出（元） 资金级别划分（按单笔成交金额）: 超大单: > 100 万元 @@ -24,7 +26,7 @@ DataFrame 列说明: 数据特点: - start 为偏移量，0=最近交易日，count 为请求数量 - 金额单位为元 - - 部分服务器不支持 Category 22，此时自动回退到逐笔重算模式（较慢） + - 当日行情：盘中走当日实时逐笔接口（收盘清算后历史逐笔接口才有当日数据） """ from easy_tdx import Market, TdxClient diff --git a/src/easy_tdx/client.py b/src/easy_tdx/client.py index df2f022..1bf7022 100644 --- a/src/easy_tdx/client.py +++ b/src/easy_tdx/client.py @@ -39,7 +39,6 @@ from .commands.base import BaseCommand from .commands.block_info import GetBlockInfoCmd, GetBlockInfoMetaCmd from .commands.company_info import GetCompanyInfoCategoryCmd, GetCompanyInfoContentCmd from .commands.finance_info import GetFinanceInfoCmd -from .commands.fund_flow import GetHistoryFundFlowCmd from .commands.minute_time import GetHistoryMinuteTimeDataCmd from .commands.report_file import GetReportFileCmd from .commands.security_bars import GetIndexBarsCmd, GetSecurityBarsCmd @@ -87,6 +86,26 @@ def _today_in_shanghai() -> int: return int(datetime.now(_SHANGHAI_TZ).strftime("%Y%m%d")) +def _fund_flow_df_with_net(df: pd.DataFrame) -> pd.DataFrame: + """为资金流 DataFrame 物化主力净额列。 + + ``HistoricalFundFlow.main_net_inflow`` / ``FundFlow.main_net_inflow`` 是 + dataclass property，``_to_df`` 的 asdict 会静默丢弃（Issue #52：用户 + "取不到主力净额"的直接原因），这里显式物化为 ``main_net_inflow`` 列 + （单位：元，正=净流入）。放在 date 列之后（无 date 时放首列）。 + """ + if df.empty or "super_in" not in df.columns: + return df + out = df.copy() + pos = 1 if "date" in out.columns else 0 + out.insert( + pos, + "main_net_inflow", + (out["super_in"] + out["large_in"]) - (out["super_out"] + out["large_out"]), + ) + return out + + def _record_signature( record: TransactionRecord, ) -> tuple[int, int, float, int, int, int]: @@ -871,31 +890,32 @@ class TdxClient: return all_recs def get_fund_flow(self, market: Market, code: str) -> pd.DataFrame: - """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" + """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。 + + 返回列含 ``main_net_inflow``（主力净流入，单位元）。 + """ records = self._collect_transaction_records( lambda start, page_size: self._execute( GetTransactionDataCmd(market, code, start, page_size) ), 2000, ) - return _to_df(_classify_fund_flow(records)) + return _fund_flow_df_with_net(_to_df(_classify_fund_flow(records))) def _fetch_fund_flow_records( self, market: Market, code: str, start: int, count: int ) -> list[HistoricalFundFlow]: - """在当前 host 上获取资金流记录（直连 + K 线回退）。 + """在当前 host 上获取资金流记录（日 K 线取日期 + 逐笔成交重算）。 - 优先走 Category 22 直连接口；空则回退为"日 K 线取日期 + 历史逐笔成交重算"。 - 返回空列表代表该 host 既无直连数据也无 K 线数据（或解析失败）。 + 通达信标准行情服务器没有"历史资金流向"专用指令：曾经的 Category 22 + 直连请求实测在全部已知服务器上仅返回 2 字节空包（Issue #52），已移除。 + 资金流一律由逐笔成交重算：历史日期走历史逐笔接口；当日成交在历史逐笔 + 接口里要收盘清算后才有，当日 bar 盘中改走当日实时逐笔接口——此前当日 + 行恒为全 0，用户"取不到最新主力净额"的直接原因（Issue #52）。 + 返回空列表代表该 host 无 K 线数据（或解析失败）。 """ - try: - direct = self._execute(GetHistoryFundFlowCmd(market, code, start, count)) - except Exception: - direct = [] - if direct: - return list(direct) - bars = self._execute(GetSecurityBarsCmd(market, code, KlineCategory.DAY, start, count)) + today = _today_in_shanghai() results: list[HistoricalFundFlow] = [] for bar in bars: date = _date_from_bar(bar) @@ -904,9 +924,12 @@ class TdxClient: def _fetch_page( page_start: int, page_size: int, _d: int = date ) -> list[TransactionRecord]: - return self._execute( - GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) - ) + cmd: BaseCommand[list[TransactionRecord]] + if _d == today: + cmd = GetTransactionDataCmd(market, code, page_start, page_size) + else: + cmd = GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) + return self._execute(cmd) records = self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) @@ -917,10 +940,13 @@ class TdxClient: ) -> pd.DataFrame: """获取个股历史日线资金流向序列。 - 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 - "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + 实现："日 K 线取日期 + 逐笔成交重算资金流"。通达信标准服务器无 + 资金流专用指令（Category 22 实测全空，见 Issue #52）；当日 bar 盘中 + 走当日实时逐笔接口，收盘清算后走历史逐笔接口。 - 空数据故障转移（v1.20.5，Issue #41）：当前 host 直连与 K 线回退均空时， + 返回列含 ``main_net_inflow``（主力净流入 = 超大单+大单净额，单位元）。 + + 空数据故障转移（v1.20.5，Issue #41）：当前 host 无 K 线数据时， 按延迟顺序逐台实测找首台返回有效数据的服务器。部分服务器对常见标的也会 返回 ret_count 撒谎的空 body（日志"K线响应为空（声称 800 条...）"）， 此前直接返回空 DataFrame，用户拿不到数据；现复用 K 线故障转移的同源逻辑。 @@ -931,14 +957,14 @@ class TdxClient: # 空数据故障转移：与 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) + return _fund_flow_df_with_net(_to_df(results)) def _fund_flow_failover( self, market: Market, code: str, start: int, count: int ) -> list[HistoricalFundFlow]: """资金流空数据故障转移：逐台实测找首台返回有效数据的服务器。 - 与 ``_find_host_returning_data`` 区别：资金流获取涉及多命令（直连 / K 线 + + 与 ``_find_host_returning_data`` 区别：资金流获取涉及多命令（K 线 + 逐笔），无法用单个 cmd 复用泛化版；这里以内联 ``_try`` 在每台候选上跑完 整 ``_fetch_fund_flow_records``，返回首台非空结果。全失败返回空列表。 """ @@ -1536,33 +1562,31 @@ class AsyncTdxClient(AsyncHeartbeatMixin): return all_recs async def get_fund_flow(self, market: Market, code: str) -> pd.DataFrame: - """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" + """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。 + + 返回列含 ``main_net_inflow``（主力净流入，单位元）。 + """ records = await self._collect_transaction_records( lambda start, page_size: self._execute( GetTransactionDataCmd(market, code, start, page_size) ), 2000, ) - return _to_df(_classify_fund_flow(records)) + return _fund_flow_df_with_net(_to_df(_classify_fund_flow(records))) async def _fetch_fund_flow_records( self, market: Market, code: str, start: int, count: int ) -> list[HistoricalFundFlow]: - """在当前 host 上获取资金流记录（直连 + K 线回退，async）。 + """在当前 host 上获取资金流记录（日 K 线取日期 + 逐笔成交重算，async）。 - 优先走 Category 22 直连接口；空则回退为"日 K 线取日期 + 历史逐笔成交重算"。 - 返回空列表代表该 host 既无直连数据也无 K 线数据（或解析失败）。 + 同步版说明：无 Category 22 直连（实测全空，Issue #52）；当日 bar 盘中 + 走当日实时逐笔接口，收盘清算后走历史逐笔接口。 + 返回空列表代表该 host 无 K 线数据（或解析失败）。 """ - try: - direct = await self._execute(GetHistoryFundFlowCmd(market, code, start, count)) - except Exception: - direct = [] - if direct: - return list(direct) - bars = await self._execute( GetSecurityBarsCmd(market, code, KlineCategory.DAY, start, count) ) + today = _today_in_shanghai() results: list[HistoricalFundFlow] = [] for bar in bars: date = _date_from_bar(bar) @@ -1571,9 +1595,12 @@ class AsyncTdxClient(AsyncHeartbeatMixin): async def _fetch_page( page_start: int, page_size: int, _d: int = date ) -> list[TransactionRecord]: - return await self._execute( - GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) - ) + cmd: BaseCommand[list[TransactionRecord]] + if _d == today: + cmd = GetTransactionDataCmd(market, code, page_start, page_size) + else: + cmd = GetHistoryTransactionDataCmd(market, code, _d, page_start, page_size) + return await self._execute(cmd) records = await self._collect_transaction_records(_fetch_page, 800) results.append(_historical_fund_flow_from_records(date, records)) @@ -1584,24 +1611,25 @@ class AsyncTdxClient(AsyncHeartbeatMixin): ) -> pd.DataFrame: """获取个股历史日线资金流向序列。 - 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 - "日 K 线取日期 + 历史逐笔成交重算资金流"的兼容实现。 + 实现："日 K 线取日期 + 逐笔成交重算资金流"；当日 bar 盘中走当日实时 + 逐笔接口。返回列含 ``main_net_inflow``（主力净流入，单位元）。 - 空数据故障转移（v1.20.5，Issue #41）：当前 host 直连与 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) + return _fund_flow_df_with_net(_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``。 + 与 ``_find_host_returning_data`` 区别：资金流获取涉及多命令（K 线 + + 逐笔），无法用单个 cmd 复用泛化版；这里以内联 ``_try`` 在每台候选上 + 跑完整 ``_fetch_fund_flow_records``。 """ bad_host = self._host ranked = await asyncio.to_thread(ping_all, get_known_hosts(), self._port, 5.0) diff --git a/src/easy_tdx/commands/fund_flow.py b/src/easy_tdx/commands/fund_flow.py deleted file mode 100644 index c11900b..0000000 --- a/src/easy_tdx/commands/fund_flow.py +++ /dev/null @@ -1,79 +0,0 @@ -"""历史资金流向命令 (Category 22)。""" - -import struct - -from ..codec.volume import _decode_volume -from ..models.enums import Market -from ..models.stats import HistoricalFundFlow -from .base import BaseCommand - - -class GetHistoryFundFlowCmd(BaseCommand[list[HistoricalFundFlow]]): - """获取历史日线资金流向序列。""" - - def __init__(self, market: Market, code: str, start: int, count: int) -> None: - self.market = market - self.code = code.encode("utf-8") - self.start = start - self.count = count - - def build_request(self) -> bytes: - # Header (12 bytes) + Payload (28 bytes) = 40 bytes - return struct.pack( - " list[HistoricalFundFlow]: - # 响应格式：9字节头 + 2字节数量 + 每条记录 36 字节 - if len(body) < 11: - return [] - - (num,) = struct.unpack(" 0: + return [] + return realtime_txn + if isinstance(cmd, GetHistoryTransactionDataCmd): + if cmd.start > 0: + return [] + return history_txn.get(cmd.date, []) + return [] + + with patch.object(TdxClient, "_execute", side_effect=mock_execute): + flows = client.get_history_fund_flow(Market.SH, "600000", 0, 2) + + assert len(flows) == 2 + assert "GetTransactionDataCmd" in seen_cmds + today_row = flows.iloc[-1] + # 今日行来自实时逐笔：100 元 × 101 手 × 100 = 超大单流入 1010000 + assert today_row["super_in"] == 1010000.0 + assert today_row["main_net_inflow"] == 1010000.0 + # 昨日行来自历史逐笔：小单流入 10000 + assert flows.iloc[0]["small_in"] == 10000.0 @patch("easy_tdx.client.TdxConnection") diff --git a/tests/unit/test_protocol_fixes.py b/tests/unit/test_protocol_fixes.py index d6de902..1116c1a 100644 --- a/tests/unit/test_protocol_fixes.py +++ b/tests/unit/test_protocol_fixes.py @@ -1,10 +1,8 @@ """协议底层修复验证（针对 2026-04-15 审查结论）。""" import struct -from unittest.mock import patch from easy_tdx.codec.price_rules import compute_price_limits -from easy_tdx.commands.fund_flow import GetHistoryFundFlowCmd from easy_tdx.commands.security_bars import GetSecurityBarsCmd from easy_tdx.commands.security_list import GetSecurityListCmd from easy_tdx.commands.security_quotes import GetSecurityQuotesCmd @@ -38,33 +36,6 @@ def test_security_bars_exact_layout(): assert len(req) == 38 -def test_history_fund_flow_exact_layout(): - """验证历史资金流请求包布局与 K 线一致，只差 category=22。""" - cmd = GetHistoryFundFlowCmd(Market.SH, "600000", 0, 10) - req = cmd.build_request() - # Header: 0x010C, 0x01016408, 0x1C, 0x1C - # Payload: 0x052D, 1 (Market.SH), "600000", 22, 1, 0, 10, 0, 0, 0 - expected = struct.pack( - " float: - seen.append(raw) - return float(raw) - - with patch("easy_tdx.commands.fund_flow._decode_volume", side_effect=fake_decode): - records = GetHistoryFundFlowCmd(Market.SH, "600000", 0, 1).parse_response(bytes(body)) - - assert seen == raw_words - assert records[0].small_out == float(raw_words[-1])