aeroxw/easy-tdx
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/easy-tdx.git synced 2026-09-12 16:54:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add examples (01-08), fix index bars parsing, add ruff hook

Browse Source 
- Add example scripts for all API categories (connection, market info,
  kline, minute, transaction, finance, block, fund flow)
- Fix GetIndexBarsCmd: index bar records have 4 extra bytes (advance/
  decline counts) that were not consumed, causing pos drift and
  corrupted dates/volumes for all records after the first
- Fix price_limits.py example (SecurityQuote has no name attr)
- Fix finance_info.py display (scientific notation -> formatted numbers)
- Add PostToolUse ruff hook (scripts/ruff_hook.py)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
GitHub
2026-05-21 18:36:50 +08:00
co-authored by Claude Opus 4.7
parent ace1099ab0
commit 7fd6e610cf
27 changed files with 735 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/01_connection/async_connect.py
+24
View File
@@ -0,0 +1,24 @@
"""演示：异步客户端连接与基本用法。"""
import asyncio
from xmtdx import AsyncTdxClient, Market, KlineCategory
async def main():
# 手动指定服务器
async with AsyncTdxClient("180.153.18.170") as c:
count = await c.get_security_count(Market.SH)
print(f"沪市证券总数: {count}")
# 自动优选服务器
async with AsyncTdxClient.from_best_host() as c:
bars = await c.get_security_bars(
Market.SH, "600000", KlineCategory.DAY, 0, 5
)
for bar in bars:
print(f"{bar.year}-{bar.month:02d}-{bar.day:02d} "
f"开:{bar.open:.2f} 高:{bar.high:.2f} "
f"低:{bar.low:.2f} 收:{bar.close:.2f}")
asyncio.run(main())
examples/01_connection/connect_best_host.py
+13
View File
@@ -0,0 +1,13 @@
"""演示：自动从候选服务器中选延迟最低的建立连接。"""
from xmtdx import TdxClient, Market
# 方式一：手动指定服务器
with TdxClient("180.153.18.170") as c:
print(f"已连接到 {c._host}:{c._port}")
# 方式二：自动优选最低延迟服务器
with TdxClient.from_best_host() as c:
print(f"已自动选择最优服务器: {c._host}:{c._port}")
count = c.get_security_count(Market.SH)
print(f"沪市证券总数: {count}")
examples/01_connection/ping_servers.py
+9
View File
@@ -0,0 +1,9 @@
"""演示：测量多台通达信服务器延迟并排序。"""
import pandas as pd
from xmtdx import TdxClient
results = TdxClient.ping_all()
df = pd.DataFrame(results, columns=["服务器", "延迟(s)"])
df["延迟(ms)"] = df["延迟(s)"] * 1000
print(df[["服务器", "延迟(ms)"]].to_string(index=False))
examples/02_market_info/market_stat.py
+19
View File
@@ -0,0 +1,19 @@
"""演示：获取全市场涨跌统计概况。"""
import pandas as pd
from xmtdx import TdxClient
with TdxClient.from_best_host() as c:
stat = c.get_market_stat()
df = pd.DataFrame([{
"上涨": stat.up_count,
"下跌": stat.down_count,
"平盘": stat.neutral_count,
"停牌(估算)": stat.suspended_count,
"总计": stat.total_count,
"成交额(亿)": round(stat.total_amount / 1e8, 2),
"总市值(万亿)": round(stat.total_market_cap / 1e12, 4),
"涨停": stat.limit_up_count,
"跌停": stat.limit_down_count,
}])
print(df.T.to_string(header=False))
examples/02_market_info/security_count.py
+9
View File
@@ -0,0 +1,9 @@
"""演示：获取市场证券总数。"""
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
sh_count = c.get_security_count(Market.SH)
sz_count = c.get_security_count(Market.SZ)
print(f"沪市证券总数: {sh_count}")
print(f"深市证券总数: {sz_count}")
examples/02_market_info/security_list.py
+40
View File
@@ -0,0 +1,40 @@
"""演示：获取市场证券列表（分页）。
展示 SecurityInfo 全部字段的中文映射与表结构。
"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
stocks = c.get_security_list(Market.SH, start=0)
# 表结构说明
print("=" * 70)
print("SecurityInfo 表结构（字段中英文对照）")
print("=" * 70)
schema = pd.DataFrame([
{"英文字段": "market", "中文含义": "市场", "类型": "Market", "说明": "SZ=深圳 SH=上海 BJ=北京"},
{"英文字段": "code", "中文含义": "证券代码", "类型": "str", "说明": "6位代码，如 600000"},
{"英文字段": "name", "中文含义": "证券名称", "类型": "str", "说明": "GBK 解码"},
{"英文字段": "volunit", "中文含义": "成交量单位", "类型": "int", "说明": "1手 = volunit 股"},
{"英文字段": "decimal_point", "中文含义": "价格小数位", "类型": "int", "说明": "通常为 2"},
{"英文字段": "pre_close", "中文含义": "昨收价", "类型": "float", "说明": "通达信自定义浮点"},
{"英文字段": "industry_tdx", "中文含义": "通达信行业", "类型": "str", "说明": "需 get_security_list_all()"},
{"英文字段": "industry_sw", "中文含义": "申万行业", "类型": "str", "说明": "需 get_security_list_all()"},
])
print(schema.to_string(index=False))
# 全字段中文 DataFrame
print(f"\n沪市第 1 页，共 {len(stocks)} 只:")
df = pd.DataFrame([{
"市场": s.market.name,
"代码": s.code,
"名称": s.name,
"成交量单位(股/手)": s.volunit,
"价格小数位": s.decimal_point,
"昨收价": s.pre_close,
"通达信行业": s.industry_tdx or "",
"申万行业": s.industry_sw or "",
} for s in stocks])
print(df.head(20).to_string(index=False))
examples/02_market_info/security_list_all.py
+46
View File
@@ -0,0 +1,46 @@
"""演示：获取沪深 A 股完整列表（含行业映射）。
展示 SecurityInfo 全部字段（含扩展行业字段）的中文映射。
注意：此方法需要拉取 tdxhy.cfg 并遍历全部证券，耗时较长。
"""
import logging
import pandas as pd
from xmtdx import TdxClient
# 启用日志，查看分页进度
logging.basicConfig(level=logging.INFO, format="%(message)s")
# timeout 调大到 30 秒，避免全量拉取时分页请求超时
with TdxClient.from_best_host(timeout=30.0) as c:
all_stocks = c.get_security_list_all()
# 表结构说明
print("=" * 70)
print("SecurityInfo 表结构（字段中英文对照）")
print("=" * 70)
schema = pd.DataFrame([
{"英文字段": "market", "中文含义": "市场", "类型": "Market", "说明": "SZ=深圳 SH=上海 BJ=北京"},
{"英文字段": "code", "中文含义": "证券代码", "类型": "str", "说明": "6位代码，如 600000"},
{"英文字段": "name", "中文含义": "证券名称", "类型": "str", "说明": "GBK 解码"},
{"英文字段": "volunit", "中文含义": "成交量单位", "类型": "int", "说明": "1手 = volunit 股"},
{"英文字段": "decimal_point", "中文含义": "价格小数位", "类型": "int", "说明": "通常为 2"},
{"英文字段": "pre_close", "中文含义": "昨收价", "类型": "float", "说明": "通达信自定义浮点"},
{"英文字段": "industry_tdx", "中文含义": "通达信行业", "类型": "str", "说明": "如 T1001，来自 tdxhy.cfg"},
{"英文字段": "industry_sw", "中文含义": "申万行业", "类型": "str", "说明": "如 X500102，来自 tdxhy.cfg"},
])
print(schema.to_string(index=False))
# 全字段中文 DataFrame
print(f"\n沪深 A 股总数: {len(all_stocks)}")
df = pd.DataFrame([{
"市场": s.market.name,
"代码": s.code,
"名称": s.name,
"成交量单位(股/手)": s.volunit,
"价格小数位": s.decimal_point,
"昨收价": s.pre_close,
"通达信行业": s.industry_tdx or "",
"申万行业": s.industry_sw or "",
} for s in all_stocks])
print(df.head(20).to_string(index=False))
examples/02_market_info/security_quotes.py
+25
View File
@@ -0,0 +1,25 @@
"""演示：批量获取实时五档行情。最多支持 80 只/次。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
stocks = [
(Market.SH, "600000"), # 浦发银行
(Market.SH, "600519"), # 贵州茅台
(Market.SZ, "000001"), # 平安银行
(Market.SZ, "000858"), # 五粮液
]
quotes = c.get_security_quotes(stocks)
df = pd.DataFrame([{
"代码": q.code,
"现价": q.price,
"涨跌幅%": (q.price - q.pre_close) / q.pre_close * 100,
"今开": q.open,
"最高": q.high,
"最低": q.low,
"昨收": q.pre_close,
"成交量(手)": q.vol,
"成交额": q.amount,
} for q in quotes])
print(df.to_string(index=False))
examples/03_kline/index_bars.py
+25
View File
@@ -0,0 +1,25 @@
"""演示：获取指数 K 线数据。
常用指数代码:
上证指数: Market.SH, "000001"
深证成指: Market.SZ, "399001"
创业板指: Market.SZ, "399006"
"""
import pandas as pd
from xmtdx import TdxClient, Market, KlineCategory
with TdxClient.from_best_host() as c:
bars = c.get_index_bars(Market.SH, "999999", KlineCategory.DAY, 0, 10)
df = pd.DataFrame([{
"日期": f"{b.year}-{b.month:02d}-{b.day:02d}",
"开盘": b.open,
"最高": b.high,
"最低": b.low,
"收盘": b.close,
"成交量": b.vol,
"成交额": b.amount,
} for b in reversed(bars)])
print("上证指数 日K线:")
fmt = {"成交量": lambda x: f"{x:,.0f}", "成交额": lambda x: f"{x:,.0f}"}
print(df.to_string(index=False, formatters=fmt))
examples/03_kline/security_bars.py
+23
View File
@@ -0,0 +1,23 @@
"""演示：获取个股 K 线数据。
K 线类别:
KlineCategory.MIN_1 / MIN_5 / MIN_15 / MIN_30 / MIN_60
KlineCategory.DAY / WEEK / MONTH / YEAR
"""
import pandas as pd
from xmtdx import TdxClient, Market, KlineCategory
with TdxClient.from_best_host() as c:
bars = c.get_security_bars(Market.SZ, "002176", KlineCategory.DAY, 0, 100)
df = pd.DataFrame([{
"日期": f"{b.year}-{b.month:02d}-{b.day:02d}",
"开盘": b.open,
"最高": b.high,
"最低": b.low,
"收盘": b.close,
"成交量": b.vol,
"成交额": b.amount,
} for b in reversed(bars)])
print("上证指数 日K线:")
print(df.to_string(index=False))
examples/04_minute/history_minute_data.py
+15
View File
@@ -0,0 +1,15 @@
"""演示：获取历史某日分时数据。date 参数为 YYYYMMDD 格式的整数。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
date = 20250110
bars = c.get_history_minute_time_data(Market.SH, "600000", date)
df = pd.DataFrame([{
"序号": i + 1,
"价格": bar.price,
"成交量": bar.vol,
} for i, bar in enumerate(bars)])
print(f"浦发银行 {date} 分时数据，共 {len(df)} 条:")
print(df.to_string(index=False))
examples/04_minute/minute_time_data.py
+14
View File
@@ -0,0 +1,14 @@
"""演示：获取今日分时数据（240 条）。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
bars = c.get_minute_time_data(Market.SH, "600000")
df = pd.DataFrame([{
"序号": i + 1,
"价格": bar.price,
"成交量": bar.vol,
} for i, bar in enumerate(bars)])
print(f"浦发银行今日分时，共 {len(df)} 条:")
print(df.to_string(index=False))
examples/05_transaction/history_transaction.py
+16
View File
@@ -0,0 +1,16 @@
"""演示：获取历史逐笔成交数据。date 参数为 YYYYMMDD 格式的整数。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
date = 20250110
records = c.get_history_transaction_data(Market.SH, "600000", date, 0, 20)
df = pd.DataFrame([{
"时间": f"{r.hour:02d}:{r.minute:02d}",
"成交价": r.price,
"成交量": r.vol,
"方向": "买" if r.buyorsell == 0 else "卖",
} for r in records])
print(f"浦发银行 {date} 最近 {len(df)} 笔成交:")
print(df.to_string(index=False))
examples/05_transaction/transaction_data.py
+15
View File
@@ -0,0 +1,15 @@
"""演示：获取当日逐笔成交数据。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
records = c.get_transaction_data(Market.SH, "600000", 0, 20)
df = pd.DataFrame([{
"时间": f"{r.hour:02d}:{r.minute:02d}",
"成交价": r.price,
"成交量": r.vol,
"方向": "买" if r.buyorsell == 0 else "卖",
} for r in records])
print(f"浦发银行最近 {len(df)} 笔成交:")
print(df.to_string(index=False))
examples/06_finance/company_info.py
+78
View File
@@ -0,0 +1,78 @@
"""演示：获取公司信息目录与各个分类的详细内容。"""
import pandas as pd
from xmtdx import TdxClient, Market
CODE = "600519"
NAME = "贵州茅台"
MARKET = Market.SH
# 要展示的分类，按需注释/取消注释
SHOW_CATEGORIES = [
"最新提示",
"公司概况",
"财务分析",
"股本结构",
"股东研究",
"机构持股",
"分红融资",
"高管治理",
"资金动向",
"资本运作",
"热点题材",
"公司公告",
"公司报道",
"经营分析",
"行业分析",
"研报评级",
]
def show_categories(categories):
"""显示公司信息目录。"""
df = pd.DataFrame([{
"目录名": cat.name,
"文件名": cat.filename,
"起始偏移": cat.start,
"内容长度": cat.length,
} for cat in categories])
print(f"{NAME} 公司信息目录:")
print(df.to_string(index=False))
def show_category_content(client, categories, category_name, max_chars=500):
"""获取并展示指定分类的内容。"""
cat = next((c for c in categories if c.name == category_name), None)
if not cat:
print(f" 未找到分类: {category_name}")
return
content = client.get_company_info_content(
MARKET, CODE, cat.filename, cat.start, cat.length
)
text = content.strip()
if len(text) > max_chars:
text = text[:max_chars] + f"\n... (共 {len(content.strip())} 字，仅显示前 {max_chars} 字)"
print(f"\n{'='*60}")
print(f"【{cat.name}】 (共 {cat.length} 字节)")
print(f"{'='*60}")
print(text)
def show_all_categories(client, categories):
"""依次展示所有 SHOW_CATEGORIES 中列出的分类内容。"""
for name in SHOW_CATEGORIES:
show_category_content(client, categories, name)
with TdxClient.from_best_host() as c:
categories = c.get_company_info_category(MARKET, CODE)
# 1. 显示目录
show_categories(categories)
# 2. 显示所有分类内容（每个分类默认只显示前500字）
show_all_categories(c, categories)
# 3. 也可以单独获取某个分类的完整内容，例如：
# show_category_content(c, categories, "公司概况", max_chars=99999)
examples/06_finance/finance_info.py
+21
View File
@@ -0,0 +1,21 @@
"""演示：获取最新财务数据。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
info = c.get_finance_info(Market.SH, "600519")
df = pd.DataFrame([
{"项目": "总股本(万股)", "数值": info.zong_guben},
{"项目": "流通股本(万股)", "数值": info.liutong_guben},
{"项目": "每股净资产", "数值": info.meigujing_zichan},
{"项目": "净利润(元)", "数值": info.jing_lirun},
{"项目": "主营收入(元)", "数值": info.zhuying_shouru},
{"项目": "主营利润(元)", "数值": info.zhuying_lirun},
{"项目": "净资产(元)", "数值": info.jing_zichan},
{"项目": "总资产(元)", "数值": info.zong_zichan},
{"项目": "股东人数", "数值": info.gudong_renshu},
{"项目": "上市日期", "数值": info.ipo_date},
])
print("贵州茅台 最新财务数据:")
print(df.to_string(index=False, formatters={"数值": lambda x: f"{x:,.0f}"}))
examples/06_finance/price_limits.py
+23
View File
@@ -0,0 +1,23 @@
"""演示：计算个股涨跌停价格。"""
import pandas as pd
from xmtdx import TdxClient, Market
CODE = "600519"
NAME = "贵州茅台"
with TdxClient.from_best_host() as c:
quotes = c.get_security_quotes([(Market.SH, CODE)])
if quotes:
q = quotes[0]
limit_up, limit_down = c.get_price_limits(
Market.SH, CODE, NAME, q.pre_close
)
df = pd.DataFrame([{
"代码": CODE,
"名称": NAME,
"昨收": q.pre_close,
"涨停价": limit_up,
"跌停价": limit_down,
}])
print(df.to_string(index=False))
examples/06_finance/xdxr_info.py
+17
View File
@@ -0,0 +1,17 @@
"""演示：获取除权除息历史记录。"""
import pandas as pd
from xmtdx import TdxClient, Market, XDXR_CATEGORY_NAMES
with TdxClient.from_best_host() as c:
records = c.get_xdxr_info(Market.SH, "600519")
df = pd.DataFrame([{
"日期": f"{r.year}-{r.month:02d}-{r.day:02d}",
"类型": XDXR_CATEGORY_NAMES.get(r.category, f"未知({r.category})"),
"每股分红(元)": r.fenhong,
"送转股比例": r.songzhuangu,
"配股价": r.peigujia,
"配股比例": r.peigu,
} for r in records])
print(f"贵州茅台 除权除息记录，共 {len(df)} 条:")
print(df.tail(10).to_string(index=False))
examples/07_block/block_info.py
+21
View File
@@ -0,0 +1,21 @@
"""演示：获取板块信息（行业、概念、风格）。
常用板块文件:
'block_zs.dat' - 行业/指数板块
'block_gn.dat' - 概念板块
'block_fg.dat' - 风格板块
"""
import pandas as pd
from xmtdx import TdxClient
with TdxClient.from_best_host() as c:
blocks = c.get_block_info("block_gn.dat")
df = pd.DataFrame([{
"板块名称": b.name,
"分类": b.category,
"成分股数": b.count,
"代码(前5)": ", ".join(b.codes[:5]),
} for b in blocks])
print(f"概念板块，共 {len(df)} 个:")
print(df.head(20).to_string(index=False))
examples/08_fund_flow/fund_flow.py
+19
View File
@@ -0,0 +1,19 @@
"""演示：获取个股当日资金流向（基于 L1 逐笔数据统计）。
资金分为四级: 超大(>100万)、大(20-100万)、中(4-20万)、小(<4万)。
"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
flow = c.get_fund_flow(Market.SH, "600519")
df = pd.DataFrame([
{"级别": "超大单", "流入(亿)": flow.super_in / 1e8, "流出(亿)": flow.super_out / 1e8},
{"级别": "大单", "流入(亿)": flow.large_in / 1e8, "流出(亿)": flow.large_out / 1e8},
{"级别": "中单", "流入(亿)": flow.medium_in / 1e8, "流出(亿)": flow.medium_out / 1e8},
{"级别": "小单", "流入(亿)": flow.small_in / 1e8, "流出(亿)": flow.small_out / 1e8},
])
df["净流入(亿)"] = df["流入(亿)"] - df["流出(亿)"]
print("贵州茅台 当日资金流向:")
print(df.to_string(index=False))
examples/08_fund_flow/history_fund_flow.py
+15
View File
@@ -0,0 +1,15 @@
"""演示：获取个股历史日线资金流向序列。"""
import pandas as pd
from xmtdx import TdxClient, Market
with TdxClient.from_best_host() as c:
flows = c.get_history_fund_flow(Market.SH, "600519", 0, 10)
df = pd.DataFrame([{
"日期": f"{f.year}-{f.month:02d}-{f.day:02d}",
"超大单净流入(亿)": (f.super_in - f.super_out) / 1e8,
"大单净流入(亿)": (f.large_in - f.large_out) / 1e8,
"主力净流入(亿)": f.main_net_inflow / 1e8,
} for f in flows])
print(f"贵州茅台 历史资金流向，共 {len(df)} 天:")
print(df.to_string(index=False))
pyproject.toml
+1 -1
View File
@@ -8,7 +8,7 @@ version = "0.1.1"
description = "通达信 TCP 协议 A 股行情数据客户端"
readme = "README.md"
requires-python = ">=3.10"
dependencies = []
dependencies = ["tzdata>=2024.1"]
[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-cov", "mypy>=1.9", "ruff>=0.4"]
scripts/ruff_hook.py
+43
View File
@@ -0,0 +1,43 @@
"""PostToolUse hook: 对 Edit/Write 修改的 .py 文件自动运行 ruff check + format。
stdin 接收 JSON: {"tool_name": "Edit"|"Write", "tool_input": {"file_path": "..."}}
"""
import json
import subprocess
import sys
def main():
try:
data = json.load(sys.stdin)
except (json.JSONDecodeError, EOFError):
return
file_path = data.get("tool_input", {}).get("file_path", "")
if not file_path.endswith(".py"):
return
# ruff check --fix（自动修复 lint 问题）
r = subprocess.run(
["ruff", "check", "--fix", file_path],
capture_output=True,
text=True,
timeout=15,
)
if r.returncode != 0 and r.stdout.strip():
print(f"[ruff check] {file_path}:\n{r.stdout.strip()}")
# ruff format
r = subprocess.run(
["ruff", "format", file_path],
capture_output=True,
text=True,
timeout=15,
)
if r.returncode != 0 and r.stdout.strip():
print(f"[ruff format] {file_path}:\n{r.stdout.strip()}")
if __name__ == "__main__":
main()
src/xmtdx/__init__.py
+6
View File
@@ -35,6 +35,8 @@ from .models import (
TransactionRecord,
XdxrRecord,
)
from .ex.client import AsyncExTdxClient, ExTdxClient
from .ex.models import KNOWN_EX_HOSTS
from .transport.sync import KNOWN_HOSTS, ping_all
__all__ = [
@@ -59,6 +61,10 @@ __all__ = [
"TdxConnectionError",
"TdxDecodeError",
"TdxCommandError",
# 扩展行情
"ExTdxClient",
"AsyncExTdxClient",
"KNOWN_EX_HOSTS",
# 工具
"ping_all",
"KNOWN_HOSTS",
src/xmtdx/client.py
+143 -41
View File
@@ -1,7 +1,11 @@
"""高层行情 API：TdxClient（同步）和 AsyncTdxClient（asyncio）。"""
import json
import logging
import asyncio
from dataclasses import asdict
from datetime import datetime
from pathlib import Path
from collections.abc import Awaitable, Callable
from types import TracebackType
from typing import TypeVar
@@ -124,6 +128,43 @@ def _historical_fund_flow_from_records(
# 同步客户端
# ============================================================
_CACHE_DIR = Path.home() / ".xmtdx" / "cache"
_CACHE_MAX_AGE = 86400 # 1 天
def _serialize_stocks(stocks: list[SecurityInfo]) -> list[dict]:
return [{k: v for k, v in asdict(s).items() if k != "_raw"} for s in stocks]
def _deserialize_stocks(data: list[dict]) -> list[SecurityInfo]:
return [SecurityInfo(**{**d, "market": Market(d["market"])}) for d in data]
def _load_cache() -> list[SecurityInfo] | None:
path = _CACHE_DIR / "security_list_all.json"
if not path.exists():
return None
try:
raw = json.loads(path.read_text("utf-8"))
updated = datetime.fromisoformat(raw["updated"])
if (datetime.now() - updated).total_seconds() > _CACHE_MAX_AGE:
return None
return _deserialize_stocks(raw["data"])
except Exception:
return None
def _save_cache(stocks: list[SecurityInfo]) -> None:
_CACHE_DIR.mkdir(parents=True, exist_ok=True)
data = {
"updated": datetime.now().isoformat(),
"count": len(stocks),
"data": _serialize_stocks(stocks),
}
(_CACHE_DIR / "security_list_all.json").write_text(
json.dumps(data, ensure_ascii=False), "utf-8"
)
class TdxClient:
"""同步通达信行情客户端，支持 IP 优选与断线自动重连。
@@ -233,46 +274,68 @@ class TdxClient:
"""获取证券列表（每页约1000条，按 start 分页）。"""
return self._execute(GetSecurityListCmd(market, start))
def get_security_list_all(self) -> list[SecurityInfo]:
def get_security_list_all(self, pages: int | str = "all") -> list[SecurityInfo]:
"""获取沪深 A 股完整证券列表，并自动挂载行业信息。
Args:
pages: 拉取页数。每个市场每页 1000 条。
"all" 拉取全部（默认，结果会缓存到本地文件）。
整数 N 表示每个市场只拉前 N 页，不缓存。
注意：
`Market.BJ` 的证券列表请求长期存在服务器超时问题，当前版本暂不纳入此方法。
若需 BJ 名单，应改由 `base_info.zip` 等文件离线解析获得。
"""
# 1. 尝试获取行业配置
industry_map = {}
log = logging.getLogger(__name__)
if pages == "all":
cached = _load_cache()
if cached is not None:
log.info("从缓存加载沪深 A 股列表，共 %d 只", len(cached))
return cached
# 计算每个市场的最大起始偏移
def _max_start(count: int) -> int:
if pages == "all":
return count
return min(count, int(pages) * 1000)
# 尝试获取行业配置
industry_map: dict[str, tuple[str, str]] = {}
try:
cfg_data = self.get_report_file("tdxhy.cfg")
if cfg_data:
industry_map = parse_tdxhy_cfg(cfg_data)
log.info("行业配置已加载，共 %d 条映射", len(industry_map))
except Exception:
pass
log.warning("无法获取 tdxhy.cfg，行业字段将为空")
all_stocks: list[SecurityInfo] = []
# 注意：Market.BJ 证券列表请求常年超时，短期降级为仅 SH/SZ；
# BJ 列表需解析 base_info.zip 获得（待实现）。
for market in [Market.SH, Market.SZ]:
count = self.get_security_count(market)
for start in range(0, count, 1000):
limit = _max_start(count)
total_pages = (limit + 999) // 1000
for page_idx, start in enumerate(range(0, limit, 1000)):
try:
stocks = self.get_security_list(market, start)
except Exception:
log.warning("%s 第 %d/%d 页获取失败，跳过", market.name, page_idx + 1, total_pages)
continue
log.info("%s 第 %d/%d 页: %d 条", market.name, page_idx + 1, total_pages, len(stocks))
for s in stocks:
# 精确 A 股过滤规则
is_a_share = False
if market == Market.SH:
# 沪市 A 股：60xxxx, 68xxxx
if s.code.startswith(("60", "68")):
is_a_share = True
elif market == Market.SZ:
# 深市 A 股：00xxxx, 30xxxx
if s.code.startswith(("00", "30")):
is_a_share = True
is_a_share = (
(market == Market.SH and s.code.startswith(("60", "68")))
or (market == Market.SZ and s.code.startswith(("00", "30")))
)
if is_a_share:
# 挂载行业信息
if s.code in industry_map:
s.industry_tdx, s.industry_sw = industry_map[s.code]
all_stocks.append(s)
log.info("沪深 A 股总数: %d", len(all_stocks))
if pages == "all":
_save_cache(all_stocks)
return all_stocks
def get_security_quotes(
@@ -440,15 +503,20 @@ class TdxClient:
`suspended_count` 是 `total - up - down - neutral` 的残差估算值，
用于保证计数守恒，不应视为协议已明确验证的停牌字段。
"""
# 通达信中 880005 是全市场行情统计代码
quotes = self.get_security_quotes([(Market.SH, "880005")])
# 通达信中 880005 是全市场行情统计，880001 是总市值指数，880006 是涨跌停统计
quotes = self.get_security_quotes([
(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006"),
])
if not quotes:
raise RuntimeError("无法获取市场统计数据")
q = quotes[0]
up = int(q.price)
down = int(q.pre_close)
down = int(q.open)
neutral = int(q.low)
total = int(q.high)
market_cap = quotes[1].price * 1e10 if len(quotes) > 1 else 0.0
limit_down = int(quotes[2].open) if len(quotes) > 2 else 0
limit_up = int(quotes[2].price) if len(quotes) > 2 else 0
return MarketStat(
up_count=up,
down_count=down,
@@ -457,6 +525,9 @@ class TdxClient:
total_count=total,
total_amount=q.amount,
total_volume=q.vol,
total_market_cap=market_cap,
limit_up_count=limit_up,
limit_down_count=limit_down,
)
def _collect_transaction_records(
@@ -668,41 +739,64 @@ class AsyncTdxClient:
async def get_security_list(self, market: Market, start: int) -> list[SecurityInfo]:
return await self._execute(GetSecurityListCmd(market, start))
async def get_security_list_all(self) -> list[SecurityInfo]:
async def get_security_list_all(self, pages: int | str = "all") -> list[SecurityInfo]:
"""获取沪深 A 股完整证券列表，并自动挂载行业信息。
Args:
pages: 拉取页数。每个市场每页 1000 条。
"all" 拉取全部（默认，结果会缓存到本地文件）。
整数 N 表示每个市场只拉前 N 页，不缓存。
注意：
`Market.BJ` 的证券列表请求长期存在服务器超时问题，当前版本暂不纳入此方法。
若需 BJ 名单，应改由 `base_info.zip` 等文件离线解析获得。
"""
industry_map = {}
log = logging.getLogger(__name__)
if pages == "all":
cached = _load_cache()
if cached is not None:
log.info("从缓存加载沪深 A 股列表，共 %d 只", len(cached))
return cached
def _max_start(count: int) -> int:
if pages == "all":
return count
return min(count, int(pages) * 1000)
industry_map: dict[str, tuple[str, str]] = {}
try:
cfg_data = await self.get_report_file("tdxhy.cfg")
if cfg_data:
industry_map = parse_tdxhy_cfg(cfg_data)
log.info("行业配置已加载，共 %d 条映射", len(industry_map))
except Exception:
pass
log.warning("无法获取 tdxhy.cfg，行业字段将为空")
all_stocks: list[SecurityInfo] = []
# 注意：Market.BJ 证券列表请求常年超时，短期降级为仅 SH/SZ；
# BJ 列表需解析 base_info.zip 获得（待实现）。
for market in [Market.SH, Market.SZ]:
count = await self.get_security_count(market)
for start in range(0, count, 1000):
limit = _max_start(count)
total_pages = (limit + 999) // 1000
for page_idx, start in enumerate(range(0, limit, 1000)):
try:
stocks = await self.get_security_list(market, start)
except Exception:
log.warning("%s 第 %d/%d 页获取失败，跳过", market.name, page_idx + 1, total_pages)
continue
log.info("%s 第 %d/%d 页: %d 条", market.name, page_idx + 1, total_pages, len(stocks))
for s in stocks:
is_a_share = False
if market == Market.SH:
if s.code.startswith(("60", "68")):
is_a_share = True
elif market == Market.SZ:
if s.code.startswith(("00", "30")):
is_a_share = True
is_a_share = (
(market == Market.SH and s.code.startswith(("60", "68")))
or (market == Market.SZ and s.code.startswith(("00", "30")))
)
if is_a_share:
if s.code in industry_map:
s.industry_tdx, s.industry_sw = industry_map[s.code]
all_stocks.append(s)
log.info("沪深 A 股总数: %d", len(all_stocks))
if pages == "all":
_save_cache(all_stocks)
return all_stocks
async def get_security_quotes(
@@ -836,15 +930,20 @@ class AsyncTdxClient:
`suspended_count` 是 `total - up - down - neutral` 的残差估算值，
用于保证计数守恒，不应视为协议已明确验证的停牌字段。
"""
# 通达信中 880005 是全市场行情统计代码
quotes = await self.get_security_quotes([(Market.SH, "880005")])
# 通达信中 880005 是全市场行情统计，880001 是总市值指数，880006 是涨跌停统计
quotes = await self.get_security_quotes([
(Market.SH, "880005"), (Market.SH, "880001"), (Market.SH, "880006"),
])
if not quotes:
raise RuntimeError("无法获取市场统计数据")
q = quotes[0]
up = int(q.price)
down = int(q.pre_close)
down = int(q.open)
neutral = int(q.low)
total = int(q.high)
market_cap = quotes[1].price * 1e10 if len(quotes) > 1 else 0.0
limit_down = int(quotes[2].open) if len(quotes) > 2 else 0
limit_up = int(quotes[2].price) if len(quotes) > 2 else 0
return MarketStat(
up_count=up,
down_count=down,
@@ -853,6 +952,9 @@ class AsyncTdxClient:
total_count=total,
total_amount=q.amount,
total_volume=q.vol,
total_market_cap=market_cap,
limit_up_count=limit_up,
limit_down_count=limit_down,
)
async def _collect_transaction_records(
src/xmtdx/commands/security_bars.py
+50 -3
View File
@@ -103,8 +103,55 @@ class GetSecurityBarsCmd(BaseCommand[list[SecurityBar]]):
class GetIndexBarsCmd(GetSecurityBarsCmd):
"""获取指数 K 线（请求格式与股票 K 线相同，服务器端按指数逻辑处理）。
"""获取指数 K 线。
实际上通达信服务器对股票代码前缀自动判断指数/股票，
此子类仅作语义区分，无额外逻辑。
请求格式与股票 K 线相同，但响应每条记录在 vol+amt 后多 4 字节
（上涨家数 uint16 + 下跌家数 uint16），必须跳过否则后续记录错位。
"""
def parse_response(self, body: bytes) -> list[SecurityBar]:
(ret_count,) = unpack_from("<H", body, 0, "security_bars header")
pos = 2
bars: list[SecurityBar] = []
pre_diff_base = 0
cat = int(self.category)
for _ in range(ret_count):
record_start = pos
year, month, day, hour, minute, pos = get_datetime(cat, body, pos)
open_diff, pos = get_price(body, pos)
close_diff, pos = get_price(body, pos)
high_diff, pos = get_price(body, pos)
low_diff, pos = get_price(body, pos)
vol, pos = get_volume(body, pos)
amount, pos = get_volume(body, pos)
# 指数记录额外 4 字节：上涨家数 + 下跌家数（各 uint16 LE）
pos += 4
open_abs = open_diff + pre_diff_base
close_abs = open_abs + close_diff
high_abs = open_abs + high_diff
low_abs = open_abs + low_diff
pre_diff_base = open_abs + close_diff
bars.append(
SecurityBar(
open=open_abs / 1000.0,
close=close_abs / 1000.0,
high=high_abs / 1000.0,
low=low_abs / 1000.0,
vol=vol,
amount=amount,
year=year,
month=month,
day=day,
hour=hour,
minute=minute,
_raw=body[record_start:pos],
)
)
return bars
src/xmtdx/models/stats.py
+3
View File
@@ -12,6 +12,9 @@ class MarketStat:
total_count: int # 总计（包含停牌）
total_amount: float # 总成交额
total_volume: float # 总成交量
total_market_cap: float # 总市值（亿元），来自 880001 收盘价，÷100 得万亿
limit_up_count: int # 涨停家数，来自 880006 close
limit_down_count: int # 跌停家数，来自 880006 open
@dataclass