如何在 QMT 策略运行前检查历史数据的完整性，避免回测陷阱？

在 QMT 策略回测开始时（即 init 阶段），可以通过尝试获取指定长度的历史数据来检查本地数据是否完整。如果获取的数据为空或长度不足，说明数据未准备好或未下载。

QMT 没有内置的“一键检查”函数，但我们可以编写一段逻辑来实现这一功能。

核心思路

1. 确定标的和周期：明确策略需要用到的股票池（stock_list）和周期（如 '1d'）。
2. 尝试获取数据：使用 ContextInfo.get_market_data_ex 接口请求一定数量的历史数据（例如策略计算指标需要过去 100 根 K 线）。
3. 验证数据完整性：遍历返回的数据结果，检查是否为空（None 或 empty）以及数据长度是否满足最低要求。
4. 输出警告：如果有标的缺失数据，打印日志提示用户去客户端“数据管理”界面补充数据。

代码实现

以下是一个完整的示例代码，展示了如何在 init 函数中执行此检查：

# -*- coding: gbk -*-
import pandas as pd

def init(ContextInfo):
    # 1. 设置股票池和策略参数
    stock_list = ['600000.SH', '000001.SZ', '600519.SH']
    ContextInfo.set_universe(stock_list)
    
    # 假设策略至少需要 100 根 K 线来计算均线等指标
    required_count = 100
    period = '1d'
    
    print("正在检查数据完整性...")
    
    # 2. 尝试获取历史数据
    # 注意：subscribe=False 表示只读取本地数据，不订阅实时推送，适合在 init 中做检查
    data_map = ContextInfo.get_market_data_ex(
        fields=[],  # 获取所有字段
        stock_code=stock_list,
        period=period,
        start_time='', 
        end_time='',
        count=required_count,
        dividend_type='follow',
        fill_data=True,
        subscribe=False 
    )
    
    # 3. 检查数据是否缺失
    missing_data_stocks = []
    insufficient_len_stocks = []
    
    for stock in stock_list:
        # 检查是否在返回的字典中
        if stock not in data_map:
            missing_data_stocks.append(stock)
            continue
            
        df = data_map[stock]
        
        # 检查 DataFrame 是否为空
        if df is None or df.empty:
            missing_data_stocks.append(stock)
        # 检查数据长度是否足够
        elif len(df) < required_count:
            insufficient_len_stocks.append(f"{stock}(现有{len(df)}根)")

    # 4. 输出检查结果
    if missing_data_stocks:
        print("=" * 50)
        print(f"[严重警告] 以下股票完全缺失本地数据 ({period})：")
        print(missing_data_stocks)
        print("请在 QMT 客户端 ->【数据管理】->【补充数据】中下载对应品种的历史数据。")
        print("=" * 50)
        
    if insufficient_len_stocks:
        print("=" * 50)
        print(f"[风险提示] 以下股票数据长度不足 {required_count} 根，可能影响指标计算：")
        print(insufficient_len_stocks)
        print("=" * 50)
        
    if not missing_data_stocks and not insufficient_len_stocks:
        print(f"数据检查通过：所有标的均包含至少 {required_count} 根 {period} 数据。")

def handlebar(ContextInfo):
    # 策略逻辑
    if not ContextInfo.is_last_bar():
        return
    # ... 交易逻辑 ...
    pass

代码详解

1. ContextInfo.get_market_data_ex:

   • 这是获取行情的推荐接口。
   • count=required_count: 指定我们需要往前追溯多少根 K 线。
   • subscribe=False: 在 init 中我们只需要读取本地历史数据进行校验，不需要建立实时订阅连接，这样效率更高。

2. 校验逻辑:

   • 完全缺失: 如果 stock 不在返回的 data_map 键中，或者对应的 DataFrame 为空，说明本地完全没有该品种的数据。
   • 长度不足: 如果 len(df) < required_count，说明虽然有数据，但不足以计算长周期的指标（例如你需要计算 MA120，但本地只有 50 根 K 线），这会导致策略报错或计算不准。

3. 解决数据缺失问题:

   • QMT 的 Python API 无法通过代码自动从服务器下载历史数据。
   • 如果代码提示数据缺失，必须人工操作 QMT 客户端界面：点击菜单栏的 “数据管理” -> 选择对应的市场和周期 -> 点击 “补充数据”（下载）。

Q&A

Q: 为什么不在 handlebar 里检查数据？
A: handlebar 会随着 K 线推进被频繁调用。数据完整性检查只需要在策略启动时（init）做一次即可，放在 handlebar 中会严重拖慢回测速度并产生大量重复日志。

Q: get_market_data_ex 返回的数据是最新的吗？
A: 在回测模式下，它返回的是截止到回测结束时间的数据。在实盘模式下，它返回的是本地已有的最新数据。如果本地数据未更新（例如未下载当天的盘后数据），获取到的可能不是最新的，因此“补充数据”这一步至关重要。

Q: 如果数据缺失，我可以在代码里自动下载吗？
A: 目前 QMT Python API 暂未开放直接通过代码触发客户端下载历史数据的接口。必须通过客户端界面的“数据管理”功能进行下载。