From 283682f6b4de050c3879252bb3f25dc4ff1ba57d Mon Sep 17 00:00:00 2001 From: minionszyw Date: Sat, 11 Apr 2026 19:55:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=88=9D=E5=A7=8B=E5=AE=9E=E7=8E=B0=20?= =?UTF-8?q?xmtdx=20=E2=80=94=E2=80=94=20=E4=BB=8E=E9=9B=B6=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=E9=80=9A=E8=BE=BE=E4=BF=A1=20TCP=20A=20=E8=82=A1?= =?UTF-8?q?=E8=A1=8C=E6=83=85=E5=AE=A2=E6=88=B7=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 替代年久失修的 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 --- .gitignore | 12 ++ pyproject.toml | 28 +++ src/xmtdx/__init__.py | 63 ++++++ src/xmtdx/client.py | 277 ++++++++++++++++++++++++++ src/xmtdx/codec/__init__.py | 18 ++ src/xmtdx/codec/datetime_.py | 68 +++++++ src/xmtdx/codec/frame.py | 42 ++++ src/xmtdx/codec/price.py | 58 ++++++ src/xmtdx/codec/volume.py | 56 ++++++ src/xmtdx/commands/__init__.py | 10 + src/xmtdx/commands/base.py | 29 +++ src/xmtdx/commands/company_info.py | 76 +++++++ src/xmtdx/commands/finance_info.py | 84 ++++++++ src/xmtdx/commands/minute_time.py | 69 +++++++ src/xmtdx/commands/security_bars.py | 106 ++++++++++ src/xmtdx/commands/security_count.py | 24 +++ src/xmtdx/commands/security_list.py | 67 +++++++ src/xmtdx/commands/security_quotes.py | 194 ++++++++++++++++++ src/xmtdx/commands/setup.py | 18 ++ src/xmtdx/commands/transaction.py | 104 ++++++++++ src/xmtdx/commands/xdxr_info.py | 99 +++++++++ src/xmtdx/exceptions.py | 17 ++ src/xmtdx/models/__init__.py | 25 +++ src/xmtdx/models/bar.py | 28 +++ src/xmtdx/models/enums.py | 23 +++ src/xmtdx/models/finance.py | 127 ++++++++++++ src/xmtdx/models/quote.py | 68 +++++++ src/xmtdx/models/security.py | 19 ++ src/xmtdx/models/timeseries.py | 39 ++++ src/xmtdx/transport/__init__.py | 4 + src/xmtdx/transport/async_.py | 123 ++++++++++++ src/xmtdx/transport/sync.py | 121 +++++++++++ tests/__init__.py | 0 tests/integration/__init__.py | 0 tests/unit/__init__.py | 0 tests/unit/test_codec_datetime.py | 64 ++++++ tests/unit/test_codec_frame.py | 39 ++++ tests/unit/test_codec_price.py | 63 ++++++ tests/unit/test_codec_volume.py | 36 ++++ 39 files changed, 2298 insertions(+) create mode 100644 .gitignore create mode 100644 pyproject.toml create mode 100644 src/xmtdx/__init__.py create mode 100644 src/xmtdx/client.py create mode 100644 src/xmtdx/codec/__init__.py create mode 100644 src/xmtdx/codec/datetime_.py create mode 100644 src/xmtdx/codec/frame.py create mode 100644 src/xmtdx/codec/price.py create mode 100644 src/xmtdx/codec/volume.py create mode 100644 src/xmtdx/commands/__init__.py create mode 100644 src/xmtdx/commands/base.py create mode 100644 src/xmtdx/commands/company_info.py create mode 100644 src/xmtdx/commands/finance_info.py create mode 100644 src/xmtdx/commands/minute_time.py create mode 100644 src/xmtdx/commands/security_bars.py create mode 100644 src/xmtdx/commands/security_count.py create mode 100644 src/xmtdx/commands/security_list.py create mode 100644 src/xmtdx/commands/security_quotes.py create mode 100644 src/xmtdx/commands/setup.py create mode 100644 src/xmtdx/commands/transaction.py create mode 100644 src/xmtdx/commands/xdxr_info.py create mode 100644 src/xmtdx/exceptions.py create mode 100644 src/xmtdx/models/__init__.py create mode 100644 src/xmtdx/models/bar.py create mode 100644 src/xmtdx/models/enums.py create mode 100644 src/xmtdx/models/finance.py create mode 100644 src/xmtdx/models/quote.py create mode 100644 src/xmtdx/models/security.py create mode 100644 src/xmtdx/models/timeseries.py create mode 100644 src/xmtdx/transport/__init__.py create mode 100644 src/xmtdx/transport/async_.py create mode 100644 src/xmtdx/transport/sync.py create mode 100644 tests/__init__.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/test_codec_datetime.py create mode 100644 tests/unit/test_codec_frame.py create mode 100644 tests/unit/test_codec_price.py create mode 100644 tests/unit/test_codec_volume.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9891dad --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.eggs/ +.mypy_cache/ +.ruff_cache/ +.pytest_cache/ +*.egg +.venv/ +venv/ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a08ea21 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "xmtdx" +version = "0.1.0" +description = "通达信 TCP 协议 A 股行情数据客户端" +requires-python = ">=3.10" +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest>=8.0", "pytest-cov", "mypy>=1.9", "ruff>=0.4"] +pandas = ["pandas>=2.0"] + +[tool.hatch.build.targets.wheel] +packages = ["src/xmtdx"] + +[tool.mypy] +strict = true +python_version = "3.10" + +[tool.ruff] +target-version = "py310" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "UP"] diff --git a/src/xmtdx/__init__.py b/src/xmtdx/__init__.py new file mode 100644 index 0000000..a984ba8 --- /dev/null +++ b/src/xmtdx/__init__.py @@ -0,0 +1,63 @@ +"""xmtdx — 通达信 TCP 协议 A 股行情数据客户端。 + +快速开始:: + + from xmtdx import TdxClient, Market, KlineCategory + + with TdxClient("180.153.18.170") as c: + count = c.get_security_count(Market.SH) + bars = c.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 5) + +asyncio 版本:: + + import asyncio + from xmtdx import AsyncTdxClient, Market, KlineCategory + + async def main(): + async with AsyncTdxClient("180.153.18.170") as c: + bars = await c.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 5) + + asyncio.run(main()) +""" + +from .client import AsyncTdxClient, TdxClient +from .exceptions import TdxCommandError, TdxConnectionError, TdxDecodeError, TdxError +from .models import ( + XDXR_CATEGORY_NAMES, + CompanyInfoCategory, + FinanceInfo, + KlineCategory, + Market, + MinuteBar, + SecurityBar, + SecurityInfo, + SecurityQuote, + TransactionRecord, + XdxrRecord, +) + +__all__ = [ + # 客户端 + "TdxClient", + "AsyncTdxClient", + # 枚举 + "Market", + "KlineCategory", + # 数据模型 + "SecurityBar", + "SecurityQuote", + "SecurityInfo", + "MinuteBar", + "TransactionRecord", + "XdxrRecord", + "XDXR_CATEGORY_NAMES", + "FinanceInfo", + "CompanyInfoCategory", + # 异常 + "TdxError", + "TdxConnectionError", + "TdxDecodeError", + "TdxCommandError", +] + +__version__ = "0.1.0" diff --git a/src/xmtdx/client.py b/src/xmtdx/client.py new file mode 100644 index 0000000..a6a620b --- /dev/null +++ b/src/xmtdx/client.py @@ -0,0 +1,277 @@ +"""高层行情 API：TdxClient（同步）和 AsyncTdxClient（asyncio）。""" + +from types import TracebackType + +from .commands.company_info import GetCompanyInfoCategoryCmd, GetCompanyInfoContentCmd +from .commands.finance_info import GetFinanceInfoCmd +from .commands.minute_time import GetHistoryMinuteTimeDataCmd, GetMinuteTimeDataCmd +from .commands.security_bars import GetIndexBarsCmd, GetSecurityBarsCmd +from .commands.security_count import GetSecurityCountCmd +from .commands.security_list import GetSecurityListCmd +from .commands.security_quotes import GetSecurityQuotesCmd +from .commands.transaction import GetHistoryTransactionDataCmd, GetTransactionDataCmd +from .commands.xdxr_info import GetXdxrInfoCmd +from .models.bar import SecurityBar +from .models.enums import KlineCategory, Market +from .models.finance import CompanyInfoCategory, FinanceInfo, XdxrRecord +from .models.quote import SecurityQuote +from .models.security import SecurityInfo +from .models.timeseries import MinuteBar, TransactionRecord +from .transport.async_ import AsyncTdxConnection +from .transport.sync import TdxConnection + +_DEFAULT_HOST = "180.153.18.170" +_DEFAULT_PORT = 7709 + + +# ============================================================ +# 同步客户端 +# ============================================================ + + +class TdxClient: + """同步通达信行情客户端。 + + 使用示例:: + + with TdxClient("180.153.18.170") as c: + bars = c.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 100) + """ + + def __init__( + self, + host: str = _DEFAULT_HOST, + port: int = _DEFAULT_PORT, + timeout: float = 15.0, + ) -> None: + self._conn = TdxConnection(host, port, timeout) + + def connect(self) -> None: + self._conn.connect() + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> "TdxClient": + self.connect() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + self.close() + + # ------------------------------------------------------------------ # + # 市场信息 + # ------------------------------------------------------------------ # + + def get_security_count(self, market: Market) -> int: + """获取市场证券总数。""" + return self._conn.execute(GetSecurityCountCmd(market)) + + def get_security_list(self, market: Market, start: int) -> list[SecurityInfo]: + """获取证券列表（每页约1000条，按 start 分页）。""" + return self._conn.execute(GetSecurityListCmd(market, start)) + + def get_security_quotes( + self, stocks: list[tuple[Market, str]] + ) -> list[SecurityQuote]: + """批量获取实时五档行情（最多80只/次）。""" + return self._conn.execute(GetSecurityQuotesCmd(stocks)) + + # ------------------------------------------------------------------ # + # K 线 + # ------------------------------------------------------------------ # + + def get_security_bars( + self, + market: Market, + code: str, + category: KlineCategory, + start: int, + count: int = 800, + ) -> list[SecurityBar]: + """获取 K 线数据（最多800条/次，按 start 分页）。""" + return self._conn.execute(GetSecurityBarsCmd(market, code, category, start, count)) + + def get_index_bars( + self, + market: Market, + code: str, + category: KlineCategory, + start: int, + count: int = 800, + ) -> list[SecurityBar]: + """获取指数 K 线数据。""" + return self._conn.execute(GetIndexBarsCmd(market, code, category, start, count)) + + # ------------------------------------------------------------------ # + # 分时 + # ------------------------------------------------------------------ # + + def get_minute_time_data(self, market: Market, code: str) -> list[MinuteBar]: + """获取今日分时数据（240条）。""" + return self._conn.execute(GetMinuteTimeDataCmd(market, code)) + + def get_history_minute_time_data( + self, market: Market, code: str, date: int + ) -> list[MinuteBar]: + """获取历史某日分时数据（date: YYYYMMDD）。""" + return self._conn.execute(GetHistoryMinuteTimeDataCmd(market, code, date)) + + # ------------------------------------------------------------------ # + # 逐笔成交 + # ------------------------------------------------------------------ # + + def get_transaction_data( + self, market: Market, code: str, start: int, count: int = 800 + ) -> list[TransactionRecord]: + """获取当日逐笔成交（分页）。""" + return self._conn.execute(GetTransactionDataCmd(market, code, start, count)) + + def get_history_transaction_data( + self, market: Market, code: str, date: int, start: int, count: int = 800 + ) -> list[TransactionRecord]: + """获取历史逐笔成交（date: YYYYMMDD，分页）。""" + return self._conn.execute( + GetHistoryTransactionDataCmd(market, code, date, start, count) + ) + + # ------------------------------------------------------------------ # + # 财务 / 公司 + # ------------------------------------------------------------------ # + + def get_xdxr_info(self, market: Market, code: str) -> list[XdxrRecord]: + """获取除权除息历史记录。""" + return self._conn.execute(GetXdxrInfoCmd(market, code)) + + def get_finance_info(self, market: Market, code: str) -> FinanceInfo: + """获取最新财务数据。""" + return self._conn.execute(GetFinanceInfoCmd(market, code)) + + def get_company_info_category( + self, market: Market, code: str + ) -> list[CompanyInfoCategory]: + """获取公司信息文件目录。""" + return self._conn.execute(GetCompanyInfoCategoryCmd(market, code)) + + def get_company_info_content( + self, market: Market, code: str, filename: str, offset: int, length: int + ) -> str: + """读取公司信息文本。""" + return self._conn.execute( + GetCompanyInfoContentCmd(market, code, filename, offset, length) + ) + + +# ============================================================ +# 异步客户端 +# ============================================================ + + +class AsyncTdxClient: + """异步通达信行情客户端（asyncio）。 + + 使用示例:: + + async with AsyncTdxClient("180.153.18.170") as c: + bars = await c.get_security_bars(Market.SH, "600000", KlineCategory.DAY, 0, 100) + """ + + def __init__( + self, + host: str = _DEFAULT_HOST, + port: int = _DEFAULT_PORT, + timeout: float = 15.0, + ) -> None: + self._conn = AsyncTdxConnection(host, port, timeout) + + async def connect(self) -> None: + await self._conn.connect() + + async def close(self) -> None: + await self._conn.close() + + async def __aenter__(self) -> "AsyncTdxClient": + await self.connect() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + await self.close() + + async def get_security_count(self, market: Market) -> int: + return await self._conn.execute(GetSecurityCountCmd(market)) + + async def get_security_list(self, market: Market, start: int) -> list[SecurityInfo]: + return await self._conn.execute(GetSecurityListCmd(market, start)) + + async def get_security_quotes( + self, stocks: list[tuple[Market, str]] + ) -> list[SecurityQuote]: + return await self._conn.execute(GetSecurityQuotesCmd(stocks)) + + async def get_security_bars( + self, + market: Market, + code: str, + category: KlineCategory, + start: int, + count: int = 800, + ) -> list[SecurityBar]: + return await self._conn.execute(GetSecurityBarsCmd(market, code, category, start, count)) + + async def get_index_bars( + self, + market: Market, + code: str, + category: KlineCategory, + start: int, + count: int = 800, + ) -> list[SecurityBar]: + return await self._conn.execute(GetIndexBarsCmd(market, code, category, start, count)) + + async def get_minute_time_data(self, market: Market, code: str) -> list[MinuteBar]: + return await self._conn.execute(GetMinuteTimeDataCmd(market, code)) + + async def get_history_minute_time_data( + self, market: Market, code: str, date: int + ) -> list[MinuteBar]: + return await self._conn.execute(GetHistoryMinuteTimeDataCmd(market, code, date)) + + async def get_transaction_data( + self, market: Market, code: str, start: int, count: int = 800 + ) -> list[TransactionRecord]: + return await self._conn.execute(GetTransactionDataCmd(market, code, start, count)) + + async def get_history_transaction_data( + self, market: Market, code: str, date: int, start: int, count: int = 800 + ) -> list[TransactionRecord]: + return await self._conn.execute( + GetHistoryTransactionDataCmd(market, code, date, start, count) + ) + + async def get_xdxr_info(self, market: Market, code: str) -> list[XdxrRecord]: + return await self._conn.execute(GetXdxrInfoCmd(market, code)) + + async def get_finance_info(self, market: Market, code: str) -> FinanceInfo: + return await self._conn.execute(GetFinanceInfoCmd(market, code)) + + async def get_company_info_category( + self, market: Market, code: str + ) -> list[CompanyInfoCategory]: + return await self._conn.execute(GetCompanyInfoCategoryCmd(market, code)) + + async def get_company_info_content( + self, market: Market, code: str, filename: str, offset: int, length: int + ) -> str: + return await self._conn.execute( + GetCompanyInfoContentCmd(market, code, filename, offset, length) + ) diff --git a/src/xmtdx/codec/__init__.py b/src/xmtdx/codec/__init__.py new file mode 100644 index 0000000..4c57027 --- /dev/null +++ b/src/xmtdx/codec/__init__.py @@ -0,0 +1,18 @@ +from .datetime_ import get_datetime, get_datetime_day, get_datetime_minute, get_time +from .frame import HEADER_SIZE, FrameHeader, decompress_body, parse_header +from .price import get_price, put_price +from .volume import get_volume + +__all__ = [ + "get_price", + "put_price", + "get_volume", + "get_datetime", + "get_datetime_minute", + "get_datetime_day", + "get_time", + "parse_header", + "decompress_body", + "FrameHeader", + "HEADER_SIZE", +] diff --git a/src/xmtdx/codec/datetime_.py b/src/xmtdx/codec/datetime_.py new file mode 100644 index 0000000..0ea365c --- /dev/null +++ b/src/xmtdx/codec/datetime_.py @@ -0,0 +1,68 @@ +"""日期时间解码（通达信 TCP 两种格式）。 + +分钟级（category < 4 或 == 7/8）：4 字节 = 2 字节压缩日期 + 2 字节分钟数 + zipday: year=(>>11)+2004, month=(% 2048)//100, day=(% 2048)%100 + tminutes: hour=//60, minute=%60 + +日线及以上（其余 category）：4 字节 YYYYMMDD 整数 +""" + +import struct + + +def get_datetime_minute( + data: bytes | bytearray, pos: int +) -> tuple[int, int, int, int, int, int]: + """解析分钟级时间戳（4 字节）。 + + Returns: + (year, month, day, hour, minute, new_pos) + """ + zipday, tminutes = struct.unpack_from("> 11) + 2004 + month = (zipday % 2048) // 100 + day = (zipday % 2048) % 100 + hour = tminutes // 60 + minute = tminutes % 60 + return year, month, day, hour, minute, pos + 4 + + +def get_datetime_day( + data: bytes | bytearray, pos: int +) -> tuple[int, int, int, int]: + """解析日期（4 字节 YYYYMMDD）。 + + Returns: + (year, month, day, new_pos) + """ + (zipday,) = struct.unpack_from(" tuple[int, int, int, int, int, int]: + """根据 KlineCategory 选择解析格式。 + + Returns: + (year, month, day, hour, minute, new_pos) + 日线及以上时 hour=15, minute=0（收盘时间，与 pytdx 保持一致） + """ + if category < 4 or category in (7, 8): + return get_datetime_minute(data, pos) + else: + year, month, day, new_pos = get_datetime_day(data, pos) + return year, month, day, 15, 0, new_pos + + +def get_time(data: bytes | bytearray, pos: int) -> tuple[int, int, int]: + """解析 2 字节时间（分钟数）。 + + Returns: + (hour, minute, new_pos) + """ + (tminutes,) = struct.unpack_from(" FrameHeader: + """解析 16 字节响应帧头。""" + u0, u1, u2, zipsize, unzipsize = struct.unpack_from(_HEADER_FMT, buf) + return FrameHeader(u0, u1, u2, zipsize, unzipsize) + + +def decompress_body(header: FrameHeader, raw_body: bytes) -> bytes: + """按需 zlib 解压 body。 + + zipsize == unzipsize 时直接返回原始字节；否则 zlib 解压。 + """ + if header.zipsize == header.unzipsize: + return raw_body + return zlib.decompress(raw_body) diff --git a/src/xmtdx/codec/price.py b/src/xmtdx/codec/price.py new file mode 100644 index 0000000..2faa2b2 --- /dev/null +++ b/src/xmtdx/codec/price.py @@ -0,0 +1,58 @@ +"""变长有符号整数编解码（通达信 TCP 价格编码）。 + +协议规则： + - 首字节：bit7=继续标记，bit6=符号（1=负），bit5~0=低6位数据 + - 后续字节：bit7=继续标记，bit6~0=7位数据 + - 所有数据位低位在前（小端 bit 顺序） + +典型用途：价格差分、成交量差分、买卖档位数量。 +""" + + +def get_price(data: bytes | bytearray, pos: int) -> tuple[int, int]: + """解码一个变长有符号整数。 + + Returns: + (value, new_pos) + """ + bit_shift = 6 + b = data[pos] + value = b & 0x3F + negative = bool(b & 0x40) + + if b & 0x80: + while True: + pos += 1 + b = data[pos] + value |= (b & 0x7F) << bit_shift + bit_shift += 7 + if not (b & 0x80): + break + + pos += 1 + return (-value if negative else value), pos + + +def put_price(value: int) -> bytes: + """将整数编码为变长格式（用于构造请求包）。""" + negative = value < 0 + value = abs(value) + + # 首字节：低6位数据 + 符号位 + first = value & 0x3F + value >>= 6 + if negative: + first |= 0x40 + if value: + first |= 0x80 + + result = bytearray([first]) + + while value: + b = value & 0x7F + value >>= 7 + if value: + b |= 0x80 + result.append(b) + + return bytes(result) diff --git a/src/xmtdx/codec/volume.py b/src/xmtdx/codec/volume.py new file mode 100644 index 0000000..749a356 --- /dev/null +++ b/src/xmtdx/codec/volume.py @@ -0,0 +1,56 @@ +"""通达信 4 字节自定义浮点格式解码（成交量专用）。 + +格式：4 字节小端 uint32，分三段： + [3] logpoint — 指数部分 + [2] hleax — 高精度部分 + [1] lheax — 中精度部分 + [0] lleax — 低精度部分 + +警告：此函数专为成交量设计，不可用于价格字段（pytdx Bug #3）。 +""" + +import struct + + +def get_volume(data: bytes | bytearray, pos: int) -> tuple[float, int]: + """从 data[pos:pos+4] 解码成交量。 + + Returns: + (volume_float, new_pos) + """ + (ivol,) = struct.unpack_from(" float: + if ivol == 0: + return 0.0 + + logpoint = (ivol >> 24) & 0xFF + hleax = (ivol >> 16) & 0xFF + lheax = (ivol >> 8) & 0xFF + lleax = ivol & 0xFF + + exp = logpoint * 2 - 0x7F + base = _pow2(exp) + + exp_h = logpoint * 2 - 0x86 + if hleax > 0x80: + hi = _pow2(exp_h) * 128 + (hleax & 0x7F) * _pow2(exp_h + 1) + else: + hi = _pow2(exp_h) * hleax + + mid = _pow2(logpoint * 2 - 0x8E) * lheax + lo = _pow2(logpoint * 2 - 0x96) * lleax + + if hleax & 0x80: + mid *= 2.0 + lo *= 2.0 + + return base + hi + mid + lo + + +def _pow2(exp: int) -> float: + if exp >= 0: + return float(1 << exp) if exp < 63 else 2.0 ** exp + return 1.0 / (1 << (-exp)) if -exp < 63 else 2.0 ** exp diff --git a/src/xmtdx/commands/__init__.py b/src/xmtdx/commands/__init__.py new file mode 100644 index 0000000..89a324e --- /dev/null +++ b/src/xmtdx/commands/__init__.py @@ -0,0 +1,10 @@ +from .base import BaseCommand +from .setup import SETUP_CMD1, SETUP_CMD2, SETUP_CMD3, SETUP_COMMANDS + +__all__ = [ + "BaseCommand", + "SETUP_CMD1", + "SETUP_CMD2", + "SETUP_CMD3", + "SETUP_COMMANDS", +] diff --git a/src/xmtdx/commands/base.py b/src/xmtdx/commands/base.py new file mode 100644 index 0000000..bb53986 --- /dev/null +++ b/src/xmtdx/commands/base.py @@ -0,0 +1,29 @@ +"""命令基类：只含请求构造与响应解析，不含任何 IO。 + +transport 层负责：发送请求、接收帧头、接收 body、解压， +然后调用 command.parse_response(body) 得到结果。 +""" + +from abc import ABC, abstractmethod +from typing import Generic, TypeVar + +T = TypeVar("T") + + +class BaseCommand(ABC, Generic[T]): + """所有行情命令的基类。 + + 子类实现： + build_request() → 返回要发送的原始字节 + parse_response() → 从解压后的 body 返回强类型结果 + """ + + @abstractmethod + def build_request(self) -> bytes: + """构造请求包（含完整帧头）。""" + ... + + @abstractmethod + def parse_response(self, body: bytes) -> T: + """解析解压后的响应 body，返回强类型结果。""" + ... diff --git a/src/xmtdx/commands/company_info.py b/src/xmtdx/commands/company_info.py new file mode 100644 index 0000000..3cf226e --- /dev/null +++ b/src/xmtdx/commands/company_info.py @@ -0,0 +1,76 @@ +"""公司信息目录与内容命令。""" + +import struct + +from ..models.enums import Market +from ..models.finance import CompanyInfoCategory +from .base import BaseCommand + + +class GetCompanyInfoCategoryCmd(BaseCommand[list[CompanyInfoCategory]]): + """获取公司信息文件目录（文件名列表 + 每段偏移/长度）。""" + + def __init__(self, market: Market, code: str) -> None: + self.market = market + self.code = code.encode("utf-8") + + def build_request(self) -> bytes: + header = bytes.fromhex("0c0f109b00010e000e00cf02".replace(" ", "")) + return header + struct.pack(" list[CompanyInfoCategory]: + if len(body) < 2: + return [] + (num,) = struct.unpack_from(" len(body): + break + name_b, filename_b, start, length = struct.unpack_from("<64s80sII", body, pos) + pos += _RECORD_SIZE + + def _decode(b: bytes) -> str: + nul = b.find(b"\x00") + raw = b[:nul] if nul != -1 else b + return raw.decode("gbk", errors="replace") + + results.append(CompanyInfoCategory( + filename=_decode(filename_b), + start=start, + length=length, + )) + + return results + + +class GetCompanyInfoContentCmd(BaseCommand[str]): + """按文件名、偏移、长度读取公司信息文本（GBK 编码）。""" + + def __init__( + self, market: Market, code: str, filename: str, offset: int, length: int + ) -> None: + self.market = market + self.code = code.encode("utf-8") + self.filename = filename.encode("gbk") + self.offset = offset + self.length = length + + def build_request(self) -> bytes: + fname_padded = (self.filename + b"\x00" * 80)[:80] + header = bytes.fromhex("0c07109c0001680068 00d002".replace(" ", "")) + return header + struct.pack( + " str: + # 前12字节：10字节未知 + 2字节长度 + if len(body) < 12: + return "" + _, length = struct.unpack_from("<10sH", body, 0) + content = body[12 : 12 + length] + return content.decode("gbk", errors="replace") diff --git a/src/xmtdx/commands/finance_info.py b/src/xmtdx/commands/finance_info.py new file mode 100644 index 0000000..030101e --- /dev/null +++ b/src/xmtdx/commands/finance_info.py @@ -0,0 +1,84 @@ +"""最新财务数据命令。""" + +import struct + +from ..models.enums import Market +from ..models.finance import FinanceInfo +from .base import BaseCommand + +# 财务字段 struct 格式：1f + 2H + 2I + 30f +_FIN_FMT = " None: + self.market = market + self.code = code.encode("utf-8") + + def build_request(self) -> bytes: + header = bytes.fromhex("0c1f18760001 0b000b001000 0100".replace(" ", "")) + return header + struct.pack(" FinanceInfo: + pos = 2 # 跳过前2字节（记录数） + market_b, code_b = struct.unpack_from(" None: + self.market = market + self.code = code.encode("utf-8") + + def build_request(self) -> bytes: + header = bytes.fromhex("0c1b08000101 0e000e001d05".replace(" ", "")) + return header + struct.pack(" list[MinuteBar]: + return _parse_minute_body(body, skip=4) + + +class GetHistoryMinuteTimeDataCmd(BaseCommand[list[MinuteBar]]): + """获取历史某日分时数据（date 格式 YYYYMMDD）。""" + + def __init__(self, market: Market, code: str, date: int) -> None: + self.market = market + self.code = code.encode("utf-8") + self.date = date + + def build_request(self) -> bytes: + # 历史分时：header + pack(" list[MinuteBar]: + # 历史分时：pytdx 中 pos 跳过 6 字节（2 num + 4 未知） + return _parse_minute_body(body, skip=6) + + +def _parse_minute_body(body: bytes, skip: int = 4) -> list[MinuteBar]: + (num,) = struct.unpack_from(" None: + self.market = market + self.code = code.encode("utf-8") + self.category = category + self.start = start + self.count = count + + def build_request(self) -> bytes: + return struct.pack( + " list[SecurityBar]: + (ret_count,) = struct.unpack_from(" None: + self.market = market + + def build_request(self) -> bytes: + header = bytes.fromhex("0c0c186c000108000800 4e04".replace(" ", "")) + return header + struct.pack(" int: + (count,) = struct.unpack_from(" None: + self.market = market + self.start = start + + def build_request(self) -> bytes: + header = bytes.fromhex("0c01186401010600060050 04".replace(" ", "")) + return header + struct.pack(" list[SecurityInfo]: + (num,) = struct.unpack_from(" str: + """将 reversed_bytes0 整数转换为 HH:MM:SS.mmm 字符串。 + + 方法来自 pytdx issue #187。raw 为 14999212 → "14:59:57.163" + """ + s = str(raw) + if len(s) < 6: + return s + # 最后6位：前两位=秒，后四位=毫秒的某种编码 + time_part = s[:-6] + ":" + last6 = int(s[-6:]) + if int(s[-6:-4]) < 60: + time_part += s[-6:-4] + ":" + time_part += f"{last6 % 10000 * 60 / 10000.0:06.3f}" + else: + mins = last6 * 60 // 1000000 + secs = (last6 * 60 % 1000000) * 60 / 1000000.0 + time_part += f"{mins:02d}:{secs:06.3f}" + return time_part + + +class GetSecurityQuotesCmd(BaseCommand[list[SecurityQuote]]): + """批量获取实时行情（最多 80 只）。 + + Args: + stocks: [(market, code), ...] 列表 + """ + + def __init__(self, stocks: list[tuple[Market, str]]) -> None: + if not stocks: + raise ValueError("stocks 不能为空") + if len(stocks) > 80: + raise ValueError("单次最多查询 80 只股票") + self.stocks = stocks + + def build_request(self) -> bytes: + n = len(self.stocks) + payload_len = n * 7 + 12 + header = struct.pack( + " list[SecurityQuote]: + pos = 0 + # pytdx 跳过前2字节（b1 cb 魔数） + pos += 2 + (num,) = struct.unpack_from(" None: + self.market = market + self.code = code.encode("utf-8") + self.start = start + self.count = count + + def build_request(self) -> bytes: + header = bytes.fromhex("0c170801010 10e000e00c50f".replace(" ", "")) + return header + struct.pack(" list[TransactionRecord]: + return _parse_transaction_body(body, skip=2) + + +class GetHistoryTransactionDataCmd(BaseCommand[list[TransactionRecord]]): + """获取历史某日逐笔成交（date 格式 YYYYMMDD，分页）。""" + + def __init__( + self, market: Market, code: str, date: int, start: int, count: int = 800 + ) -> None: + self.market = market + self.code = code.encode("utf-8") + self.date = date + self.start = start + self.count = count + + def build_request(self) -> bytes: + # 历史逐笔：header + pack(" list[TransactionRecord]: + # 历史逐笔：num(2) + 4字节填充；无"成交笔数"字段 + return _parse_history_transaction_body(body) + + +def _parse_transaction_body(body: bytes) -> list[TransactionRecord]: + """当日逐笔：time + price + vol + num_orders + buyorsell + unknown""" + (num,) = struct.unpack_from(" list[TransactionRecord]: + """历史逐笔：num(2) + skip(4) + [time + price + vol + buyorsell + unknown]""" + (num,) = struct.unpack_from(" None: + self.market = market + self.code = code.encode("utf-8") + + def build_request(self) -> bytes: + header = bytes.fromhex("0c1f18760001 0b000b000f000100".replace(" ", "")) + return header + struct.pack(" list[XdxrRecord]: + if len(body) < 11: + return [] + + pos = 9 # 跳过9字节（market+code+未知） + (num,) = struct.unpack_from(" len(body): + break + market_b, code_b = struct.unpack_from(" len(body): + break + + chunk = body[pos : pos + 16] + pos += 16 + + rec = XdxrRecord( + market=Market(market_b), + code=code_b.decode("utf-8").rstrip("\x00"), + year=year, + month=month, + day=day, + category=category, + name=XDXR_CATEGORY_NAMES.get(category, str(category)), + _raw=body[record_start:pos], + ) + + if category == 1: + fenhong, peigujia, songzhuangu, peigu = struct.unpack(" float: + """股本数量解码（uint32 → 股数）。 + + pytdx 使用 get_volume 但会产生错误结果（xdxr Bug #1 注释）。 + 目前保持与服务器原始整数一致，待进一步逆向确认正确解码方式。 + """ + return float(raw) diff --git a/src/xmtdx/exceptions.py b/src/xmtdx/exceptions.py new file mode 100644 index 0000000..25b0581 --- /dev/null +++ b/src/xmtdx/exceptions.py @@ -0,0 +1,17 @@ +"""xmtdx 异常层次""" + + +class TdxError(Exception): + """所有 xmtdx 异常的基类""" + + +class TdxConnectionError(TdxError): + """TCP 连接失败或超时""" + + +class TdxDecodeError(TdxError): + """响应报文解析失败""" + + +class TdxCommandError(TdxError): + """命令执行失败（服务器返回错误）""" diff --git a/src/xmtdx/models/__init__.py b/src/xmtdx/models/__init__.py new file mode 100644 index 0000000..d926189 --- /dev/null +++ b/src/xmtdx/models/__init__.py @@ -0,0 +1,25 @@ +from .bar import SecurityBar +from .enums import KlineCategory, Market +from .finance import ( + XDXR_CATEGORY_NAMES, + CompanyInfoCategory, + FinanceInfo, + XdxrRecord, +) +from .quote import SecurityQuote +from .security import SecurityInfo +from .timeseries import MinuteBar, TransactionRecord + +__all__ = [ + "Market", + "KlineCategory", + "SecurityBar", + "SecurityQuote", + "SecurityInfo", + "MinuteBar", + "TransactionRecord", + "XdxrRecord", + "XDXR_CATEGORY_NAMES", + "FinanceInfo", + "CompanyInfoCategory", +] diff --git a/src/xmtdx/models/bar.py b/src/xmtdx/models/bar.py new file mode 100644 index 0000000..104c1a5 --- /dev/null +++ b/src/xmtdx/models/bar.py @@ -0,0 +1,28 @@ +"""K 线数据模型""" + +from dataclasses import dataclass, field + + +@dataclass +class SecurityBar: + """单根 K 线（适用于 1m/5m/15m/30m/60m/日/周/月/季/年）""" + + open: float + close: float + high: float + low: float + vol: float # 成交量（股） + amount: float # 成交额（元） + + year: int + month: int + day: int + hour: int + minute: int + + # 原始字节，供字段逆向分析使用 + _raw: bytes = field(default=b"", repr=False, compare=False) + + @property + def datetime_str(self) -> str: + return f"{self.year}-{self.month:02d}-{self.day:02d} {self.hour:02d}:{self.minute:02d}" diff --git a/src/xmtdx/models/enums.py b/src/xmtdx/models/enums.py new file mode 100644 index 0000000..e740250 --- /dev/null +++ b/src/xmtdx/models/enums.py @@ -0,0 +1,23 @@ +"""市场代码与 K 线周期枚举""" + +from enum import IntEnum + + +class Market(IntEnum): + SZ = 0 # 深圳 + SH = 1 # 上海 + + +class KlineCategory(IntEnum): + MIN_5 = 0 + MIN_15 = 1 + MIN_30 = 2 + MIN_60 = 3 + DAY = 4 + WEEK = 5 + MONTH = 6 + MIN_1 = 7 + MIN_3 = 8 # 通达信内部用，实际同 MIN_1 + YEAR = 9 + SEASON = 10 + YEAR_ALT = 11 diff --git a/src/xmtdx/models/finance.py b/src/xmtdx/models/finance.py new file mode 100644 index 0000000..6b6a75c --- /dev/null +++ b/src/xmtdx/models/finance.py @@ -0,0 +1,127 @@ +"""财务与公司信息模型""" + +from dataclasses import dataclass, field + +from .enums import Market + + +@dataclass +class XdxrRecord: + """除权除息记录（一只股票可有多条） + + pytdx Bug #1 已修复：循环内不再从 body[:7] 读 market/code， + 而是从当前 pos 正确读取。 + """ + + market: Market + code: str + year: int + month: int + day: int + category: int # 事件类型（见下方 CATEGORY_NAMES） + name: str # 事件类型名称 + + # category == 1（除权除息） + fenhong: float | None = None # 每股分红（元） + peigujia: float | None = None # 配股价（元） + songzhuangu: float | None = None # 送转股比例 + peigu: float | None = None # 配股比例 + + # category in [11, 12]（扩缩股） + suogu: float | None = None # 缩股比例 + + # category in [13, 14]（权证） + xingquanjia: float | None = None # 行权价 + fenshu: float | None = None # 分数 + + # category in [2..10]（股本变动类） + panqian_liutong: float | None = None # 盘前流通股本 + panhou_liutong: float | None = None # 盘后流通股本 + qian_zongguben: float | None = None # 前总股本 + hou_zongguben: float | None = None # 后总股本 + + _raw: bytes = field(default=b"", repr=False, compare=False) + + +XDXR_CATEGORY_NAMES: dict[int, str] = { + 1: "除权除息", + 2: "送配股上市", + 3: "非流通股上市", + 4: "未知股本变动", + 5: "股本变化", + 6: "增发新股", + 7: "股份回购", + 8: "增发新股上市", + 9: "转配股上市", + 10: "可转债上市", + 11: "扩缩股", + 12: "非流通股缩股", + 13: "送认购权证", + 14: "送认沽权证", +} + + +@dataclass +class FinanceInfo: + """最新财务数据（单只股票）""" + + market: Market + code: str + + # 股本（万股） + liutong_guben: float # 流通股本 + zong_guben: float # 总股本 + guojia_gu: float # 国家股 + faqiren_faren_gu: float # 发起人法人股 + faren_gu: float # 法人股 + b_gu: float # B股 + h_gu: float # H股 + zhigong_gu: float # 职工股 + + # 基本信息 + province: int # 所属省份代码 + industry: int # 所属行业代码 + updated_date: int # 财务更新日期 YYYYMMDD + ipo_date: int # 上市日期 YYYYMMDD + gudong_renshu: float # 股东人数 + + # 资产负债（元） + zong_zichan: float # 总资产 + liudong_zichan: float # 流动资产 + guding_zichan: float # 固定资产 + wuxing_zichan: float # 无形资产 + liudong_fuzhai: float # 流动负债 + changqi_fuzhai: float # 长期负债 + ziben_gongjijin: float # 资本公积金 + jing_zichan: float # 净资产 + + # 利润（元） + zhuying_shouru: float # 主营收入 + zhuying_lirun: float # 主营利润 + yingshou_zhangkuan: float # 应收账款 + yingye_lirun: float # 营业利润 + touzi_shouyu: float # 投资收益 + jingying_xianjinliu: float # 经营现金流 + zong_xianjinliu: float # 总现金流 + cunhuo: float # 存货 + lirun_zonghe: float # 利润总额 + shuihou_lirun: float # 税后利润 + jing_lirun: float # 净利润 + weifen_lirun: float # 未分配利润 + + # 每股指标 + meigujing_zichan: float # 每股净资产（原 baoliu1） + + # 协议保留字段（含义未完全确认） + reserve2: float = field(default=0.0, repr=False) # 原 baoliu2 + + _raw: bytes = field(default=b"", repr=False, compare=False) + + +@dataclass +class CompanyInfoCategory: + """公司信息文件目录条目""" + + filename: str # 文件名（如 '600000.txt'） + start: int # 内容起始偏移 + length: int # 内容长度（字节） diff --git a/src/xmtdx/models/quote.py b/src/xmtdx/models/quote.py new file mode 100644 index 0000000..3f66e26 --- /dev/null +++ b/src/xmtdx/models/quote.py @@ -0,0 +1,68 @@ +"""实时行情五档报价模型""" + +from dataclasses import dataclass, field + +from .enums import Market + + +@dataclass +class SecurityQuote: + """单只股票实时五档行情。 + + 带 unknown_ 前缀的字段为协议中尚未明确含义的字段，保留以供逆向分析。 + _raw 为该股票记录的原始字节切片。 + """ + + market: Market + code: str + + # 价格 + price: float # 现价 + pre_close: float # 昨收 + open: float # 今开 + high: float # 最高 + low: float # 最低 + + # 量额 + vol: float # 总成交量（手） + cur_vol: float # 当前成交量 + amount: float # 成交额（元） + s_vol: float # 内盘（主动卖） + b_vol: float # 外盘（主动买） + + # 活跃度指标（含义来自社区逆向，仅供参考） + active1: int + active2: int + + # 买盘五档 + bid1: float; bid_vol1: float + bid2: float; bid_vol2: float + bid3: float; bid_vol3: float + bid4: float; bid_vol4: float + bid5: float; bid_vol5: float + + # 卖盘五档 + ask1: float; ask_vol1: float + ask2: float; ask_vol2: float + ask3: float; ask_vol3: float + ask4: float; ask_vol4: float + ask5: float; ask_vol5: float + + # 已确认含义 + rise_speed: float # 涨速（原 reversed_bytes9 / 100） + + # 未知字段：买卖量之后的两个变长整数 + unknown_2: int = field(default=0, repr=False) # 原 reversed_bytes2 + unknown_3: int = field(default=0, repr=False) # 原 reversed_bytes3 + + # 未知字段：尾部四个变长整数 + unknown_5: int = field(default=0, repr=False) # 原 reversed_bytes5 + unknown_6: int = field(default=0, repr=False) # 原 reversed_bytes6 + unknown_7: int = field(default=0, repr=False) # 原 reversed_bytes7 + unknown_8: int = field(default=0, repr=False) # 原 reversed_bytes8 + + # 服务器时间字符串（从 unknown_0 原始整数解析，格式 HH:MM:SS.mmm） + server_time: str = field(default="", repr=True) + + # 原始字节（该股票记录切片） + _raw: bytes = field(default=b"", repr=False, compare=False) diff --git a/src/xmtdx/models/security.py b/src/xmtdx/models/security.py new file mode 100644 index 0000000..b6990d8 --- /dev/null +++ b/src/xmtdx/models/security.py @@ -0,0 +1,19 @@ +"""证券基本信息模型""" + +from dataclasses import dataclass, field + +from .enums import Market + + +@dataclass +class SecurityInfo: + """证券列表条目（来自 get_security_list）""" + + market: Market + code: str + name: str # 股票名称（GBK 解码，截断字节用 replacement char 替代） + volunit: int # 成交量单位（手 = volunit 股） + decimal_point: int # 价格小数位数 + pre_close: float # 昨收价（已修复 pytdx Bug #3：改用正确价格解码） + + _raw: bytes = field(default=b"", repr=False, compare=False) diff --git a/src/xmtdx/models/timeseries.py b/src/xmtdx/models/timeseries.py new file mode 100644 index 0000000..985a006 --- /dev/null +++ b/src/xmtdx/models/timeseries.py @@ -0,0 +1,39 @@ +"""分时与逐笔成交模型""" + +from dataclasses import dataclass, field + + +@dataclass +class MinuteBar: + """今日/历史分时（每分钟一条，共 240 条） + + unknown_1: 协议中第二个变长整数，含义未明（疑似均价的编码形式）。 + """ + + price: float # 价格 + vol: int # 成交量 + + # pytdx 中被完全丢弃的字段，保留以供分析 + unknown_1: int = field(default=0, repr=False) # 原 reversed1 + + _raw: bytes = field(default=b"", repr=False, compare=False) + + +@dataclass +class TransactionRecord: + """逐笔成交记录 + + unknown_last: pytdx 中被 _ 丢弃的最后一个变长整数，保留以供分析。 + 时间精度仅到分钟（协议限制），unknown_last 可能含秒或序号信息。 + """ + + hour: int + minute: int + price: float + vol: int + buyorsell: int # 0=卖, 1=买, 2=中性/撮合 + + # pytdx 中被丢弃的字段 + unknown_last: int = field(default=0, repr=False) + + _raw: bytes = field(default=b"", repr=False, compare=False) diff --git a/src/xmtdx/transport/__init__.py b/src/xmtdx/transport/__init__.py new file mode 100644 index 0000000..4b84324 --- /dev/null +++ b/src/xmtdx/transport/__init__.py @@ -0,0 +1,4 @@ +from .async_ import AsyncTdxConnection +from .sync import TdxConnection + +__all__ = ["TdxConnection", "AsyncTdxConnection"] diff --git a/src/xmtdx/transport/async_.py b/src/xmtdx/transport/async_.py new file mode 100644 index 0000000..80fc22a --- /dev/null +++ b/src/xmtdx/transport/async_.py @@ -0,0 +1,123 @@ +"""异步 TCP 连接（基于 asyncio）。""" + +import asyncio +from types import TracebackType +from typing import TYPE_CHECKING, TypeVar + +from ..codec.frame import HEADER_SIZE, decompress_body, parse_header +from ..commands.setup import SETUP_COMMANDS +from ..exceptions import TdxConnectionError + +if TYPE_CHECKING: + from ..commands.base import BaseCommand + +T = TypeVar("T") + +_DEFAULT_HOST = "180.153.18.170" +_DEFAULT_PORT = 7709 +_DEFAULT_TIMEOUT = 15.0 + + +class AsyncTdxConnection: + """异步通达信 TCP 连接（asyncio）。 + + 使用示例:: + + async with AsyncTdxConnection("180.153.18.170") as conn: + result = await conn.execute(SomeCommand(...)) + """ + + def __init__( + self, + host: str = _DEFAULT_HOST, + port: int = _DEFAULT_PORT, + timeout: float = _DEFAULT_TIMEOUT, + ) -> None: + self.host = host + self.port = port + self.timeout = timeout + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + + async def connect(self) -> None: + """建立 TCP 连接并完成握手。""" + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(self.host, self.port), + timeout=self.timeout, + ) + except (OSError, asyncio.TimeoutError) as e: + raise TdxConnectionError(f"无法连接 {self.host}:{self.port}: {e}") from e + self._reader = reader + self._writer = writer + await self._send_setup() + + async def close(self) -> None: + """关闭连接。""" + if self._writer is not None: + try: + self._writer.close() + await self._writer.wait_closed() + except OSError: + pass + self._reader = None + self._writer = None + + async def execute(self, cmd: "BaseCommand[T]") -> T: + """执行一条命令（异步版本）。""" + if self._writer is None or self._reader is None: + raise TdxConnectionError("未连接，请先调用 connect()") + request = cmd.build_request() + try: + self._writer.write(request) + await self._writer.drain() + header_buf = await self._recv_exact(HEADER_SIZE) + header = parse_header(header_buf) + raw_body = await self._recv_exact(header.zipsize) + except (OSError, asyncio.IncompleteReadError) as e: + raise TdxConnectionError(f"通信错误: {e}") from e + body = decompress_body(header, raw_body) + return cmd.parse_response(body) + + # ------------------------------------------------------------------ # + # context manager + # ------------------------------------------------------------------ # + + async def __aenter__(self) -> "AsyncTdxConnection": + await self.connect() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + await self.close() + + # ------------------------------------------------------------------ # + # internals + # ------------------------------------------------------------------ # + + async def _send_setup(self) -> None: + """按序发送三条握手命令并丢弃响应。""" + assert self._writer is not None + assert self._reader is not None + for cmd_bytes in SETUP_COMMANDS: + self._writer.write(cmd_bytes) + await self._writer.drain() + try: + hdr_buf = await asyncio.wait_for( + self._recv_exact(HEADER_SIZE), timeout=5.0 + ) + hdr = parse_header(hdr_buf) + if hdr.zipsize > 0: + await self._recv_exact(hdr.zipsize) + except (OSError, asyncio.TimeoutError, asyncio.IncompleteReadError): + pass + + async def _recv_exact(self, n: int) -> bytes: + """读满 n 字节。""" + assert self._reader is not None + data = await self._reader.readexactly(n) + return data diff --git a/src/xmtdx/transport/sync.py b/src/xmtdx/transport/sync.py new file mode 100644 index 0000000..c432dc7 --- /dev/null +++ b/src/xmtdx/transport/sync.py @@ -0,0 +1,121 @@ +"""同步 TCP 连接（基于 socket）。""" + +import socket +from types import TracebackType +from typing import TYPE_CHECKING, TypeVar + +from ..codec.frame import HEADER_SIZE, decompress_body, parse_header +from ..commands.setup import SETUP_COMMANDS +from ..exceptions import TdxConnectionError + +if TYPE_CHECKING: + from ..commands.base import BaseCommand + +T = TypeVar("T") + +_DEFAULT_HOST = "180.153.18.170" +_DEFAULT_PORT = 7709 +_DEFAULT_TIMEOUT = 15.0 + + +class TdxConnection: + """同步通达信 TCP 连接。 + + 使用示例:: + + with TdxConnection("180.153.18.170") as conn: + result = conn.execute(SomeCommand(...)) + """ + + def __init__( + self, + host: str = _DEFAULT_HOST, + port: int = _DEFAULT_PORT, + timeout: float = _DEFAULT_TIMEOUT, + ) -> None: + self.host = host + self.port = port + self.timeout = timeout + self._sock: socket.socket | None = None + + def connect(self) -> None: + """建立 TCP 连接并完成握手（发送3条 setup 命令）。""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + try: + sock.connect((self.host, self.port)) + except OSError as e: + sock.close() + raise TdxConnectionError(f"无法连接 {self.host}:{self.port}: {e}") from e + self._sock = sock + self._send_setup() + + def close(self) -> None: + """关闭连接。""" + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + self._sock = None + + def execute(self, cmd: "BaseCommand[T]") -> T: + """执行一条命令：发送请求，接收并解压响应，返回解析结果。""" + if self._sock is None: + raise TdxConnectionError("未连接，请先调用 connect()") + request = cmd.build_request() + try: + self._sock.sendall(request) + header_buf = self._recv_exact(HEADER_SIZE) + header = parse_header(header_buf) + raw_body = self._recv_exact(header.zipsize) + except OSError as e: + raise TdxConnectionError(f"通信错误: {e}") from e + body = decompress_body(header, raw_body) + return cmd.parse_response(body) + + # ------------------------------------------------------------------ # + # context manager + # ------------------------------------------------------------------ # + + def __enter__(self) -> "TdxConnection": + self.connect() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + self.close() + + # ------------------------------------------------------------------ # + # internals + # ------------------------------------------------------------------ # + + def _send_setup(self) -> None: + """按序发送三条握手命令并丢弃响应。""" + assert self._sock is not None + for cmd_bytes in SETUP_COMMANDS: + self._sock.sendall(cmd_bytes) + # 读取并丢弃握手响应 + try: + hdr_buf = self._recv_exact(HEADER_SIZE) + hdr = parse_header(hdr_buf) + if hdr.zipsize > 0: + self._recv_exact(hdr.zipsize) + except OSError: + # 部分服务器的握手无响应，忽略错误 + pass + + def _recv_exact(self, n: int) -> bytes: + """循环 recv 直到读满 n 字节。""" + assert self._sock is not None + buf = bytearray() + while len(buf) < n: + chunk = self._sock.recv(n - len(buf)) + if not chunk: + raise TdxConnectionError("连接被服务器关闭") + buf.extend(chunk) + return bytes(buf) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_codec_datetime.py b/tests/unit/test_codec_datetime.py new file mode 100644 index 0000000..08fe660 --- /dev/null +++ b/tests/unit/test_codec_datetime.py @@ -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(" bytes: + return struct.pack(" bytes: + return struct.pack(" bytes: + return struct.pack(" 3.0" 含义：raw 4098 对应真实股数 3.0亿， + # 但 get_volume(4098) ≈ 5.88e-39（接近零），说明 xdxr_info 里对股本字段 + # 调用 get_volume 是错误用法。xmtdx 在 xdxr_info 命令中会用正确的解码方式。 + val, pos = get_volume(_pack(4098), 0) + assert abs(val) < 1e-30 # 接近零，与 pytdx 行为一致 + + def test_advances_pos(self): + data = _pack(0) + _pack(0) + _, pos = get_volume(data, 0) + assert pos == 4 + _, pos2 = get_volume(data, pos) + assert pos2 == 8 + + def test_nonnegative(self): + # 成交量不应为负 + for raw in [0, 1000, 0x10000, 0x1000000, 0x7FFFFFFF]: + val, _ = get_volume(_pack(raw), 0) + assert val >= 0.0