讯投QMT使用小技巧: 历史行情数据下载与本地缓存方案

说明：本文包含 AI 创作内容，请自行判断是否适用。文中包含的代码片段，切勿直接使用，需要根据实际情况修改。

概述

写策略离不开历史数据：算指标要 K 线，做回测要序列，分析 tick 要分笔。但 QMT 的历史数据不是天然就躺在本地的——它需要你主动「下载」并「订阅」。很多新手第一次跑策略报「数据长度不足」或者拿到一堆空值，十有八九就是没下载历史数据，或者用错了取数函数。

本文把下载、订阅、取数、缓存四件事一次讲清楚。

一、下载历史数据：download_history_data

QMT 的历史数据是按需下载的。你不去下载，get_history_data / get_market_data_ex 拿到的可能就是空或者很短的片段。

def init(C):
    stock = '600000.SH'
    # 下载日线历史数据，从2020-01-01到今天
    ContextInfo.download_history_data(stock, '1d', '2020-01-01', '')

ContextInfo.download_history_data(stockcode, period, start_time, end_time)

• stockcode：带后缀的代码，如 '600000.SH'；也可以传 '000001.SZ' 这种。
• period：周期，'1d' 日线、'1m' 分钟线、'tick' 分笔。
• start_time / end_time：格式 '2020-01-01' 或 '20200101'，end_time 传空串表示到最新。

下载时机

• 回测前：把回测区间内的数据一次性下载好，否则回测会因缺数据报错或拿空值。
• 实盘 init 时：下载策略所需的历史窗口（比如 60 日均线就至少下载 60 个交易日）。
• 多股票时：循环下载，但别在 handlebar 里下载，那会阻塞行情处理。下载放 init 或定时任务里。

批量下载的小技巧

下载是同步且较慢的操作，几百只股票一个个下会很卡。可以做成「下载进度可恢复」：

def download_batch(C, stocks, period='1d', start='2020-01-01'):
    done = []
    for i, s in enumerate(stocks):
        try:
            C.download_history_data(s, period, start, '')
            done.append(s)
        except Exception as e:
            print(f"下载失败 {s}: {e}")
        if i % 50 == 0:
            print(f"进度 {i}/{len(stocks)}")
    return done

二、订阅行情：set_universe vs subscribe_quote

下载的是历史数据，实盘要拿实时行情还需要订阅。

set_universe（订阅K线行情）

def init(C):
    C.set_universe(['600000.SH', '000001.SZ'])

set_universe 订阅之后，handlebar 才会被这些品种的 K 线推动，get_market_data_ex 才能拿到实时值。这是绝大多数策略用的方式。

subscribe_quote（订阅实时tick/盘口）

def init(C):
    C.subscribe_quote('600000.SH', 'tick', count=10)

适合需要逐笔或盘口深度数据的策略。注意全市场订阅 tick（全推）数据量巨大，按需订阅即可。

订阅和下载是两件事：下载管历史，订阅管实时。实盘策略通常两者都要。

三、取数：get_history_data 与 get_market_data_ex

get_history_data（取单个品种的某个字段序列）

来自官方示例的写法：

data = C.get_history_data(30, '1d', 'close', dividend_type='front_ratio')
close_list = data['600000.SH']   # 长度30的收盘价序列

• 第一个参数：取多少根。
• 第二个：周期。
• 第三个：字段，'close'/'open'/'high'/'low'/'volume'/'amount'。
• dividend_type：复权类型，详见 复权数据怎么选。

get_market_data_ex（多品种多字段的 DataFrame）

bars = C.get_market_data_ex(['close', 'volume'], ['600000.SH', '000001.SZ'],
                            period='1d', count=60)
df = bars['600000.SH']   # 一个DataFrame，含close、volume列

适合一次取多只股票、多个字段、要 DataFrame 形态的场景。

四、本地缓存：避免重复取数

get_history_data 每次调用都会走一次数据接口，在 handlebar 高频路径里反复取 60 日数据非常浪费。建议把算好的指标或取好的序列缓存起来，只在需要时刷新。

class a(): pass
A = a()

def init(C):
    A.cache = {}          # stock -> {'data':..., 'barpos':...}
    C.set_universe(['600000.SH'])

def handlebar(C):
    if not C.is_last_bar():
        return
    stock = '600000.SH'
    barpos = C.barpos
    # 同一根bar只取一次数
    if A.cache.get(stock, {}).get('barpos') != barpos:
        close = C.get_history_data(60, '1d', 'close', dividend_type='front_ratio')[stock]
        A.cache[stock] = {'data': close, 'barpos': barpos}
    close = A.cache[stock]['data']
    ma60 = sum(close) / len(close)
    # ...

这种「按 barpos 缓存」的写法，保证一根 K 线内多次访问只取数一次，是实盘性能优化的关键。

落地到本地文件/数据库

如果数据量更大（比如全市场日线、多年 tick），可以下载后落到本地 parquet/csv 或 SQLite，下次直接读本地，不再依赖 QMT 接口：

import pandas as pd

def save_local(stock, df, path='./data/'):
    df.to_parquet(f"{path}{stock.replace('.','_')}.parquet")

def load_local(stock, path='./data/'):
    try:
        return pd.read_parquet(f"{path}{stock.replace('.','_')}.parquet")
    except FileNotFoundError:
        return None

把数据落本地后，配合 公共代码的编写与引用，可以让多个策略共享同一份缓存数据。

五、常见坑

1. 没下载就拿空值：get_history_data 返回 None 或长度为 0，先检查 download_history_data 是否执行过。
2. 在 handlebar 里下载：会严重阻塞，下载只放 init 或定时任务。
3. 复权类型用错：算指标前后复权结果不同，详见 复权数据怎么选。
4. 新上市/停牌数据不足：取数后要判断长度，不够就跳过（官方双均线示例就是这么做的）。
5. 全推 tick 撑爆内存：按需订阅，别无脑全推。

总结

历史数据的流程是：download（下载历史） → set_universe/subscribe（订阅实时） → get_history_data/get_market_data_ex（取数） → 缓存（按 barpos 或落本地）。把这四步顺序记牢，再配上按 barpos 的缓存，策略的数据层就稳了。

说明：本文包含 AI 创作内容，请自行判断是否适用。文中包含的代码片段，切勿直接使用，需要根据实际情况修改。

---

相关文章

常用数据接口速查（股票列表/财务/指数成分股） 开始阅读

复权数据前复权后复权该怎么选 开始阅读

迅投QMT自带策略-双均线实盘示例PY策略-解读 开始阅读

讯投QMT使用小技巧: 公共代码的编写与引用 开始阅读

QMT 知识库 XtQuant知识库 使用文档 pdf 开始阅读

所有与QMT相关的文章查看目录