简介这是一套面向金融数据分析从业者与Python进阶开发者的通达信数据接入工具旨在降低pytdx原生库的使用门槛解决手动连接行情服务器、解析二进制数据格式、管理多市场配置等复杂问题。资源包共203个文件体量54.79MB涵盖79个Python脚本实现行情获取、K线合成、板块分类、指标计算等核心逻辑、33个.dat数据文件含标准通达信格式的L2快照与历史日线、19个.cfg配置文件如tdxzs.cfg、tdxhy.cfg等预置A股、期货、港股等市场参数、19个.md文档含快速上手指南、API说明与调用示例以及Dockerfile、Makefile等工程化支持文件。已有477人学习下载。用户可直接复用封装后的统一接口读取实时/历史行情快速构建量化策略原型完整目录结构体现模块化设计思想便于二次开发与本地化适配配套配置文件与标准化命名规范显著提升多源数据管理效率。 这阵子有朋友问我手里有通达信客户端数据也能看但想写点程序自动拉行情、做选股回测量一大就卡壳。这其实是很多做量化、做数据分析的人都会撞上的墙卡在“数据源”这一关。我当时解决这个问题的思路很直接找一个能直接跟通达信行情服务器对话的Python库把它复杂的地方封装掉做成自己团队随取随用的数据接口。今天就把这套基于pytdx的源码封装工具的设计思路、核心细节和踩坑过程完整拆出来希望能让同样卡在这堵墙前的朋友少走弯路。先说结论pytdx没有官方文档但协议本身不复杂难的是把连接管理、容错、数据格式统一这些脏活累活做好。全文围绕这个库展开适合刚接触Python量化、想做行情数据落地的读者。我会从为什么要封装讲起再到核心API的使用、封装架构设计、常见兼容性问题排查一层层扒开。1. 项目背景与整体设计思路要讲清楚这个封装工具为什么存在得先回到一个原始问题行情数据从哪里来市面上要么是免费接口但字段不全、要么是商业数据源价格不低、要么是抓取网页接口但极其不稳定。剩下的一个选择就是自己直连行情服务器。1.1 为什么选择pytdx而不是其他数据方案我在选型时只考虑一个问题谁的数据最接近“原生、完整、无中间加工”答案很明确通达信行情服务器。它是国内股票行情数据最主流的下游分发端之一数据覆盖沪深京A股、B股、基金、债券、期权等。而pytdx是纯Python实现的通达信行情协议客户端没有额外dll依赖直接解析底层数据包。对比几个常见思路方案优点缺点直接抓取网页接口无需额外依赖字段有限、封IP、接口随版本变动pandas-datareader代码简单数据源在国外、时延高、字段无法定制商业量化数据源数据干净稳定成本高、部署重pytdx数据直连、字段全、实时性高需自行封装、协议无官方文档选择pytdx的核心逻辑在于它把最难的“协议编码/解码”部分处理掉了让我可以把精力放在上层封装、数据格式化和容错处理上。但同时它也把另一块难题留给了我——连接不稳定、断线重连、增量更新、复权处理这些事不自己写就没人帮你写。1.2 这个封装工具到底解决了什么问题直接裸用pytdx写业务代码你会遇到几个非常现实的问题每次获取数据都要手动创建连接用完还得记得关闭漏了就会积攒大量TIME_WAIT连接。服务端动不动就断开连接代码里全是重复的异常捕获和重连逻辑。不同接口返回的字段命名、类型不一致有的返回str有的返回float有的返回int清洗工作重复且容易出错。hq、ex扩展行情两套API使用方式不同业务代码需要分别适配切换成本高。这个封装工具要做的不是重新发明pytdx而是在它上面做一层真正的“数据服务层”统一管理连接池、统一错误重试、统一数据格式输出、提供简洁的调用入口。让使用方写一行代码就能拿到干净的DataFrame。2. pytdx核心功能拆解与原理在动手封装之前我必须先把pytdx的底层能力摸清楚。pytdx分为两套APITdxHq_API标准行情协议和TdxExHq_API扩展行情协议很多人只用过前者用后者的人少但恰恰是后者能补齐不少数据盲区。2.1 标准协议TdxHq_API这套API是绝大多数人的入门选择它会创建一个连接到通达信标准行情服务器的客户端支持的方法包括证券列表、K线、分时、分笔、财务、除权信息等。先看两个基础用法我直接写出最核心的部分from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): # 获取股票列表 stocks api.get_security_list(0, 0) # 获取日K线 bars api.get_security_bars(9, 0, 000001, 0, 100) # 获取分时数据 ticks api.get_minute_time_data(0, 000001) # 获取财务数据 finance api.get_finance_info(0, 000001) # 获取除权除息信息 xdxr api.get_xdxr_info(0, 000001)这段代码看起来简单实际上里面有几个关键参数不是字面上那么好理解的。get_security_list的第二个参数是起始位置传入0代表从第0条开始。这个方法一次最多返回1000条记录所以A股5000多只股票需要分6次拉取用循环控制偏移量。get_security_bars的第一个参数category9代表日K线0代表5分钟K线1代表15分钟K线2代表30分钟K线3代表1小时K线4代表日K线和9几乎一样但周期定义略有差异7代表1分钟K线8代表周K线。第二个参数是市场代码0是深圳1是上海。第三个参数是股票代码注意是字符串。第四个参数是起始位置第五个参数是最大数量一次最多800根。get_xdxr_info返回的是除权除息数据在做复权计算时是必需的但它有个坑它返回的是原始字符串数值比如10派1.5元这种文本需要自己解析成可计算的数值。2.2 扩展协议TdxExHq_API扩展协议设计的初衷是获取期货、外盘、港股等非A股数据。我当时一度忽略了这个模块直到需要做全市场扫描时才发现它的用处它支持获取板块内的成分股列表这是标准API里没有的功能。from pytdx.exhq import TdxExHq_API ex_api TdxExHq_API() with ex_api.connect(221.231.141.60, 7727): # 获取板块数量 block_count ex_api.get_block_count() # 获取板块列表 blocks ex_api.get_block_info(0, 100) # 获取板块内成分股 stocks ex_api.get_security_list(0, 0)扩展协议的服务器端口一般是7727它返回的板块信息包括板块代码、板块名称、板块类型等。做行业轮动分析时这个接口比维护一份静态板块映射文件靠谱得多因为它是实时拉取的最新数据。2.3 连接管理为什么需要连接池pytdx设计上是一个连接对应一个会话如果你的程序需要高频访问行情接口频繁开合连接的开销是不可忽略的。我测试过新建一个连接平均耗时300-500ms而真正获取一次K线数据只需要20ms。如果业务代码每次调用都重新连接效率会慢到一个不可接受的程度。封装工具的核心就是做一个长连接池。启动时预先建立N个连接调用方从池中借用、使用、归还。连接断开时自动重建。我在这上面吃过亏一开始只建了一个连接程序里多线程调用时把同一个连接同时执行多个请求导致数据包错乱。后来改成池化之后这个现象彻底消失了。3. 源码封装工具的分层架构封装工具用到的设计思路并不复杂可以概括为“入口统一、数据格式化、连接自动管理”。我把它拆成了四个层级。3.1 接口层统一入口设计与函数清单这层负责把pytdx各种方法映射成统一风格、统一参数、返回DataFrame的接口。调用方不需要知道底层是标准协议还是扩展协议只要调get_daily_kline(000001, 2020-01-01, 2024-12-31)就能拿到预期数据。我定义的几个核心方法方法名底层实现返回内容get_daily_klineget_security_bars(category9)日K线DataFrameget_minute_klineget_security_bars(category0/1/2/3/7)分钟K线DataFrameget_realtime_quoteget_security_quotes实时五档行情get_trade_detailget_transaction_data分笔成交明细get_finance_summaryget_finance_info财务指标摘要get_dividend_infoget_xdxr_info除权除息列表get_stock_listget_security_list全市场股票列表get_block_stocksTdxExHq_API指定板块成分股每个方法内部统一做四件事拿到连接、请求数据、解析成DataFrame、返回。如果是失败交给你重试层处理。3.2 配置层服务器列表管理与动态切换通达信行情服务器不止一个而且不同服务器在不同网络环境下的连通性差异很大。有的快得飞起有的连上就断。所以封装工具必须有一个服务器列表配置和健康检查机制。我在配置文件中维护了一份服务器列表每个服务器包含IP、端口、协议类型hq/ex、区域华南/华北/华东等启动时对所有服务器做一次ping和连接测试按延迟从低到高排序优先使用低延迟的。配置文件的格式如下servers: - host: 119.147.212.81 port: 7709 protocol: hq region: 深圳 - host: 123.125.108.14 port: 7709 protocol: hq region: 北京 - host: 221.231.141.60 port: 7727 protocol: ex region: 上海自动切换的逻辑是如果连接某个服务器连续失败三次就把它标记为不可用换到下一个每隔一段时间做一次健康检查把恢复的服务器重新加入候选池。这是解决“网络环境变化导致连接不上”这个老大难问题的最有效手段。3.3 数据模型层统一字段名与类型转换pytdx返回的数据在可读性方面很原始具体表现是字段名极其精简比如open、high、low、close这是英文缩写还好但有的接口返回的是price、vol这种含义模糊的字段。另外还有下面这些问题返回的字段类型不一致vol有时是int有时是str。价格和时间戳精度不统一有的接口返回秒级时间戳有的返回毫秒级。股票代码格式不统一有的返回000001有的返回SZ000001。封装层要做的第一件事就是把这些数据全部转成标准格式def normalize_stock_code(code, marketNone): code str(code).strip().zfill(6) if market is None: market 0 if code.startswith((0, 2, 3)) else 1 return pd.Series({code: code, market: market})所有K线数据统一转成datetime格式的索引、float格式的价格、int格式的成交量字段名统一为open/high/low/close/volume/amount。这样下游不管做什么分析都不需要再操心“这个字段是不是字符串”“这个量是手还是股”这种琐碎问题。3.4 工具层复权因子计算与K线合成这块是整个封装里我认为技术含量最高的部分。通达信原生接口返回的K线数据是不复权的。如果直接用不复权数据做回测遇到除权除息日价格会出现跳空导致技术指标计算失真。所以封装必须提供复权能力。复权因子的计算依赖除权除息数据。思路是先获取该股票从上市到现在的所有除权除息记录解析出“每10股派X元”“每10股送Y股”这些信息换算出一个累计复权因子序列然后在K线数据上做前复权或后复权。计算公式很直接后复权价 不复权价 × 累计复权因子前复权价 不复权价 × 最新复权因子 / 当日复权因子封装工具里我专门写了adjust_price_factor方法输入是原始K线DataFrame和除权除息DataFrame输出是带复权因子的完整DataFrame。这个逻辑看起来简单但一旦遇到多次分红送股、公积金转增股本手工计算的出错率极高交给代码处理是唯一可靠的方式。4. 核心实操过程与关键代码解析理论讲再多最终还是要落地跑起来。下面我按完整流程走一遍从安装到核心功能跑通每一步都附上代码和必要的解释。4.1 环境准备与安装安装pytdx很简单用pip就可以pip install pytdx它会自动安装依赖的pandas和numpy这两个是数据处理的底座。如果你已经有Python环境这一步基本不会有问题。我实测的是Python 3.8-3.11都能正常运行这个库。一个容易踩的坑是Python 3.12以上版本因为pytdx源码中用到了一些旧版datetime库的写法在新版本Python上可能会报模块迁移警告。如果遇到建议建一个3.10的虚拟环境使用更省心。4.2 建立第一个行情连接直接写一个最小的示例验证环境是否正常from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): # 获取上证指数最新行情 quote api.get_security_quotes([(1, 000001)]) print(quote)如果连接成功会返回一个包含最新价的字典。这个(1, 000001)是一个元组第一个元素是市场代码1表示上海0表示深圳。很多人第一次接触时会弄反结果拿到的是“平安银行”的数据而不是“上证指数”的因为000001在深圳市场是平安银行在上海市场是上证指数。验证成功后就可以正式使用封装工具了。下面我把整个封装工具的核心代码按模块拆出来讲。4.3 核心封装类TdxDataWrapper我做了个小设计把整个连接池和重试逻辑收敛到一个类里对外暴露的方法是直接可用的业务方法。这个类是整个封装工具的地基。import time import random import pandas as pd from pytdx.hq import TdxHq_API from pytdx.exhq import TdxExHq_API class TdxDataWrapper: def __init__(self, server_list, max_connections3): self.server_list server_list self.max_connections max_connections self._connections [] self._conn_index 0 self.max_retry 3 def _get_connection(self): if not self._connections: self._create_connections() conn self._connections[self._conn_index % len(self._connections)] self._conn_index 1 if not conn[connected]: conn[api] self._create_api(conn[server]) conn[connected] True return conn[api] def _create_connections(self): for i in range(self.max_connections): server self.server_list[i % len(self.server_list)] api self._create_api(server) self._connections.append({server: server, api: api, connected: True}) staticmethod def _create_api(server): if server[protocol] ex: api TdxExHq_API() else: api TdxHq_API() api.connect(server[host], server[port]) return api def _execute(self, func, *args, **kwargs): last_exc None for attempt in range(self.max_retry): try: api self._get_connection() result getattr(api, func)(*args, **kwargs) if result is None: raise ValueError(返回结果为空可能连接已失效) return result except Exception as exc: last_exc exc # 标记连接失效下次重新创建 self._connections[self._conn_index % len(self._connections)][connected] False time.sleep(0.5 * (attempt 1)) raise last_exc def get_daily_kline(self, market, code, start0, count800): df self._execute(get_security_bars, 9, market, code, start, count) return self._format_kline(df) staticmethod def _format_kline(df): df df.rename(columns{ datetime: datetime, open: open, high: high, low: low, close: close, vol: volume, amount: amount }) df[datetime] pd.to_datetime(df[datetime]) df df.set_index(datetime) for col in [open, high, low, close]: df[col] df[col].astype(float) df[volume] df[volume].astype(float) return df这个类解决的最关键问题就是连接状态管理。_execute方法内部会捕获所有异常包括网络异常、返回空、超时等并自动重连重试。同时它实现了简单的连接池轮询round-robin把请求分散到不同连接上避免单连接过热。4.4 验证封装后的调用效果封装后业务代码调用变成了这个样子wrapper TdxDataWrapper(server_list) # 获取平安银行近800个交易日的日K线 kline wrapper.get_daily_kline(0, 000001) print(kline.head()) # 获取上证指数最新的实时行情 quote wrapper._execute(get_security_quotes, [(1, 000001)]) print(quote) # 获取深市股票列表第一页 stocks wrapper._execute(get_security_list, 0, 0) print(stocks.head())打印出来的K线DataFrame长这样open high low close volume amount datetime 2023-06-01 10.52 10.78 10.48 10.72 820356.0 8.75e08 2023-06-02 10.73 10.85 10.61 10.80 630276.0 6.79e08 2023-06-05 10.79 11.02 10.75 10.98 980452.0 1.07e09这样的数据结构对pandas来说非常友好后续做移动平均、macd计算、回测都不需要做额外的数据清洗。4.5 全市场股票列表获取与增量更新做选股回测时第一步通常是拿到全市场的股票列表。pytdx的get_security_list一次最多返回1000条而全市场股票数量已远超这个数所以必须分批拉取。def get_all_stocks(wrapper, market0): all_stocks [] start 0 while True: batch wrapper._execute(get_security_list, market, start) if batch is None or len(batch) 0: break all_stocks.append(batch) start len(batch) if len(batch) 1000: break # 控制频次别把服务器打爆 time.sleep(0.05) return pd.concat(all_stocks, ignore_indexTrue)取到列表后建议落一份到本地CSV或SQLite做增量更新。我采用的做法是每天收盘后增量更新一次每次只拉取当日有交易的股票数据与已有数据合并去重后写回存储。4.6 分钟K线拼接从800根限制到全量数据没有接触过pytdx的人大概率遇到过这样一个问题get_security_bars单次能返回的K线数量是800根上限。对日K线来说800根约等于3年多的数据勉强能看但对分钟K线来说800根5分钟K线只覆盖大约7个交易日完全不够用。解决方案是做分段拉取也就是以800根为单位来回翻页。这里有一个比较核心的思路是循环调用多次然后把结果拼接起来。def get_all_kline(wrapper, market, code, category9, total_bars2000): all_bars [] start 0 while start total_bars: count min(800, total_bars - start) bars wrapper._execute(get_security_bars, category, market, code, start, count) if bars is None or len(bars) 0: break all_bars.append(bars) start count time.sleep(0.02) return pd.concat(all_bars, ignore_indexTrue)这里有个细节需要说明分钟K线如果从0开始拉拉的是最近的数据但顺序是最新的在前还是最旧的在前实测下来pytdx返回的数据是时间正序的所以concat不需要额外排序。如果你自己写的时候发现顺序反了可以用df.sort_values(datetime)修正一下。4.7 复权数据计算完整实现复权这块我单独拿出来细说因为它直接影响回测真实度。通达信除权除息数据返回的格式里比较核心的字段是category如1代表除权除息4代表配股和pan每10股送转股数、pei每10股配股数、peiprice配股价、fenshu每10股派现金额这些。我需要先自己写个解析函数def parse_xdxr(df_xdxr): records [] for _, row in df_xdxr.iterrows(): category int(row[category]) if category 1: # 除权除息 records.append({ date: str(row[datetime])[:8], dividend: float(row[pan]) if row[pan] else 0, shares: float(row[fenshu]) if row[fenshu] else 0, peigujia: float(row[peiprice]) if row[peiprice] else 0, peigu: float(row[pei]) if row[pei] else 0, }) return records然后计算累计复权因子。核心思路是从过去到现在依次累乘def calc_factor(records): records.sort(keylambda x: x[date]) factor 1.0 result [] for rec in records: # 除权因子 1 / (1 送转比例) factor / (1 rec[shares] / 10) # 分红对价格影响若配股则需考虑配股价 result.append((rec[date], factor)) return result有了每天的复权因子前复权价的计算就是close * latest_factor / factor_on_that_day。封装里我把这套逻辑写成了get_adjusted_kline方法直接返回前复权K线省去下游二次处理。5. 常见问题与排查技巧实录在使用pytdx和封装工具的过程中我踩过不少坑。整理成速查表方便大家针对性地定位问题。现象根因解决方案连接成功但get_security_bars返回None股票代码与市场代码不匹配核对市场代码0深、1沪连接超时或ConnectionResetError单连接长时间高频请求使用连接池、限制请求频率、增加重试获取股票列表数量不完整未分页拉取用循环按1000条增量拉取获取分钟K线只有最近几天count限制为800分段拉取后拼接复权价格计算错误除权除息文本解析不全用解析函数统一处理所有类别Python 3.12版本导入pytdx报错库未适配新版本切换到3.8-3.11环境获取财务数据字段错位部分服务器返回的字段顺序有差异切换备用服务器测试5.1 连接不稳定最隐性的坑行情服务器对连接频率有限制。如果你单线程循环请求1000次大概率会在中间某个节点被服务器断开连接。ConnectionResetError就是这种情况的直接信号。处理办法有两条路径一是主动降低请求频率time.sleep(0.05-0.1)是相对安全的二是做自动重连。我在封装里两种都做了正常请求间隔50ms异常时对失效连接做重建。这种做法有个副作用慢。拉一只股票全年日K线可能只要几秒钟但如果拉全市场4000只股票的日K线就需要几分钟。但在免费方案的前提下稳定性比速度优先级更高。5.2 关于“无法获取非深圳市场股票列表”的求助这个问题是我见过问得最多的一个。很多人运行get_security_list(1, 0)时返回空或者报错就以为pytdx不支持上海市场。实际上这里有个历史背景部分早期版本的通达信服务器对上海市场股票列表接口做了调整返回的数据结构有变化。解决思路是优先连接较新的行情服务器同时用market1和market0分别拉取并对比。如果实在不行切换另一个服务器地址测试。备选服务器列表可以在多个社区和项目文档里获取到多试几个总有一个能通。另外注意pytdx的上海市场列表接口返回的是上证A股、B股、指数等混合在一起的数据你需要用security_type字段筛选出想要的类型。我封装时默认过滤掉指数、基金等品种只保留普通股。5.3 财务数据字段错位问题get_finance_info返回的字段比较多包括总股本、流通股本、每股净资产、每股收益、营业收入等几十个。我在对比多个服务器返回结果时发现不同服务器的字段顺序并不完全相同。如果直接按位置索引取值就会出现张冠李戴。解决方案是不要用位置索引而是用字段名访问。pytdx返回的是list[dict]结构天然可以通过row[liutongguben]这种方式安全取值。封装时统一转成DataFrame用列名操作从根上消除这个问题。6. 总结以一个实战技巧收尾最后分享一个我实际使用中的小技巧虽然很多人直接在本地跑行情拉取脚本但更建议把它封装成一个小型HTTP服务或定时任务每天早上9点前自动拉取昨日全市场日K线并写入SQLite或ClickHouse。这样回测的时候直接从本地数据库读取速度远快于每次实时拉取。我现在的流程就是定时任务调用这个封装工具把数据落库然后回测框架只跟数据库会话。如果后续想把工具做得更强可以考虑增加一个增量更新模块让本地数据库与服务器实时同步。但目前这个方案下日级更新的频率已经覆盖了我大部分使用场景。先把核心封装做好其他的都是锦上添花。本文还有配套的精品资源点击获取 基于pytdx封装通达信行情数据接口：从连接到复权的完整方案
简介这是一套面向金融数据分析从业者与Python进阶开发者的通达信数据接入工具旨在降低pytdx原生库的使用门槛解决手动连接行情服务器、解析二进制数据格式、管理多市场配置等复杂问题。资源包共203个文件体量54.79MB涵盖79个Python脚本实现行情获取、K线合成、板块分类、指标计算等核心逻辑、33个.dat数据文件含标准通达信格式的L2快照与历史日线、19个.cfg配置文件如tdxzs.cfg、tdxhy.cfg等预置A股、期货、港股等市场参数、19个.md文档含快速上手指南、API说明与调用示例以及Dockerfile、Makefile等工程化支持文件。已有477人学习下载。用户可直接复用封装后的统一接口读取实时/历史行情快速构建量化策略原型完整目录结构体现模块化设计思想便于二次开发与本地化适配配套配置文件与标准化命名规范显著提升多源数据管理效率。 这阵子有朋友问我手里有通达信客户端数据也能看但想写点程序自动拉行情、做选股回测量一大就卡壳。这其实是很多做量化、做数据分析的人都会撞上的墙卡在“数据源”这一关。我当时解决这个问题的思路很直接找一个能直接跟通达信行情服务器对话的Python库把它复杂的地方封装掉做成自己团队随取随用的数据接口。今天就把这套基于pytdx的源码封装工具的设计思路、核心细节和踩坑过程完整拆出来希望能让同样卡在这堵墙前的朋友少走弯路。先说结论pytdx没有官方文档但协议本身不复杂难的是把连接管理、容错、数据格式统一这些脏活累活做好。全文围绕这个库展开适合刚接触Python量化、想做行情数据落地的读者。我会从为什么要封装讲起再到核心API的使用、封装架构设计、常见兼容性问题排查一层层扒开。1. 项目背景与整体设计思路要讲清楚这个封装工具为什么存在得先回到一个原始问题行情数据从哪里来市面上要么是免费接口但字段不全、要么是商业数据源价格不低、要么是抓取网页接口但极其不稳定。剩下的一个选择就是自己直连行情服务器。1.1 为什么选择pytdx而不是其他数据方案我在选型时只考虑一个问题谁的数据最接近“原生、完整、无中间加工”答案很明确通达信行情服务器。它是国内股票行情数据最主流的下游分发端之一数据覆盖沪深京A股、B股、基金、债券、期权等。而pytdx是纯Python实现的通达信行情协议客户端没有额外dll依赖直接解析底层数据包。对比几个常见思路方案优点缺点直接抓取网页接口无需额外依赖字段有限、封IP、接口随版本变动pandas-datareader代码简单数据源在国外、时延高、字段无法定制商业量化数据源数据干净稳定成本高、部署重pytdx数据直连、字段全、实时性高需自行封装、协议无官方文档选择pytdx的核心逻辑在于它把最难的“协议编码/解码”部分处理掉了让我可以把精力放在上层封装、数据格式化和容错处理上。但同时它也把另一块难题留给了我——连接不稳定、断线重连、增量更新、复权处理这些事不自己写就没人帮你写。1.2 这个封装工具到底解决了什么问题直接裸用pytdx写业务代码你会遇到几个非常现实的问题每次获取数据都要手动创建连接用完还得记得关闭漏了就会积攒大量TIME_WAIT连接。服务端动不动就断开连接代码里全是重复的异常捕获和重连逻辑。不同接口返回的字段命名、类型不一致有的返回str有的返回float有的返回int清洗工作重复且容易出错。hq、ex扩展行情两套API使用方式不同业务代码需要分别适配切换成本高。这个封装工具要做的不是重新发明pytdx而是在它上面做一层真正的“数据服务层”统一管理连接池、统一错误重试、统一数据格式输出、提供简洁的调用入口。让使用方写一行代码就能拿到干净的DataFrame。2. pytdx核心功能拆解与原理在动手封装之前我必须先把pytdx的底层能力摸清楚。pytdx分为两套APITdxHq_API标准行情协议和TdxExHq_API扩展行情协议很多人只用过前者用后者的人少但恰恰是后者能补齐不少数据盲区。2.1 标准协议TdxHq_API这套API是绝大多数人的入门选择它会创建一个连接到通达信标准行情服务器的客户端支持的方法包括证券列表、K线、分时、分笔、财务、除权信息等。先看两个基础用法我直接写出最核心的部分from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): # 获取股票列表 stocks api.get_security_list(0, 0) # 获取日K线 bars api.get_security_bars(9, 0, 000001, 0, 100) # 获取分时数据 ticks api.get_minute_time_data(0, 000001) # 获取财务数据 finance api.get_finance_info(0, 000001) # 获取除权除息信息 xdxr api.get_xdxr_info(0, 000001)这段代码看起来简单实际上里面有几个关键参数不是字面上那么好理解的。get_security_list的第二个参数是起始位置传入0代表从第0条开始。这个方法一次最多返回1000条记录所以A股5000多只股票需要分6次拉取用循环控制偏移量。get_security_bars的第一个参数category9代表日K线0代表5分钟K线1代表15分钟K线2代表30分钟K线3代表1小时K线4代表日K线和9几乎一样但周期定义略有差异7代表1分钟K线8代表周K线。第二个参数是市场代码0是深圳1是上海。第三个参数是股票代码注意是字符串。第四个参数是起始位置第五个参数是最大数量一次最多800根。get_xdxr_info返回的是除权除息数据在做复权计算时是必需的但它有个坑它返回的是原始字符串数值比如10派1.5元这种文本需要自己解析成可计算的数值。2.2 扩展协议TdxExHq_API扩展协议设计的初衷是获取期货、外盘、港股等非A股数据。我当时一度忽略了这个模块直到需要做全市场扫描时才发现它的用处它支持获取板块内的成分股列表这是标准API里没有的功能。from pytdx.exhq import TdxExHq_API ex_api TdxExHq_API() with ex_api.connect(221.231.141.60, 7727): # 获取板块数量 block_count ex_api.get_block_count() # 获取板块列表 blocks ex_api.get_block_info(0, 100) # 获取板块内成分股 stocks ex_api.get_security_list(0, 0)扩展协议的服务器端口一般是7727它返回的板块信息包括板块代码、板块名称、板块类型等。做行业轮动分析时这个接口比维护一份静态板块映射文件靠谱得多因为它是实时拉取的最新数据。2.3 连接管理为什么需要连接池pytdx设计上是一个连接对应一个会话如果你的程序需要高频访问行情接口频繁开合连接的开销是不可忽略的。我测试过新建一个连接平均耗时300-500ms而真正获取一次K线数据只需要20ms。如果业务代码每次调用都重新连接效率会慢到一个不可接受的程度。封装工具的核心就是做一个长连接池。启动时预先建立N个连接调用方从池中借用、使用、归还。连接断开时自动重建。我在这上面吃过亏一开始只建了一个连接程序里多线程调用时把同一个连接同时执行多个请求导致数据包错乱。后来改成池化之后这个现象彻底消失了。3. 源码封装工具的分层架构封装工具用到的设计思路并不复杂可以概括为“入口统一、数据格式化、连接自动管理”。我把它拆成了四个层级。3.1 接口层统一入口设计与函数清单这层负责把pytdx各种方法映射成统一风格、统一参数、返回DataFrame的接口。调用方不需要知道底层是标准协议还是扩展协议只要调get_daily_kline(000001, 2020-01-01, 2024-12-31)就能拿到预期数据。我定义的几个核心方法方法名底层实现返回内容get_daily_klineget_security_bars(category9)日K线DataFrameget_minute_klineget_security_bars(category0/1/2/3/7)分钟K线DataFrameget_realtime_quoteget_security_quotes实时五档行情get_trade_detailget_transaction_data分笔成交明细get_finance_summaryget_finance_info财务指标摘要get_dividend_infoget_xdxr_info除权除息列表get_stock_listget_security_list全市场股票列表get_block_stocksTdxExHq_API指定板块成分股每个方法内部统一做四件事拿到连接、请求数据、解析成DataFrame、返回。如果是失败交给你重试层处理。3.2 配置层服务器列表管理与动态切换通达信行情服务器不止一个而且不同服务器在不同网络环境下的连通性差异很大。有的快得飞起有的连上就断。所以封装工具必须有一个服务器列表配置和健康检查机制。我在配置文件中维护了一份服务器列表每个服务器包含IP、端口、协议类型hq/ex、区域华南/华北/华东等启动时对所有服务器做一次ping和连接测试按延迟从低到高排序优先使用低延迟的。配置文件的格式如下servers: - host: 119.147.212.81 port: 7709 protocol: hq region: 深圳 - host: 123.125.108.14 port: 7709 protocol: hq region: 北京 - host: 221.231.141.60 port: 7727 protocol: ex region: 上海自动切换的逻辑是如果连接某个服务器连续失败三次就把它标记为不可用换到下一个每隔一段时间做一次健康检查把恢复的服务器重新加入候选池。这是解决“网络环境变化导致连接不上”这个老大难问题的最有效手段。3.3 数据模型层统一字段名与类型转换pytdx返回的数据在可读性方面很原始具体表现是字段名极其精简比如open、high、low、close这是英文缩写还好但有的接口返回的是price、vol这种含义模糊的字段。另外还有下面这些问题返回的字段类型不一致vol有时是int有时是str。价格和时间戳精度不统一有的接口返回秒级时间戳有的返回毫秒级。股票代码格式不统一有的返回000001有的返回SZ000001。封装层要做的第一件事就是把这些数据全部转成标准格式def normalize_stock_code(code, marketNone): code str(code).strip().zfill(6) if market is None: market 0 if code.startswith((0, 2, 3)) else 1 return pd.Series({code: code, market: market})所有K线数据统一转成datetime格式的索引、float格式的价格、int格式的成交量字段名统一为open/high/low/close/volume/amount。这样下游不管做什么分析都不需要再操心“这个字段是不是字符串”“这个量是手还是股”这种琐碎问题。3.4 工具层复权因子计算与K线合成这块是整个封装里我认为技术含量最高的部分。通达信原生接口返回的K线数据是不复权的。如果直接用不复权数据做回测遇到除权除息日价格会出现跳空导致技术指标计算失真。所以封装必须提供复权能力。复权因子的计算依赖除权除息数据。思路是先获取该股票从上市到现在的所有除权除息记录解析出“每10股派X元”“每10股送Y股”这些信息换算出一个累计复权因子序列然后在K线数据上做前复权或后复权。计算公式很直接后复权价 不复权价 × 累计复权因子前复权价 不复权价 × 最新复权因子 / 当日复权因子封装工具里我专门写了adjust_price_factor方法输入是原始K线DataFrame和除权除息DataFrame输出是带复权因子的完整DataFrame。这个逻辑看起来简单但一旦遇到多次分红送股、公积金转增股本手工计算的出错率极高交给代码处理是唯一可靠的方式。4. 核心实操过程与关键代码解析理论讲再多最终还是要落地跑起来。下面我按完整流程走一遍从安装到核心功能跑通每一步都附上代码和必要的解释。4.1 环境准备与安装安装pytdx很简单用pip就可以pip install pytdx它会自动安装依赖的pandas和numpy这两个是数据处理的底座。如果你已经有Python环境这一步基本不会有问题。我实测的是Python 3.8-3.11都能正常运行这个库。一个容易踩的坑是Python 3.12以上版本因为pytdx源码中用到了一些旧版datetime库的写法在新版本Python上可能会报模块迁移警告。如果遇到建议建一个3.10的虚拟环境使用更省心。4.2 建立第一个行情连接直接写一个最小的示例验证环境是否正常from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): # 获取上证指数最新行情 quote api.get_security_quotes([(1, 000001)]) print(quote)如果连接成功会返回一个包含最新价的字典。这个(1, 000001)是一个元组第一个元素是市场代码1表示上海0表示深圳。很多人第一次接触时会弄反结果拿到的是“平安银行”的数据而不是“上证指数”的因为000001在深圳市场是平安银行在上海市场是上证指数。验证成功后就可以正式使用封装工具了。下面我把整个封装工具的核心代码按模块拆出来讲。4.3 核心封装类TdxDataWrapper我做了个小设计把整个连接池和重试逻辑收敛到一个类里对外暴露的方法是直接可用的业务方法。这个类是整个封装工具的地基。import time import random import pandas as pd from pytdx.hq import TdxHq_API from pytdx.exhq import TdxExHq_API class TdxDataWrapper: def __init__(self, server_list, max_connections3): self.server_list server_list self.max_connections max_connections self._connections [] self._conn_index 0 self.max_retry 3 def _get_connection(self): if not self._connections: self._create_connections() conn self._connections[self._conn_index % len(self._connections)] self._conn_index 1 if not conn[connected]: conn[api] self._create_api(conn[server]) conn[connected] True return conn[api] def _create_connections(self): for i in range(self.max_connections): server self.server_list[i % len(self.server_list)] api self._create_api(server) self._connections.append({server: server, api: api, connected: True}) staticmethod def _create_api(server): if server[protocol] ex: api TdxExHq_API() else: api TdxHq_API() api.connect(server[host], server[port]) return api def _execute(self, func, *args, **kwargs): last_exc None for attempt in range(self.max_retry): try: api self._get_connection() result getattr(api, func)(*args, **kwargs) if result is None: raise ValueError(返回结果为空可能连接已失效) return result except Exception as exc: last_exc exc # 标记连接失效下次重新创建 self._connections[self._conn_index % len(self._connections)][connected] False time.sleep(0.5 * (attempt 1)) raise last_exc def get_daily_kline(self, market, code, start0, count800): df self._execute(get_security_bars, 9, market, code, start, count) return self._format_kline(df) staticmethod def _format_kline(df): df df.rename(columns{ datetime: datetime, open: open, high: high, low: low, close: close, vol: volume, amount: amount }) df[datetime] pd.to_datetime(df[datetime]) df df.set_index(datetime) for col in [open, high, low, close]: df[col] df[col].astype(float) df[volume] df[volume].astype(float) return df这个类解决的最关键问题就是连接状态管理。_execute方法内部会捕获所有异常包括网络异常、返回空、超时等并自动重连重试。同时它实现了简单的连接池轮询round-robin把请求分散到不同连接上避免单连接过热。4.4 验证封装后的调用效果封装后业务代码调用变成了这个样子wrapper TdxDataWrapper(server_list) # 获取平安银行近800个交易日的日K线 kline wrapper.get_daily_kline(0, 000001) print(kline.head()) # 获取上证指数最新的实时行情 quote wrapper._execute(get_security_quotes, [(1, 000001)]) print(quote) # 获取深市股票列表第一页 stocks wrapper._execute(get_security_list, 0, 0) print(stocks.head())打印出来的K线DataFrame长这样open high low close volume amount datetime 2023-06-01 10.52 10.78 10.48 10.72 820356.0 8.75e08 2023-06-02 10.73 10.85 10.61 10.80 630276.0 6.79e08 2023-06-05 10.79 11.02 10.75 10.98 980452.0 1.07e09这样的数据结构对pandas来说非常友好后续做移动平均、macd计算、回测都不需要做额外的数据清洗。4.5 全市场股票列表获取与增量更新做选股回测时第一步通常是拿到全市场的股票列表。pytdx的get_security_list一次最多返回1000条而全市场股票数量已远超这个数所以必须分批拉取。def get_all_stocks(wrapper, market0): all_stocks [] start 0 while True: batch wrapper._execute(get_security_list, market, start) if batch is None or len(batch) 0: break all_stocks.append(batch) start len(batch) if len(batch) 1000: break # 控制频次别把服务器打爆 time.sleep(0.05) return pd.concat(all_stocks, ignore_indexTrue)取到列表后建议落一份到本地CSV或SQLite做增量更新。我采用的做法是每天收盘后增量更新一次每次只拉取当日有交易的股票数据与已有数据合并去重后写回存储。4.6 分钟K线拼接从800根限制到全量数据没有接触过pytdx的人大概率遇到过这样一个问题get_security_bars单次能返回的K线数量是800根上限。对日K线来说800根约等于3年多的数据勉强能看但对分钟K线来说800根5分钟K线只覆盖大约7个交易日完全不够用。解决方案是做分段拉取也就是以800根为单位来回翻页。这里有一个比较核心的思路是循环调用多次然后把结果拼接起来。def get_all_kline(wrapper, market, code, category9, total_bars2000): all_bars [] start 0 while start total_bars: count min(800, total_bars - start) bars wrapper._execute(get_security_bars, category, market, code, start, count) if bars is None or len(bars) 0: break all_bars.append(bars) start count time.sleep(0.02) return pd.concat(all_bars, ignore_indexTrue)这里有个细节需要说明分钟K线如果从0开始拉拉的是最近的数据但顺序是最新的在前还是最旧的在前实测下来pytdx返回的数据是时间正序的所以concat不需要额外排序。如果你自己写的时候发现顺序反了可以用df.sort_values(datetime)修正一下。4.7 复权数据计算完整实现复权这块我单独拿出来细说因为它直接影响回测真实度。通达信除权除息数据返回的格式里比较核心的字段是category如1代表除权除息4代表配股和pan每10股送转股数、pei每10股配股数、peiprice配股价、fenshu每10股派现金额这些。我需要先自己写个解析函数def parse_xdxr(df_xdxr): records [] for _, row in df_xdxr.iterrows(): category int(row[category]) if category 1: # 除权除息 records.append({ date: str(row[datetime])[:8], dividend: float(row[pan]) if row[pan] else 0, shares: float(row[fenshu]) if row[fenshu] else 0, peigujia: float(row[peiprice]) if row[peiprice] else 0, peigu: float(row[pei]) if row[pei] else 0, }) return records然后计算累计复权因子。核心思路是从过去到现在依次累乘def calc_factor(records): records.sort(keylambda x: x[date]) factor 1.0 result [] for rec in records: # 除权因子 1 / (1 送转比例) factor / (1 rec[shares] / 10) # 分红对价格影响若配股则需考虑配股价 result.append((rec[date], factor)) return result有了每天的复权因子前复权价的计算就是close * latest_factor / factor_on_that_day。封装里我把这套逻辑写成了get_adjusted_kline方法直接返回前复权K线省去下游二次处理。5. 常见问题与排查技巧实录在使用pytdx和封装工具的过程中我踩过不少坑。整理成速查表方便大家针对性地定位问题。现象根因解决方案连接成功但get_security_bars返回None股票代码与市场代码不匹配核对市场代码0深、1沪连接超时或ConnectionResetError单连接长时间高频请求使用连接池、限制请求频率、增加重试获取股票列表数量不完整未分页拉取用循环按1000条增量拉取获取分钟K线只有最近几天count限制为800分段拉取后拼接复权价格计算错误除权除息文本解析不全用解析函数统一处理所有类别Python 3.12版本导入pytdx报错库未适配新版本切换到3.8-3.11环境获取财务数据字段错位部分服务器返回的字段顺序有差异切换备用服务器测试5.1 连接不稳定最隐性的坑行情服务器对连接频率有限制。如果你单线程循环请求1000次大概率会在中间某个节点被服务器断开连接。ConnectionResetError就是这种情况的直接信号。处理办法有两条路径一是主动降低请求频率time.sleep(0.05-0.1)是相对安全的二是做自动重连。我在封装里两种都做了正常请求间隔50ms异常时对失效连接做重建。这种做法有个副作用慢。拉一只股票全年日K线可能只要几秒钟但如果拉全市场4000只股票的日K线就需要几分钟。但在免费方案的前提下稳定性比速度优先级更高。5.2 关于“无法获取非深圳市场股票列表”的求助这个问题是我见过问得最多的一个。很多人运行get_security_list(1, 0)时返回空或者报错就以为pytdx不支持上海市场。实际上这里有个历史背景部分早期版本的通达信服务器对上海市场股票列表接口做了调整返回的数据结构有变化。解决思路是优先连接较新的行情服务器同时用market1和market0分别拉取并对比。如果实在不行切换另一个服务器地址测试。备选服务器列表可以在多个社区和项目文档里获取到多试几个总有一个能通。另外注意pytdx的上海市场列表接口返回的是上证A股、B股、指数等混合在一起的数据你需要用security_type字段筛选出想要的类型。我封装时默认过滤掉指数、基金等品种只保留普通股。5.3 财务数据字段错位问题get_finance_info返回的字段比较多包括总股本、流通股本、每股净资产、每股收益、营业收入等几十个。我在对比多个服务器返回结果时发现不同服务器的字段顺序并不完全相同。如果直接按位置索引取值就会出现张冠李戴。解决方案是不要用位置索引而是用字段名访问。pytdx返回的是list[dict]结构天然可以通过row[liutongguben]这种方式安全取值。封装时统一转成DataFrame用列名操作从根上消除这个问题。6. 总结以一个实战技巧收尾最后分享一个我实际使用中的小技巧虽然很多人直接在本地跑行情拉取脚本但更建议把它封装成一个小型HTTP服务或定时任务每天早上9点前自动拉取昨日全市场日K线并写入SQLite或ClickHouse。这样回测的时候直接从本地数据库读取速度远快于每次实时拉取。我现在的流程就是定时任务调用这个封装工具把数据落库然后回测框架只跟数据库会话。如果后续想把工具做得更强可以考虑增加一个增量更新模块让本地数据库与服务器实时同步。但目前这个方案下日级更新的频率已经覆盖了我大部分使用场景。先把核心封装做好其他的都是锦上添花。本文还有配套的精品资源点击获取