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

MOOTDX 通达信数据接口实战6 步从零搭建个人量化数据底座【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdxMOOTDX 是通达信行情数据接口的 Python 封装库它把「实时行情、本地历史数据、财务报表」三类数据源统一收敛到一套简洁 API 中让你不必手写 TCP 协议解析、二进制文件解码和服务器选路逻辑。读完本文你将基于它从零搭建一个名为 TinyQuant 的个人量化数据底座一条命令跑通行情抓取一套类封装管好重连与缓存最后把数据导出为可上线的服务。 第一步5 分钟跑通最小闭环目标安装 MOOTDX并拿到第一份真实行情数据。python -m venv mootdx_env source mootdx_env/bin/activate pip install -U githttps://gitcode.com/GitHub_Trending/mo/mootdx验证安装并获取首个行情快照from mootdx.quotes import Quotes # bestipTrue 表示自动探测并选择最快的行情服务器 client Quotes.factory(marketstd, bestipTrue, timeout15) df client.quotes(symbol[000001, 600000]) print(df.head())这段代码背后发生了三件事自动从服务器池里测速选路mootdx/server.py、建立并保持 TCP 长连接、把返回结果统一转换为 pandas DataFrame。你会看到代码、名称、最新价、成交量等字段输出格式与通达信客户端基本一致。要点marketstd是沪深股票市场marketext是期货、黄金等扩展市场首次运行会在~/.mootdx/config.json生成选路结果之后直接复用命令行也可完成选路python -m mootdx bestip -vv 第二步搭一层「线上行情服务」目标把单次调用升级为带重连、带日志、可复用的行情客户端支撑盘中高频轮询。import logging from mootdx.quotes import Quotes logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) log logging.getLogger(tinyquant) class MarketClient: 线上行情客户端连接复用 断线检测 自动重连 def __init__(self, serverNone, timeout10): self._server server self._timeout timeout self.client self._connect() def _connect(self): return Quotes.factory( marketstd, serverself._server, heartbeatTrue, # 心跳保活避免空闲断连 auto_retryTrue, # 内部自动重试 timeoutself._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(symbolsymbols) def bars(self, symbol, frequency9, offset200, adjustNone): kwargs {adjust: adjust} if adjust else {} return self.client.bars(symbolsymbol, frequencyfrequency, offsetoffset, **kwargs) def close(self): self.client.close() client MarketClient() df client.quotes(symbol[000001, 600000, 000858]) print(df[[code, price, vol]])K 线频率用数字或字符串别名均可常用对照表如下frequency周期字符串别名05 分钟5m115 分钟15m230 分钟30m360 分钟1h4 / 9日线days/day5周线week6月线mon71 分钟1m10 / 11季线 / 年线3mon/year要点adjustqfq或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(marketstd, tdxdirtdxdir) def daily(self, symbol, startNone, endNone): df self.reader.daily(symbolsymbol) 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(symbolsymbol) # 1 分钟线 def fzline(self, symbol): return self.reader.fzline(symbolsymbol) # 5 分钟线 reader LocalReader(/path/to/tdx) daily reader.daily(600036, start2023-01-01) print(daily.tail())Reader 的路径匹配是自动的mootdx/reader.py中的find_path会根据代码前缀判断 sh / sz / 板块指数 / 扩展市场再拼接vipdoc/lday、vipdoc/minline等子目录你只需给出六位代码。本地数据的两个额外收益全量历史网络接口通常只能拉取最近几百根 K 线本地lday文件里是上市以来的全部日线板块数据reader.block(symbolblock_gn)可读取概念板块成分reader.block_new(namemy_block, symbol[...])能创建自定义板块 第四步打通「财务数据管道」目标通达信财务文件gpcw*.zip是二进制压缩包MOOTDX 的 Affair 模块负责下载、解压、解析并输出中文表头 DataFrame。from pathlib import Path from mootdx.affair import Affair class FinancialSync: 财务数据增量同步 解析 def __init__(self, downdirfinancial): self.downdir Path(downdir) self.downdir.mkdir(parentsTrue, exist_okTrue) def remote_files(self): 远程财务文件清单filename / hash / filesize return Affair.files() def parse(self, filename): # 文件不存在时 Affair.parse 会自动先下载再解析 return Affair.parse(downdirstr(self.downdir), filenamefilename) 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(downdirfinancial)内部用 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(maxsize256) # 第一级进程内内存缓存 pd_cache(expired600) # 第二级磁盘缓存10 分钟过期 def daily_bars(symbol, frequency9, offset200): client Quotes.factory(marketstd) return client.bars(symbolsymbol, frequencyfrequency, offsetoffset) 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]) # 二次调用走缓存实际收益对比调用方式首次耗时二次耗时说明直接请求网络200~500 ms200~500 ms每次都要走服务器内存缓存命中300 ms 1 ms命中率约 100%零网络开销磁盘缓存命中300 ms~5 ms进程重启后仍生效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_retries3, delay1.0, backoff2.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_retries3, delay1) def fetch_snapshot(client, symbols): client.ensure_alive() return client.quotes(symbolsymbols)配合ensure_alive()断线后会自动重连再重试指数退避策略避免对服务器造成瞬时冲击。6.2 一键导出from mootdx.utils import to_file to_file(daily, data/600036.csv) # 支持 csv / xlsx / xls / json / h5to_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(syncTrue) # 重新测速选路 print(2) 本地数据目录) print( 存在 if Path(tdxdir).is_dir() else 不存在, tdxdir) print(3) 行情接口冒烟测试) client Quotes.factory(marketstd) 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 数据底座已经具备线上行情服务批量快照、任意周期 K 线、分钟线、分笔、指数、F10、除权除息支持前/后复权本地数据仓库日线 / 1 分钟 / 5 分钟线、板块与自定义板块离线秒级读取财务数据管道全量报告期清单、增量下载MD5 校验、中文列名 DataFrame性能优化内存 磁盘两级缓存、函数级耗时统计、批量并发工程健壮性心跳保活、断线重连、指数退避重试、结构化日志上线能力多格式导出、连通性诊断、CLI 子命令想继续深入建议按这个顺序进阶先基于adjust复权参数做因子回测再用tools/customize.py维护自选板块最后把缓存层替换为 Redis 或 ClickHouse接入定时调度cron / Airflow即可升级为团队级数据服务。动手跑一遍上面的代码你的量化之路就正式开始了。【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考