feat: 补全 A 股历史资金流向序列功能

1. 新增命令:实现 GetHistoryFundFlowCmd (0x052d Category 22) 用于拉取历史资金分布。
2. 新增模型:增加 HistoricalFundFlow 结构,支持超大/大/中/小单的双向统计。
3. 客户端 API:TdxClient/AsyncTdxClient 增加 get_history_fund_flow() 接口。
4. 单元测试:在 test_a_share_extensions.py 中增加响应包解析逻辑验证。
5. 文档更新:README.md 同步 API 及数据模型定义。
This commit is contained in:
M
2026-04-15 15:26:51 +08:00
parent 4dabbf6d15
commit daaba7dc13
5 changed files with 158 additions and 17 deletions
+9
View File
@@ -106,6 +106,7 @@ client = AsyncTdxClient.from_best_host(ping_timeout=5.0)
| `get_transaction_data(market, code, start, count=800)` | 当日逐笔成交(分页) |
| `get_history_transaction_data(market, code, date, start, count=800)` | 历史逐笔成交 |
| `get_fund_flow(market, code)` | 当日资金流向统计(超大/大/中/小单) |
| `get_history_fund_flow(market, code, start, count)` | 历史日线资金流向序列(Category 22 |
| `get_xdxr_info(market, code)` | 除权除息历史 |
| `get_finance_info(market, code)` | 最新财务数据 |
| `get_company_info_category(market, code)` | 公司信息文件目录 |
@@ -205,6 +206,14 @@ super_in/out large_in/out medium_in/out small_in/out
main_net_inflow total_net_inflow
```
### HistoricalFundFlow(历史资金流序列)
```
year month day
super_in/out large_in/out medium_in/out small_in/out
main_net_inflow
```
## 修复的 pytdx Bug
+14 -1
View File
@@ -8,6 +8,7 @@ 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.report_file import GetReportFileCmd
from .commands.minute_time import GetHistoryMinuteTimeDataCmd, GetMinuteTimeDataCmd
from .commands.security_bars import GetIndexBarsCmd, GetSecurityBarsCmd
@@ -24,7 +25,7 @@ from .models.enums import KlineCategory, Market
from .models.finance import CompanyInfoCategory, FinanceInfo, TdxBlock, XdxrRecord
from .models.quote import SecurityQuote
from .models.security import SecurityInfo
from .models.stats import FundFlow, MarketStat
from .models.stats import FundFlow, HistoricalFundFlow, MarketStat
from .models.timeseries import MinuteBar, TransactionRecord
from .transport.async_ import AsyncTdxConnection
from .transport.sync import KNOWN_HOSTS, TdxConnection, ping_all
@@ -362,6 +363,12 @@ class TdxClient:
return FundFlow(**stats)
def get_history_fund_flow(
self, market: Market, code: str, start: int, count: int
) -> list[HistoricalFundFlow]:
"""获取个股历史日线资金流向序列(Category 22)。"""
return self._execute(GetHistoryFundFlowCmd(market, code, start, count))
# ============================================================
# 异步客户端
@@ -664,3 +671,9 @@ class AsyncTdxClient:
stats[f"small_{direction}"] += amount
return FundFlow(**stats)
async def get_history_fund_flow(
self, market: Market, code: str, start: int, count: int
) -> list[HistoricalFundFlow]:
"""获取个股历史日线资金流向序列。"""
return await self._execute(GetHistoryFundFlowCmd(market, code, start, count))
+83
View File
@@ -0,0 +1,83 @@
"""历史资金流向命令 (Category 22)。"""
import struct
from .._binary import slice_bytes, unpack_from
from ..codec.volume import _decode_volume
from ..exceptions import TdxDecodeError
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:
# 使用 0x052d 指令(K 线类指令)
# 负载长度固定为 28 字节 (0x1c)
payload_len = 0x1c
header = struct.pack(
"<HIHHH",
0x010c,
0x01016408, # 注意此处标志位与普通行情略有不同
payload_len,
payload_len,
0x052d,
)
# 参数包:Market(B), Code(6s), Category(H=22), Unknown(H=1), Start(I), Count(I), 3个Unknown(H)
params = struct.pack(
"<B6sHHIIHHH",
int(self.market),
self.code,
22, # Category 22
1, # Unknown
self.start,
self.count,
0, 0, 0
)
return header + params
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]流出(超/大/中/小)
# 这里的 i 是 uint32 原始字节,随后需用 _decode_volume 解码
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
+24
View File
@@ -38,3 +38,27 @@ class FundFlow:
"""全单净流入。"""
return (self.super_in + self.large_in + self.medium_in + self.small_in) - \
(self.super_out + self.large_out + self.medium_out + self.small_out)
@dataclass
class HistoricalFundFlow:
"""历史日线资金流向条目。"""
year: int
month: int
day: int
# 金额项 (单位:元)
super_in: float
super_out: float
large_in: float
large_out: float
medium_in: float
medium_out: float
small_in: float
small_out: float
@property
def main_net_inflow(self) -> float:
"""当日主力净流入。"""
return (self.super_in + self.large_in) - (self.super_out + self.large_out)
+28 -16
View File
@@ -1,11 +1,13 @@
"""针对本轮 A 股增强功能的单元测试。"""
import pytest
import struct
from unittest.mock import patch, MagicMock, AsyncMock
from xmtdx import TdxClient, Market
from xmtdx.models.security import SecurityInfo
from xmtdx.models.timeseries import TransactionRecord
from xmtdx.models.quote import SecurityQuote
from xmtdx.models.stats import FundFlow, HistoricalFundFlow, MarketStat
@patch("xmtdx.client.TdxConnection")
def test_get_fund_flow_logic(mock_conn_cls):
@@ -14,14 +16,10 @@ def test_get_fund_flow_logic(mock_conn_cls):
client = TdxClient("127.0.0.1")
# 构造模拟 Tick 数据
# A股 1手=100股。
# 1. 超大单: 100元 * 100手 * 100 = 100万 (Buy)
# 2. 大单: 10元 * 250手 * 100 = 25万 (Sell)
# 3. 小单: 10元 * 10手 * 100 = 1万 (Buy)
mock_recs = [
TransactionRecord(10, 0, 100.0, 100, 0, 0), # super_in
TransactionRecord(10, 1, 10.0, 250, 1, 0), # large_out
TransactionRecord(10, 2, 10.0, 10, 0, 0), # small_in
TransactionRecord(10, 0, 100.0, 100, 0, 0), # super_in (100*100*100 = 100w)
TransactionRecord(10, 1, 10.0, 250, 1, 0), # large_out (10*250*100 = 25w)
TransactionRecord(10, 2, 10.0, 10, 0, 0), # small_in (10*10*100 = 1w)
]
with patch.object(TdxClient, "get_transaction_data", return_value=mock_recs):
@@ -31,7 +29,6 @@ def test_get_fund_flow_logic(mock_conn_cls):
assert flow.large_out == 250000.0
assert flow.small_in == 10000.0
assert flow.main_net_inflow == 1000000.0 - 250000.0
assert flow.total_net_inflow == (1000000.0 + 10000.0) - 250000.0
@patch("xmtdx.client.TdxConnection")
def test_get_security_list_all_filtering(mock_conn_cls):
@@ -60,25 +57,20 @@ def test_get_security_list_all_filtering(mock_conn_cls):
all_stocks = client.get_security_list_all()
# 应该只保留 3 只 A 股 (600000, 000001, 830000)
assert len(all_stocks) == 3
codes = [s.code for s in all_stocks]
assert "600000" in codes
assert "000001" in codes
assert "830000" in codes
assert "999999" not in codes
# 检查行业挂载
s0 = next(s for s in all_stocks if s.code == "600000")
assert s0.industry_tdx == "T01"
assert s0.industry_sw == "X01"
@patch("xmtdx.client.TdxConnection")
def test_get_market_stat_mapping(mock_conn_cls):
"""测试市场统计字段映射。"""
client = TdxClient("127.0.0.1")
# 模拟 880005 行情返回
mock_quote = SecurityQuote(
Market.SH, "880005",
price=3000.0, # up
@@ -95,7 +87,27 @@ def test_get_market_stat_mapping(mock_conn_cls):
with patch.object(TdxClient, "get_security_quotes", return_value=[mock_quote]):
stat = client.get_market_stat()
assert stat.up_count == 3000
assert stat.down_count == 2000
assert stat.neutral_count == 500
assert stat.total_count == 5500
assert stat.total_amount == 50000000.0
def test_get_history_fund_flow_parsing():
"""测试历史资金流序列解析逻辑。"""
from xmtdx.commands.fund_flow import GetHistoryFundFlowCmd
# 模拟 Category 22 响应 (Header 9 + Count 2 + Body 36)
body = bytearray(9)
body.extend(struct.pack("<H", 1)) # 1 record
# Record: Date(I) + 8 * custom_float(i)
# 2025-01-08
date = 20250108
# 模拟 8 个流向金额
record = struct.pack("<Iiiiiiiii", date, 100, 200, 300, 400, 500, 600, 700, 800)
body.extend(record)
cmd = GetHistoryFundFlowCmd(Market.SH, "600000", 0, 1)
res = cmd.parse_response(bytes(body))
assert len(res) == 1
assert res[0].year == 2025
assert res[0].month == 1
assert res[0].day == 8