场景示例¶
每个片段都是完整可跑的。共同前提：
import asyncio
from pathlib import Path
from pytdxdata import TdxData
from pytdxdata.models import Adjust, KlinePeriod
全市场日线批量落库¶
传多标的时，连接池会把不同股票分配到不同服务器——这正是它比逐个 await 快的原因。
async def dump_all_daily():
async with TdxData(cache_dir=Path("./tdx_cache")) as td:
infos = await td.get_universe("sz") # 缓存 1 天
symbols = [i.symbol for i in infos] # ["sz000001", ...]
bars = await td.get_bars(
symbols,
KlinePeriod.DAY,
count=800,
adjust=Adjust.QFQ,
concurrency=12,
)
got = {b.symbol for b in bars}
print(f"{len(got)} / {len(symbols)} 只拿到数据")
# bars 是扁平 list[SecurityBar]（按标的/时间排序），落库 / 写 parquet 都行
if bars:
print(bars[-1].symbol, bars[-1].datetime, bars[-1].close)
分页与并发的取舍
concurrency 不是越大越好。单台服务器有连接上限，concurrency 超过 pool_max × 服务器数
之后只会排队。实践值：8~16。
分钟线增量更新¶
只拉「上次之后」的部分，用 get_bars(start_date=...) 省掉全量重拉。
async def update_minute(symbol: str, last_dt):
async with TdxData(cache_dir=Path("./tdx_cache")) as td:
rows = await td.get_bars(symbol, KlinePeriod.MIN_1, start_date=last_dt)
return [r for r in rows if r.datetime > last_dt]
1 分钟 K 线走 MAC 单连接串行分页
这是刻意设计：并发分页会把不同页分到不同服务器，而各服务器历史深度不一致，
拼接后会出现非确定性数据空洞。所以 MIN_1 强制单连接串行（见 通道与数据源）。
报价轮询¶
async def poll_quotes(symbols, interval=5.0):
async with TdxData() as td:
while True:
quotes = await td.get_quotes(symbols) # >80 只自动切批；缓存 TTL 5s
for q in quotes:
print(q.code, q.price, q.bid[0], q.ask[0])
await asyncio.sleep(interval)
缓存 TTL 与轮询间隔对齐时，重复轮询几乎 0 网络开销。但注意通达信公开源本身是 15 分钟延时数据，别拿它做实时交易决策。
板块轮动 / 领涨股¶
async def sector_rotation():
async with TdxData() as td:
# 近 5 日概念板块涨幅排行
top = await td.get_board_change_ranking(kind="concept", days=5, top_n=10)
for row in top:
print(row)
# 拿第一名板块的成分股
symbol = top[0]["symbol"] if "symbol" in top[0] else "881001"
members = await td.get_board_members(symbol, count=20)
for m in members[:5]:
print(m.code, m.name, m.fields)
# 板块资金汇总
print(await td.get_board_summary(symbol))
个股所属板块（反向查询）¶
async def belong(symbol: str):
async with TdxData() as td:
for b in await td.get_board_of(symbol):
print(b)
逐笔复盘¶
async def review_transactions(symbol: str, ymd: int):
async with TdxData() as td:
rows = await td.get_ticks(symbol, date=ymd, count=2000)
total = sum(r.trade_count or 0 for r in rows)
buy = sum(r.vol for r in rows if r.bs_flag == 0) # 主动买
sell = sum(r.vol for r in rows if r.bs_flag == 1) # 主动卖
print(f"{ymd} 总成交笔数={total} 主动买={buy} 主动卖={sell}")
财务因子快照¶
async def finance_snapshot(symbols):
async with TdxData() as td:
for fin in await td.get_finance(symbols): # 标准通道，最新快照
print(fin.symbol, fin)
历史专业财报（gpcw*，含多报告期）走文件接口。
索引里的文件名与服务器实际不一致
tdxfin/gpcw.txt 索引里写的是 gpcwYYYYMMDD.zip，但服务器实际只提供未压缩的
tdxfin/gpcwYYYYMMDD.dat —— 这些差异已由 get_financial_records_parsed() 内部处理，
直接把索引里的 filename 传进去即可。
另外索引里含有尚未生成的占位条目（filesize 只有百来字节），取它们会得到空列表。
想拿最新一期可用数据，先按体积过滤：
async def history_financials(code: str):
async with TdxData(cache_dir=Path("./tdx_cache")) as td:
infos = await td.get_financial_file_infos() # 索引
records = await td.get_financial_records_parsed(infos[-1].filename)
row = next((r for r in records if r.code == code), None)
if row:
print(row.report_date, row.fields[:8])
行业分类映射¶
async def industry_lookup(codes):
async with TdxData(cache_dir=Path("./tdx_cache")) as td:
hy = await td.get_industry_map() # {6位代码: IndustryInfo}
for c in codes:
info = hy.get(c)
if info:
print(c, info.tdx_industry, info.sw_industry)
板块成分（离线解析）¶
get_block_parsed 拿到的是纯 dataclass，落成本地 JSON 后完全脱网可用——
适合做「板块成分快照」的定时任务。
import json
async def snapshot_blocks():
async with TdxData(cache_dir=Path("./tdx_cache")) as td:
all_blocks = []
for fn in ("block_zs.dat", "block_gn.dat", "block_fg.dat"):
for b in await td.get_block_parsed(fn):
all_blocks.append({
"name": b.name,
"category": b.category,
"codes": b.codes,
})
Path("blocks.json").write_text(
json.dumps(all_blocks, ensure_ascii=False), encoding="utf-8"
)
港股 / 美股 / 期货 / 期权¶
async def ex_markets():
async with TdxData() as td:
hk = await td.get_bars("hk00700", KlinePeriod.DAY, count=700)
us = await td.get_bars("usAAPL", KlinePeriod.DAY, count=700)
fut = await td.get_bars("cffex:IFL0", KlinePeriod.DAY, count=700)
opt = await td.get_bars("sh_stock_option:10010971", KlinePeriod.DAY, count=700)
print(len(hk), len(us), len(fut), len(opt))
# 港股全量逐笔（自动翻页，≤50 页 / 9 万条）
ticks = await td.get_ticks("hk00700", count=None)
print(len(ticks))
转成 pandas¶
包本身不带 pandas 依赖，但转换是一行的事：
import pandas as pd
async def to_dataframe(symbol: str):
async with TdxData() as td:
bars = await td.get_bars(symbol, KlinePeriod.DAY, count=800)
df = pd.DataFrame([{
"dt": b.datetime, "open": b.open, "high": b.high,
"low": b.low, "close": b.close, "vol": b.vol, "amount": b.amount,
} for b in bars])
return df.set_index("dt")
长驻服务（FastAPI）¶
不要每次请求都新建 TdxData
池与缓存都挂在实例上，每请求新建 = 每请求重建连接池 + 缓存全部失效。
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
async with TdxData(cache_dir=Path("./tdx_cache"), pool_min=4, pool_max=16) as td:
app.state.td = td
yield
app = FastAPI(lifespan=lifespan)
@app.get("/kline/{symbol}")
async def kline(symbol: str, request: Request):
td = request.app.state.td
bars = await td.get_bars(symbol, KlinePeriod.DAY, count=30)
return [{"dt": str(b.datetime), "close": b.close} for b in bars]