策略运行机制

本章深入讲解 QMT 内置 Python 的策略运行机制，包括生命周期、回调函数和运行模式。

策略生命周期

QMT 策略从启动到停止，经历以下阶段：

加载策略 → 调用 init() → 等待行情 → 调用 handlebar() → 策略停止

生命周期详解

  1. 加载阶段：QMT 读取策略文件，检查语法
  2. 初始化阶段：调用 init(ContextInfo)，仅执行一次
  3. 运行阶段：每根 K 线触发 handlebar(ContextInfo)
  4. 停止阶段：用户手动停止或客户端关闭
提示

init 函数中适合做一次性初始化工作，如设置股票池、初始化变量等。

核心回调函数

init

def init(ContextInfo):
    """策略初始化，仅调用一次"""
    # 设置股票池
    ContextInfo.set_universe(['000001.SZ'])
    # 初始化全局变量
    ContextInfo.counter = 0

handlebar

def handlebar(ContextInfo):
    """每根 K 线触发一次"""
    # 获取当前 K 线数据
    close = ContextInfo.get_close_price()
    # 判断是否为最新 K 线
    if ContextInfo.is_last_bar():
        print(f'最新收盘价: {close}')

回调执行顺序

init() → handlebar(K线1) → handlebar(K线2) → ... → handlebar(K线N)

运行模式

回测模式

  • 使用历史数据
  • 按时间顺序逐根回放 K 线
  • 不产生真实委托
  • 可查看收益曲线和统计指标

模拟信号模式

  • 使用实时行情
  • 产生模拟委托信号
  • 不执行真实交易
  • 适合验证策略实时表现

真实交易模式

  • 使用实时行情
  • 产生真实委托
  • 涉及真实资金
  • 需要充分测试后使用
注意

从回测切换到真实交易前，务必经过模拟信号模式的充分验证。

K 线触发机制

触发时机

handlebar 在每根 K 线完成时触发：

  • 日线策略：每个交易日收盘后触发
  • 分钟线策略：每根分钟 K 线结束时触发
  • Tick 策略：每个 Tick 数据到达时触发

is_last_bar 判断

def handlebar(ContextInfo):
    # 回测模式下，所有 K 线都会触发
    # 模拟/实盘模式下，只有最新的 K 线需要处理
    if not ContextInfo.is_last_bar():
        return  # 跳过历史 K 线
    
    # 处理最新 K 线
    close = ContextInfo.get_close_price()
    print(f'最新价格: {close}')

多周期触发

如果策略订阅了多个周期的数据，handlebar 会按各周期分别触发。

ContextInfo 对象

ContextInfo 是策略运行的核心上下文对象，贯穿整个生命周期。

常用属性

属性 类型 说明
stockcode str 当前品种代码
period str 当前周期
timetag int 当前 K 线时间戳
barpos int 当前 K 线位置
do_back_test bool 是否回测模式

常用方法

# 设置股票池
ContextInfo.set_universe(['000001.SZ', '600000.SH'])

# 获取收盘价
close = ContextInfo.get_close_price()

# 判断是否最新 K 线
is_last = ContextInfo.is_last_bar()

# 获取历史数据
history = ContextInfo.get_history_data('close', 20)

# 自定义变量存储
ContextInfo.my_var = '自定义数据'

策略状态管理

全局变量

在 ContextInfo 上挂载自定义变量，可在不同回调间共享：

def init(ContextInfo):
    ContextInfo.position = {}  # 持仓字典
    ContextInfo.last_signal = None  # 上次信号

def handlebar(ContextInfo):
    if not ContextInfo.is_last_bar():
        return
    # 读取上次信号
    if ContextInfo.last_signal:
        print(f'上次信号: {ContextInfo.last_signal}')
    # 更新信号
    ContextInfo.last_signal = 'buy'

持久化

策略重启后，ContextInfo 上的自定义变量会丢失。需要持久化的数据应保存到文件：

import json

def save_state(ContextInfo):
    """保存策略状态"""
    state = {'position': ContextInfo.position}
    with open('strategy_state.json', 'w') as f:
        json.dump(state, f)

def load_state(ContextInfo):
    """加载策略状态"""
    try:
        with open('strategy_state.json', 'r') as f:
            state = json.load(f)
            ContextInfo.position = state['position']
    except FileNotFoundError:
        ContextInfo.position = {}

错误处理

常见错误类型

  • 数据获取错误：品种不存在、数据未下载
  • 交易接口错误：权限不足、资金不够
  • 语法/运行时错误：代码 bug、类型错误

错误处理最佳实践

def handlebar(ContextInfo):
    try:
        close = ContextInfo.get_close_price()
        if close is None:
            print('警告：获取收盘价失败')
            return
        # 策略逻辑...
    except Exception as e:
        print(f'策略异常: {e}')
平台：QuantX