这段时间把通达信的数据接口在 Python 里的用法从头到尾过了一遍趁着细节还没忘赶紧整理成一篇笔记。如果你跟我一样没有 wind 账号又不想为数据终端花大几千还想靠 Python 拿 A 股行情数据做研究那 tdxpy 这类数据接口 API 应该是目前性价比最高的方案之一。这篇笔记会从选型逻辑、环境准备、核心 API 拆解、真实踩坑到工程化落地完整记录我自己的实操过程希望能帮你少走一段冤枉路。1. 为什么要折腾 tdxpy免费行情接口的选型逻辑1.1 商业数据源的价格门槛做量化研究或者个人投研数据是第一关。市面上的商业数据源wind 一年几万起步choice、iFinD 虽然稍微便宜点但对一个还没产生稳定策略的个人研究者来说这笔开销怎么看都不划算。更尴尬的是很多开源方案虽然免费但要么数据范围受限要么更新时效满足不了日内行情研究。我自己的需求很明确日频和分钟频的 K 线、实时五档盘口、全市场股票列表最好还能拿到板块成分。这几点如果全走商业数据源月费轻松上千如果走免费接口最关键的是要搞清楚哪一类接口能满足其中大部分需求。1.2 tdxpy 这类接口到底属于什么定位tdxpy 简单理解就是通达信数据接口在 Python 里的封装。通达信这套行情体系在券商行情里覆盖得非常广很多券商的行情软件底层都在用。tdxpy 做的事情本质上是把行情服务器返回的数据转成 Python 对象让你能像操作普通 list/dict 一样处理行情数据。这里要说明一下tdxpy 在社区里不同的分支、不同的作者手里API 命名和一些调用细节会略有差异有的版本需要本机装了通达信客户端才能跑有的版本是走协议直连。但核心思路是一样的——连上行情服务器发请求收数据解析。所以与其纠结某一个具体函数名不如先把这套接口的共性逻辑吃透。相比之下akshare 是聚合型数据源爬的是公开网页tushare 靠积分制数据质量也不错baostock 免费但历史数据偏日频。tdxpy 这类接口最大的优势是原始行情、实时性好、不做过多加工拿到手的数据干净适合自己掌控股利计算和复权逻辑。1.3 选型的时候我主要看这几点我的选型标准其实就四条你可以直接套用数据时效能不能拿到当天甚至当下的行情而不是延迟一天。数据粒度有没有分钟线、五档盘口还是只有日线。成本与门槛是否免费需不需要本机装额外软件安装配置复杂度如何。扩展性能不能把数据存到本地方便回测和研究反复用。按这个标准筛下来tdxpy 这类直连通达信行情服务器的接口在实时性和原始性上是明显占优的。这也是我最终决定花时间折腾它而不是无脑用现成聚合库的原因。2. 环境准备与第一次连通先把通道跑通2.1 准备一个干净的 Python 环境很多人在装这类接口时遇到问题先别急着怀疑库本身很大概率是 Python 环境版本太新或者本机装了好几个 Python 导致依赖装错地方。我建议你用 Python 3.9 到 3.11 之间的版本配合 venv 或 conda 单独建一个环境。太新的 Python 版本部分涉及 c 扩展或者二进制依赖的库可能还没有对应轮子安装时会当场编译失败太老的版本又容易和 pandas、numpy 的新版本冲突。conda create -n tdx python3.10 -y conda activate tdx pip install pandas numpy先把 pandas 和 numpy 装好后面不管是数据解析还是落盘保存都用得上。2.2 安装与最小连接验证装好基础依赖后再装数据接口库本身。因为 tdxpy 的具体封装在不同项目里可能有差异我在这里不写死某个安装命令建议你直接到项目的 README 或者 PyPI 页面确认官方推荐的安装方式。装完之后第一步不是写复杂逻辑而是跑一个最简连接脚本确认网络层和协议层是通的。# 以常见的协议直连封装为例具体类名/方法名以你实际安装的版本为准 from tdxpy import TdxClient client TdxClient() # 服务器地址可以从社区公开的行情服务器列表中找延迟最低的 connected client.connect(行情服务器地址, 7709) if connected: print(连接成功) print(client.get_server_time()) # 先拿服务器时间确认链路没问题 else: print(连接失败检查网络和服务器地址)这里我特别提醒两点。第一行情服务器地址的变化非常频繁网上能找到很多公开的列表建议自己写个小脚本批量测延迟选延迟最低的。不要长期硬编码一个地址失效是常态。第二有的封装版本连接成功不代表能拿到数据还得验证一下服务器时间或者指数行情确认返回结果不是空值。这一步忘掉的话后面写再多代码都是白搭。2.3 为什么第一次连通如此重要我最初犯过的错误是跳过自检直接写全量拉数据脚本结果跑了几分钟才发现其中一个服务器地址早就失效了报错信息又很隐晦排查了半天。其实问题不在代码逻辑而在最底层网络层根本没通。所以我的经验是先花十分钟把最小链路跑通再往上层加逻辑。连接成功、能拿到服务器时间、能取到任意一只股票的快照这三步确认完后面再写批量脚本腰杆都硬了。3. 核心 API 拆解行情快照、K 线与板块列表3.1 连接管理与会话机制tdxpy 这类接口的使用方式一句话总结就是一次连接、多次请求。连接一次之后可以反复发查询请求用完再断开千万不要每条数据都重新连一次。这就像打电话不是发短信——拨号有成本保持通话状态才能高效沟通。工程上的做法一般是在脚本开始时建立连接然后将连接对象作为公共组件传入需要取数的函数里脚本结束统一关闭。如果担心长时间挂着连接被服务端断开可以隔一段时间发一个轻量查询保持活跃。# 示意连接生命周期管理 with client as conn: data1 conn.get_stock_quotes(000001) data2 conn.get_kline(000001, perioddaily, count100) # 退出 with 块时自动断开3.2 行情快照接口行情快照是最常用的接口返回某只股票当前的实时状态包括最新价、涨跌幅、成交量、成交额、五档盘口等。它的特点是一次请求多只股票你可以把几十只股票代码放在一个列表里传给接口而不是一只一只地查这样能显著减少请求次数。关于返回值我提醒大家不要想当然。不同封装对字段的单位处理不一样有的价格按元直接返回有的需要你手动除以 100 或 1000成交量有的按股、有的按手成交额有的单位是元有的单位是万元。拿到数据后的第一件事是找一只已知行情的股票做校验把字段单位全部确认清楚不然后面的计算全错。# 示意批量获取快照 codes [000001, 600519, 300750] quotes client.get_stock_quotes(codes) for item in quotes: # 先打印原始字段人工确认单位再写后续逻辑 print(item)3.3 K 线历史数据接口K 线是回测研究最核心的数据。这个接口的参数一般包括市场代码、证券代码、K 线周期、起始位置和数量。这里有个非常关键的细节市场代码和证券代码是分开传的。A 股里上海市场和深圳市场的代码规则不同接口通常单列一个 market 参数0 代表深圳1 代表上海写代码的时候要注意别把两个市场搞混。如果你直接传完整的 6 位代码而不指定市场部分接口会返回空数据。另一个细节是单次请求的数量上限。不同周期上限不同日线单次能取的数量通常比分钟线多但都不可能一次取全量历史。所以正确的取数姿势是先确认上限再分段拉取最后拼接。# 示意分段拉取日K def fetch_kline(client, code, market, period, total): bars [] batch_size 800 # 视具体接口上限调整 remaining total while remaining 0: count min(batch_size, remaining) part client.get_kline( code, marketmarket, periodperiod, startremaining - count, # 从最早处往前补 countcount, ) if not part: break bars part bars remaining - count return bars3.4 列表与板块数据除了单只股票的行情做策略研究通常还需要全市场股票列表和板块成分。这类接口的返回值往往是一个大的列表每条包含代码、名称、所属市场等信息。我的习惯是全市场列表不频繁拉取每天收盘后拉一次存下来就够了。板块数据同理板块分类标准各家人为因素很多如果你要做行业轮动建议以自己长期跟踪的分类为准而不是频繁切换数据源导致分类口径不一致。4. 免费接口的坑这些细节文档里不会写4.1 复权数据的处理复权是免费接口最大的坑没有之一。很多接口返回的 K 线是不复权的原始价遇到除权除息的日子K 线图上会出现明显的价格断层直接拿来算收益率或者画均线结果会非常离谱。我的处理思路是本地存原始价同时把每次除权除息的事件记录下来自己计算复权因子。回测时如果策略是短周期我一般用后复权保证历史价格序列可比如果只看当前形态再用前复权。这样做的问题是代码量会多不少但好处是数据完全可控不会被接口的默认策略绑架。如果实在不想自己算复权因子至少也要确认你用的接口有没有提供指定复权类型的参数并在拉数时显式传入而不是依赖默认值。好多坑都是我没指定它默认不复权造成的。4.2 停牌股的缺失值处理停牌股会让数据出现缺失 K 线。问题在于如果你按股票代码去拼 DataFrame缺的 K 线会导致索引错位一次错位后面全乱。正确做法是以交易日历为基准把所有股票的数据做 outer join缺失日期填充 NaN 或前值。这样虽然多占点存储但后续做截面分析时的数据结构是统一的比事后补错位容易得多。# 示意以交易日历为基准对齐 df df.set_index(date).reindex(trade_calendar) df[close] df[close].fillna(methodffill) # 停牌期间沿用前收盘价这种处理方式在回测时还有一个好处遇到停牌股策略会自动跳过不可交易的日子不会因为假数据而误入场。4.3 请求频率与连接稳定性免费接口没有 SLA请求太猛会被服务端限流甚至断连。我自己的实测经验是普通行情服务器每秒请求控制在 2 到 4 次比较稳批量拉 K 线时要主动加 sleep 控制节奏。import time # 示意批量拉数时控制节奏 for i, code in enumerate(all_codes): bars client.get_kline(code, perioddaily, count500) store(code, bars) time.sleep(0.3) # 平滑请求速率另外长连接偶尔会被服务端断开这是正常现象。工程上必须做断线重连机制——捕获连接异常重新 connect 后继续跑而不是整个脚本崩溃退出。我在第 5 节会讲一个完整的重试思路。4.4 数据精度与字段单位这一条看似基础实际上最容易埋雷。有些接口用 int 返回价格数值上等于真实价格乘以 100 或 1000你如果直接把 int 当价格用算出来的收益率差出好几倍。成交量同理有的按手、有的按股。我的建议是写一个统一的数据清洗层把接口返回的原始数据全部换算成标准单位再进入下一环节。这样数据清洗只在这一层发生后面回测或计算用的数据是一致的不会这个模块按手、那个模块按股地互相打架。4.5 时区与时间戳接口返回的时间大多数是北京时间但不少字段是字符串或者无时区的时间对象直接和本地 datetime 比较容易出问题。我的做法是入库时统一转成无时区的交易时间或者统一存成固定时区的字符串。这里的关键不是绝对标准而是全项目统一标准别在一个库里混着不同时区语义的时间否则做时间切片的时候非常痛苦。5. 从笔记到工具把采集脚本做成可复用模块5.1 三层模块结构能跑和能稳定跑是两码事。我从笔记阶段过渡到工具阶段时把采集脚本重构成了三层结构连接层、请求层、存储层。连接层专门负责建立连接、心跳探测、断线重连。请求层把各种 API 调用封装成语义化函数比如 get_daily_bars(code)、get_realtime_quote(code)上层代码不用关心数据是从哪个服务器来的。存储层负责读写本地数据暴露 save_bars、load_bars 这样的接口。这样设计的好处是哪天想换数据源只需要改连接层和请求层的实现策略代码完全不用动反过来想从本地存储换成数据库也只动存储层。5.2 增量更新策略拉数据的正确姿势不是全量重拉而是首次全量建底稿之后增量更新。全量拉一次全市场日线大概要跑很久但增量更新就快得多。每天收盘后只需要拉最近几天留足缓冲的 K 线跟本地已有的数据做去重拼接。这样每天运行时间从小时级降到分钟级。# 示意增量更新逻辑 last_date get_local_last_date(code) # 本地已有的最后日期 latest client.get_kline(code, perioddaily, count10) new_bars [b for b in latest if b.date last_date] if new_bars: append_to_local(code, new_bars)这里要注意增量拉的时候多拉几天做缓冲是为了防止除权除息或数据修正导致最后几根 K 线和前一天的存储不一致。宁可多做一次覆盖更新也别留缺口。5.3 本地存储选型数据量不大SQLite 加 CSV 就够用如果做全市场多年分钟线研究建议直接上 Parquet。我用的是按股票代码分文件的目录结构每天数据存一个 Parquet 文件。原因是分钟线数据量很大按代码分文件能避免单个文件过大跨天读取时再用多线程批量加载性能完全够用。SQLite 适合按日期范围查询的场景比如2024 年 1 月到 3 月所有股票的日线数据。如果你主要做截面分析优先考虑它。5.4 容错与重试机制免费接口的运行环境里网络抖动和数据为空都是常态。我的做法是给请求层加一个带重试的装饰器遇到连接异常、超时、空数据先等一秒再重试最多重试三次还失败就把代码写进失败队列下一轮启动时补拉。import functools import time def retry(max_retries3, delay1): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): for i in range(max_retries): try: result func(*args, **kwargs) if result: return result except Exception as e: print(frequest error: {e}, retry {i 1}) time.sleep(delay) return None return wrapper return decorator retry(max_retries3, delay1) def safe_get_kline(client, code): return client.get_kline(code, perioddaily, count500)有了这层保障一个采集任务挂机跑几个小时中途偶尔报几个错脚本也能自己恢复而不是直接死掉。6. 和其它 Python 行情接口的横向对比与选型建议6.1 主流方案的差异对比用 tdxpy 之前我也把市面上常见的 Python 行情接口过了一遍这里给出我自己的横向对比方便你做选型参考。方案数据时效数据范围成本上手难度适合场景tdxpy/协议直连类实时行情盘中可取行情、板块为主财务少免费中本地量化研究需原始行情akshare准实时/日频范围极广含宏观、财报免费低数据探索飞快拿数据tushare日频/分钟财务数据质量好免费额度有限低需要财报行情结合的轻量项目baostock有延迟日线为主免费低长周期回测数据稳定6.2 为什么我最终选择混用而非单一方案把表拉出来后你会发现没有一个方案是完美的。tdxpy 类接口强在实时行情和原始数据弱在财务数据几乎为零akshare 数据范围广但毕竟是聚合爬取稳定性和数据规范性差一些tushare 数据质量不错但免费积分有额度限制baostock 稳定但分钟线基本不用想。我现在的实践是混用用 tdxpy 类接口拉日线和分钟线底稿用 akshare 补财报和板块数据用 baostock 做历史行情数据的交叉校验。三套数据源互相印证关键数据不只看单边很大程度上避免了单一免费源突然挂掉或者数据出错的风险。这种做法虽然前期配置麻烦一点但长期来看当你发现某个策略的收益率异常得可疑时可以快速交叉验证是不是数据源出了问题而不是在策略逻辑里浪费时间排查。行情数据这件事免费不等于廉价花点心思把数据链路做扎实后面做研究的心态会完全不一样。