三行 Python 拉取 A 股 K 线：mootdx 通达信数据封装实战

三行 Python 拉取 A 股 K 线mootdx 通达信数据封装实战【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx要拿到某只股票最近两百根日线、再叠一条均线绕不开一个前置问题K 线、分时、财务这些通达信数据从哪里来、怎么读进 DataFrame。mootdx 就是解决这一步的——它把通达信客户端背后的网络协议和本地文件解析打包成 Python 接口业务代码只负责调函数。定位一个通达信数据的 Python 封装层mootdx 本质是通达信数据源的 Python 封装把 pytdx 协议的连接、重连、解码等底层脏活收拢到内部对外暴露统一风格的方法。它的目标用户很明确写策略的量化初学者、做行情分析的数据工程师以及要搭盘中监控系统的人。装上之后实时行情、离线 K 线、财务指标都走同一套取数路径。核心能力集中在全周期 K 线与实时五档通达信本地文件离线解析财务数据抓取与落盘复权因子与交易日历工厂模式一行建连最小可运行路径安装分三步先拉源码、进目录、装依赖全部完成即可导入使用git clone https://gitcode.com/GitHub_Trending/mo/mootdx cd mootdx pip install -U mootdx[all]下面这段代码建连并打印 000001 的最新价与昨收from mootdx.quotes import Quotes # 标准市场客户端, factory 会自动挑选可用服务器 client Quotes.factory(marketstd) # quotes 返回 DataFrame, 一次可传多只代码 row client.quotes([000001]) # 取首行, 打印最新价、昨收、开盘 print(最新价:, row.iloc[0][price]) print(昨收价:, row.iloc[0][last_close]) print(开盘价:, row.iloc[0][open])quotes的返回就是 pandas 的 DataFrame空结果会返回空表而非抛错取数前先判断长度可以避免下标越界。按数据维度展开能力A 股 K 线数据获取bars 与 quotes 接口线上取数的主力是barsK 线和quotes实时frequency决定周期offset控制条数。from mootdx.quotes import Quotes client Quotes.factory(marketstd) # frequency9 对应日K, offset 单次上限 800 daily client.bars(symbol600036, frequency9, offset200) # 前复权直接传 adjustqfq, 内部自动处理除权 adjusted client.bars(symbol600036, frequency9, adjustqfq) # 按日期取历史分笔成交 ticks client.transactions(symbol600036, date20240510, offset100) print(daily[[close, vol]].tail())需要复权行情时优先走bars的adjust参数比拿到原始数据后自己按除权除息回算更省事。通达信离线数据读取Reader 解析本地文件有本地通达信安装目录时Reader直接读磁盘上的.day/.lc1文件无需联网适合大批量回刷历史。from mootdx.reader import Reader # tdxdir 指向通达信安装目录, 按本机路径修改 reader Reader.factory(marketstd, tdxdirC:/new_tdx) # 本地日K daily reader.daily(symbol600036) # 本地 1 分钟线, suffix1 一分钟, 5 五分钟 minute reader.minute(symbol600036, suffix1) print(日线, len(daily), 分钟, len(minute))离线读取依赖本地已下载的数据文件目录里没有对应股票的.day时会返回空路径拼错是最常见的坑。财务数据抓取与文件管理Affair 模块Affair负责从远端拉历史专业财务数据能先列清单、按文件名取、再解析落盘。from mootdx.affair import Affair # 列出远端全部财务文件(文件名/哈希/大小) files Affair.files() # 取最新一个, 按文件名单独下载 latest files[0][filename] Affair.fetch(downdir./cw_data, filenamelatest) # 解析为 DataFrame 并落盘 df Affair.parse(downdir./cw_data, filenamelatest) df.to_csv(gpcw.csv, indexFalse)files()返回的哈希可用来做增量同步本地已有相同哈希的文件就没必要重复下载。交易日历与复权工具utils 子包utils下挂了日历、复权、格式转换三类小工具常和上面的取数结果配合使用。from mootdx.utils import holiday from mootdx.utils import adjust from mootdx.tools import tdx2csv # 判断某天是否法定节假日 is_off holiday.holiday(date2024-10-01) # 拉取指定股票的复权因子 factor adjust.fq_factor(symbol600036, methodqfq) # 把 TDX 导出的 txt 转成标准 csv tdx2csv.txt2csv(infilesz#000001.txt, outfilesz#000001.csv) print(节假日:, is_off)复权因子、交易日历这类是横切所有股票的公共数据取一次缓存起来比每次现算更划算。生产环境要点连接容错重试与指数退避策略通达信免费服务器并不稳定长时间跑任务时会遇到断连和空返回稳妥做法是套一层带退避的重试。import time from mootdx.quotes import Quotes MAX_RETRY 3 def fetch_kline(symbol): client Quotes.factory(marketstd) for attempt in range(MAX_RETRY): try: # 拉取日K线, frequency9 为日线 return client.bars(symbolsymbol, frequency9, offset200) except Exception as exc: # 网络或连接异常统一兜底 wait 2 ** attempt # 指数退避: 1s, 2s, 4s print(f第 {attempt 1} 次失败, {wait}s 后重试: {exc}) time.sleep(wait) client.reconnect() # 断开后重建连接 raise RuntimeError(f{symbol} 连续 {MAX_RETRY} 次获取失败)reconnect()只在连接真正断开时才有意义配合closed属性判断可以省掉多余的握手。请求缓存与节流减少重复调用同一批股票在一分钟内被反复取数时直接打网络既慢又容易触发限流加一层带过期时间的本地缓存最直观。import time from mootdx.quotes import Quotes client Quotes.factory(marketstd) _cache {} def get_bars(symbol, ttl300): now time.time() # 缓存未过期直接返回, 不发起网络请求 if symbol in _cache and now - _cache[symbol][1] ttl: return _cache[symbol][0] data client.bars(symbolsymbol, frequency9, offset200) _cache[symbol] (data, now) return data缓存只适合盘中变化慢的数据tick 级实时行情别进缓存否则会拿到过期快照。与周边工具衔接取数之后的链路几乎是现成的bars、quotes、Reader返回的都是 pandas DataFrame可以直接接 rolling、pct_change 做技术指标接 matplotlib 画图也可以作为 Backtrader、Zipline 这类回测框架的自定义数据源把load指到上面任一函数即可不用自己写协议层。下一步装完依赖后先跑通 sample/ 里的报价与读取示例确认能取到数据再改逻辑接口细节看 docs/quick.md 与 docs/api/ 两个入口想抠实现就去tests/quotes对着真实返回结构调试。建议从一只股票的日 K 做起跑通取数—复权—落盘整条链路再扩到多股票批量。【免费下载链接】mootdx通达信数据读取的一个简便使用封装项目地址: https://gitcode.com/GitHub_Trending/mo/mootdx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考