feat: 初始实现 xmtdx —— 从零实现通达信 TCP A 股行情客户端

替代年久失修的 pytdx,修复已知 bug,保留未解字段供逆向分析。

主要内容:
- codec 层:get_price 变长编码、get_volume 自定义浮点、datetime/frame 解析
- transport 层:同步(socket)+ 异步(asyncio)双实现,共用命令层
- 命令层(11 条):security_count/list/quotes/bars、minute_time(今日+历史)、
  transaction(当日+历史)、xdxr_info、finance_info、company_info
- 高层 API:TdxClient + AsyncTdxClient
- 单元测试 26 条,全部通过;真实服务器集成测试覆盖全部命令

修复 pytdx Bug #1–5:xdxr 循环读取错误位置、GBK 截断崩溃、
pre_close 误用 get_volume、逐笔/分时未解字段被丢弃

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
minionszyw
2026-04-11 19:55:19 +08:00
co-authored by Claude Sonnet 4.6
commit 283682f6b4
39 changed files with 2298 additions and 0 deletions
+64
View File
@@ -0,0 +1,64 @@
"""日期时间解码单元测试。"""
import struct
from xmtdx.codec.datetime_ import get_datetime, get_datetime_day, get_datetime_minute, get_time
def _pack_minute(year: int, month: int, day: int, hour: int, minute: int) -> bytes:
zipday = ((year - 2004) << 11) | (month * 100 + day)
tminutes = hour * 60 + minute
return struct.pack("<HH", zipday, tminutes)
def _pack_day(year: int, month: int, day: int) -> bytes:
return struct.pack("<I", year * 10000 + month * 100 + day)
class TestGetDatetimeMinute:
def test_basic(self):
data = _pack_minute(2024, 4, 10, 14, 30)
y, mo, d, h, mi, pos = get_datetime_minute(data, 0)
assert (y, mo, d, h, mi) == (2024, 4, 10, 14, 30)
assert pos == 4
def test_open_time(self):
data = _pack_minute(2026, 1, 5, 9, 30)
y, mo, d, h, mi, pos = get_datetime_minute(data, 0)
assert h == 9 and mi == 30
def test_close_time(self):
data = _pack_minute(2026, 1, 5, 15, 0)
y, mo, d, h, mi, _ = get_datetime_minute(data, 0)
assert h == 15 and mi == 0
class TestGetDatetimeDay:
def test_basic(self):
data = _pack_day(2026, 4, 10)
y, mo, d, pos = get_datetime_day(data, 0)
assert (y, mo, d) == (2026, 4, 10)
assert pos == 4
class TestGetDatetime:
def test_minute_category(self):
data = _pack_minute(2026, 3, 15, 10, 0)
for cat in (0, 1, 2, 3, 7, 8):
y, mo, d, h, mi, _ = get_datetime(cat, data, 0)
assert h == 10 and mi == 0
def test_day_category(self):
data = _pack_day(2026, 3, 15)
for cat in (4, 5, 6, 9):
y, mo, d, h, mi, _ = get_datetime(cat, data, 0)
assert (y, mo, d) == (2026, 3, 15)
assert h == 15 and mi == 0
class TestGetTime:
def test_basic(self):
data = struct.pack("<H", 14 * 60 + 30) # 14:30
h, mi, pos = get_time(data, 0)
assert h == 14 and mi == 30
assert pos == 2