mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 14:34:15 +08:00
cleanup: 删除开发过程中的中间验证脚本
This commit is contained in:
@@ -1,19 +0,0 @@
|
||||
"""测试 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()
|
||||
@@ -1,36 +0,0 @@
|
||||
"""实测板块信息获取。"""
|
||||
import sys
|
||||
import pathlib
|
||||
|
||||
# 添加 src 到 path
|
||||
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent / "src"))
|
||||
|
||||
from xmtdx import TdxClient
|
||||
|
||||
def main():
|
||||
print("正在寻找最优服务器...")
|
||||
try:
|
||||
with TdxClient.from_best_host() as c:
|
||||
print(f"已连接到: {c._host}")
|
||||
|
||||
# 尝试获取概念板块 (block_gn.dat)
|
||||
filename = "block_gn.dat"
|
||||
print(f"正在获取 {filename} ...")
|
||||
blocks = c.get_block_info(filename)
|
||||
|
||||
print(f"成功获取 {len(blocks)} 个板块。")
|
||||
|
||||
# 打印前 5 个板块及其前 3 个股票
|
||||
for b in blocks[:5]:
|
||||
print(f"板块名称: {b.name:<12} 股票数: {b.count:<5} 样例: {b.codes[:3]}")
|
||||
|
||||
if not blocks:
|
||||
print("警告:未获取到任何板块数据。")
|
||||
|
||||
except Exception as e:
|
||||
print(f"实测失败: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,27 +0,0 @@
|
||||
"""验证涨跌停价。"""
|
||||
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()
|
||||
@@ -1,36 +0,0 @@
|
||||
"""验证大文件拉取。"""
|
||||
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()
|
||||
Reference in New Issue
Block a user