QMT 策略如何引用外部计算好的指标？ext_data 函数使用详解

在 QMT 中，将外部计算好的自定义指标导入并在策略中使用的最灵活、最常用的方法是利用 Python 的 Pandas 库读取外部文件（如 CSV、Excel），将其存储在 ContextInfo 全局对象中，然后在 handlebar 中根据当前的时间和标的进行调用。

这种方式不需要通过 QMT 的界面进行繁琐的数据导入操作，完全通过代码控制，非常适合量化研究。

核心思路

1. 准备数据：确保你的外部数据包含日期（或时间戳）和股票代码作为索引列。
2. init 函数：在策略初始化时，使用 pandas.read_csv 或 read_excel 一次性读取数据，并将其转换为方便查询的格式（如字典或 MultiIndex DataFrame），挂载到 ContextInfo 上。
3. handlebar 函数：在每根 K 线运行时，获取当前的时间和代码，去内存中的数据里查找对应的值。

代码实现

假设你有一个 CSV 文件 D:/my_strategy_data/custom_indicators.csv，内容格式如下：

date,stock_code,my_signal
20231027,600000.SH,1.5
20231027,000001.SZ,0.8
...

以下是完整的策略代码示例：

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

def init(ContextInfo):
    """
    初始化函数：负责读取外部数据并处理成易于查询的格式
    """
    # 1. 设置股票池和回测参数（示例）
    ContextInfo.set_universe(['600000.SH', '000001.SZ'])
    ContextInfo.start = '20230101'
    ContextInfo.end = '20231231'
    ContextInfo.period = '1d'
    
    # 2. 读取外部数据文件
    # 请确保文件路径正确，且文件编码与读取方式匹配（通常为 utf-8 或 gbk）
    file_path = r'D:/my_strategy_data/custom_indicators.csv'
    
    # 检查文件是否存在，避免报错
    if not os.path.exists(file_path):
        print(f"错误：找不到文件 {file_path}")
        ContextInfo.custom_data = None
        return

    try:
        # 读取 CSV
        # 假设 CSV 列名为: date, stock_code, my_signal
        # dtype={'date': str} 确保日期被读取为字符串，方便后续匹配
        df = pd.read_csv(file_path, dtype={'date': str, 'stock_code': str})
        
        # 3. 数据预处理：构建索引以便快速查找
        # 将 date 和 stock_code 设为索引，这样可以通过 (date, code) 直接取值
        # sort_index 可以提高查询效率
        df = df.set_index(['date', 'stock_code']).sort_index()
        
        # 将处理好的数据挂载到 ContextInfo 上，使其成为全局变量
        ContextInfo.custom_data = df
        print("外部数据导入成功，前5行如下：")
        print(df.head())
        
    except Exception as e:
        print(f"读取数据发生错误: {e}")
        ContextInfo.custom_data = None

def handlebar(ContextInfo):
    """
    K线执行函数：获取当前K线对应的自定义指标数据
    """
    # 如果数据没加载成功，直接返回
    if ContextInfo.custom_data is None:
        return

    # 1. 获取当前 K 线的时间和代码
    # 获取当前 K 线的时间戳（毫秒）
    timetag = ContextInfo.get_bar_timetag(ContextInfo.barpos)
    # 将时间戳转换为字符串，格式必须与 CSV 中的日期格式一致 (这里假设 CSV 是 YYYYMMDD)
    current_date = timetag_to_datetime(timetag, '%Y%m%d')
    
    # 获取当前主图或运行到的股票代码
    current_stock = ContextInfo.stockcode + '.' + ContextInfo.market
    
    # 2. 从全局数据中查找指标值
    try:
        # 使用 .loc[日期, 代码] 进行索引查找
        # 假设我们要获取的列名叫 'my_signal'
        if (current_date, current_stock) in ContextInfo.custom_data.index:
            signal_value = ContextInfo.custom_data.loc[(current_date, current_stock), 'my_signal']
            
            # 3. 使用指标进行逻辑判断 (示例)
            print(f"时间: {current_date}, 代码: {current_stock}, 自定义指标值: {signal_value}")
            
            # 示例策略逻辑：如果指标大于 1，且当前没有持仓，则买入
            # 注意：这里仅为演示，实际交易需完善风控和资金判断
            # position = ContextInfo.get_position(current_stock) # 需自行封装获取持仓函数
            # if signal_value > 1.0:
            #     order_shares(current_stock, 100, 'fix', -1, ContextInfo, ContextInfo.accid)
                
        else:
            # 数据缺失处理
            # print(f"警告: 未找到 {current_date} {current_stock} 的外部数据")
            pass
            
    except Exception as e:
        print(f"查询数据出错: {e}")

关键点说明

1. 数据对齐 (Data Alignment)：

   • 这是最重要的一步。QMT 的 timetag_to_datetime 转换出的时间格式必须与你 CSV 文件中的 date 列格式完全一致。
   • 如果 CSV 是 2023-10-27，则转换格式应为 '%Y-%m-%d'。
   • 如果 CSV 是 20231027，则转换格式应为 '%Y%m%d'。

2. 性能优化：

   • 不要在 handlebar 中读取文件。handlebar 会在每根 K 线被调用，频繁读取文件会导致策略极慢甚至卡死。务必在 init 中读取并常驻内存。
   • 使用 Pandas 的 set_index 和 sort_index 可以将查询时间复杂度降低，极大提升回测速度。

3. 股票代码格式：

   • QMT 的标准格式是 代码.市场（例如 600000.SH）。请确保你的 CSV 文件中股票代码格式与此一致，或者在 init 读取后用 Pandas 进行字符串处理转换成一致的格式。

4. 替代方案 (扩展数据)：

   • 如果你希望在 K 线图的副图中画出这个指标，或者数据量极其庞大（几个 G），可以使用 QMT 的 “扩展数据” (Extended Data) 功能。这需要通过 QMT 客户端界面的“数据管理” -> “导入数据”功能将数据导入到本地数据库，然后使用 ext_data() 函数调用。但对于一般的策略计算，上述 Pandas 读取 CSV 的方法更加直接和便于调试。

Q&A

Q: 如果我的数据是分钟频的，怎么匹配？
A: 在 init 中读取数据时，确保 CSV 有包含时间的列（如 20231027 093000）。在 handlebar 中，使用 timetag_to_datetime(timetag, '%Y%m%d %H%M%S') 转换出相同格式的字符串，然后在 DataFrame 中进行索引查找。

Q: 这种方法支持实盘吗？
A: 支持。在实盘模式下，init 会在策略启动时运行一次。如果你的外部数据是盘中实时更新的（例如另一个脚本在不断写 CSV），你需要在 handlebar 中定期（例如每隔几分钟，或检测到文件修改时间变化时）重新读取文件更新 ContextInfo.custom_data，或者使用 ContextInfo.run_time 设置定时任务来刷新数据。

Q: 数据量太大导致内存溢出怎么办？
A: 如果 CSV 文件非常大（例如全市场高频数据），直接读入内存可能会崩溃。此时建议：

1. 使用 HDF5 或 Parquet 格式替代 CSV，读取速度更快且支持分块。
2. 仅读取当前策略需要的股票代码和时间段的数据。
3. 使用数据库（如 MySQL, SQLite, MongoDB），在 init 建立连接，在 handlebar 中查询（注意数据库查询延迟）。