mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 16:54:17 +08:00
MandClaude Sonnet 4.6
74ae6b6975
股本字段（category 2–10 的 panqian_liutong / panhou_liutong / qian_zongguben / hou_zongguben）与成交量字段使用相同的通达信自定义 4 字节浮点格式；原 float(uint32) 直解差约 374 倍。 改用 _decode_volume 后，最近一条 hou_zongguben（3,330,583.75 万股） 与 FinanceInfo.zong_guben / 10000（3,330,583.75 万股）完全吻合。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
336 lines
11 KiB
Python
336 lines
11 KiB
Python
"""离线 fixture 测试：将录制的原始 body 字节喂给各命令 parse_response，验证解析结果。
|
||
|
||
fixtures/ 目录下每个 .hex 文件是一次真实服务器响应的 body（已解压），
|
||
对应的 .json 文件记录关键预期值，供手工核对。
|
||
此测试文件直接断言解析结果，无需网络连接。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import pathlib
|
||
|
||
FIXTURES = pathlib.Path(__file__).parent.parent / "fixtures"
|
||
|
||
|
||
def load_hex(name: str) -> bytes:
|
||
return bytes.fromhex((FIXTURES / f"{name}.hex").read_text().strip())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# security_count
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_security_count_parse():
|
||
from xmtdx.commands.security_count import GetSecurityCountCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("security_count")
|
||
cmd = GetSecurityCountCmd(Market.SH)
|
||
count = cmd.parse_response(body)
|
||
|
||
assert isinstance(count, int)
|
||
assert count > 0
|
||
# 体积固定为 2 字节，结果与录制时完全一致
|
||
assert count == 26885
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# security_list
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_security_list_parse():
|
||
from xmtdx.commands.security_list import GetSecurityListCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("security_list")
|
||
cmd = GetSecurityListCmd(Market.SH, 0)
|
||
records = cmd.parse_response(body)
|
||
|
||
assert len(records) == 1000
|
||
|
||
r0 = records[0]
|
||
assert r0.code == "999999"
|
||
assert r0.name == "上证指数"
|
||
# pre_close is a float; the index level is stored ×100 and has many decimal places
|
||
assert abs(r0.pre_close - 11654847.33) < 1.0
|
||
|
||
# _raw present and non-empty for every record
|
||
assert all(len(r._raw) > 0 for r in records)
|
||
|
||
|
||
def test_security_list_gbk_no_crash():
|
||
"""Bug #2 修复验证：GBK 解码不崩溃，所有记录均有 code。"""
|
||
from xmtdx.commands.security_list import GetSecurityListCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("security_list")
|
||
cmd = GetSecurityListCmd(Market.SH, 0)
|
||
records = cmd.parse_response(body)
|
||
assert all(r.code for r in records)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# security_bars
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_security_bars_parse():
|
||
from xmtdx.commands.security_bars import GetSecurityBarsCmd
|
||
from xmtdx.models.enums import KlineCategory, Market
|
||
|
||
body = load_hex("security_bars")
|
||
cmd = GetSecurityBarsCmd(Market.SH, "600000", KlineCategory.DAY, 0, 5)
|
||
bars = cmd.parse_response(body)
|
||
|
||
assert len(bars) == 5
|
||
|
||
b0 = bars[0]
|
||
assert abs(b0.open - 10.25) < 0.01
|
||
assert abs(b0.high - 10.25) < 0.01
|
||
assert abs(b0.low - 10.08) < 0.01
|
||
assert abs(b0.close - 10.12) < 0.01
|
||
assert b0.vol > 0
|
||
|
||
# OHLC sanity: high ≥ open,close,low; low ≤ open,close
|
||
for bar in bars:
|
||
assert bar.high >= bar.open - 0.001
|
||
assert bar.high >= bar.close - 0.001
|
||
assert bar.low <= bar.open + 0.001
|
||
assert bar.low <= bar.close + 0.001
|
||
assert bar.vol > 0
|
||
assert len(bar._raw) > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# security_quotes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_security_quotes_parse():
|
||
from xmtdx.commands.security_quotes import GetSecurityQuotesCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("security_quotes")
|
||
cmd = GetSecurityQuotesCmd([(Market.SH, "600000")])
|
||
quotes = cmd.parse_response(body)
|
||
|
||
assert len(quotes) == 1
|
||
|
||
q = quotes[0]
|
||
assert q.code == "600000"
|
||
assert abs(q.pre_close - 9.93) < 0.01
|
||
|
||
# unknown fields are captured (not discarded)
|
||
assert hasattr(q, "unknown_2")
|
||
assert hasattr(q, "unknown_3")
|
||
assert hasattr(q, "unknown_5")
|
||
assert hasattr(q, "unknown_6")
|
||
assert hasattr(q, "unknown_7")
|
||
assert hasattr(q, "unknown_8")
|
||
assert hasattr(q, "rise_speed")
|
||
assert len(q._raw) > 0
|
||
|
||
# fixed values from frozen fixture
|
||
assert q.unknown_2 == -1
|
||
assert q.unknown_3 == 22694
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# minute_time
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_minute_time_parse():
|
||
from xmtdx.commands.minute_time import GetMinuteTimeDataCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("minute_time")
|
||
cmd = GetMinuteTimeDataCmd(Market.SH, "600000")
|
||
bars = cmd.parse_response(body)
|
||
|
||
assert len(bars) == 240
|
||
|
||
b0 = bars[0]
|
||
assert isinstance(b0.price, float)
|
||
assert isinstance(b0.vol, int)
|
||
# Bug #5 fix: unknown_1 is preserved, not discarded
|
||
assert hasattr(b0, "unknown_1")
|
||
assert isinstance(b0.unknown_1, int)
|
||
assert len(b0._raw) > 0
|
||
|
||
# fixed values
|
||
assert abs(b0.price - 0.01) < 0.001
|
||
assert b0.vol == 48
|
||
assert b0.unknown_1 == 54
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# history_minute_time
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_history_minute_time_parse():
|
||
from xmtdx.commands.minute_time import GetHistoryMinuteTimeDataCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("history_minute_time")
|
||
cmd = GetHistoryMinuteTimeDataCmd(Market.SH, "600000", 20250108)
|
||
bars = cmd.parse_response(body)
|
||
|
||
assert len(bars) == 240
|
||
|
||
b0 = bars[0]
|
||
assert abs(b0.price - 10.29) < 0.01
|
||
assert b0.vol == 10044
|
||
assert hasattr(b0, "unknown_1")
|
||
assert len(b0._raw) > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# transaction (current day)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_transaction_parse():
|
||
from xmtdx.commands.transaction import GetTransactionDataCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("transaction")
|
||
cmd = GetTransactionDataCmd(Market.SH, "600000", 0, 10)
|
||
recs = cmd.parse_response(body)
|
||
|
||
assert len(recs) == 10
|
||
|
||
r0 = recs[0]
|
||
assert r0.hour == 14
|
||
assert r0.minute == 59
|
||
assert abs(r0.price - 9.9) < 0.01
|
||
assert r0.vol == 0
|
||
|
||
# Bug #4 fix: unknown_last captured
|
||
assert hasattr(r0, "unknown_last")
|
||
assert len(r0._raw) > 0
|
||
|
||
# buyorsell: 0=buy, 1=sell, 2=neutral, 8=auction — field is an int
|
||
for r in recs:
|
||
assert isinstance(r.buyorsell, int)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# history_transaction
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_history_transaction_parse():
|
||
from xmtdx.commands.transaction import GetHistoryTransactionDataCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("history_transaction")
|
||
cmd = GetHistoryTransactionDataCmd(Market.SH, "600000", 20250108, 0, 10)
|
||
recs = cmd.parse_response(body)
|
||
|
||
assert len(recs) == 10
|
||
|
||
r0 = recs[0]
|
||
assert r0.hour == 14
|
||
assert r0.minute == 56
|
||
assert abs(r0.price - 10.3) < 0.01
|
||
assert r0.vol == 50
|
||
|
||
assert hasattr(r0, "unknown_last")
|
||
assert len(r0._raw) > 0
|
||
|
||
for r in recs:
|
||
assert isinstance(r.buyorsell, int)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# xdxr_info
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_xdxr_info_parse():
|
||
from xmtdx.commands.xdxr_info import GetXdxrInfoCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("xdxr_info")
|
||
cmd = GetXdxrInfoCmd(Market.SH, "600000")
|
||
recs = cmd.parse_response(body)
|
||
|
||
assert len(recs) == 87
|
||
|
||
r0 = recs[0]
|
||
assert r0.year == 1999
|
||
assert r0.month == 11
|
||
assert r0.day == 10
|
||
assert r0.category == 5
|
||
|
||
# Bug #1 fix: each record has a unique date (not all reading from body[:7])
|
||
dates = {(r.year, r.month, r.day) for r in recs}
|
||
assert len(dates) > 1, "All records have the same date — Bug #1 not fixed!"
|
||
|
||
assert all(len(r._raw) > 0 for r in recs)
|
||
|
||
# share count decode: 通达信自定义浮点，单位万股，与 FinanceInfo.zong_guben/10000 一致
|
||
stock_recs = [r for r in recs if 2 <= r.category <= 10]
|
||
last = stock_recs[-1]
|
||
# 最近一条 hou_zongguben ≈ 3_330_583.75 万股（与 FinanceInfo.zong_guben 33_305_837_500 ÷ 10000 完全吻合）
|
||
assert last.hou_zongguben is not None
|
||
assert abs(last.hou_zongguben - 3_330_583.75) < 1.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# finance_info
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_finance_info_parse():
|
||
from xmtdx.commands.finance_info import GetFinanceInfoCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("finance_info")
|
||
cmd = GetFinanceInfoCmd(Market.SH, "600000")
|
||
info = cmd.parse_response(body)
|
||
|
||
# Check key fields are present and reasonable
|
||
assert info.liutong_guben > 0
|
||
assert info.zong_guben > 0
|
||
assert info.meigujing_zichan > 0
|
||
|
||
# Fixed values from frozen fixture
|
||
assert abs(info.liutong_guben - 33305837500.0) < 1e6
|
||
assert abs(info.zong_guben - 33305837500.0) < 1e6
|
||
assert abs(info.meigujing_zichan - 22.13) < 0.1
|
||
|
||
assert len(info._raw) > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# company_info_category
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_company_info_category_parse():
|
||
from xmtdx.commands.company_info import GetCompanyInfoCategoryCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("company_info_category")
|
||
cmd = GetCompanyInfoCategoryCmd(Market.SH, "600000")
|
||
cats = cmd.parse_response(body)
|
||
|
||
assert len(cats) == 16
|
||
|
||
c0 = cats[0]
|
||
assert c0.name == "最新提示"
|
||
assert c0.filename == "600000.txt"
|
||
assert c0.start == 0
|
||
assert c0.length == 11426
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# company_info_content
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_company_info_content_parse():
|
||
from xmtdx.commands.company_info import GetCompanyInfoContentCmd
|
||
from xmtdx.models.enums import Market
|
||
|
||
body = load_hex("company_info_content")
|
||
cmd = GetCompanyInfoContentCmd(Market.SH, "600000", "600000.txt", 0, 11426)
|
||
text = cmd.parse_response(body)
|
||
|
||
assert isinstance(text, str)
|
||
assert len(text) == 8070
|
||
assert "600000" in text
|
||
assert "浦发银行" in text
|