feat: 完善实时行情字段并增加大文件拉取功能

1. 行情增强:解析 SecurityQuote 中的 limit_up (涨停价) 和 limit_down (跌停价)。
2. 新增命令:实现 GetReportFileCmd (0x06B9),支持从服务器拉取 base_info.zip 等大文件。
3. 客户端 API:TdxClient/AsyncTdxClient 增加 get_report_file 方法,支持自动分片。
4. 验证脚本:增加 verify_limits.py 和 verify_report_file.py 实测脚本。
5. 文档更新:README.md 同步 API 变更及数据模型更新。
This commit is contained in:
M
2026-04-14 17:57:54 +08:00
parent fcd14c1d49
commit 2fbc7b9782
8 changed files with 160 additions and 6 deletions
+19
View File
@@ -0,0 +1,19 @@
"""测试 get_report_file 拉取板块文件。"""
import sys
import pathlib
# 添加 src 到 path
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "src"))
from xmtdx import TdxClient
def main():
print("正在拉取 block_gn.dat (使用 get_report_file) ...")
with TdxClient.from_best_host() as c:
data = c.get_report_file("block_gn.dat")
print(f"成功拉取 {len(data)} 字节。")
if data:
print(f"前 16 字节 hex: {data[:16].hex()}")
if __name__ == "__main__":
main()
+27
View File
@@ -0,0 +1,27 @@
"""验证涨跌停价。"""
import sys
import pathlib
# 添加 src 到 path
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "src"))
from xmtdx import TdxClient, Market
def main():
print("正在连接服务器验证涨跌停...")
with TdxClient.from_best_host() as c:
# 选取一些知名 A 股
stocks = [(Market.SH, "600000"), (Market.SZ, "000001"), (Market.SZ, "300750")]
quotes = c.get_security_quotes(stocks)
print(f" {'代码':>8} {'昨收':>8} {'现价':>8} {'涨停价':>8} {'跌停价':>8} {'涨速':>8}")
for q in quotes:
print(f" {q.code:>8} {q.pre_close:>8.2f} {q.price:>8.2f} {q.limit_up:>8.2f} {q.limit_down:>8.2f} {q.rise_speed:>8.2f}")
# 校验涨停价是否大约为昨收 * 1.1 (主板/深证) 或 * 1.2 (创业板)
expected_up = round(q.pre_close * (1.2 if q.code.startswith("300") else 1.1), 2)
if abs(q.limit_up - expected_up) > 0.02:
print(f" [!] 涨停价异常:预期约 {expected_up}")
if __name__ == "__main__":
main()
+36
View File
@@ -0,0 +1,36 @@
"""验证大文件拉取。"""
import sys
import pathlib
import zipfile
import io
# 添加 src 到 path
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "src"))
from xmtdx import TdxClient
def main():
print("正在拉取 base_info.zip ...")
with TdxClient.from_best_host() as c:
data = c.get_report_file("base_info.zip")
print(f"成功拉取 {len(data)} 字节。")
if not data:
print("失败:返回数据为空。")
return
# 尝试作为 zip 打开验证
try:
with zipfile.ZipFile(io.BytesIO(data)) as z:
files = z.namelist()
print(f"Zip 文件包含 {len(files)} 个文件:{files[:5]}...")
# 检查是否包含 base_info.txt
if any("base_info.txt" in f for f in files):
print("验证成功:包含 base_info.txt")
except Exception as e:
print(f"Zip 校验失败: {e}")
# 打印前 16 字节 hex
print(f"Header hex: {data[:16].hex()}")
if __name__ == "__main__":
main()