简介本资源是一套面向金融数据分析开发者与量化交易爱好者的Python通达信数据接口实现方案解决通达信本地行情数据难以直接接入Python生态的技术痛点适用于股票历史/实时行情获取、因子计算、策略回测等场景。压缩包共29个文件含24个Python源文件涵盖行情获取、K线解析、公司信息提取、除权除息处理等核心模块、2个XML配置文件用于协议参数与服务端地址管理、2个文本文件含变更日志与基础说明及1个.gitignore整体仅65KB轻量易集成。已有1744人学习下载代码结构清晰模块职责分明如hqget.py负责行情请求调度parser目录下各py文件专注不同数据类型的解析base.py与log.py提供底层通信与日志支持。读者可直接复用其网络通信封装、数据解码逻辑与缓存机制快速构建自己的量化分析流水线。 做量化的朋友应该都有这个体会看盘、复盘用通达信确实顺手但真要把数据拿来做策略回测、跑机器学习模型立马就卡住了。行情软件里看得见的数据想批量导出来用Python处理要么手动导出Excel累死人要么去接券商的收费接口要么就得绕一圈爬网页。我折腾了很久之后干脆自己动手写了一套基于Python的通达信数据接口把本地行情文件解析、实时行情获取、复权计算这些事都封装成统一的调用方式。这篇文章就把这套接口的设计思路、核心源码和踩坑记录完整拆出来分享给同样被数据折腾过的朋友。先说清楚这套东西到底能干什么它相当于在你本地行情终端和Python之间架了一座桥让pandas、NumPy、回测框架、可视化库都能直接拿到结构化的行情数据。不管你是想跑一个双均线策略还是想做全市场选股扫描又或者只是想快速把某只股票的历史K线拉出来画个图这套接口都能直接服务。适合的人群很明确有一定Python基础、熟悉通达信操作、正在被数据获取问题卡住的个人量化研究者。我在这里分享的是自己实际在用的方案不是那种只能跑通Demo的教学代码。里面包括了文件格式解析的细节、pytdx实时通道的封装、前复权因子计算还有一堆文档里不会写的坑。照着这套设计去搭你不仅能拿到数据还能真正理解这些数据是怎么从通达信里流出来的。1. 整体设计与技术选型为什么“两条腿走路”最稳1.1 数据获取的三种主流方案对比在动手写代码之前必须先想清楚数据从哪来。市面上常见的通达信数据获取方案大致有三种直接解析本地数据文件、用pytdx库连接行情服务器、通过通达信的DLL接口做扩展。本地文件解析是最传统的方式。通达信软件会把行情数据落地到安装目录下的文件里日线、5分钟线都有固定的二进制格式。这种方式的好处是数据完整、读取速度快、不依赖网络缺点是你得先让通达信把数据下载完而且数据文件格式会随版本变化。pytdx是Python生态里很成熟的通达信协议库它直接实现了通达信的通信协议不需要本地安装通达信也能连上行情服务器拿数据。优点是部署灵活、支持历史K线和实时行情缺点是免费服务器有时候不稳定高频调用容易被限流。通达信DLL扩展则是另外一条路。通达信的公式系统允许通过DLL调用外部函数你可以在DLL里写逻辑去调Python接口或者反过来用Python去调通达信内部的数据。这条路适合做指标级的数据交换但开发门槛高调试也麻烦。1.2 我最终选择的“双通道缓存”架构三种方案都有各自的价值但我实际使用后发现单一方案根本不够用。实时选股和盘中预警需要pytdx这样的网络通道因为本地文件在盘中不会频繁刷新但做历史回测时需要大量、稳定的历史K线数据这时候本地文件解析反而更可靠因为数据已经完整落盘不会有网络波动。所以我最终的架构是“双通道缓存”。pytdx负责实时行情和当日数据本地文件解析负责历史数据和分钟线两者共用一个数据缓存层。上层策略代码只面向一个统一的接口不管数据来自哪里返回的都是标准化的pandas DataFrame。这个架构图在代码里体现得很直接一个DataCenter类作为门面内部根据数据的时间范围和数据粒度自动决定走哪条通道。比如你请求某只股票最近三天的5分钟线它直接走pytdx你请求过去一年的日线它优先读本地文件如果本地文件不存在或数据不完整再退回pytdx补数据。这样设计的好处有三个。第一是冗余一条通道挂了另一条还能顶上第二是效率历史数据走本地文件实时数据走网络各自干自己最擅长的事第三是代码解耦以后哪怕换了数据源上层的策略逻辑一行都不用改。2. 核心细节解析文件格式、复权计算与分钟线补全2.1 通达信本地数据文件格式完全拆解通达信的日线文件一般存放在安装目录下的vipdoc或T0002子目录中按市场分为sh、sz、bj等文件夹文件名就是股票代码加上扩展名比如sh600000.day。每条日线记录固定占32字节字段排列顺序是日期、开盘价、最高价、最低价、收盘价、成交额、成交量、保留字段。价格字段在文件里是整数实际价格要除以100也就是文件里存的其实是“价格乘以100”后的值。成交额是浮点数单位是元成交量是整数单位是股。这里有个容易踩的坑通达信软件界面上显示的成交量单位可能是手但文件里存的是股换算的时候要记得除以100。分钟线文件的格式类似5分钟线在T0002目录下的fzline或new_pd目录里文件名类似sh600000.lc5每条记录同样是32字节。字段包括日期、时间、开盘、最高、最低、收盘、成交额、成交量其中时间字段是整数格式是HHMM比如930代表9点30分。用Python的struct模块解析这类二进制文件非常方便。核心代码就一段import struct import pandas as pd from datetime import datetime def read_day_file(filepath): records [] with open(filepath, rb) as f: while True: chunk f.read(32) if len(chunk) 32: break date, open_, high, low, close, amount, volume, _ struct.unpack(IIIIIfII, chunk) records.append({ date: datetime.strptime(str(date), %Y%m%d), open: open_ / 100.0, high: high / 100.0, low: low / 100.0, close: close / 100.0, amount: amount, volume: volume }) df pd.DataFrame(records) df.set_index(date, inplaceTrue) return df这段代码不是空谈它可以直接跑。唯一需要注意的是struct.unpack的格式字符串我用的是IIIIIfII对应两个无符号整数加一个浮点再加两个无符号整数正好32字节。不同版本的通达信可能存在细微差异解析前最好用十六进制编辑器看一条记录验证一下。2.2 前复权与后复权回测数据必须跨过的坎拿到原始行情数据后直接拿去回测会出大问题。原因是A股市场经常有分红送股除权除权当天股价会跳空不复权的数据在除权日会出现一个明显的价格断层。如果策略信号依赖价格均线或波动率这种断层会造成大量假信号。处理除权跳空的标准做法是计算复权因子。前复权以当前价格为基准把历史价格按比例缩小后复权以最早价格为基准把后来的价格放大。回测里最常用的是前复权因为它的价格序列和软件里看到的当前价格一致。复权因子的计算需要除权除息数据也就是送股、配股、分红、配股价这些信息。pytdx的get_xdxr_info接口可以直接拿到这些数据。拿到之后每一期的因子按下面公式计算factor (close_before - dividend price_of_rights * ratio_of_rights) / (close_before * (1 ratio_of_splits ratio_of_rights))然后用这个因子做累计连乘得到每一日的前复权因子。用pytdx获取除权信息的核心代码如下from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): xdxr api.get_xdxr_info(0, 600000) df_xdxr api.to_df(xdxr)拿到除权信息后按日期排序从最早的记录开始逐一计算累计因子再乘到原始价格上就能得到前复权价格。这里我强烈建议复权计算的结果一定要缓存下来因为除权信息不会频繁变动没必要每次回测都重复计算很费时间。2.3 分钟线数据不完整时怎么补全和拼接本地文件的分钟线有一个很尴尬的问题通达信默认可能不会把每一根分钟线都落地到本地默认一般是下载5分钟线1分钟线要看具体设置。而且如果你某段时间没打开通达信这段期间的分钟线就断了。这时候就要靠pytdx来补。pytdx的get_history_minute_time_data接口可以获取最近的分钟线但它的限制是一次最多返回800根左右。想拿全一天的1分钟线240根没问题要拿好几天就得分段请求。我的数据补全策略是这样先读本地文件把已有的分钟线加载进来再判断本地数据覆盖到哪个时间点然后从时间点之后用pytdx分段拉取最后做去重合并。合并的关键是按时间戳去重因为本地文件和网络数据在边界上可能重叠一两根。分钟线的字段和日线基本一致唯一多出来的是时间字段解析的时候要特别注意time字段是int型格式是HHMM。直接转成字符串再补零不然会得到像930这样的数字没法直接转datetime。3. 实操过程与核心源码实现3.1 环境准备Python版本、依赖库与初始化配置这套接口对Python版本没有太苛刻的要求3.8及以上都行。核心依赖是pandas、numpy和pytdx这三件套装好就能跑。如果你还要做策略回测可以额外装一个backtrader或vectorbt不过那不是数据接口的必要组件。安装命令很简单pip install pandas numpy pytdx如果是Windows系统我建议把通达信安装在默认路径下这样数据文件的路径不用改。如果你用的是绿色免安装版记得在配置里指定通达信的安装目录。我的代码里会用一个配置文件去管理这些路径避免写死在代码里。# config.py class Config: TDX_PATH rD:\new_tdx # 通达信安装目录 LOCAL_DAY_DIR TDX_PATH r\vipdoc LOCAL_MINUTE_DIR TDX_PATH r\vipdoc CACHE_DIR ./data_cache3.2 本地文件读取模块日线与分钟线统一入口我封装了一个TdxFileReader类专门负责解析本地文件。它对外暴露两个主要方法read_daily(stock_code)和read_minute(stock_code, freq)。股票代码兼容两种格式纯数字600000和带市场前缀sh600000内部统一转换成文件名格式。这里有一个很容易忽略的细节交易日判断。本地文件不会包含非交易日的数据所以不用额外过滤但解析出来的日期索引最好转成DatetimeIndex方便后面做时间切片。另外通达信日线文件的最后一条记录往往是当天未收盘的实时数据盘中读取时需要注意这一点否则容易把半根K线当成完整K线用。class TdxFileReader: def __init__(self, config: Config): self.config config def _resolve_filepath(self, stock_code: str, category: str, ext: str): if stock_code.startswith((sh, sz, bj)): market stock_code[:2] code stock_code[2:] else: code stock_code market sh if code.startswith(6) else sz return f{self.config.LOCAL_DAY_DIR}/{market}/{market}{code}.{ext} def read_daily(self, stock_code: str) - pd.DataFrame: filepath self._resolve_filepath(stock_code, day, day) if not os.path.exists(filepath): raise FileNotFoundError(f{filepath} not found, please download data in TDX first) return read_day_file(filepath) def read_minute(self, stock_code: str, freq: str 5) - pd.DataFrame: ext lc5 if freq 5 else lc1 filepath self._resolve_filepath(stock_code, min, ext) if not os.path.exists(filepath): raise FileNotFoundError(f{filepath} not found) return read_minute_file(filepath, freq)3.3 pytdx通道封装历史K线、实时快照、财务数据一把抓pytdx相当于一个Python版的通达信客户端它直接和行情服务器通信速度很快。我封装了一个TdxApi类把常用的操作包起来对外提供get_kline、get_realtime、get_finance这三个方法。get_kline方法用于获取历史K线内部自动处理分页。pytdx的get_security_bars接口一次最多返回800根K线如果要获取更长的历史需要按照日期顺序逐段拼接。封装后上层只管传股票代码、周期、起始日期和结束日期返回的就是规整的DataFrame。from pytdx.hq import TdxHq_API class TdxApi: def __init__(self, host119.147.212.81, port7709): self.host host self.port port self.api TdxHq_API() self.connected False def connect(self): self.api.connect(self.host, self.port) self.connected True def close(self): if self.connected: self.api.disconnect() self.connected False def get_kline(self, market, code, category9, start0, count800): # category: 9日线, 05分钟, 115分钟, 230分钟, 360分钟 if not self.connected: self.connect() data self.api.get_security_bars(category, market, code, start, count) return self.api.to_df(data) def get_realtime(self, market, code): if not self.connected: self.connect() data self.api.get_security_quotes([(market, code)]) return self.api.to_df(data) def get_finance(self, market, code): if not self.connected: self.connect() data self.api.get_finance_info(market, code) return self.api.to_df(data)这段代码里有几个关键点要说明。market参数是按市场区分的0代表深圳1代表上海。上证股票代码以6开头对应market1深证0开头对应market0。北交所有单独的代码规则pytdx新版本也有对应支持。get_realtime返回的是实时五档行情在盘中做监控非常有用。3.4 数据中心层统一接口、磁盘缓存与自动回退上层策略不关心数据来自哪里这是这个模块的设计核心。DataCenter类统一暴露get_price方法参数包含股票代码、周期、开始日期、结束日期、是否复权返回的永远是一个价格字段齐全的DataFrame。这个类内部有一个简单的三级数据策略先查磁盘缓存再从本地文件解析最后才走pytdx网络。磁盘缓存用的是pkl或parquet格式读取快也节省内存。每次请求数据时如果缓存目录里已经有对应文件并且文件时间戳比行情日期新就直接读缓存否则重新获取并写入缓存。class DataCenter: def __init__(self, config: Config): self.config config self.file_reader TdxFileReader(config) self.tdx_api TdxApi() os.makedirs(config.CACHE_DIR, exist_okTrue) def get_price(self, stock_code, freqdaily, startNone, endNone, adjustqfq): cache_path f{self.config.CACHE_DIR}/{stock_code}_{freq}_{adjust}.pkl if os.path.exists(cache_path): df pd.read_pickle(cache_path) # 简单判断缓存是否满足时间范围 if df.index.min() pd.Timestamp(start) and df.index.max() pd.Timestamp(end): return df.loc[start:end] if freq daily: try: df self.file_reader.read_daily(stock_code) except FileNotFoundError: df self.tdx_api.get_kline(...) else: try: df self.file_reader.read_minute(stock_code, freq) except FileNotFoundError: df self.tdx_api.fetch_history_minute_data(...) if adjust qfq: df apply_qfq_factor(stock_code, df) df.to_pickle(cache_path) return df.loc[start:end]这里我只是把结构写出来实际使用时还需要补充参数解析、股票代码到market的映射等细节。不过核心思路已经清楚了所有获取数据的复杂性都被封装在里面调用者只需要面对一个简单的get_price函数。3.5 通达信DLL扩展把Python计算能力搬进指标里除了拿数据有些场景需要反过来在通达信的公式系统里调用Python的计算能力。比如你有一个复杂的机器学习模型想直接在通达信指标里面看到模型输出这就需要一个桥梁。通达信的公式系统支持外部DLL调用。你可以用C/C写一个DLL封装一批自定义函数然后在公式里通过“函数名(参数)”的形式调用。如果你想在DLL内部去请求Python服务也有办法DLL可以通过HTTP请求或者共享内存和本机的Python进程通信。这种方案适合把Python训练好的模型参数灌进DLL在盘中实时计算信号。但说实话开发难度不低而且调试不方便。我自己的经验是如果不是特别依赖通达信界面展示自定义指标建议还是把计算全部放在Python侧通达信只负责显示最终结果。比如我可以在Python里算好买卖点后把结果写进一个自定义板块或通过文件传给通达信。3.6 数据安全与合规提醒最后说一个和代码无关但很重要的事。网上有些教程会教你破解通达信的指标密码、绕过公式加密这类操作我不建议碰。一方面这些属于灰色地带可能违反软件使用协议另一方面做量化研究最重要的是数据和逻辑的可验证性你用别人的加密黑盒指标出了问题都不知道该从哪里排查。我在这套体系里的做法是所有数据都来自正规授权渠道本地已安装的软件文件、公开的行情协议和自己运行的计算逻辑每一步都可追溯。这样的体系用起来才踏实出了问题也能自己定位。4. 常见问题与排查技巧实录4.1 高频问题速查表问题现象可能原因解决办法读本地日线文件报文件不存在通达信版本不同数据目录不在vipdoc下搜索*.day文件定位实际目录读出的价格比软件显示大100倍忘记除以100解析时统一用open / 100.0分钟线数据只有最近几天通达信未完整下载历史分钟数据在软件里执行“盘后数据下载”pytdx连接经常超时免费行情服务器不稳定配置多个备用服务器地址自动切换复权后价格出现负数除权因子计算顺序错误检查除权信息是否按日期升序日期索引不是datetime类型解析时没有转换日期格式用pd.to_datetime统一转换盘中读取本地日线最后一条异常本地文件最后的未收盘K线被误读盘中优先走pytdx实时接口4.2 三个让人崩溃的坑文件锁、缓存陈旧与合并重复第一个坑是文件占用。通达信运行的时候会持续写入数据文件如果这时候你用Python去读有可能读到半个记录或者读到还没来得及落盘的数据。解决方法是尽量在非交易时段读本地文件盘中数据走pytdx。实在要盘中读可以先复制一份文件到临时目录再解析这样能避免文件锁问题。第二个坑是缓存陈旧。我最初设计缓存的时候没有考虑“当天数据更新”的情况结果盘中请求数据时缓存里还是昨天的数据导致信号迟钝。后来我加了一个小逻辑当请求的结束日期是当天时强制绕过缓存直接获取最新数据。这样既保证了效率又不会拿旧数据做盘中决策。第三个坑是分钟线拼接重复。pytdx返回的最后几根K线和本地文件可能重叠叠加的时候如果没去重数据里就会出现相同时间戳的两行策略计算均线的时候会莫名其妙多出一根“假K线”。解决方法是合并后按时间戳去重并且保留后获取的版本。4.3 调试技巧用日志记录每个数据请求接口写完以后我强烈建议你加日志。每个数据请求都记录一下来源本地文件/pytdx/缓存、耗时、数据条数这样线上跑策略的时候一旦数据出了问题你能快速定位是哪一层的问题。我用的方案是Python标准库的logging模块配置好输出格式把日志写到文件里。DataCenter的每个get_price调用都会输出一行日志包含股票代码、周期、数据来源、数据条数和耗时。不要小看这个习惯它帮我排查了不少诡异的问题。5. 扩展应用从数据接口到可运行的量化小系统5.1 用这套接口跑一个最简单的双均线回测数据接口的价值最终要落到策略上。我用这套接口写了一个最简单的双均线策略回测代码算是一个完整示例。策略逻辑很经典5日均线上穿20日均线买入下穿卖出全部用收盘价计算信号。def backtest_ma(stock_code, fast5, slow20): dc DataCenter(Config()) df dc.get_price(stock_code, freqdaily, start2020-01-01, end2023-12-31, adjustqfq) df[ma_fast] df[close].rolling(fast).mean() df[ma_slow] df[close].rolling(slow).mean() df[position] 0 df.loc[df[ma_fast] df[ma_slow], position] 1 df[position] df[position].shift(1).fillna(0) df[daily_return] df[close].pct_change().fillna(0) df[strategy_return] df[daily_return] * df[position] df[cum_return] (1 df[strategy_return]).cumprod() return df回测结果跑出来之后你会立刻发现数据质量对最终收益曲线的影响有多大。如果用的不复权数据除权日会直接让策略产生虚假止损信号如果分钟线和日线数据没对齐回测区间里的收益计算就是错的。这些细节通通要归功于前面数据接口的打磨。5.2 盘中实时监控调用pytdx快照做预警数据接口不只服务回测还能做盘中监控。我用get_realtime封装了一个简单的预警脚本每5秒拉取一次自选股的实时快照如果涨幅超过阈值或者成交量突然放大就通过微信推送通知。这个功能跑起来很简单但很实用。需要注意的坑是不要用同一个pytdx连接在高频轮询几分钟内太频繁会被服务器拉黑。我的做法是设置一个合理的轮询间隔多只股票通过一次get_security_quotes批量获取而不是一只一只地请求。5.3 服务化改造把数据接口封装成HTTP服务如果多台机器、多个策略都要用这套数据接口直接在每台机器上各跑一套Python代码显然不合理。我后来把DataCenter封装成了一个FastAPI服务对外暴露一个GET接口输入股票代码和周期返回JSON格式的K线数据。这样其他服务只要发HTTP请求就能拿数据彻底解耦了数据获取逻辑和策略逻辑。服务化之后还有个附带好处可以集中做数据缓存和权限控制所有策略共享同一份数据缓存避免重复拉取数据浪费网络带宽。5.4 后续优化方向这套接口目前已经能覆盖我日常80%的需求但还有一些可以继续优化的地方。比如把缓存从本地磁盘换到数据库SQLite或者ClickHouse支持更多数据频率比如tick数据再比如自动判断交易时段在盘中切换实时通道、盘后自动刷新历史数据。这些都是很自然的演进方向按需去加就行。做了这么久的量化数据工作我最大的体会是数据接口这种事情看似是个工具活但细节极多做得好不好直接决定了后面策略开发的效率。把数据层面彻底打通后面写策略、做回测、上模拟盘都会顺畅很多。本文还有配套的精品资源点击获取
Python通达信数据接口实战：本地文件解析与pytdx双通道架构
简介本资源是一套面向金融数据分析开发者与量化交易爱好者的Python通达信数据接口实现方案解决通达信本地行情数据难以直接接入Python生态的技术痛点适用于股票历史/实时行情获取、因子计算、策略回测等场景。压缩包共29个文件含24个Python源文件涵盖行情获取、K线解析、公司信息提取、除权除息处理等核心模块、2个XML配置文件用于协议参数与服务端地址管理、2个文本文件含变更日志与基础说明及1个.gitignore整体仅65KB轻量易集成。已有1744人学习下载代码结构清晰模块职责分明如hqget.py负责行情请求调度parser目录下各py文件专注不同数据类型的解析base.py与log.py提供底层通信与日志支持。读者可直接复用其网络通信封装、数据解码逻辑与缓存机制快速构建自己的量化分析流水线。 做量化的朋友应该都有这个体会看盘、复盘用通达信确实顺手但真要把数据拿来做策略回测、跑机器学习模型立马就卡住了。行情软件里看得见的数据想批量导出来用Python处理要么手动导出Excel累死人要么去接券商的收费接口要么就得绕一圈爬网页。我折腾了很久之后干脆自己动手写了一套基于Python的通达信数据接口把本地行情文件解析、实时行情获取、复权计算这些事都封装成统一的调用方式。这篇文章就把这套接口的设计思路、核心源码和踩坑记录完整拆出来分享给同样被数据折腾过的朋友。先说清楚这套东西到底能干什么它相当于在你本地行情终端和Python之间架了一座桥让pandas、NumPy、回测框架、可视化库都能直接拿到结构化的行情数据。不管你是想跑一个双均线策略还是想做全市场选股扫描又或者只是想快速把某只股票的历史K线拉出来画个图这套接口都能直接服务。适合的人群很明确有一定Python基础、熟悉通达信操作、正在被数据获取问题卡住的个人量化研究者。我在这里分享的是自己实际在用的方案不是那种只能跑通Demo的教学代码。里面包括了文件格式解析的细节、pytdx实时通道的封装、前复权因子计算还有一堆文档里不会写的坑。照着这套设计去搭你不仅能拿到数据还能真正理解这些数据是怎么从通达信里流出来的。1. 整体设计与技术选型为什么“两条腿走路”最稳1.1 数据获取的三种主流方案对比在动手写代码之前必须先想清楚数据从哪来。市面上常见的通达信数据获取方案大致有三种直接解析本地数据文件、用pytdx库连接行情服务器、通过通达信的DLL接口做扩展。本地文件解析是最传统的方式。通达信软件会把行情数据落地到安装目录下的文件里日线、5分钟线都有固定的二进制格式。这种方式的好处是数据完整、读取速度快、不依赖网络缺点是你得先让通达信把数据下载完而且数据文件格式会随版本变化。pytdx是Python生态里很成熟的通达信协议库它直接实现了通达信的通信协议不需要本地安装通达信也能连上行情服务器拿数据。优点是部署灵活、支持历史K线和实时行情缺点是免费服务器有时候不稳定高频调用容易被限流。通达信DLL扩展则是另外一条路。通达信的公式系统允许通过DLL调用外部函数你可以在DLL里写逻辑去调Python接口或者反过来用Python去调通达信内部的数据。这条路适合做指标级的数据交换但开发门槛高调试也麻烦。1.2 我最终选择的“双通道缓存”架构三种方案都有各自的价值但我实际使用后发现单一方案根本不够用。实时选股和盘中预警需要pytdx这样的网络通道因为本地文件在盘中不会频繁刷新但做历史回测时需要大量、稳定的历史K线数据这时候本地文件解析反而更可靠因为数据已经完整落盘不会有网络波动。所以我最终的架构是“双通道缓存”。pytdx负责实时行情和当日数据本地文件解析负责历史数据和分钟线两者共用一个数据缓存层。上层策略代码只面向一个统一的接口不管数据来自哪里返回的都是标准化的pandas DataFrame。这个架构图在代码里体现得很直接一个DataCenter类作为门面内部根据数据的时间范围和数据粒度自动决定走哪条通道。比如你请求某只股票最近三天的5分钟线它直接走pytdx你请求过去一年的日线它优先读本地文件如果本地文件不存在或数据不完整再退回pytdx补数据。这样设计的好处有三个。第一是冗余一条通道挂了另一条还能顶上第二是效率历史数据走本地文件实时数据走网络各自干自己最擅长的事第三是代码解耦以后哪怕换了数据源上层的策略逻辑一行都不用改。2. 核心细节解析文件格式、复权计算与分钟线补全2.1 通达信本地数据文件格式完全拆解通达信的日线文件一般存放在安装目录下的vipdoc或T0002子目录中按市场分为sh、sz、bj等文件夹文件名就是股票代码加上扩展名比如sh600000.day。每条日线记录固定占32字节字段排列顺序是日期、开盘价、最高价、最低价、收盘价、成交额、成交量、保留字段。价格字段在文件里是整数实际价格要除以100也就是文件里存的其实是“价格乘以100”后的值。成交额是浮点数单位是元成交量是整数单位是股。这里有个容易踩的坑通达信软件界面上显示的成交量单位可能是手但文件里存的是股换算的时候要记得除以100。分钟线文件的格式类似5分钟线在T0002目录下的fzline或new_pd目录里文件名类似sh600000.lc5每条记录同样是32字节。字段包括日期、时间、开盘、最高、最低、收盘、成交额、成交量其中时间字段是整数格式是HHMM比如930代表9点30分。用Python的struct模块解析这类二进制文件非常方便。核心代码就一段import struct import pandas as pd from datetime import datetime def read_day_file(filepath): records [] with open(filepath, rb) as f: while True: chunk f.read(32) if len(chunk) 32: break date, open_, high, low, close, amount, volume, _ struct.unpack(IIIIIfII, chunk) records.append({ date: datetime.strptime(str(date), %Y%m%d), open: open_ / 100.0, high: high / 100.0, low: low / 100.0, close: close / 100.0, amount: amount, volume: volume }) df pd.DataFrame(records) df.set_index(date, inplaceTrue) return df这段代码不是空谈它可以直接跑。唯一需要注意的是struct.unpack的格式字符串我用的是IIIIIfII对应两个无符号整数加一个浮点再加两个无符号整数正好32字节。不同版本的通达信可能存在细微差异解析前最好用十六进制编辑器看一条记录验证一下。2.2 前复权与后复权回测数据必须跨过的坎拿到原始行情数据后直接拿去回测会出大问题。原因是A股市场经常有分红送股除权除权当天股价会跳空不复权的数据在除权日会出现一个明显的价格断层。如果策略信号依赖价格均线或波动率这种断层会造成大量假信号。处理除权跳空的标准做法是计算复权因子。前复权以当前价格为基准把历史价格按比例缩小后复权以最早价格为基准把后来的价格放大。回测里最常用的是前复权因为它的价格序列和软件里看到的当前价格一致。复权因子的计算需要除权除息数据也就是送股、配股、分红、配股价这些信息。pytdx的get_xdxr_info接口可以直接拿到这些数据。拿到之后每一期的因子按下面公式计算factor (close_before - dividend price_of_rights * ratio_of_rights) / (close_before * (1 ratio_of_splits ratio_of_rights))然后用这个因子做累计连乘得到每一日的前复权因子。用pytdx获取除权信息的核心代码如下from pytdx.hq import TdxHq_API api TdxHq_API() with api.connect(119.147.212.81, 7709): xdxr api.get_xdxr_info(0, 600000) df_xdxr api.to_df(xdxr)拿到除权信息后按日期排序从最早的记录开始逐一计算累计因子再乘到原始价格上就能得到前复权价格。这里我强烈建议复权计算的结果一定要缓存下来因为除权信息不会频繁变动没必要每次回测都重复计算很费时间。2.3 分钟线数据不完整时怎么补全和拼接本地文件的分钟线有一个很尴尬的问题通达信默认可能不会把每一根分钟线都落地到本地默认一般是下载5分钟线1分钟线要看具体设置。而且如果你某段时间没打开通达信这段期间的分钟线就断了。这时候就要靠pytdx来补。pytdx的get_history_minute_time_data接口可以获取最近的分钟线但它的限制是一次最多返回800根左右。想拿全一天的1分钟线240根没问题要拿好几天就得分段请求。我的数据补全策略是这样先读本地文件把已有的分钟线加载进来再判断本地数据覆盖到哪个时间点然后从时间点之后用pytdx分段拉取最后做去重合并。合并的关键是按时间戳去重因为本地文件和网络数据在边界上可能重叠一两根。分钟线的字段和日线基本一致唯一多出来的是时间字段解析的时候要特别注意time字段是int型格式是HHMM。直接转成字符串再补零不然会得到像930这样的数字没法直接转datetime。3. 实操过程与核心源码实现3.1 环境准备Python版本、依赖库与初始化配置这套接口对Python版本没有太苛刻的要求3.8及以上都行。核心依赖是pandas、numpy和pytdx这三件套装好就能跑。如果你还要做策略回测可以额外装一个backtrader或vectorbt不过那不是数据接口的必要组件。安装命令很简单pip install pandas numpy pytdx如果是Windows系统我建议把通达信安装在默认路径下这样数据文件的路径不用改。如果你用的是绿色免安装版记得在配置里指定通达信的安装目录。我的代码里会用一个配置文件去管理这些路径避免写死在代码里。# config.py class Config: TDX_PATH rD:\new_tdx # 通达信安装目录 LOCAL_DAY_DIR TDX_PATH r\vipdoc LOCAL_MINUTE_DIR TDX_PATH r\vipdoc CACHE_DIR ./data_cache3.2 本地文件读取模块日线与分钟线统一入口我封装了一个TdxFileReader类专门负责解析本地文件。它对外暴露两个主要方法read_daily(stock_code)和read_minute(stock_code, freq)。股票代码兼容两种格式纯数字600000和带市场前缀sh600000内部统一转换成文件名格式。这里有一个很容易忽略的细节交易日判断。本地文件不会包含非交易日的数据所以不用额外过滤但解析出来的日期索引最好转成DatetimeIndex方便后面做时间切片。另外通达信日线文件的最后一条记录往往是当天未收盘的实时数据盘中读取时需要注意这一点否则容易把半根K线当成完整K线用。class TdxFileReader: def __init__(self, config: Config): self.config config def _resolve_filepath(self, stock_code: str, category: str, ext: str): if stock_code.startswith((sh, sz, bj)): market stock_code[:2] code stock_code[2:] else: code stock_code market sh if code.startswith(6) else sz return f{self.config.LOCAL_DAY_DIR}/{market}/{market}{code}.{ext} def read_daily(self, stock_code: str) - pd.DataFrame: filepath self._resolve_filepath(stock_code, day, day) if not os.path.exists(filepath): raise FileNotFoundError(f{filepath} not found, please download data in TDX first) return read_day_file(filepath) def read_minute(self, stock_code: str, freq: str 5) - pd.DataFrame: ext lc5 if freq 5 else lc1 filepath self._resolve_filepath(stock_code, min, ext) if not os.path.exists(filepath): raise FileNotFoundError(f{filepath} not found) return read_minute_file(filepath, freq)3.3 pytdx通道封装历史K线、实时快照、财务数据一把抓pytdx相当于一个Python版的通达信客户端它直接和行情服务器通信速度很快。我封装了一个TdxApi类把常用的操作包起来对外提供get_kline、get_realtime、get_finance这三个方法。get_kline方法用于获取历史K线内部自动处理分页。pytdx的get_security_bars接口一次最多返回800根K线如果要获取更长的历史需要按照日期顺序逐段拼接。封装后上层只管传股票代码、周期、起始日期和结束日期返回的就是规整的DataFrame。from pytdx.hq import TdxHq_API class TdxApi: def __init__(self, host119.147.212.81, port7709): self.host host self.port port self.api TdxHq_API() self.connected False def connect(self): self.api.connect(self.host, self.port) self.connected True def close(self): if self.connected: self.api.disconnect() self.connected False def get_kline(self, market, code, category9, start0, count800): # category: 9日线, 05分钟, 115分钟, 230分钟, 360分钟 if not self.connected: self.connect() data self.api.get_security_bars(category, market, code, start, count) return self.api.to_df(data) def get_realtime(self, market, code): if not self.connected: self.connect() data self.api.get_security_quotes([(market, code)]) return self.api.to_df(data) def get_finance(self, market, code): if not self.connected: self.connect() data self.api.get_finance_info(market, code) return self.api.to_df(data)这段代码里有几个关键点要说明。market参数是按市场区分的0代表深圳1代表上海。上证股票代码以6开头对应market1深证0开头对应market0。北交所有单独的代码规则pytdx新版本也有对应支持。get_realtime返回的是实时五档行情在盘中做监控非常有用。3.4 数据中心层统一接口、磁盘缓存与自动回退上层策略不关心数据来自哪里这是这个模块的设计核心。DataCenter类统一暴露get_price方法参数包含股票代码、周期、开始日期、结束日期、是否复权返回的永远是一个价格字段齐全的DataFrame。这个类内部有一个简单的三级数据策略先查磁盘缓存再从本地文件解析最后才走pytdx网络。磁盘缓存用的是pkl或parquet格式读取快也节省内存。每次请求数据时如果缓存目录里已经有对应文件并且文件时间戳比行情日期新就直接读缓存否则重新获取并写入缓存。class DataCenter: def __init__(self, config: Config): self.config config self.file_reader TdxFileReader(config) self.tdx_api TdxApi() os.makedirs(config.CACHE_DIR, exist_okTrue) def get_price(self, stock_code, freqdaily, startNone, endNone, adjustqfq): cache_path f{self.config.CACHE_DIR}/{stock_code}_{freq}_{adjust}.pkl if os.path.exists(cache_path): df pd.read_pickle(cache_path) # 简单判断缓存是否满足时间范围 if df.index.min() pd.Timestamp(start) and df.index.max() pd.Timestamp(end): return df.loc[start:end] if freq daily: try: df self.file_reader.read_daily(stock_code) except FileNotFoundError: df self.tdx_api.get_kline(...) else: try: df self.file_reader.read_minute(stock_code, freq) except FileNotFoundError: df self.tdx_api.fetch_history_minute_data(...) if adjust qfq: df apply_qfq_factor(stock_code, df) df.to_pickle(cache_path) return df.loc[start:end]这里我只是把结构写出来实际使用时还需要补充参数解析、股票代码到market的映射等细节。不过核心思路已经清楚了所有获取数据的复杂性都被封装在里面调用者只需要面对一个简单的get_price函数。3.5 通达信DLL扩展把Python计算能力搬进指标里除了拿数据有些场景需要反过来在通达信的公式系统里调用Python的计算能力。比如你有一个复杂的机器学习模型想直接在通达信指标里面看到模型输出这就需要一个桥梁。通达信的公式系统支持外部DLL调用。你可以用C/C写一个DLL封装一批自定义函数然后在公式里通过“函数名(参数)”的形式调用。如果你想在DLL内部去请求Python服务也有办法DLL可以通过HTTP请求或者共享内存和本机的Python进程通信。这种方案适合把Python训练好的模型参数灌进DLL在盘中实时计算信号。但说实话开发难度不低而且调试不方便。我自己的经验是如果不是特别依赖通达信界面展示自定义指标建议还是把计算全部放在Python侧通达信只负责显示最终结果。比如我可以在Python里算好买卖点后把结果写进一个自定义板块或通过文件传给通达信。3.6 数据安全与合规提醒最后说一个和代码无关但很重要的事。网上有些教程会教你破解通达信的指标密码、绕过公式加密这类操作我不建议碰。一方面这些属于灰色地带可能违反软件使用协议另一方面做量化研究最重要的是数据和逻辑的可验证性你用别人的加密黑盒指标出了问题都不知道该从哪里排查。我在这套体系里的做法是所有数据都来自正规授权渠道本地已安装的软件文件、公开的行情协议和自己运行的计算逻辑每一步都可追溯。这样的体系用起来才踏实出了问题也能自己定位。4. 常见问题与排查技巧实录4.1 高频问题速查表问题现象可能原因解决办法读本地日线文件报文件不存在通达信版本不同数据目录不在vipdoc下搜索*.day文件定位实际目录读出的价格比软件显示大100倍忘记除以100解析时统一用open / 100.0分钟线数据只有最近几天通达信未完整下载历史分钟数据在软件里执行“盘后数据下载”pytdx连接经常超时免费行情服务器不稳定配置多个备用服务器地址自动切换复权后价格出现负数除权因子计算顺序错误检查除权信息是否按日期升序日期索引不是datetime类型解析时没有转换日期格式用pd.to_datetime统一转换盘中读取本地日线最后一条异常本地文件最后的未收盘K线被误读盘中优先走pytdx实时接口4.2 三个让人崩溃的坑文件锁、缓存陈旧与合并重复第一个坑是文件占用。通达信运行的时候会持续写入数据文件如果这时候你用Python去读有可能读到半个记录或者读到还没来得及落盘的数据。解决方法是尽量在非交易时段读本地文件盘中数据走pytdx。实在要盘中读可以先复制一份文件到临时目录再解析这样能避免文件锁问题。第二个坑是缓存陈旧。我最初设计缓存的时候没有考虑“当天数据更新”的情况结果盘中请求数据时缓存里还是昨天的数据导致信号迟钝。后来我加了一个小逻辑当请求的结束日期是当天时强制绕过缓存直接获取最新数据。这样既保证了效率又不会拿旧数据做盘中决策。第三个坑是分钟线拼接重复。pytdx返回的最后几根K线和本地文件可能重叠叠加的时候如果没去重数据里就会出现相同时间戳的两行策略计算均线的时候会莫名其妙多出一根“假K线”。解决方法是合并后按时间戳去重并且保留后获取的版本。4.3 调试技巧用日志记录每个数据请求接口写完以后我强烈建议你加日志。每个数据请求都记录一下来源本地文件/pytdx/缓存、耗时、数据条数这样线上跑策略的时候一旦数据出了问题你能快速定位是哪一层的问题。我用的方案是Python标准库的logging模块配置好输出格式把日志写到文件里。DataCenter的每个get_price调用都会输出一行日志包含股票代码、周期、数据来源、数据条数和耗时。不要小看这个习惯它帮我排查了不少诡异的问题。5. 扩展应用从数据接口到可运行的量化小系统5.1 用这套接口跑一个最简单的双均线回测数据接口的价值最终要落到策略上。我用这套接口写了一个最简单的双均线策略回测代码算是一个完整示例。策略逻辑很经典5日均线上穿20日均线买入下穿卖出全部用收盘价计算信号。def backtest_ma(stock_code, fast5, slow20): dc DataCenter(Config()) df dc.get_price(stock_code, freqdaily, start2020-01-01, end2023-12-31, adjustqfq) df[ma_fast] df[close].rolling(fast).mean() df[ma_slow] df[close].rolling(slow).mean() df[position] 0 df.loc[df[ma_fast] df[ma_slow], position] 1 df[position] df[position].shift(1).fillna(0) df[daily_return] df[close].pct_change().fillna(0) df[strategy_return] df[daily_return] * df[position] df[cum_return] (1 df[strategy_return]).cumprod() return df回测结果跑出来之后你会立刻发现数据质量对最终收益曲线的影响有多大。如果用的不复权数据除权日会直接让策略产生虚假止损信号如果分钟线和日线数据没对齐回测区间里的收益计算就是错的。这些细节通通要归功于前面数据接口的打磨。5.2 盘中实时监控调用pytdx快照做预警数据接口不只服务回测还能做盘中监控。我用get_realtime封装了一个简单的预警脚本每5秒拉取一次自选股的实时快照如果涨幅超过阈值或者成交量突然放大就通过微信推送通知。这个功能跑起来很简单但很实用。需要注意的坑是不要用同一个pytdx连接在高频轮询几分钟内太频繁会被服务器拉黑。我的做法是设置一个合理的轮询间隔多只股票通过一次get_security_quotes批量获取而不是一只一只地请求。5.3 服务化改造把数据接口封装成HTTP服务如果多台机器、多个策略都要用这套数据接口直接在每台机器上各跑一套Python代码显然不合理。我后来把DataCenter封装成了一个FastAPI服务对外暴露一个GET接口输入股票代码和周期返回JSON格式的K线数据。这样其他服务只要发HTTP请求就能拿数据彻底解耦了数据获取逻辑和策略逻辑。服务化之后还有个附带好处可以集中做数据缓存和权限控制所有策略共享同一份数据缓存避免重复拉取数据浪费网络带宽。5.4 后续优化方向这套接口目前已经能覆盖我日常80%的需求但还有一些可以继续优化的地方。比如把缓存从本地磁盘换到数据库SQLite或者ClickHouse支持更多数据频率比如tick数据再比如自动判断交易时段在盘中切换实时通道、盘后自动刷新历史数据。这些都是很自然的演进方向按需去加就行。做了这么久的量化数据工作我最大的体会是数据接口这种事情看似是个工具活但细节极多做得好不好直接决定了后面策略开发的效率。把数据层面彻底打通后面写策略、做回测、上模拟盘都会顺畅很多。本文还有配套的精品资源点击获取
