通达信数据接口Python封装实战四步从零搭建你的首个行情分析小工具【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx如果你刚接触量化或者只是想写个脚本自动盯盘第一步通常都是卡在“数据从哪来”上。通达信数据接口是 A 股数据最老牌也最实用的来源之一而 mootdx 这个 Python 封装库正是把“通达信数据读取”这件事变得足够简单一个工厂方法连服务器、一个对象读本地文件、几行代码同步财报。本文不堆概念直接带你从零手写一个叫“盯盘助手”的小工具把行情、财报、复权、缓存这些模块一个个装进去。先看一眼项目结构方便后面按图索骥核心行情代码在 mootdx/quotes.py离线数据读取在 mootdx/reader.py财务数据在 mootdx/financial/完整的接口说明可以翻 docs/ 目录。 第一步把环境跑通让“盯盘助手”活过来2.1 创建虚拟环境并安装依赖量化系统环境搭建老规矩先隔离环境避免把系统 Python 搞乱# 创建并激活虚拟环境 python -m venv kanpan_env source kanpan_env/bin/activate # 安装 mootdx 全家桶含命令行和全部扩展依赖 pip install -U mootdx[all]装完顺手验证一下版本同时确认 pandas 这些数据处理的“左膀右臂”都在import pandas as pd from mootdx import __version__ print(fmootdx 版本: {__version__}) print(fpandas 版本: {pd.__version__}) # 输出示例: # mootdx 版本: 0.12.5 # pandas 版本: 2.0.3 小贴士如果后面要用命令行工具比如测速记得装mootdx[cli]拿不准就统一用[all]最省心。2.2 自动挑选最快的行情服务器行情服务器遍布全国网络环境不同最优节点也不同。mootdx 提供了内置的测速工具直接让它在候选列表里“掐表”把响应最快的几个挑出来python -m mootdx bestip -vv命令跑完会在配置里写入最优服务器之后每次创建客户端都会自动使用。如果你想把选中的服务器打印成表格看效果可以在 Python 里调用from mootdx.server import bestip # 返回按响应速度排序的 (地址, 端口) 列表前 5 名 fastest bestip(limit5, consoleTrue) print(fastest)输出大致长这样时间单位为毫秒---------------------------------------- | Name | Addr | Port | Time | ---------------------------------------- | 华泰证券 | 112.74.214.43 | 7709 | 12.35 ms | | 国泰君安 | 180.153.39.51 | 7709 | 18.02 ms | | ... | ... | ... | ... | ----------------------------------------⚠️ 常见坑首次使用如果提示找不到配置多半是bestip没有成功执行过。先手动跑一遍python -m mootdx bestip把配置落盘再继续后面的代码。⚡ 第二步装上“行情雷达”盘中数据随手取2.3 一行代码连上实时行情服务我们要做的第一件事是让“盯盘助手”能实时看到报价。用工厂方法创建客户端打开心跳保活和多线程网络抖动时连接不容易掉线from mootdx.quotes import Quotes client Quotes.factory( marketstd, # std标准市场(股票)ext扩展市场(期货/黄金等) bestipTrue, # 自动使用测速结果里的最优服务器 multithreadTrue, # 开启多线程批量请求更快 heartbeatTrue, # 心跳保活长连接更稳定 timeout15, # 单次请求超时时间(秒) ) # 获取五档报价 quote client.quotes(symbol600519) print(quote[[price, last_close, vol, amount]])price last_close vol amount 0 1688.88 1690.00 1265412 2.136470e09看到这行输出说明你的 Python 已经能实时获取 A 股行情了。核心封装就这一个工厂方法后面的bars、minute、index都是同一个客户端上的“兄弟接口”。2.4 批量拉取K线与分时数据Python获取实时行情盘中盯盘通常不是看一只票而是看一个自选列表。我们把它封装成一个批量方法顺便加上超时保护def watch_list(symbols): 批量获取多只股票的实时行情快照 data [] for code in symbols: try: row client.quotes(symbolcode) data.append(row[[code, price, vol]].iloc[0]) except Exception as exc: print(f[跳过] {code} 获取失败: {exc}) return pd.concat(data, axis1).T my_pool [600519, 300750, 601318] watch_list(my_pool)K 线和分时数据同样简单frequency9表示日线换成别的数字可以拿到周线、月线甚至分钟线# 日K最近 30 根 daily client.bars(symbol600519, frequency9, offset30) # 当日分时 minute_today client.minute(symbol600519) # 指数行情 index_df client.index(symbol000001, frequency9) 小贴士offset单次最多 800 根需要更多历史就往回翻页循环叠加即可分笔成交用client.transaction(symbol...)历史某一天的分笔用transactions(symbol..., date20240115)。️ 第三步补上“历史记忆”离线读取通达信本地数据盘中数据是“现在”但回测和选股需要“过去”。如果你本机装有通达信客户端本地vipdoc目录下就有完整的日线、分钟线文件mootdx 的 Reader 可以离线直接解析不占网络、速度飞快。2.5 适配多市场路径的本地数据读取from mootdx.reader import Reader # tdxdir 换成你自己的通达信安装目录 reader Reader.factory(marketstd, tdxdir/path/to/tdx) # 日线 df_daily reader.daily(symbol600519) # 分钟线当日 df_min reader.minute(symbol600519) # 五分钟线复权线 df_fz reader.fzline(symbol600519) print(df_daily.tail(3))open high low close vol amount 2024-01-10 1660 1692 1648 1685 3845216 6.5e09 2024-01-11 1680 1710 1672 1705 4120870 7.0e09 2024-01-12 1702 1718 1690 1708 3982210 6.8e09Reader 会自动判断市场前缀sh/sz、识别 88 开头的板块指数、甚至支持带#的扩展市场代码路径匹配这种脏活完全不用你操心。想要日线做回测再把复权因子补上复权相关的实现可以参考 mootdx/tools/reversion.py 和 mootdx/utils/factor.py。from mootdx.utils.adjust import to_adjust # 前复权处理 adjusted to_adjust(df_daily, symbol600519, adjustqfq) print(adjusted.head(2))⚠️ 常见坑tdxdir目录不存在会直接抛异常。拿到一台新机器先确认vipdoc/sh/lday这类子目录真实存在再初始化 Reader。 第四步接入“财务体检”财报数据自动同步与解析光看价格不看基本面等于闭着眼睛开车。通达信的财务数据以季度压缩包形式发布形如gpcw20231231.zipmootdx 把“查列表→下载→解析”整条链路都封装好了。2.6 用 Affair 完成财报自动同步from mootdx.affair import Affair # 1) 获取远程财务文件清单含文件名、MD5、大小 files Affair.files() print([f[filename] for f in files[:5]]) # [gpcw19960630.zip, gpcw19961231.zip, ...] # 2) 只下载指定季度 Affair.fetch(downdirfinancial_store, filenamegpcw20231231.zip) # 3) 下载并直接解析成 DataFrame df_fin Affair.parse(downdirfinancial_store, filenamegpcw20231231.zip) print(df_fin.columns.tolist())解析出来的字段都是规范化的财务指标每股收益、净资产、主营收入等可以直接 join 到行情数据上。这里有个很贴心的细节fetch会校验本地文件的 MD5文件已存在且完整就直接跳过不会重复下载——财报数据自动解析这一步基本是“设置一次长期免维护”。 小贴士批量补历史财报直接Affair.fetch(downdirfinancial_store)不带 filename它会并发把清单里的文件全部拉下来。数据格式细节见 mootdx/financial/。️ 第五步让工具“皮实耐用”——缓存、重试与日志三板斧网络请求难免偶发抖动生产环境的代码必须学会“自己扛”。我们给“盯盘助手”加三件护甲。2.7 内存磁盘的混合缓存策略同一个标的的数据经常被重复查询每次都走网络很浪费。先建一个本地缓存目录用 pickle 落盘再用lru_cache做内存层import os import pickle from functools import lru_cache CACHE_DIR ./hq_cache os.makedirs(CACHE_DIR, exist_okTrue) lru_cache(maxsize512) def get_bars_cached(symbol, freq): 内存缓存优先未命中再查磁盘缓存 cache_file os.path.join(CACHE_DIR, f{symbol}_{freq}.pkl) if os.path.exists(cache_file): with open(cache_file, rb) as f: return pickle.load(f) data client.bars(symbolsymbol, frequencyfreq, offset200) with open(cache_file, wb) as f: pickle.dump(data, f) return data项目中还自带了基于 pandas 的缓存装饰器嫌手写麻烦可以直接用参考 mootdx/utils/pandas_cache.py 里的pd_cache。2.8 重试机制与监控日志针对连接中断这类瞬时错误写一个指数退避的重试装饰器同时用项目自带的 logger 记录每次请求的耗时方便事后复盘import time from functools import wraps from mootdx.exceptions import MootdxException from mootdx.logger import logger def retry_guard(times3, wait1.0): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(times): try: start time.perf_counter() result func(*args, **kwargs) cost (time.perf_counter() - start) * 1000 logger.info(f{func.__name__} 耗时 {cost:.1f} ms) return result except MootdxException as exc: logger.warning(f第 {attempt 1} 次重试: {exc}) time.sleep(wait * (attempt 1)) raise RuntimeError(f{func.__name__} 重试 {times} 次仍失败) return wrapper return decorator retry_guard(times3, wait0.5) def safe_bars(symbol): return client.bars(symbolsymbol, frequency9, offset50)效果对比不加重试时一次网络抖动可能让整个盘中任务中断加上重试日志后同样的抖动只会留下一行 warning任务照常完成。 第六步让数据“会思考”——复权K线 技术指标数据齐了下一步自然是算指标。以经典 MACD 和 RSI 为例用 pandas 手算即可不依赖额外第三方库def add_indicators(df): 给日K数据追加 MACD / RSI 指标列 close df[close] ema12 close.ewm(span12, adjustFalse).mean() ema26 close.ewm(span26, adjustFalse).mean() df[dif] ema12 - ema26 df[dea] df[dif].ewm(span9, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2 delta close.diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / loss df[rsi] 100 - 100 / (1 rs) return df kline client.bars(symbol600519, frequency9, offset120) kline add_indicators(kline) print(kline[[close, dif, dea, macd, rsi]].tail(3))之后无论是“金叉死叉”信号还是 RSI 超买超卖过滤都是在这个 DataFrame 上加两行判断的事。整个“行情→指标→信号”的链路就闭环了。 第七步常见问题排查清单故障诊断速查工具上线前把这几个“体检项目”过一遍能少踩一半坑import os from mootdx.server import bestip def diagnostics(tdx_path): # 1) 服务器连通性 try: bestip(limit3, consoleTrue) print(✅ 行情服务器连通正常) except Exception as exc: print(f❌ 服务器连接异常: {exc}) # 2) 本地数据目录 if os.path.isdir(tdx_path): print(✅ 通达信数据目录存在) else: print(❌ 数据目录不存在请检查 tdxdir 参数) # 3) 缓存目录权限 if os.access(./hq_cache, os.W_OK): print(✅ 缓存目录可写) else: print(❌ 缓存目录不可写检查磁盘权限) diagnostics(/path/to/tdx)另外两个高频问题单独说明M1 Mac 上报 PyMiniRacer 错误这是苹果芯片与底层 JS 引擎的兼容问题属于已知 issue换用 Linux 或 x86 环境即可绕开。quotes()返回空先确认股票代码没带市场前缀如600519即可不要写sh600519再检查bestip是否已执行。 写在最后动手跑通你的第一版工具从环境搭建、实时行情、离线历史、财报解析到缓存、重试、指标计算你的“盯盘助手”已经具备了一个迷你量化工具的全部骨架。回顾一下今天埋下的关键钩子实时行情与 K 线Quotes.factory(...)bars/minute/index离线本地数据Reader.factory(...)daily/fzline财报自动同步Affair.files/fetch/parse可靠性三件套pickle 磁盘缓存、重试装饰器、logger 监控指标计算pandas 手写 MACD/RSImootdx 的源码非常克制值得精读的地方我都用链接标出来了行情客户端 mootdx/quotes.py、本地读取 mootdx/reader.py、财务模块 mootdx/financial/。如果你用的是 git 安装方式仓库地址是https://gitcode.com/GitHub_Trending/mo/mootdx克隆到本地边看边改最方便。别让文章停在收藏夹里——现在就把pip install mootdx[all]跑起来用 20 行代码让你的“盯盘助手”输出第一张行情表。跑通了你就是半个量化工程师了。【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考