fix(client): 历史资金流当日行全零 + 主力净额列缺失（issue #52）

三个根因（全部实测核实）： 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 双客户端同步修改；更新示例与三份文档；重写/新增回归测试（当日实时逐笔路径、主力净额列断言）。
2026-09-12 21:34:21 +08:00 · 2026-08-26 15:00:53 +08:00
parent 7c9e19de93
commit 574ffdd2a4
8 changed files with 158 additions and 227 deletions
@@ -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)
@@ -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(
-            "<HIHHHH6sHHHHIIH",
-            0x010C,
-            0x01016408,
-            0x001C,
-            0x001C,
-            0x052D,
-            int(self.market),
-            self.code,
-            22,
-            1,
-            self.start,
-            self.count,
-            0,
-            0,
-            0,
-        )
-
-    def parse_response(self, body: bytes) -> list[HistoricalFundFlow]:
-        # 响应格式：9字节头 + 2字节数量 + 每条记录 36 字节
-        if len(body) < 11:
-            return []
-
-        (num,) = struct.unpack("<H", body[9:11])
-        pos = 11
-        results = []
-
-        for _ in range(num):
-            if len(body) < pos + 36:
-                break
-
-            # 记录格式：4字节日期 + 8个4字节自定义浮点金额
-            # [0]日期, [1..4]流入(超/大/中/小), [5..8]流出(超/大/中/小)
-            raw_data = struct.unpack("<IIIIIIIII", body[pos : pos + 36])
-
-            raw_date = raw_data[0]
-            year = raw_date // 10000
-            month = (raw_date // 100) % 100
-            day = raw_date % 100
-
-            results.append(
-                HistoricalFundFlow(
-                    year=year,
-                    month=month,
-                    day=day,
-                    super_in=_decode_volume(raw_data[1]),
-                    large_in=_decode_volume(raw_data[2]),
-                    medium_in=_decode_volume(raw_data[3]),
-                    small_in=_decode_volume(raw_data[4]),
-                    super_out=_decode_volume(raw_data[5]),
-                    large_out=_decode_volume(raw_data[6]),
-                    medium_out=_decode_volume(raw_data[7]),
-                    small_out=_decode_volume(raw_data[8]),
-                )
-            )
-            pos += 36
-
-        return results