先说实话做量化研究或者日常行情分析的人最头疼的往往不是策略本身而是“数据从哪里来”。国内行情数据没有对个人开放的官方 API以前常见的路子无非是打开通达信客户端手动导出、写爬虫去抓网页数据、或者买第三方数据服务。直到我在 GitHub 上翻到 tdxpy 这个库发现它能直接用 Python 连接通达信的行情服务器把日K线、分笔成交、财务数据一口气拉下来整个过程干净利落。这篇文章是我踩过不少坑之后整理的 API 使用笔记覆盖基础原理、常用接口参数、批量拉取和问题排查适合想自己搭一套免费行情数据管线的量化研究者、数据分析和 Python 开发同学。1. 先把底摸清tdxpy 到底是什么和 Web API 有什么不一样1.1 通达信协议一个“野生”但长期可用的行情源通达信是国内最普及的行情软件之一很多券商直接拿它做交易终端内核。它背后有一套基于 TCP 的私有行情协议客户端通过它向行情服务器请求股票列表、K线、分笔、财务指标等数据。早年开源社区的人通过抓包、逆向把协议格式逐步摸清楚了于是出现了多个语言的实现。tdxpy 就是这套协议在 Python 生态里一个很顺手的封装。很多人第一次接触“接口”这个词脑子里浮现的是 HTTP 接口发一个 GET 请求服务器回一段 JSON像调 DeepSeek、GPT 的 API 那样。tdxpy 完全不是这个路数它是直接在 Python 进程里与行情服务器建立 TCP 长连接请求数据的时候发送的是二进制报文返回的也是二进制流然后库内部帮你解析成 Python 对象。所以它本质上也是一套 API只是它没有 REST 风格、没有 URL、没有 token只有一组类方法和一套协议参数。理解这点很重要。正因为它是直连接行情服务器所以数据实时性、完整度都非常好而且不依赖第三方转发没有按次计费的压力。代价是协议不是官方保证的服务器偶尔会调整需要维护一个可用的服务器地址池。1.2 tdxpy 与 pytdx 的关系老库的新生提到通达信协议很多人会先遇到 pytdx这是更早出现、传播更广的开源库。tdxpy 是在 pytdx 基础上重构出来的一个版本继承了核心协议能力但把 API 改得更加“现代 Python”支持with上下文管理器连接对象能自动管理生命周期返回结果直接用to_df()转成 pandas DataFrame代码更简洁不用手动处理一堆底层细节兼容 Python 3.7 以上的版本安装就能用。我的建议很直接如果只是做行情数据处理直接优先尝试 tdxpy。很多网上教程还在用 pytdx 的旧写法不是不能用而是 tdxpy 在开发体验上确实舒服不少后面所有示例我都会用 tdxpy 来写。1.3 和传统取数方式放在一起比一比为了说清楚“为什么值得用 tdxpy”我把它和常见的几种取数方式放在一起对比。取数方式优点缺点通达信客户端手动导出数据完整复权功能成熟手动操作毫无自动化能力爬虫抓网页数据不需要懂协议反爬严重字段不稳定容易被封第三方 HTTP 数据API接入简单文档规范收费、限流、数据质量依赖服务商tdxpy 直连行情服务器免费、实时、原始数据齐全协议非官方服务器地址需要维护从表里能看出来tdxpy 最大的优势是“免费可自动化原生K线/分笔/财务数据全都有”。适合自己研究、做离线分析、跑回测的场景。但它也确实有门槛服务器地址要从社区获取偶尔会失效协议细节没有官方文档遇到问题得自己试验。2. 环境准备安装 tdxpy 并用三行代码连上行情服务器2.1 安装与依赖既然是一个 Python 库安装就一行命令pip install tdxpy它内部依赖 pandas如果你的环境里没有 pandas安装时会一起带上。建议在 Python 3.8 以上的环境里用我自己在 3.10、3.11 上都跑过没有兼容问题。装好之后验证一下能不能 import 成功顺便看看版本import tdxpy print(tdxpy.__version__)正常输出版本号就说明环境没问题了。2.2 行情服务器地址从哪里来tdxpy 连接行情服务器需要 ip 和端口官网和通达信客户端都不会直接给你一份“开放服务器列表”但社区里常年有人整理。你可以从几个途径获取在 GitHub 上搜 tdxpy 或 pytdx 相关的 issue、wiki里面会有网友共享的服务器列表从本机安装的通达信客户端配置目录里找服务器列表文件用现成的开源脚本对候选 IP 段做探测把能连上的筛出来。我把自己常用的几个地址放这里仅供学习测试地址随时可能变动119.147.212.81:7709 218.108.98.244:7709 124.71.187.122:7709 115.238.56.198:7709 123.125.108.14:7709提示这些不是永久有效的用之前最好先探活。我每次批量任务前都会写一个探活脚本把能连通的服务器挑出来再开始正式拉数据。2.3 连接测试一分钟验证能不能用连接代码非常简单from tdxpy import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): print(连接成功)connect接收三个参数ip、端口、超时时间默认 15 秒可以显式传。连接成功后会进入 with 块说明 TCP 握手和协议握手都完成了。如果超时或报错多半是服务器挂了换一个地址再试。连接这一步要特别注意免费行情服务器一般只支持短连接和低频请求不要在程序里长时间保持一个连接做高频轮询低延迟场景不适合用 tdxpy这类协议接口更适合批量拉历史数据。2.4 我刚上手时遇见的连接问题第一次连接失败是很正常的常见原因我整理一下服务器地址失效或端口不对换一个地址就通网络环境有防火墙TCP 7709 端口出去被拦connect的超时时间设太短默认 15 秒一般够用但网络波动时建议调到 20 秒以上。我自己写了一个简单的探活函数循环测试一批地址连通即返回import socket def ping_server(ip, port, timeout5): try: with socket.create_connection((ip, port), timeouttimeout): return True except OSError: return False这个只是测 TCP 是否通真正要确认行情协议能不能用还是得用 tdxpy 的 connect 去试。3. 核心 API 逐个拆解这些接口参数别搞混3.1 K线数据get_security_bars 是使用频率最高的接口拿到行情数据第一步通常是拉 K线整个 tdxpy 里最常用的就是get_security_bars。它有两个容易踩坑的参数category和start。category是K线周期常见值如下category含义4日K线5周K线6月K线71分钟K线81分钟K线9日K线10季K线11年K线你没看错日K线在通达信协议里有两个值4 和 9。两者返回的内容基本一致有些服务器对 9 支持更好我习惯直接用 9。另外分钟线里 7 和 8 都对应 1 分钟不同版本协议略有差异。最稳妥的方式是写代码时先拉一条数据看一下返回的时间戳确认周期再继续。start表示偏移量0 代表最新的那根K线1 代表往前一根依此类推。这个参数和很多人的直觉相反它不是你理解的“起始日期”而是“从最新一根往前数多少根”。要拉历史K线就得通过不断增大 start 来往前翻。拉最近 10 根日K的示例from tdxpy import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): bars api.get_security_bars(9, 0, 000001, 0, 10) df api.to_df(bars) print(df)参数依次是category、market、code、start、count。market 只有两个值0 代表深圳市场1 代表上海市场。股票代码要补足 6 位比如平安银行是000001浦发银行是600000。count最大是 800超过 800 根K线要自己分段拿。例如要拿最近 2000 根日K就要分别请求 start0、800、1600。这个在后面批量拉历史数据的章节会展开。返回的字段里包含 open、high、low、close、vol、amount以及 year、month、day、hour、minute 等时间字段。用api.to_df()转换后时间字段还在通常我会再把它们合并成一个 datetime 列方便后续按时间排序和索引。3.2 指数K线get_index_bars 的用途与差异指数数据和个股K线在通达信协议里是分开的接口也不一样。拉上证指数、深证成指这类指数用get_index_bars参数和get_security_bars一模一样bars api.get_index_bars(9, 1, 000001, 0, 100) df api.to_df(bars)这里 market 1 对应上海市场代码000001是指上证指数而不是平安银行。很多初学者在这里被坑过同样的代码在个股接口里是平安银行在指数接口里是上证指数。所以写代码时一定要区分清楚。指数数据在分析大盘趋势、做行业轮动研究时很有用比如同时拉上证指数、深证成指、创业板指组合成自己的市场情绪指标。就我个人的经验指数K线和个股K线在数据量上没差别拉取速度和稳定性也一样。3.3 全市场证券列表get_security_list 让你不再手动维护股票池做全市场扫描或批量分析时第一步要拿到所有股票的代码和名称。手动维护一份股票池不是不行但容易漏新股也容易踩退市股。用get_security_list直接从服务器拿省心很多。它的参数只有两个market 和 start。market 同样是 0 表示深圳、1 表示上海start 是起始偏移量从 0 开始。一次性不会返回全部股票需要循环读取直到拿完。大致代码如下def get_all_stocks(api, market0): stocks [] start 0 while True: data api.get_security_list(market, start) if not data: break df api.to_df(data) stocks.extend(df.to_dict(records)) if len(df) 1000: break start 1000 return stocks返回的字段里有 code、name、pre_close 等信息。我把沪深两市的股票列表拉下来后一般会保存成一份本地 CSV每周更新一次用来做股票池管理。这里有个容易忽略的点返回的列表包含的不止是A股还有基金、债券、B股等需要根据代码前缀或者字段做过滤。比如纯A股可以过滤掉 B 股和基金代码段避免后续拉K线时出现一堆无效数据。3.4 实时行情快照get_security_quotes 做实时快照很方便虽然批处理大多用的是K线接口但偶尔需要看当前的实时行情比如做盘中监控、验证某个数据是否更新就要用get_security_quotes。它和前面的接口不同支持一次请求多只股票。调用方式是把 market 和 code 组成元组放进一个列表quotes api.get_security_quotes([(0, 000001), (1, 600000)]) df api.to_df(quotes)返回里包含当前价、涨跌幅、成交量、买卖五档等字段。这个接口在批量任务里有一个用途拉K线前先拉一次快照确认服务器还活着、数据有没有刷新。需要注意的是免费服务器的实时快照延迟无法保证不适合做毫秒级高频交易但做分钟级监控基本够用。3.5 分笔成交与财务数据get_transaction_data 和 get_finance_info除了K线分笔成交和财务数据也是研究里常用的。分笔成交用get_transaction_data拉的是最近的分笔明细参数顺序和K线接口类似transactions api.get_transaction_data(0, 000001, 0, 100) df api.to_df(transactions)这里注意分笔成交接口没有 category 参数market 和 code 后面直接接 start 和 count。分笔数据包含成交时间、价格、成交量、买卖方向等字段做盘口分析和日内微观结构研究时会用到。财务数据用get_finance_info入参是 market 和 codefinance api.get_finance_info(0, 000001) df api.to_df(finance)返回的是财报摘要字段比如总股本、流通股本、每股净资产等适合做基础面筛选。不过我实测发现这个接口的字段更新频率不高做精细的财务分析最好还是结合定期报告文本和行情数据一起看。除权除息信息用get_xdxr_info返回历史上每一笔分红、送股、配股的记录。做复权计算时它很有用后面第5节我会专门讲。3.6 其他接口速查表tdxpy 还提供了不少接口我把常用的整理成一个速查表方便查阅方法功能关键参数get_security_bars个股/基金K线category, market, code, start, countget_index_bars指数K线category, market, code, start, countget_security_list证券列表market, startget_security_quotes实时快照[(market, code), ...]get_transaction_data分笔成交market, code, start, countget_history_transaction_data历史分笔market, code, start, countget_finance_info财务摘要market, codeget_xdxr_info除权除息market, codeget_company_info_category公司信息market, code这些接口的返回结果基本都能用api.to_df()转成 DataFrame。如果发现某个方法在你的 tdxpy 版本里不存在先看一下版本号可能是版本更新导致的方法名变化。4. 实操把全市场日K线批量拉到本地4.1 先设计一个明确的取数方案理论说完了直接上一个我实际做过的需求拉沪深两市全部A股最近3年的日K线存成 CSV 文件用于本地回测和指标分析。这个需求拆解下来有三个关键点拿到所有A股代码也就是先跑一遍get_security_list按代码逐只拉K线每只股票要覆盖约 750 个交易日3年需要分段拉取拉完做数据校验剔除停牌太久、数据异常的股票最后落地到本地。我的建议是写代码前先把方案想清楚别急着跑。尤其是批量任务如果中途因为服务器断开、代码异常挂了没有断点续传设计的话等于白跑。4.2 分段拉取K线的循环怎么写前面说了每次最多取 800 根所以要写一个循环。假设要取 2000 根日Kdef fetch_kline(api, market, code, category9, total2000): frames [] start 0 while start total: bars api.get_security_bars(category, market, code, start, 800) if not bars: break df api.to_df(bars) frames.append(df) start 800 return frames把分段拿到的 DataFrame 合并到一起时注意先按时间字段排序再去重import pandas as pd def merge_kline_frames(frames): if not frames: return pd.DataFrame() df pd.concat(frames, ignore_indexTrue) df[datetime] pd.to_datetime(df[[year, month, day, hour, minute]]) df df.drop_duplicates(subsetdatetime).sort_values(datetime) return df停牌日不会出现在K线数据里所以日期不连续是正常的不要误判为数据缺失。4.3 批量遍历股票池注意节奏整个流程的核心代码大致长这样from tdxpy import TdxHq_API import time api TdxHq_API() with api.connect(119.147.212.81, 7709): # 先拿股票列表 stock_list get_all_stocks(api, market0) get_all_stocks(api, market1) # 再逐只拉K线 for item in stock_list: try: frames fetch_kline(api, item[market], item[code], total750) df merge_kline_frames(frames) df.to_csv(fdata/{item[code]}.csv, indexFalse) except Exception as e: print(f股票 {item[code]} 拉取失败: {e}) time.sleep(0.3) # 控制节奏防止被服务器断开这里有个很关键的经验批量拉数据一定要加 sleep。我自己最开始没有加连续跑了几百个请求后连接就被服务器断掉了。加个 0.3 秒的间隔后整个流程稳定了很多虽然总时间变长但至少能一次性跑完。4.4 数据落地CSV 还是数据库如果数据量不大、字段固定CSV 是最简单好用的。我一般会把K线数据按股票代码存成独立 CSV 文件文件名就是代码方便随时读取。用 pandas 读取也很方便df pd.read_csv(data/000001.csv, parse_dates[datetime])如果数据量大了或者要做频繁的增量更新建议用 SQLite 或 Parquet。SQLite 适合单机小规模数据查询方便Parquet 压缩比高、读得快适合列存分析场景。我自己的数据管线是 CSV 打底跑通之后再决定要不要迁移到更复杂的存储。4.5 增量更新的基本思路全量拉取只适合第一次建库。之后每天只需要补当天的新K线数据。增量更新的逻辑很简单def update_kline(api, market, code, category9): # 先读本地已有数据的最后一根日期 df pd.read_csv(fdata/{code}.csv, parse_dates[datetime]) last_date df[datetime].max() # 拉最新数据这里先拉50根覆盖最近两个多月 bars api.get_security_bars(category, market, code, 0, 50) new_df api.to_df(bars) new_df[datetime] pd.to_datetime(new_df[[year, month, day, hour, minute]]) # 过滤掉老数据只保留新日期 new_df new_df[new_df[datetime] last_date] if not new_df.empty: df pd.concat([df, new_df], ignore_indexTrue) df.to_csv(fdata/{code}.csv, indexFalse)增量更新前要先确认服务器数据有没有刷新到最新通常收盘后一两小时数据才比较完整太早拉可能少最后一根K线。5. 踩坑记录这些问题你大概率会遇到5.1 连接超时和服务器失效免费服务器的稳定性谈不上“可靠”。我遇到过最典型的情况是前天还好好的地址第二天就连不上了。这不是 tdxpy 的问题而是服务器那边调整了策略。解决思路是维护一个服务器地址池每次任务启动时先探活从可用地址里选一个。我一般会准备 5 个以上地址探活通过后再跑正式任务。5.2 拉到一半连接断开批量任务跑太久连接断了是常有的事。第一次我跑了二十分钟后连接断开前面拉的数据没保存全部白干。现在的做法有两个一是每拉完一只股票立刻存盘不攒到最后统一写二是记录进度跑之前把已经完成的股票代码写入一个日志文件重跑时跳过。这两招配合起来基本不用担心批量任务中断。5.3 日K线数据对不上start 偏移量的坑有人拉历史K线时发现数据和通达信客户端对不上检查下来是 start 参数理解错了。start0 是最新一根不是从最早一根开始。如果直接从 start0 开始递增你会发现拉回来的数据顺序是“最新到最老”而且和日期没有直接对应关系必须依赖返回的日期字段做排序。另外不同服务器同一股票的最新K线数量可能有差别拉完多根再合并时要主动按日期排序、去重而不是假设服务器返回的数据本来就整齐。5.4 复权数据的坑这是回测和指标计算里最麻烦的问题。通达信接口返回的K线是未复权价格除权除息当天价格会出现跳空。做历史回测时直接用未复权价格算出来的收益率会明显失真。标准的做法是用get_xdxr_info拿除权除息记录然后计算复权因子。计算前复权价时需要用最新价格往前调整历史价格计算后复权价时则需要把历史价格加上累计分红和送配的影响。我自己写的复权逻辑不复杂但很繁琐。如果你不想自己实现也可以先用未复权数据做因子分析到正式回测阶段再引入复权逻辑。或者干脆用第三方数据源做对照确保复权价格算得没错。5.5 频率限制与封IP风险免费服务器没有明文的频率限制但并不意味着没有限制。我之前因为循环里忘了加 sleep短时间内打了几千个请求结果那一段 IP 被服务器临时禁掉了大概过了几个小时才恢复。我的节奏是批量任务每次请求之间 sleep 0.2 到 0.5 秒单只股票的连续拉取间隔短一点股票与股票之间稍微拉长。宁可跑得慢一点也不要弄到被封。5.6 常见错误速查表现象可能原因解决办法连接超时服务器不通或网络限端口换服务器地址调大超时时间返回数据为空市场代码或股票代码错了检查 market 和 code先拉一条数据验证K线数量不够start 偏移方式理解错了确认 start0 是最新分段递增批量任务中断连接被服务器断开每只股票即时存盘做断点续传IP 被短暂禁止请求太频繁加 sleep降低请求节奏数据里有停牌缺口正常现象不要按日期连续性判断数据错误6. 从 tdxpy 到“自己的 API”把数据服务化6.1 封装一个数据获取类而不是到处复制代码当你写了好几个脚本都在做“连接服务器、拉K线、转 DataFrame”这些重复操作时就该考虑封装了。我后来把常用逻辑收敛到一个DataFetcher类里统一管理服务器地址、连接、重试和缓存业务代码里就不再出现 tdxpy 的细节了。类的大致结构如下class DataFetcher: def __init__(self, servers): self.servers servers self.api None def connect(self): for ip, port in self.servers: try: api TdxHq_API() api.connect(ip, port, time_out10) self.api api return True except Exception: continue return False def get_kline(self, market, code, category9, total800): if not self.api: self.connect() frames [] start 0 while start total: bars self.api.get_security_bars(category, market, code, start, 800) if not bars: break frames.append(self.api.to_df(bars)) start 800 time.sleep(0.2) return merge_kline_frames(frames)封装之后你的业务代码就变成fetcher DataFetcher(SERVERS) df fetcher.get_kline(0, 000001, total750)这样即使 tdxpy 内部 API 有小变化你只需要改 DataFetcher 一处业务代码不受影响。这个习惯对你的长期效率非常重要。6.2 本地缓存同一份数据不要拉两遍如果每天跑同一个分析脚本每次全量重新从服务器拉数据就很浪费。我的方案是把每日增量数据存到本地分析之前先读本地数据只有本地缺数据时才发起请求。最简单的缓存就是文件缓存每次拉完数据按代码和日期命名存到本地目录cache_path fcache/{code}_{category}_{total}.csv if os.path.exists(cache_path): df pd.read_csv(cache_path, parse_dates[datetime]) else: df fetcher.get_kline(market, code, category, total) df.to_csv(cache_path, indexFalse)数据量再大一点可以引入 SQLite。表结构建议把 code、datetime 作为联合主键这样插入增量数据时可以靠数据库层面的唯一约束去掉重复记录。6.3 定时任务让数据每天自动更新本地数据建好后更新就变成一个定时任务。可以交给 cron也可以直接用 Python 的 schedule 库每天收盘后运行增量更新脚本。以 cron 为例每天 16:30 执行一次更新30 16 * * 1-5 /usr/bin/python3 /opt/data_pipeline/update.py脚本内部对股票列表做遍历对每只股票执行增量更新逻辑。增量更新脚本跑完可以追加一个数据质量检查步骤比如统计当天成功更新的股票数量、失败列表有异常就发日志。不用搞太复杂但有一个失败记录能省很多排查时间。6.4 再聊聊“API”这件事从数据接口到接口思想很多人刚接触 tdxpy 时会把它和调大模型 API、调地图 API 这类事情混为一谈。其实“API”的本质是一种约定你按照对方定义的参数发请求对方按照约定的格式返回结果至于底层是 HTTP JSON还是 TCP 二进制都是实现细节。tdxpy 的一个价值是让你理解这层抽象无论数据存在哪里、以什么协议传输对上层使用者来说它就是一个“提供数据的接口”。同样的思维也能用到你自己的数据工程里——当你把本地K线数据用 FastAPI 包装成 HTTP 接口给团队其他人调用时你就真正把 tdxpy 能力变成了“自己的数据接口 API”。我自己的做法是先维护一套本地行情数据仓库再用 FastAPI 暴露几个只读接口比如/api/kline?code000001periodday。团队里的同事需要行情数据时直接调这个接口不用关心我今天到底从哪个服务器拉的、有没有做缓存。这个过程很有“闭合”的感觉从用别人的接口到造自己的接口。6.5 后续可以扩展的方向数据管线跑通之后后面能做的事情很多。比如在K线数据基础上计算技术指标做成指标服务把分笔成交数据落库做日内微观结构分析结合除权除息信息实现完整的复权计算模块对接其他数据源做交叉验证比如用第三方 API 对照K线的最高最低价是否一致。慢慢你会发现数据能力上来了很多研究和开发思路都会打开。tdxpy 只是一个起点但它帮你省掉了最脏最累的那一部分取数工作。最后分享一个我自己的小习惯拉数据时永远把服务器地址、请求参数、股票代码写进日志里不要只记 DataFrame。数据出问题的时候能快速追溯到底是哪一批数据、从哪里来的、用什么参数拉的能省下大把排查时间。这也是我从“能用”到“稳定用”之间最大的改进。
Python获取通达信行情数据：tdxpy库API实战笔记
先说实话做量化研究或者日常行情分析的人最头疼的往往不是策略本身而是“数据从哪里来”。国内行情数据没有对个人开放的官方 API以前常见的路子无非是打开通达信客户端手动导出、写爬虫去抓网页数据、或者买第三方数据服务。直到我在 GitHub 上翻到 tdxpy 这个库发现它能直接用 Python 连接通达信的行情服务器把日K线、分笔成交、财务数据一口气拉下来整个过程干净利落。这篇文章是我踩过不少坑之后整理的 API 使用笔记覆盖基础原理、常用接口参数、批量拉取和问题排查适合想自己搭一套免费行情数据管线的量化研究者、数据分析和 Python 开发同学。1. 先把底摸清tdxpy 到底是什么和 Web API 有什么不一样1.1 通达信协议一个“野生”但长期可用的行情源通达信是国内最普及的行情软件之一很多券商直接拿它做交易终端内核。它背后有一套基于 TCP 的私有行情协议客户端通过它向行情服务器请求股票列表、K线、分笔、财务指标等数据。早年开源社区的人通过抓包、逆向把协议格式逐步摸清楚了于是出现了多个语言的实现。tdxpy 就是这套协议在 Python 生态里一个很顺手的封装。很多人第一次接触“接口”这个词脑子里浮现的是 HTTP 接口发一个 GET 请求服务器回一段 JSON像调 DeepSeek、GPT 的 API 那样。tdxpy 完全不是这个路数它是直接在 Python 进程里与行情服务器建立 TCP 长连接请求数据的时候发送的是二进制报文返回的也是二进制流然后库内部帮你解析成 Python 对象。所以它本质上也是一套 API只是它没有 REST 风格、没有 URL、没有 token只有一组类方法和一套协议参数。理解这点很重要。正因为它是直连接行情服务器所以数据实时性、完整度都非常好而且不依赖第三方转发没有按次计费的压力。代价是协议不是官方保证的服务器偶尔会调整需要维护一个可用的服务器地址池。1.2 tdxpy 与 pytdx 的关系老库的新生提到通达信协议很多人会先遇到 pytdx这是更早出现、传播更广的开源库。tdxpy 是在 pytdx 基础上重构出来的一个版本继承了核心协议能力但把 API 改得更加“现代 Python”支持with上下文管理器连接对象能自动管理生命周期返回结果直接用to_df()转成 pandas DataFrame代码更简洁不用手动处理一堆底层细节兼容 Python 3.7 以上的版本安装就能用。我的建议很直接如果只是做行情数据处理直接优先尝试 tdxpy。很多网上教程还在用 pytdx 的旧写法不是不能用而是 tdxpy 在开发体验上确实舒服不少后面所有示例我都会用 tdxpy 来写。1.3 和传统取数方式放在一起比一比为了说清楚“为什么值得用 tdxpy”我把它和常见的几种取数方式放在一起对比。取数方式优点缺点通达信客户端手动导出数据完整复权功能成熟手动操作毫无自动化能力爬虫抓网页数据不需要懂协议反爬严重字段不稳定容易被封第三方 HTTP 数据API接入简单文档规范收费、限流、数据质量依赖服务商tdxpy 直连行情服务器免费、实时、原始数据齐全协议非官方服务器地址需要维护从表里能看出来tdxpy 最大的优势是“免费可自动化原生K线/分笔/财务数据全都有”。适合自己研究、做离线分析、跑回测的场景。但它也确实有门槛服务器地址要从社区获取偶尔会失效协议细节没有官方文档遇到问题得自己试验。2. 环境准备安装 tdxpy 并用三行代码连上行情服务器2.1 安装与依赖既然是一个 Python 库安装就一行命令pip install tdxpy它内部依赖 pandas如果你的环境里没有 pandas安装时会一起带上。建议在 Python 3.8 以上的环境里用我自己在 3.10、3.11 上都跑过没有兼容问题。装好之后验证一下能不能 import 成功顺便看看版本import tdxpy print(tdxpy.__version__)正常输出版本号就说明环境没问题了。2.2 行情服务器地址从哪里来tdxpy 连接行情服务器需要 ip 和端口官网和通达信客户端都不会直接给你一份“开放服务器列表”但社区里常年有人整理。你可以从几个途径获取在 GitHub 上搜 tdxpy 或 pytdx 相关的 issue、wiki里面会有网友共享的服务器列表从本机安装的通达信客户端配置目录里找服务器列表文件用现成的开源脚本对候选 IP 段做探测把能连上的筛出来。我把自己常用的几个地址放这里仅供学习测试地址随时可能变动119.147.212.81:7709 218.108.98.244:7709 124.71.187.122:7709 115.238.56.198:7709 123.125.108.14:7709提示这些不是永久有效的用之前最好先探活。我每次批量任务前都会写一个探活脚本把能连通的服务器挑出来再开始正式拉数据。2.3 连接测试一分钟验证能不能用连接代码非常简单from tdxpy import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): print(连接成功)connect接收三个参数ip、端口、超时时间默认 15 秒可以显式传。连接成功后会进入 with 块说明 TCP 握手和协议握手都完成了。如果超时或报错多半是服务器挂了换一个地址再试。连接这一步要特别注意免费行情服务器一般只支持短连接和低频请求不要在程序里长时间保持一个连接做高频轮询低延迟场景不适合用 tdxpy这类协议接口更适合批量拉历史数据。2.4 我刚上手时遇见的连接问题第一次连接失败是很正常的常见原因我整理一下服务器地址失效或端口不对换一个地址就通网络环境有防火墙TCP 7709 端口出去被拦connect的超时时间设太短默认 15 秒一般够用但网络波动时建议调到 20 秒以上。我自己写了一个简单的探活函数循环测试一批地址连通即返回import socket def ping_server(ip, port, timeout5): try: with socket.create_connection((ip, port), timeouttimeout): return True except OSError: return False这个只是测 TCP 是否通真正要确认行情协议能不能用还是得用 tdxpy 的 connect 去试。3. 核心 API 逐个拆解这些接口参数别搞混3.1 K线数据get_security_bars 是使用频率最高的接口拿到行情数据第一步通常是拉 K线整个 tdxpy 里最常用的就是get_security_bars。它有两个容易踩坑的参数category和start。category是K线周期常见值如下category含义4日K线5周K线6月K线71分钟K线81分钟K线9日K线10季K线11年K线你没看错日K线在通达信协议里有两个值4 和 9。两者返回的内容基本一致有些服务器对 9 支持更好我习惯直接用 9。另外分钟线里 7 和 8 都对应 1 分钟不同版本协议略有差异。最稳妥的方式是写代码时先拉一条数据看一下返回的时间戳确认周期再继续。start表示偏移量0 代表最新的那根K线1 代表往前一根依此类推。这个参数和很多人的直觉相反它不是你理解的“起始日期”而是“从最新一根往前数多少根”。要拉历史K线就得通过不断增大 start 来往前翻。拉最近 10 根日K的示例from tdxpy import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): bars api.get_security_bars(9, 0, 000001, 0, 10) df api.to_df(bars) print(df)参数依次是category、market、code、start、count。market 只有两个值0 代表深圳市场1 代表上海市场。股票代码要补足 6 位比如平安银行是000001浦发银行是600000。count最大是 800超过 800 根K线要自己分段拿。例如要拿最近 2000 根日K就要分别请求 start0、800、1600。这个在后面批量拉历史数据的章节会展开。返回的字段里包含 open、high、low、close、vol、amount以及 year、month、day、hour、minute 等时间字段。用api.to_df()转换后时间字段还在通常我会再把它们合并成一个 datetime 列方便后续按时间排序和索引。3.2 指数K线get_index_bars 的用途与差异指数数据和个股K线在通达信协议里是分开的接口也不一样。拉上证指数、深证成指这类指数用get_index_bars参数和get_security_bars一模一样bars api.get_index_bars(9, 1, 000001, 0, 100) df api.to_df(bars)这里 market 1 对应上海市场代码000001是指上证指数而不是平安银行。很多初学者在这里被坑过同样的代码在个股接口里是平安银行在指数接口里是上证指数。所以写代码时一定要区分清楚。指数数据在分析大盘趋势、做行业轮动研究时很有用比如同时拉上证指数、深证成指、创业板指组合成自己的市场情绪指标。就我个人的经验指数K线和个股K线在数据量上没差别拉取速度和稳定性也一样。3.3 全市场证券列表get_security_list 让你不再手动维护股票池做全市场扫描或批量分析时第一步要拿到所有股票的代码和名称。手动维护一份股票池不是不行但容易漏新股也容易踩退市股。用get_security_list直接从服务器拿省心很多。它的参数只有两个market 和 start。market 同样是 0 表示深圳、1 表示上海start 是起始偏移量从 0 开始。一次性不会返回全部股票需要循环读取直到拿完。大致代码如下def get_all_stocks(api, market0): stocks [] start 0 while True: data api.get_security_list(market, start) if not data: break df api.to_df(data) stocks.extend(df.to_dict(records)) if len(df) 1000: break start 1000 return stocks返回的字段里有 code、name、pre_close 等信息。我把沪深两市的股票列表拉下来后一般会保存成一份本地 CSV每周更新一次用来做股票池管理。这里有个容易忽略的点返回的列表包含的不止是A股还有基金、债券、B股等需要根据代码前缀或者字段做过滤。比如纯A股可以过滤掉 B 股和基金代码段避免后续拉K线时出现一堆无效数据。3.4 实时行情快照get_security_quotes 做实时快照很方便虽然批处理大多用的是K线接口但偶尔需要看当前的实时行情比如做盘中监控、验证某个数据是否更新就要用get_security_quotes。它和前面的接口不同支持一次请求多只股票。调用方式是把 market 和 code 组成元组放进一个列表quotes api.get_security_quotes([(0, 000001), (1, 600000)]) df api.to_df(quotes)返回里包含当前价、涨跌幅、成交量、买卖五档等字段。这个接口在批量任务里有一个用途拉K线前先拉一次快照确认服务器还活着、数据有没有刷新。需要注意的是免费服务器的实时快照延迟无法保证不适合做毫秒级高频交易但做分钟级监控基本够用。3.5 分笔成交与财务数据get_transaction_data 和 get_finance_info除了K线分笔成交和财务数据也是研究里常用的。分笔成交用get_transaction_data拉的是最近的分笔明细参数顺序和K线接口类似transactions api.get_transaction_data(0, 000001, 0, 100) df api.to_df(transactions)这里注意分笔成交接口没有 category 参数market 和 code 后面直接接 start 和 count。分笔数据包含成交时间、价格、成交量、买卖方向等字段做盘口分析和日内微观结构研究时会用到。财务数据用get_finance_info入参是 market 和 codefinance api.get_finance_info(0, 000001) df api.to_df(finance)返回的是财报摘要字段比如总股本、流通股本、每股净资产等适合做基础面筛选。不过我实测发现这个接口的字段更新频率不高做精细的财务分析最好还是结合定期报告文本和行情数据一起看。除权除息信息用get_xdxr_info返回历史上每一笔分红、送股、配股的记录。做复权计算时它很有用后面第5节我会专门讲。3.6 其他接口速查表tdxpy 还提供了不少接口我把常用的整理成一个速查表方便查阅方法功能关键参数get_security_bars个股/基金K线category, market, code, start, countget_index_bars指数K线category, market, code, start, countget_security_list证券列表market, startget_security_quotes实时快照[(market, code), ...]get_transaction_data分笔成交market, code, start, countget_history_transaction_data历史分笔market, code, start, countget_finance_info财务摘要market, codeget_xdxr_info除权除息market, codeget_company_info_category公司信息market, code这些接口的返回结果基本都能用api.to_df()转成 DataFrame。如果发现某个方法在你的 tdxpy 版本里不存在先看一下版本号可能是版本更新导致的方法名变化。4. 实操把全市场日K线批量拉到本地4.1 先设计一个明确的取数方案理论说完了直接上一个我实际做过的需求拉沪深两市全部A股最近3年的日K线存成 CSV 文件用于本地回测和指标分析。这个需求拆解下来有三个关键点拿到所有A股代码也就是先跑一遍get_security_list按代码逐只拉K线每只股票要覆盖约 750 个交易日3年需要分段拉取拉完做数据校验剔除停牌太久、数据异常的股票最后落地到本地。我的建议是写代码前先把方案想清楚别急着跑。尤其是批量任务如果中途因为服务器断开、代码异常挂了没有断点续传设计的话等于白跑。4.2 分段拉取K线的循环怎么写前面说了每次最多取 800 根所以要写一个循环。假设要取 2000 根日Kdef fetch_kline(api, market, code, category9, total2000): frames [] start 0 while start total: bars api.get_security_bars(category, market, code, start, 800) if not bars: break df api.to_df(bars) frames.append(df) start 800 return frames把分段拿到的 DataFrame 合并到一起时注意先按时间字段排序再去重import pandas as pd def merge_kline_frames(frames): if not frames: return pd.DataFrame() df pd.concat(frames, ignore_indexTrue) df[datetime] pd.to_datetime(df[[year, month, day, hour, minute]]) df df.drop_duplicates(subsetdatetime).sort_values(datetime) return df停牌日不会出现在K线数据里所以日期不连续是正常的不要误判为数据缺失。4.3 批量遍历股票池注意节奏整个流程的核心代码大致长这样from tdxpy import TdxHq_API import time api TdxHq_API() with api.connect(119.147.212.81, 7709): # 先拿股票列表 stock_list get_all_stocks(api, market0) get_all_stocks(api, market1) # 再逐只拉K线 for item in stock_list: try: frames fetch_kline(api, item[market], item[code], total750) df merge_kline_frames(frames) df.to_csv(fdata/{item[code]}.csv, indexFalse) except Exception as e: print(f股票 {item[code]} 拉取失败: {e}) time.sleep(0.3) # 控制节奏防止被服务器断开这里有个很关键的经验批量拉数据一定要加 sleep。我自己最开始没有加连续跑了几百个请求后连接就被服务器断掉了。加个 0.3 秒的间隔后整个流程稳定了很多虽然总时间变长但至少能一次性跑完。4.4 数据落地CSV 还是数据库如果数据量不大、字段固定CSV 是最简单好用的。我一般会把K线数据按股票代码存成独立 CSV 文件文件名就是代码方便随时读取。用 pandas 读取也很方便df pd.read_csv(data/000001.csv, parse_dates[datetime])如果数据量大了或者要做频繁的增量更新建议用 SQLite 或 Parquet。SQLite 适合单机小规模数据查询方便Parquet 压缩比高、读得快适合列存分析场景。我自己的数据管线是 CSV 打底跑通之后再决定要不要迁移到更复杂的存储。4.5 增量更新的基本思路全量拉取只适合第一次建库。之后每天只需要补当天的新K线数据。增量更新的逻辑很简单def update_kline(api, market, code, category9): # 先读本地已有数据的最后一根日期 df pd.read_csv(fdata/{code}.csv, parse_dates[datetime]) last_date df[datetime].max() # 拉最新数据这里先拉50根覆盖最近两个多月 bars api.get_security_bars(category, market, code, 0, 50) new_df api.to_df(bars) new_df[datetime] pd.to_datetime(new_df[[year, month, day, hour, minute]]) # 过滤掉老数据只保留新日期 new_df new_df[new_df[datetime] last_date] if not new_df.empty: df pd.concat([df, new_df], ignore_indexTrue) df.to_csv(fdata/{code}.csv, indexFalse)增量更新前要先确认服务器数据有没有刷新到最新通常收盘后一两小时数据才比较完整太早拉可能少最后一根K线。5. 踩坑记录这些问题你大概率会遇到5.1 连接超时和服务器失效免费服务器的稳定性谈不上“可靠”。我遇到过最典型的情况是前天还好好的地址第二天就连不上了。这不是 tdxpy 的问题而是服务器那边调整了策略。解决思路是维护一个服务器地址池每次任务启动时先探活从可用地址里选一个。我一般会准备 5 个以上地址探活通过后再跑正式任务。5.2 拉到一半连接断开批量任务跑太久连接断了是常有的事。第一次我跑了二十分钟后连接断开前面拉的数据没保存全部白干。现在的做法有两个一是每拉完一只股票立刻存盘不攒到最后统一写二是记录进度跑之前把已经完成的股票代码写入一个日志文件重跑时跳过。这两招配合起来基本不用担心批量任务中断。5.3 日K线数据对不上start 偏移量的坑有人拉历史K线时发现数据和通达信客户端对不上检查下来是 start 参数理解错了。start0 是最新一根不是从最早一根开始。如果直接从 start0 开始递增你会发现拉回来的数据顺序是“最新到最老”而且和日期没有直接对应关系必须依赖返回的日期字段做排序。另外不同服务器同一股票的最新K线数量可能有差别拉完多根再合并时要主动按日期排序、去重而不是假设服务器返回的数据本来就整齐。5.4 复权数据的坑这是回测和指标计算里最麻烦的问题。通达信接口返回的K线是未复权价格除权除息当天价格会出现跳空。做历史回测时直接用未复权价格算出来的收益率会明显失真。标准的做法是用get_xdxr_info拿除权除息记录然后计算复权因子。计算前复权价时需要用最新价格往前调整历史价格计算后复权价时则需要把历史价格加上累计分红和送配的影响。我自己写的复权逻辑不复杂但很繁琐。如果你不想自己实现也可以先用未复权数据做因子分析到正式回测阶段再引入复权逻辑。或者干脆用第三方数据源做对照确保复权价格算得没错。5.5 频率限制与封IP风险免费服务器没有明文的频率限制但并不意味着没有限制。我之前因为循环里忘了加 sleep短时间内打了几千个请求结果那一段 IP 被服务器临时禁掉了大概过了几个小时才恢复。我的节奏是批量任务每次请求之间 sleep 0.2 到 0.5 秒单只股票的连续拉取间隔短一点股票与股票之间稍微拉长。宁可跑得慢一点也不要弄到被封。5.6 常见错误速查表现象可能原因解决办法连接超时服务器不通或网络限端口换服务器地址调大超时时间返回数据为空市场代码或股票代码错了检查 market 和 code先拉一条数据验证K线数量不够start 偏移方式理解错了确认 start0 是最新分段递增批量任务中断连接被服务器断开每只股票即时存盘做断点续传IP 被短暂禁止请求太频繁加 sleep降低请求节奏数据里有停牌缺口正常现象不要按日期连续性判断数据错误6. 从 tdxpy 到“自己的 API”把数据服务化6.1 封装一个数据获取类而不是到处复制代码当你写了好几个脚本都在做“连接服务器、拉K线、转 DataFrame”这些重复操作时就该考虑封装了。我后来把常用逻辑收敛到一个DataFetcher类里统一管理服务器地址、连接、重试和缓存业务代码里就不再出现 tdxpy 的细节了。类的大致结构如下class DataFetcher: def __init__(self, servers): self.servers servers self.api None def connect(self): for ip, port in self.servers: try: api TdxHq_API() api.connect(ip, port, time_out10) self.api api return True except Exception: continue return False def get_kline(self, market, code, category9, total800): if not self.api: self.connect() frames [] start 0 while start total: bars self.api.get_security_bars(category, market, code, start, 800) if not bars: break frames.append(self.api.to_df(bars)) start 800 time.sleep(0.2) return merge_kline_frames(frames)封装之后你的业务代码就变成fetcher DataFetcher(SERVERS) df fetcher.get_kline(0, 000001, total750)这样即使 tdxpy 内部 API 有小变化你只需要改 DataFetcher 一处业务代码不受影响。这个习惯对你的长期效率非常重要。6.2 本地缓存同一份数据不要拉两遍如果每天跑同一个分析脚本每次全量重新从服务器拉数据就很浪费。我的方案是把每日增量数据存到本地分析之前先读本地数据只有本地缺数据时才发起请求。最简单的缓存就是文件缓存每次拉完数据按代码和日期命名存到本地目录cache_path fcache/{code}_{category}_{total}.csv if os.path.exists(cache_path): df pd.read_csv(cache_path, parse_dates[datetime]) else: df fetcher.get_kline(market, code, category, total) df.to_csv(cache_path, indexFalse)数据量再大一点可以引入 SQLite。表结构建议把 code、datetime 作为联合主键这样插入增量数据时可以靠数据库层面的唯一约束去掉重复记录。6.3 定时任务让数据每天自动更新本地数据建好后更新就变成一个定时任务。可以交给 cron也可以直接用 Python 的 schedule 库每天收盘后运行增量更新脚本。以 cron 为例每天 16:30 执行一次更新30 16 * * 1-5 /usr/bin/python3 /opt/data_pipeline/update.py脚本内部对股票列表做遍历对每只股票执行增量更新逻辑。增量更新脚本跑完可以追加一个数据质量检查步骤比如统计当天成功更新的股票数量、失败列表有异常就发日志。不用搞太复杂但有一个失败记录能省很多排查时间。6.4 再聊聊“API”这件事从数据接口到接口思想很多人刚接触 tdxpy 时会把它和调大模型 API、调地图 API 这类事情混为一谈。其实“API”的本质是一种约定你按照对方定义的参数发请求对方按照约定的格式返回结果至于底层是 HTTP JSON还是 TCP 二进制都是实现细节。tdxpy 的一个价值是让你理解这层抽象无论数据存在哪里、以什么协议传输对上层使用者来说它就是一个“提供数据的接口”。同样的思维也能用到你自己的数据工程里——当你把本地K线数据用 FastAPI 包装成 HTTP 接口给团队其他人调用时你就真正把 tdxpy 能力变成了“自己的数据接口 API”。我自己的做法是先维护一套本地行情数据仓库再用 FastAPI 暴露几个只读接口比如/api/kline?code000001periodday。团队里的同事需要行情数据时直接调这个接口不用关心我今天到底从哪个服务器拉的、有没有做缓存。这个过程很有“闭合”的感觉从用别人的接口到造自己的接口。6.5 后续可以扩展的方向数据管线跑通之后后面能做的事情很多。比如在K线数据基础上计算技术指标做成指标服务把分笔成交数据落库做日内微观结构分析结合除权除息信息实现完整的复权计算模块对接其他数据源做交叉验证比如用第三方 API 对照K线的最高最低价是否一致。慢慢你会发现数据能力上来了很多研究和开发思路都会打开。tdxpy 只是一个起点但它帮你省掉了最脏最累的那一部分取数工作。最后分享一个我自己的小习惯拉数据时永远把服务器地址、请求参数、股票代码写进日志里不要只记 DataFrame。数据出问题的时候能快速追溯到底是哪一批数据、从哪里来的、用什么参数拉的能省下大把排查时间。这也是我从“能用”到“稳定用”之间最大的改进。
