MOOTDX 通达信数据接口实战：6 步从零搭建个人量化数据底座

MOOTDX 通达信数据接口实战：6 步从零搭建个人量化数据底座

【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx

MOOTDX 是通达信行情数据接口的 Python 封装库，它把「实时行情、本地历史数据、财务报表」三类数据源统一收敛到一套简洁 API 中，让你不必手写 TCP 协议解析、二进制文件解码和服务器选路逻辑。读完本文，你将基于它从零搭建一个名为 TinyQuant 的个人量化数据底座：一条命令跑通行情抓取，一套类封装管好重连与缓存，最后把数据导出为可上线的服务。

🚀 第一步：5 分钟跑通最小闭环

目标：安装 MOOTDX，并拿到第一份真实行情数据。

python -m venv mootdx_env source mootdx_env/bin/activate pip install -U git+https://gitcode.com/GitHub_Trending/mo/mootdx

验证安装并获取首个行情快照：

from mootdx.quotes import Quotes # bestip=True 表示自动探测并选择最快的行情服务器 client = Quotes.factory(market='std', bestip=True, timeout=15) df = client.quotes(symbol=['000001', '600000']) print(df.head())

这段代码背后发生了三件事：自动从服务器池里测速选路（mootdx/server.py）、建立并保持 TCP 长连接、把返回结果统一转换为 pandas DataFrame。你会看到代码、名称、最新价、成交量等字段，输出格式与通达信客户端基本一致。

要点：

• market='std'是沪深股票市场，market='ext'是期货、黄金等扩展市场
• 首次运行会在~/.mootdx/config.json生成选路结果，之后直接复用
• 命令行也可完成选路：python -m mootdx bestip -vv

📡 第二步：搭一层「线上行情服务」

目标：把单次调用升级为带重连、带日志、可复用的行情客户端，支撑盘中高频轮询。

import logging from mootdx.quotes import Quotes logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') log = logging.getLogger('tinyquant') class MarketClient: """线上行情客户端：连接复用 + 断线检测 + 自动重连""" def __init__(self, server=None, timeout=10): self._server = server self._timeout = timeout self.client = self._connect() def _connect(self): return Quotes.factory( market='std', server=self._server, heartbeat=True, # 心跳保活，避免空闲断连 auto_retry=True, # 内部自动重试 timeout=self._timeout, ) def ensure_alive(self): if getattr(self.client, 'closed', False): log.warning('连接已断开，正在重新建立连接...') self.client = self._connect() def quotes(self, symbols): return self.client.quotes(symbol=symbols) def bars(self, symbol, frequency=9, offset=200, adjust=None): kwargs = {'adjust': adjust} if adjust else {} return self.client.bars(symbol=symbol, frequency=frequency, offset=offset, **kwargs) def close(self): self.client.close() client = MarketClient() df = client.quotes(symbol=['000001', '600000', '000858']) print(df[['code', 'price', 'vol']])

K 线频率用数字或字符串别名均可，常用对照表如下：

要点：

• adjust='qfq'或'hfq'可让返回的 K 线自动前复权/后复权（内部走mootdx/utils/adjust.py的因子复权链路）
• 同一连接下批量查询 10 只股票一次quotes()调用即可完成，比逐只调用省掉 10 次握手
• 收盘后做策略回测建议走第三步的本地数据，盘中实时监控才走本层

🗄️ 第三步：接上「本地数据仓库」

目标：通达信客户端每天收盘后会把日线、分钟线写入本地目录，MOOTDX 的 Reader 能直接解析这些二进制文件，速度远快于网络请求。

from pathlib import Path import pandas as pd from mootdx.reader import Reader class LocalReader: """离线数据读取：日线 / 分钟线 / 五分钟线""" def __init__(self, tdxdir): # tdxdir 指向通达信安装目录，例如 'C:/new_tdx' 或 '/opt/tdx' self.reader = Reader.factory(market='std', tdxdir=tdxdir) def daily(self, symbol, start=None, end=None): df = self.reader.daily(symbol=symbol) if df is None or df.empty: return df if 'date' in df.columns: df = df.set_index(pd.to_datetime(df['date'])) if start: df = df[df.index >= start] if end: df = df[df.index <= end] return df def minute(self, symbol): return self.reader.minute(symbol=symbol) # 1 分钟线 def fzline(self, symbol): return self.reader.fzline(symbol=symbol) # 5 分钟线 reader = LocalReader('/path/to/tdx') daily = reader.daily('600036', start='2023-01-01') print(daily.tail())

Reader 的路径匹配是自动的：mootdx/reader.py中的find_path会根据代码前缀判断 sh / sz / 板块指数 / 扩展市场，再拼接vipdoc/lday、vipdoc/minline等子目录，你只需给出六位代码。

本地数据的两个额外收益：

• 全量历史：网络接口通常只能拉取最近几百根 K 线，本地lday文件里是上市以来的全部日线
• 板块数据：reader.block(symbol='block_gn')可读取概念板块成分，reader.block_new(name='my_block', symbol=[...])能创建自定义板块

📊 第四步：打通「财务数据管道」

目标：通达信财务文件gpcw*.zip是二进制压缩包，MOOTDX 的 Affair 模块负责下载、解压、解析，并输出中文表头 DataFrame。

