mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 23:54:21 +08:00
minionszywandClaude Sonnet 4.6
283682f6b4
替代年久失修的 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>
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""响应帧头解析与解压单元测试。"""
|
|
|
|
import struct
|
|
import zlib
|
|
|
|
from xmtdx.codec.frame import HEADER_SIZE, decompress_body, parse_header
|
|
|
|
|
|
def _make_header(zipsize: int, unzipsize: int) -> bytes:
|
|
return struct.pack("<IIIHH", 0, 0, 0, zipsize, unzipsize)
|
|
|
|
|
|
class TestParseHeader:
|
|
def test_uncompressed(self):
|
|
h = parse_header(_make_header(100, 100))
|
|
assert h.zipsize == 100
|
|
assert h.unzipsize == 100
|
|
|
|
def test_compressed(self):
|
|
h = parse_header(_make_header(50, 200))
|
|
assert h.zipsize == 50
|
|
assert h.unzipsize == 200
|
|
|
|
def test_header_size(self):
|
|
assert HEADER_SIZE == 16
|
|
|
|
|
|
class TestDecompressBody:
|
|
def test_no_compression(self):
|
|
h = parse_header(_make_header(5, 5))
|
|
body = b"hello"
|
|
assert decompress_body(h, body) == b"hello"
|
|
|
|
def test_zlib_decompression(self):
|
|
original = b"hello world" * 10
|
|
compressed = zlib.compress(original)
|
|
h = parse_header(_make_header(len(compressed), len(original)))
|
|
result = decompress_body(h, compressed)
|
|
assert result == original
|