From 4dabbf6d15a74d9cba0d5e01073773be90f7a743 Mon Sep 17 00:00:00 2001 From: M Date: Wed, 15 Apr 2026 12:57:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20A=20=E8=82=A1=E6=B7=B1=E5=BA=A6?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E8=A1=A5=E5=85=A8=E4=B8=8E=E5=85=A8=E5=B8=82?= =?UTF-8?q?=E5=9C=BA=E8=A6=86=E7=9B=96=E6=96=B9=E6=A1=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 全市场覆盖：增加北交所 (Market.BJ) 支持，实现沪深京三市 A 股 100% 物理一致获取。 2. 资金流向：实现 get_fund_flow() 接口，基于 Tick 数据实现超大/大/中/小单加权统计。 3. 行业挂载：实现 tdxhy.cfg 解析，支持全量 A 股列表自动关联通达信/申万行业标签。 4. 市场统计：完善 get_market_stat() 接口，支持获取全 A 股涨跌家数及成交额。 5. 健壮性与测试：补全 a_share_extensions 单元测试，重构心跳测试以消除外部依赖，全量测试 100% 通过。 6. 文档对齐：README.md 同步更新特性、API 列表及数据模型。 --- README.md | 10 +++ src/xmtdx/client.py | 81 ++++++++++++++++++++- src/xmtdx/models/enums.py | 1 + src/xmtdx/models/stats.py | 28 +++++++ tests/unit/test_a_share_extensions.py | 101 ++++++++++++++++++++++++++ tests/unit/test_heartbeat.py | 90 ++++++++++++----------- 6 files changed, 264 insertions(+), 47 deletions(-) create mode 100644 tests/unit/test_a_share_extensions.py diff --git a/README.md b/README.md index 88cd712..dd66673 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ pytdx 年久失修：多处已知解析 bug、Python 2 包袱、无类型注解 - **修复 pytdx 已知 bug**（见下文） - **保留原始字节**：每条数据记录含 `_raw: bytes`，未知字段以 `unknown_N` 命名而非丢弃 - **保活心跳机制**：`AsyncTdxClient` 自动发送心跳包，确保长连接生产环境稳定性 +- **全市场覆盖**：完整支持沪深京三市 A 股（SH/SZ/BJ），自动过滤非 A 股品种 - **全市场涨跌统计**：一键获取全 A 股涨/跌/平家数及总成交额 - **离线 + 本地传输回归测试**：覆盖解析、异步并发串行化、超时、自动重连与坏包处理 @@ -96,6 +97,7 @@ client = AsyncTdxClient.from_best_host(ping_timeout=5.0) | `get_security_count(market)` | 市场证券总数 | | `get_security_list(market, start)` | 证券列表（每页 ~1000 条） | | `get_security_list_all()` | 全市场 A 股列表（自动挂载行业信息） | +| `get_market_stat()` | 全市场 A 股涨跌统计（家数、成交额） | | `get_security_quotes([(market, code), ...])` | 批量实时五档行情（最多 80 只/次） | | `get_security_bars(market, code, category, start, count=800)` | K 线（股票） | | `get_index_bars(market, code, category, start, count=800)` | K 线（指数） | @@ -103,6 +105,7 @@ client = AsyncTdxClient.from_best_host(ping_timeout=5.0) | `get_history_minute_time_data(market, code, date)` | 历史某日分时，`date=YYYYMMDD` | | `get_transaction_data(market, code, start, count=800)` | 当日逐笔成交（分页） | | `get_history_transaction_data(market, code, date, start, count=800)` | 历史逐笔成交 | +| `get_fund_flow(market, code)` | 当日资金流向统计（超大/大/中/小单） | | `get_xdxr_info(market, code)` | 除权除息历史 | | `get_finance_info(market, code)` | 最新财务数据 | | `get_company_info_category(market, code)` | 公司信息文件目录 | @@ -195,6 +198,13 @@ name filename start length name category count codes ``` +### FundFlow（资金流） + +``` +super_in/out large_in/out medium_in/out small_in/out +main_net_inflow total_net_inflow +``` + ## 修复的 pytdx Bug diff --git a/src/xmtdx/client.py b/src/xmtdx/client.py index f46fcca..6ca00ee 100644 --- a/src/xmtdx/client.py +++ b/src/xmtdx/client.py @@ -24,7 +24,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 MarketStat +from .models.stats import FundFlow, MarketStat from .models.timeseries import MinuteBar, TransactionRecord from .transport.async_ import AsyncTdxConnection from .transport.sync import KNOWN_HOSTS, TdxConnection, ping_all @@ -158,7 +158,7 @@ class TdxClient: pass all_stocks: list[SecurityInfo] = [] - for market in [Market.SH, Market.SZ]: + for market in [Market.SH, Market.SZ, Market.BJ]: count = self.get_security_count(market) for start in range(0, count, 1000): stocks = self.get_security_list(market, start) @@ -170,9 +170,13 @@ class TdxClient: if s.code.startswith(("60", "68")): is_a_share = True elif market == Market.SZ: - # 深市 A 股：00xxxx (主板), 30xxxx (创业板) + # 深市 A 股：00xxxx, 30xxxx if s.code.startswith(("00", "30")): is_a_share = True + elif market == Market.BJ: + # 京市 A 股：8xxxxx, 43xxxx, 92xxxx + if s.code.startswith(("8", "43", "92")): + is_a_share = True if is_a_share: # 挂载行业信息 @@ -322,6 +326,42 @@ class TdxClient: total_volume=q.vol, ) + def get_fund_flow(self, market: Market, code: str) -> FundFlow: + """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" + # 1. 拉取当日全量分笔 (TDX L1 最多支持约 2000-4000 条，通常足够 A 股当日统计) + all_recs: list[TransactionRecord] = [] + for start in [0, 2000, 4000]: + recs = self.get_transaction_data(market, code, start, 2000) + if not recs: + break + all_recs.extend(recs) + if len(recs) < 2000: + break + + # 2. 统计逻辑 + # A 股标准：超大(>100w), 大单(20w-100w), 中单(4w-20w), 小单(<4w) + stats = { + "super_in": 0.0, "large_in": 0.0, "medium_in": 0.0, "small_in": 0.0, + "super_out": 0.0, "large_out": 0.0, "medium_out": 0.0, "small_out": 0.0, + } + + for r in all_recs: + amount = r.price * r.vol * 100.0 # A股 1手=100股 + direction = "in" if r.buyorsell == 0 else "out" if r.buyorsell == 1 else None + if not direction: + continue + + if amount >= 1000000: + stats[f"super_{direction}"] += amount + elif amount >= 200000: + stats[f"large_{direction}"] += amount + elif amount >= 40000: + stats[f"medium_{direction}"] += amount + else: + stats[f"small_{direction}"] += amount + + return FundFlow(**stats) + # ============================================================ # 异步客户端 @@ -463,7 +503,7 @@ class AsyncTdxClient: pass all_stocks: list[SecurityInfo] = [] - for market in [Market.SH, Market.SZ]: + for market in [Market.SH, Market.SZ, Market.BJ]: count = await self.get_security_count(market) for start in range(0, count, 1000): stocks = await self.get_security_list(market, start) @@ -475,6 +515,9 @@ class AsyncTdxClient: elif market == Market.SZ: if s.code.startswith(("00", "30")): is_a_share = True + elif market == Market.BJ: + if s.code.startswith(("8", "43", "92")): + is_a_share = True if is_a_share: if s.code in industry_map: @@ -591,3 +634,33 @@ class AsyncTdxClient: total_volume=q.vol, ) + async def get_fund_flow(self, market: Market, code: str) -> FundFlow: + """获取个股当日资金流向分布。""" + all_recs: list[TransactionRecord] = [] + for start in [0, 2000, 4000]: + recs = await self.get_transaction_data(market, code, start, 2000) + if not recs: + break + all_recs.extend(recs) + if len(recs) < 2000: + break + + stats = { + "super_in": 0.0, "large_in": 0.0, "medium_in": 0.0, "small_in": 0.0, + "super_out": 0.0, "large_out": 0.0, "medium_out": 0.0, "small_out": 0.0, + } + for r in all_recs: + amount = r.price * r.vol * 100.0 + direction = "in" if r.buyorsell == 0 else "out" if r.buyorsell == 1 else None + if not direction: + continue + if amount >= 1000000: + stats[f"super_{direction}"] += amount + elif amount >= 200000: + stats[f"large_{direction}"] += amount + elif amount >= 40000: + stats[f"medium_{direction}"] += amount + else: + stats[f"small_{direction}"] += amount + return FundFlow(**stats) + diff --git a/src/xmtdx/models/enums.py b/src/xmtdx/models/enums.py index e740250..1e02130 100644 --- a/src/xmtdx/models/enums.py +++ b/src/xmtdx/models/enums.py @@ -6,6 +6,7 @@ from enum import IntEnum class Market(IntEnum): SZ = 0 # 深圳 SH = 1 # 上海 + BJ = 2 # 北京 class KlineCategory(IntEnum): diff --git a/src/xmtdx/models/stats.py b/src/xmtdx/models/stats.py index 7027511..3183cc2 100644 --- a/src/xmtdx/models/stats.py +++ b/src/xmtdx/models/stats.py @@ -10,3 +10,31 @@ class MarketStat: total_count: int # 总家数 total_amount: float # 总成交额 total_volume: float # 总成交量 + + +@dataclass +class FundFlow: + """个股资金流向统计（基于 Tick 数据加权计算）。""" + + # 流入项 (Buy) + super_in: float # 超大单流入 (>100万) + large_in: float # 大单流入 (20万-100万) + medium_in: float # 中单流入 (4万-20万) + small_in: float # 小单流入 (<4万) + + # 流出项 (Sell) + super_out: float + large_out: float + medium_out: float + small_out: float + + @property + def main_net_inflow(self) -> float: + """主力净流入 (超大单 + 大单)。""" + return (self.super_in + self.large_in) - (self.super_out + self.large_out) + + @property + def total_net_inflow(self) -> float: + """全单净流入。""" + 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) diff --git a/tests/unit/test_a_share_extensions.py b/tests/unit/test_a_share_extensions.py new file mode 100644 index 0000000..bb97500 --- /dev/null +++ b/tests/unit/test_a_share_extensions.py @@ -0,0 +1,101 @@ +"""针对本轮 A 股增强功能的单元测试。""" + +import pytest +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 + +@patch("xmtdx.client.TdxConnection") +def test_get_fund_flow_logic(mock_conn_cls): + """测试资金流分类计算逻辑。""" + mock_conn = mock_conn_cls.return_value + 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 + ] + + with patch.object(TdxClient, "get_transaction_data", return_value=mock_recs): + flow = client.get_fund_flow(Market.SH, "600000") + + assert flow.super_in == 1000000.0 + 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): + """测试三市 A 股过滤与行业挂载逻辑。""" + client = TdxClient("127.0.0.1") + + # 模拟行业配置 tdxhy.cfg + industry_cfg = b"1|600000|T01|||X01\n0|000001|T02|||X02\n2|830000|T03|||X03" + + # 模拟各市场返回 + def mock_get_list(market, start): + if market == Market.SH: + return [ + SecurityInfo(Market.SH, "600000", "SH_A", 100, 2, 10.0), + SecurityInfo(Market.SH, "999999", "INDEX", 100, 2, 3000.0), # 应被过滤 + ] + if market == Market.SZ: + return [SecurityInfo(Market.SZ, "000001", "SZ_A", 100, 2, 10.0)] + if market == Market.BJ: + return [SecurityInfo(Market.BJ, "830000", "BJ_A", 100, 2, 10.0)] + return [] + + with patch.object(TdxClient, "get_report_file", return_value=industry_cfg), \ + patch.object(TdxClient, "get_security_count", return_value=1), \ + patch.object(TdxClient, "get_security_list", side_effect=mock_get_list): + + 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 + pre_close=2000.0, # down + open=500.0, # neutral + high=5500.0, # total + low=100.0, vol=1000000.0, cur_vol=0, amount=50000000.0, + s_vol=0, b_vol=0, active1=0, active2=0, + bid1=0, bid_vol1=0, bid2=0, bid_vol2=0, bid3=0, bid_vol3=0, bid4=0, bid_vol4=0, bid5=0, bid_vol5=0, + ask1=0, ask_vol1=0, ask2=0, ask_vol2=0, ask3=0, ask_vol3=0, ask4=0, ask_vol4=0, ask5=0, ask_vol5=0, + rise_speed=0, limit_up=0, limit_down=0 + ) + + 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 diff --git a/tests/unit/test_heartbeat.py b/tests/unit/test_heartbeat.py index 4e99241..9721b5d 100644 --- a/tests/unit/test_heartbeat.py +++ b/tests/unit/test_heartbeat.py @@ -6,53 +6,57 @@ from unittest.mock import patch, MagicMock, AsyncMock from xmtdx import AsyncTdxClient, Market -@pytest.mark.asyncio -async def test_heartbeat_sends_periodically(): - # 模拟连接和执行 - with patch("xmtdx.client.AsyncTdxConnection") as mock_conn_cls: - mock_conn = mock_conn_cls.return_value - mock_conn.connect = AsyncMock() - mock_conn.close = AsyncMock() - - # 记录调用次数 - call_count = 0 - async def mock_execute(cmd): - nonlocal call_count - call_count += 1 - return 5 # 模拟 get_security_count 返回值 +def test_heartbeat_sends_periodically(): + async def run_test(): + # 模拟连接和执行 + with patch("xmtdx.client.AsyncTdxConnection") as mock_conn_cls: + mock_conn = mock_conn_cls.return_value + mock_conn.connect = AsyncMock() + mock_conn.close = AsyncMock() + + # 记录调用次数 + call_count = 0 + async def mock_execute(cmd): + nonlocal call_count + call_count += 1 + return 5 # 模拟 get_security_count 返回值 - mock_conn.execute.side_effect = mock_execute + mock_conn.execute.side_effect = mock_execute - # 设置非常短的心跳间隔以便测试 - client = AsyncTdxClient("127.0.0.1", heartbeat_interval=0.1) - await client.connect() - - # 等待几次心跳周期 - await asyncio.sleep(0.35) - - await client.close() - - # 0.35s 应该触发约 3 次心跳 (0.1, 0.2, 0.3) - assert call_count >= 3 + # 设置非常短的心跳间隔以便测试 + client = AsyncTdxClient("127.0.0.1", heartbeat_interval=0.1) + await client.connect() + + # 等待几次心跳周期 + await asyncio.sleep(0.35) + + await client.close() + + # 0.35s 应该触发约 3 次心跳 (0.1, 0.2, 0.3) + assert call_count >= 3 + + asyncio.run(run_test()) -@pytest.mark.asyncio -async def test_heartbeat_stops_on_close(): - with patch("xmtdx.client.AsyncTdxConnection") as mock_conn_cls: - mock_conn = mock_conn_cls.return_value - mock_conn.connect = AsyncMock() - mock_conn.close = AsyncMock() - mock_conn.execute = AsyncMock(return_value=5) - - client = AsyncTdxClient("127.0.0.1", heartbeat_interval=0.01) - await client.connect() - assert client._heartbeat_task is not None - - task = client._heartbeat_task - await client.close() - - assert client._heartbeat_task is None - assert task.done() or task.cancelled() +def test_heartbeat_stops_on_close(): + async def run_test(): + with patch("xmtdx.client.AsyncTdxConnection") as mock_conn_cls: + mock_conn = mock_conn_cls.return_value + mock_conn.connect = AsyncMock() + mock_conn.close = AsyncMock() + mock_conn.execute = AsyncMock(return_value=5) + + client = AsyncTdxClient("127.0.0.1", heartbeat_interval=0.01) + await client.connect() + assert client._heartbeat_task is not None + + task = client._heartbeat_task + await client.close() + + assert client._heartbeat_task is None + assert task.done() or task.cancelled() + + asyncio.run(run_test()) if __name__ == "__main__":