from pathlib import Path from mootdx.affair import Affair class FinancialSync: """财务数据增量同步 + 解析""" def __init__(self, downdir='financial'): self.downdir = Path(downdir) self.downdir.mkdir(parents=True, exist_ok=True) def remote_files(self): """远程财务文件清单：filename / hash / filesize""" return Affair.files() def parse(self, filename): # 文件不存在时 Affair.parse 会自动先下载再解析 return Affair.parse(downdir=str(self.downdir), filename=filename) sync = FinancialSync() # 查看有哪些报告期文件 for f in sync.remote_files()[:3]: print(f['filename'], f['filesize']) # 下载并解析某一期资产负债表/利润表数据 df = sync.parse('gpcw20231231.zip') print(df[['report_date', '基本每股收益', '每股净资产', '净资产收益率', '货币资金', '资产总计']].head())

数据特点：

• 行是股票代码（索引），列是 200+ 个中文财务字段，覆盖资产负债表、利润表、现金流量表
• 下载带 MD5 校验（mootdx/affair.py中fetch_file会比对文件哈希，已存在且完整则跳过），天然支持增量更新
• 想全量拉取直接Affair.fetch(downdir='financial')，内部用 asyncio 并发下载并带进度条

⚡ 第五步：给底座装上「缓存与并发」

目标：用「内存 LRU + 磁盘 Pickle」两级缓存，把重复请求的耗时从秒级压到毫秒级。

from functools import lru_cache from mootdx.quotes import Quotes from mootdx.utils.pandas_cache import pd_cache from mootdx.utils.timer import timeit @lru_cache(maxsize=256) # 第一级：进程内内存缓存 @pd_cache(expired=600) # 第二级：磁盘缓存，10 分钟过期 def daily_bars(symbol, frequency=9, offset=200): client = Quotes.factory(market='std') return client.bars(symbol=symbol, frequency=frequency, offset=offset) @timeit def load_watchlist(symbols): return {s: daily_bars(s) for s in symbols} result = load_watchlist(['600036', '000001', '000858']) result = load_watchlist(['600036', '000001', '000858']) # 二次调用走缓存

实际收益对比：

timeit装饰器来自mootdx/utils/timer.py，会自动打印函数耗时，方便你量化每个环节的性能。磁盘缓存目录默认为.pd_cache，可用pd_cached_delete()一键清空。

🛡️ 第六步：让底座「扛得住生产环境」

目标：补齐重试、异常、日志、导出与诊断，达到可长期运行的标准。

6.1 失败重试装饰器

import time import logging from functools import wraps from mootdx.exceptions import MootdxException log = logging.getLogger('tinyquant') def retry_on_failure(max_retries=3, delay=1.0, backoff=2.0): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, max_retries + 1): try: return func(*args, **kwargs) except (MootdxException, ConnectionError, TimeoutError) as exc: log.warning('第 %s/%s 次尝试失败：%s', attempt, max_retries, exc) if attempt == max_retries: raise time.sleep(delay * (backoff ** (attempt - 1))) return wrapper return decorator @retry_on_failure(max_retries=3, delay=1) def fetch_snapshot(client, symbols): client.ensure_alive() return client.quotes(symbol=symbols)

配合ensure_alive()，断线后会自动重连再重试，指数退避策略避免对服务器造成瞬时冲击。

6.2 一键导出

from mootdx.utils import to_file to_file(daily, 'data/600036.csv') # 支持 csv / xlsx / xls / json / h5

to_file会根据扩展名自动选择 pandas 的导出方法并创建目录，落地成数据仓库的最小形态。

6.3 上线前诊断

from pathlib import Path from mootdx.quotes import Quotes from mootdx.server import check_server def diagnose(tdxdir): print('1) 服务器连通性：') check_server(sync=True) # 重新测速选路 print('2) 本地数据目录：') print(' 存在' if Path(tdxdir).is_dir() else ' 不存在', tdxdir) print('3) 行情接口冒烟测试：') client = Quotes.factory(market='std') df = client.quotes(symbol=['000001']) print(' 正常' if df is not None and not df.empty else ' 异常') diagnose('/path/to/tdx')

MOOTDX 还自带命令行工具，python -m mootdx下提供bestip、quotes、reader、affair等子命令，对应文档见docs/cli/，适合定时任务和人工排查场景。

✅ 收尾：能力清单与下一步

至此，TinyQuant 数据底座已经具备：

1. 线上行情服务：批量快照、任意周期 K 线、分钟线、分笔、指数、F10、除权除息，支持前/后复权
2. 本地数据仓库：日线 / 1 分钟 / 5 分钟线、板块与自定义板块，离线秒级读取
3. 财务数据管道：全量报告期清单、增量下载（MD5 校验）、中文列名 DataFrame
4. 性能优化：内存 + 磁盘两级缓存、函数级耗时统计、批量并发
5. 工程健壮性：心跳保活、断线重连、指数退避重试、结构化日志
6. 上线能力：多格式导出、连通性诊断、CLI 子命令

想继续深入，建议按这个顺序进阶：先基于adjust复权参数做因子回测，再用tools/customize.py维护自选板块，最后把缓存层替换为 Redis 或 ClickHouse，接入定时调度（cron / Airflow）即可升级为团队级数据服务。动手跑一遍上面的代码，你的量化之路就正式开始了。

【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx