简介本资源是一套面向金融数据分析工程师与量化策略开发者的Python通达信数据接口实现方案解决本地Python环境直连通达信行情服务器、高效获取实时及历史股票数据的技术难题。压缩包共29个文件含24个Python源码覆盖行情获取、K线解析、公司信息查询、除权除息处理等核心模块、2个XML配置文件用于协议参数与服务端地址管理、2个文本文档含变更日志与基础说明及1个.gitignore整体仅65KB轻量易集成。已有1744人学习下载适合具备Python基础并从事股票量化分析、策略回测或交易系统开发的中高级开发者。读者可直接复用完整接口架构快速调用get_security_bars、get_history_transaction_data等20余个标准化方法结合内置日志、缓存与错误处理机制大幅降低通达信数据接入门槛。1. 项目思路与接口方案选型1.1 为什么需要自建数据接口做量化或者技术分析的人第一道坎基本都是数据。我自己最早做回测的时候从通达信客户端手工导K线一天导几百只股票导完还得重新整理格式临时加个股票又要重来。最痛苦的是每天收盘后要重复一遍流程错一次图就废了。后来我才彻底想明白与其在“导出数据”上耗时间不如直接写一个数据接口层把从行情服务器取数、清洗、落库这件事自动化。这个项目要做的事简单说就是用 Python 对接通达信的行情数据源不依赖手工导出也不用打开软件。拿到日线、分钟线、实时报价、股票列表和基础财务信息。整个设计不是一条脚本写完就跑而是按照“接口化”的思路来做上层策略/分析代码只跟接口打交道不关心数据是从哪台服务器、什么协议、什么格式取回来的。这样后期换数据源的时候只需要改最底下的连接层业务逻辑完全不用动。1.2 三种常见取数方案的对比在动手写代码之前我对比过圈子里常见的取数路子大致有三类。方案免费程度数据时效批量能力主要问题pytdx 协议接口免费实时强支持批量需要维护服务器列表偶尔有连接风险通达信本地文件解析免费依赖下载中文件格式私有每次要手动更新本地数据商业数据接口如 Wind / 聚宽收费或限权限实时强年费门槛高不适合个人练手我最后选的是第一类也就是直接用 pytdx 这套开源协议实现。原因是它的数据字段和通达信客户端基本一致足够全而且免费、延迟低。pytdx 不是什么黑科技它就是帮我们省掉了自己手动构造协议包、解析返回字节流的脏活把行情服务器的数据拿回来变成了纯 Python 对象的调用。代价是服务器列表需要自己维护因为通达信行情服务器有公网入口也有一批历史遗留的IP经常出现某几个连着连着就失效的情况。1.3 接口程序的分层设计整个数据接口的源码我拆成了三层来写这样职责清楚后面扩展也方便。连接层负责服务器连接、断线重连、心跳保活。对外只暴露一个“当前是否可用”的状态。数据层负责具体的数据组装比如 K 线、实时报价、股票列表。这里统一做市场代码判断、字段映射、DataFrame 转换。存储层负责把清洗后的数据写进 SQLite 或者直接返回给调用方。这三层之间用最朴素的类和方法来衔接不引入复杂框架。为什么这么设计因为个人项目最重要的是“好维护”。我之前见过很多人的代码把服务器地址、字段映射、SQL 语句全部写在一个巨型函数里表面上跑得通但换一个需求就要动一行改三处。分层之后连接出问题查连接层字段不对查数据层入库出错查存储层问题定位快很多。2. 环境准备与连接管理实现2.1 Python 环境与依赖安装我项目的运行环境是 Python 3.9核心依赖只有两个pytdx 和 pandas。前者负责行情协议后者负责数据处理。装的时候直接用 pip 就行。pip install pytdx pandas如果你用的是 VSCode建议提前配好 Python 环境F1 调出命令面板选Python: Select Interpreter把解释器切换到项目所在的虚拟环境否则后面 pip 装的包容易跟全局环境搞混。这一步虽然基础但踩过的人都知道环境错位的问题比代码问题更难排查。2.2 连接管理类怎么写才算稳pytdx 的用法其实很直接核心是TdxHq_API但如果你每次都手动 connect、手动处理失败代码会越写越乱。我封装了一个连接管理类把服务器列表、重连逻辑和线程安全都放在一起。import threading from pytdx.hq import TdxHq_API class TdxClient: def __init__(self, hostsNone, port7709): # 服务器列表优先放自己实测延迟低的 self.hosts hosts or [ (119.147.212.81, 7709), (114.80.63.12, 7709), (218.108.98.244, 7709), ] self.api TdxHq_API(heartbeatTrue) self.lock threading.Lock() self.connected False def connect(self): for host, port in self.hosts: try: if self.api.connect(host, port): self.connected True print(fconnected to {host}:{port}) return except Exception as e: print(fconnect {host} failed: {e}) continue raise ConnectionError(所有行情服务器连接失败) def close(self): if self.api: self.api.disconnect() self.connected False这里有几个点值得说。第一是heartbeatTrue这个参数pytdx 会自动发心跳包防止长时间空闲被服务器踢掉。第二是加锁因为很多取数场景会用线程池并发拉数据如果不加锁多个线程同时 commit 同一个连接轻则数据错乱重则直接断连。第三是服务器列表要维护成“候补制”第一个连不上就换下一个实测里这个设计救了我很多次。2.3 核心 API 的记忆口诀pytdx 的接口数量不少但真正高频使用的其实就那么几个我列一张表方便你自己写的时候快速定位。方法功能关键参数get_security_bars获取个股K线category、market、code、start、countget_index_bars获取指数K线同上get_security_quotes获取实时行情传入 [(market, code), ...] 列表get_security_list获取股票列表market、startget_security_count获取市场股票数量marketget_finance_info获取财务摘要market、code记忆重点market是市场代码0 代表深圳1 代表上海category是K线周期9 代表日K0 代表5分钟K4 也是日K老版本协议建议统一用 9。count一次最大 800 根超过的部分要靠start参数往前翻。3. 核心数据模块的实现细节3.1 K线获取与 DataFrame 统一封装取K线是整套接口最核心的功能。通达信协议里K线数据结构包含开高低收、成交量、成交额、时间字段。pytdx 返回的是原始 dict 列表我先统一转成 pandas DataFrame再清洗字段。import pandas as pd def get_kline(self, code: str, category: int 9, count: int 800, market: int None): if market is None: market 1 if code.startswith(6) else 0 with self.lock: raw self.api.get_security_bars(category, market, code, 0, count) if not raw: return pd.DataFrame() df self.api.to_df(raw) if df.empty: return df # 统一时间格式把分开的年月日时分秒拼成datetime df[datetime] ( df[year].astype(str) - df[month].astype(str).str.zfill(2) - df[day].astype(str).str.zfill(2) df[hour].astype(str).str.zfill(2) : df[minute].astype(str).str.zfill(2) ) df[code] code df df[[datetime, code, open, close, high, low, vol, amount]] return df代码里的market自动判断值得单独说一句。股票代码以 6 开头是上海其他大多是深圳这是最常用的区分方法。但要注意指数、基金、转债的规则不完全一样如果你要拉指数数据建议显式传market参数不要依赖自动判断。比如上证指数代码是999999虽然首位不是6但市场是上海1。3.2 一次最多取800根怎么突破很多第一次用 pytdx 的人都会遇到一个疑问我明明要2000根日K为什么只返回800根这其实不是 bug而是服务器端的限制。单次请求最多返回 800 根K线想要更长的历史就得用start参数翻页。def get_kline_history(self, code: str, category: int 9, total: int 3000): all_data [] start 0 while start total: with self.lock: raw self.api.get_security_bars(category, market, code, start, min(800, total - start)) if not raw: break df self.api.to_df(raw) if df.empty: break all_data.append(df) start len(df) if not all_data: return pd.DataFrame() df_all pd.concat(all_data, ignore_indexTrue) # 按时间正序 df_all df_all.sort_values(datetime).reset_index(dropTrue) return df_all这里要特别注意翻页的时候start是“从最新数据往回偏移的根数”不是日期。也就是说start0是最新的800根start800是再往前推800根。方向不要搞反否则取出来的是未来数据回测就会莫名其妙地“穿越”。3.3 实时行情快照的获取做盘中监控的时候实时报价比K线更常用。get_security_quotes支持一次传多只股票接口返回的是最新价、开盘、最高、最低、买卖五档、内外盘、成交额等字段。def get_quotes(self, codes: list): # 自动判断市场 qlist [(1 if c.startswith(6) else 0, c) for c in codes] with self.lock: quotes self.api.get_security_quotes(qlist) if not quotes: return pd.DataFrame() df self.api.to_df(quotes) keep [market, code, price, last_close, open, high, low, vol, amount, bid1, ask1, bid_vol1, ask_vol1] df df[[c for c in keep if c in df.columns]] return df实测来看get_security_quotes一次传50到100只股票是比较合理的范围数量太大会触发服务器风控返回可能不完整。如果你要监控全市场4000多只股票建议分批拉取并且每批次之间做一个小延迟。3.4 全市场股票列表同步做全市场扫描或者选股的时候需要先拿到一份股票列表。pytdx 的get_security_list是按市场、按位置拉取的一次最多取 1000 条实际不同服务器可能返回 800 条所以要用循环来拉全。def get_all_securities(self): all_records [] for market in (0, 1): # 0深圳 1上海 total self.api.get_security_count(market) start 0 while start total: with self.lock: records self.api.get_security_list(market, start) if not records: break all_records.extend(records) start len(records) df pd.DataFrame(all_records) if not df.empty: df df[[code, name, pre_close, volunit, decimal_point]] return dfget_all_securities返回的字段里有code、name、pre_close等够用。注意不要每天重复拉全量列表通常开市前拉一次存到本地就足够。4. 数据持久化与增量更新策略4.1 为什么用 SQLite 而不是 CSV很多个人项目图省事直接把数据存成 CSV我发现数据量一大就成灾难。一是每次全量读进内存慢二是做增量更新的时候必须把整个文件读出来去重再写回性能和可靠性都差。所以我选 SQLite它不需要单独启动服务Python 内置支持而且支持事务和主键去重对个人量化项目来说足够了。建表语句我这样设计CREATE TABLE IF NOT EXISTS daily_kline ( code VARCHAR(10), date VARCHAR(10), open REAL, high REAL, low REAL, close REAL, vol INTEGER, amount REAL, PRIMARY KEY (code, date) ); CREATE INDEX IF NOT EXISTS idx_daily_code ON daily_kline(code);为什么主键要设置成(code, date)的组合因为同一只股票同一天只能有一条日K这样数据库层面就去重了。写入的时候用INSERT OR REPLACE就算重复拉取也不会产生脏数据。4.2 增量更新的核心逻辑全量更新全部历史数据只需要做一次之后每天只要拉最新几根K线就行。增量更新的逻辑我写成这样def update_daily(self, code: str): # 1. 查询本地最新日期 latest self.get_latest_date(code) # 2. 讨论需要拉多少根 if latest is None: total 800 # 无本地数据先拉800根再说 else: # 简单策略距离现在的自然日差 10根容错 days (pd.Timestamp.now().date() - pd.Timestamp(latest).date()).days total min(days 10, 800) # 3. 拉取并通过主键写库 df self.get_kline(code, counttotal) if df.empty: return df[date] pd.to_datetime(df[datetime]).dt.date.astype(str) self.save_to_db(df)这里有个细节total计算不能只看“缺失几天就拉几根”因为中间可能有停牌、节假日而且服务器可能还没更新最后一根完整日K。所以加一个10根左右的容错确保补全。拉多了一点没关系反正有主键去重。4.3 批量更新与并发限速单线程跑全市场更新太慢了我用ThreadPoolExecutor做并发但并发数一定要控制好。行情服务器对单个IP的连接数有硬限制实测并发超过 10 就很容易出现超时或者连接被重置。from concurrent.futures import ThreadPoolExecutor, as_completed def batch_update(self, codes: list, max_workers: int 6): with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(self.update_daily, c): c for c in codes} for future in as_completed(futures): code futures[future] try: future.result() print(f{code} updated) except Exception as e: print(f{code} failed: {e})另外并发场景下SQLite 写入需要小心。SQLite 同一时刻只允许一个写事务多个线程同时写会报database is locked。解决办法有两个一是写操作全部集中到一个专用连接里串行执行二是使用 WAL 模式减少读写锁冲突。conn sqlite3.connect(tdx.db, timeout30) conn.execute(PRAGMA journal_modeWAL;)加了 WAL 模式之后并发读写表现会好很多但仍建议把“查询数据”和“写入数据”用不同的连接对象分开避免在同一个连接上交错操作。5. 常见问题排查与稳定性优化5.1 高频问题速查表现象原因解决办法connect 返回 False服务器IP失效或端口不通换一个服务器维护候补列表某个代码取不到K线market 判断错误手动传入 market 参数返回数据只有800根单次请求上限用 start 翻页线程并发时连接被重置并发数过高降低 worker 数量加锁数据库报 locked多线程同时写 SQLite使用 WAL 模式或串行写入盘中取“当日K线”为空服务器尚未生成该周期K线改拉实时报价或者延迟取数5.2 数据可靠性检查数据接口不能只写“能跑就行”还要考虑数据对不对。我自己的代码里加了一个简单的校验把拉到的日K按时间排序后检查 open、high、low、close 四价是否异常。比如 high 必须大于等于 open 和 closelow 必须小于等于 open 和 close量不能为负数。如果发现异常宁可跳过这只股票也不把脏数据写进库里。def validate_kline(self, df: pd.DataFrame) - bool: if df.empty: return False if (df[high] df[low]).any(): return False if (df[vol] 0).any(): return False return True经验之谈行情数据偶尔会有个别异常点比如某天最高价突然比前后高出一大截但这不是接口错误而是除权除息或者服务器数据校正的问题。如果你做的是回测最好在拿到数据后做一个前复权处理否则历史收益算出来偏差会很大。5.3 稳定性优化心得这套接口上线跑了一段时间之后我总结出三个提高稳定性的关键点。一是连接心跳。pytdx 的heartbeatTrue必须开否则长时间空闲会被服务器主动断开。二是断线重连。不要在每次取数时才判断连接状态而是写一个独立的定时任务每隔几分钟检查一次连接断开了就自动重连。三是服务器列表要常更新。公网开放的行情服务器名单会变化我每隔一段时间就测一次延迟把失效的IP从列表里删掉把表现好的放前面。另外取数任务尽量避开开盘后的尖峰时段。比如交易日 9:30 到 10:00 服务器压力最大如果只是做盘后数据更新建议下午 15:30 之后再拉成功率会高很多。6. 数据接口的指标计算与扩展应用6.1 用 pandas 快速计算常用指标数据接口的价值不只是拿原始K线更重要的是能跟后续分析接上。我习惯在拿到 DataFrame 之后直接用 pandas 算指标。比如常见的 MA、RSI代码很简单。def add_ma(df, window5): df[fma{window}] df[close].rolling(window).mean() return df def add_rsi(df, window14): delta df[close].diff() gain delta.clip(lower0).rolling(window).mean() loss -delta.clip(upper0).rolling(window).mean() df[frsi{window}] 100 - 100 / (1 gain / loss) return df网上那些“98%胜率”“顶底信号”指标很多本质就是这类技术指标的组合。我自己不迷信胜率数字但有了稳定的数据接口之后你可以拿任何指标到历史数据上回测验证这是最有价值的事。6.2 把接口输出的数据用于选股策略接口设计好之后跑一个简单的选股策略就非常顺。比如选出“日K突破20日均线且成交量放大1.5倍”的股票def scan_stocks(codes): result [] for code in codes: df get_kline(code, count30) if df.empty: continue df add_ma(df, 20) latest df.iloc[-1] prev df.iloc[-2] if latest[close] latest[ma20] and prev[close] prev[ma20]: if latest[vol] prev[vol] * 1.5: result.append(code) return result配合前面说的并发批量更新全市场扫描一次只需要几分钟。这已经是个人量化研究里一个很实用的工具了。6.3 和通达信本身的配合最后说一句这套 Python 接口设计出来不是为了“替代”通达信而是补足它的短板。通达信的优势是看盘、公式编辑、板块联动数据接口的优势是自动化、批量化、可编程。我实际工作中常常是“通达信看盘、Python出数据、Excel出报表”三条线并行。把接口做成独立源码模块之后你甚至可以把它封装成一个 Python 包供团队里其他人直接 import 使用这才是接口设计的真正价值。写在最后我实际把这一套接口跑了大半年从最开始手动导出Excel到现在每天自动拉数据、入库、跑选股条件节省的时间非常可观。要说最后一点心得就是数据接口这类底层组件一定要舍得花时间把结构设计好别为了赶进度省掉异常处理或并发控制。前期多写几十行代码后期能少熬好几个夜。希望这篇拆解能帮到正打算做同类数据接口的人少踩几个我踩过的坑。本文还有配套的精品资源点击获取