QMT ContextInfo 变量状态与回退问题

ContextInfo 是 QMT 策略的上下文对象,挂载在其上的自定义变量在策略运行期间保持状态。但在回测环境中存在状态回退问题:当 Bar 回放时,某些变量的值可能被重置为之前的值,导致策略逻辑出现非预期行为。理解状态回退的原因并采取正确的管理方式,是编写可靠策略的关键。

QMT ContextInfo 作为状态容器的用法

ContextInfo 对象贯穿策略的整个生命周期,在 init、handlebar、subscribe 回调、run_time 回调之间共享。开发者可以在 ContextInfo 上挂载自定义变量来保存策略状态,如持仓记录、信号标记、计算缓存等。

ContextInfo 充当策略的"全局状态容器"。在 init 中初始化的变量(如 ContextInfo.counter = 0),在后续的 handlebar 中可以直接读取和修改。这种机制使得策略可以在不同回调之间传递状态,而无需使用 Python 全局变量。ContextInfo 上的变量在策略运行期间保持,直到策略停止。

小建议 将所有需要跨回调共享的状态统一挂载在 ContextInfo 上,而不是使用 Python 全局变量。在 init 中集中初始化所有状态变量,便于管理和排查。为变量使用清晰的命名,如 ContextInfo.positionContextInfo.last_signal 等。

def init(ContextInfo):
    # 在 init 中集中初始化所有状态变量
    ContextInfo.set_universe(['000001.SZ'])
    ContextInfo.position = {}           # 持仓记录
    ContextInfo.last_signal = None      # 上次信号
    ContextInfo.counter = 0             # 计数器
    ContextInfo.cached_ma = None        # 计算缓存
    print('状态变量初始化完成')

def handlebar(ContextInfo):
    if not ContextInfo.is_last_bar():
        return
    # 读取和修改状态
    ContextInfo.counter += 1
    close = ContextInfo.get_close_price()
    ContextInfo.last_signal = f'bar_{ContextInfo.counter}_close_{close}'
    print(f'当前状态: {ContextInfo.last_signal}')

QMT 普通全局变量与 ContextInfo 变量

QMT 策略中可以使用两种方式保存跨回调状态:Python 全局变量和 ContextInfo 自定义属性。两者在策略运行期间都能保持状态,但在管理方式和可维护性上存在差异。

Python 全局变量定义在函数外部,在整个策略文件中可访问。ContextInfo 变量挂载在 ContextInfo 对象上,通过 ContextInfo.xxx 访问。Python 全局变量的生命周期与策略文件相同,ContextInfo 变量的生命周期与 ContextInfo 对象相同。在 QMT 中,推荐使用 ContextInfo 变量,因为它与策略上下文绑定更紧密,且在调试时可以通过 ContextInfo 查看所有状态。

小建议 统一使用 ContextInfo 变量管理状态,避免混用 Python 全局变量。Python 全局变量仅用于真正的常量定义(如 MAX_POSITION = 10),不用于可变状态。这样可以使状态管理更加清晰,减少因变量作用域混淆导致的 bug。

# 常量使用 Python 全局变量(不可变)
MAX_POSITION = 10
STOP_LOSS_RATIO = 0.05

# 可变状态使用 ContextInfo 变量
def init(ContextInfo):
    ContextInfo.current_position = 0
    ContextInfo.entry_price = 0

def handlebar(ContextInfo):
    if not ContextInfo.is_last_bar():
        return
    close = ContextInfo.get_close_price()
    
    # 读取状态
    if ContextInfo.current_position > 0:
        # 使用常量计算止损价
        stop_price = ContextInfo.entry_price * (1 - STOP_LOSS_RATIO)
        if close < stop_price:
            print(f'触发止损,当前价: {close}, 止损价: {stop_price}')
            ContextInfo.current_position = 0
    
    # 修改状态
    if ContextInfo.current_position == 0 and close > 10:
        ContextInfo.current_position = MAX_POSITION
        ContextInfo.entry_price = close
        print(f'建仓,价格: {close}')

QMT ContextInfo 状态回退问题的原因

在回测环境中,QMT 按时间顺序逐根回放 K 线。回测引擎在处理每根 K 线时,会将 ContextInfo 的内置属性(如 barpos、timetag、stockcode 等)设置为当前 K 线对应的值。在某些情况下,ContextInfo 上的自定义变量可能出现状态回退——即变量的值被恢复为之前某个时刻的值。

状态回退的根本原因是回测引擎对 ContextInfo 状态的管理机制。回测引擎在处理每根 K 线时,可能会对 ContextInfo 进行状态快照和恢复。当策略使用多周期数据、多品种数据或回测引擎需要重新处理某根 K 线时,ContextInfo 的自定义变量可能被恢复到快照时的值,而非最近修改后的值。这会导致策略逻辑出现非预期行为,例如计数器被重置、持仓记录丢失等。

小建议 状态回退问题在复杂策略(多周期、多品种)中更容易出现。如果策略出现"变量值莫名其妙被重置"的现象,首先怀疑状态回退问题。通过在 handlebar 中添加日志记录关键变量的值,可以确认是否发生了状态回退。

def handlebar(ContextInfo):
    # 添加状态日志,排查回退问题
    if hasattr(ContextInfo, 'counter'):
        print(f'barpos={ContextInfo.barpos}, counter={ContextInfo.counter}')
    else:
        print(f'barpos={ContextInfo.barpos}, counter 不存在(可能被回退)')
    
    if not ContextInfo.is_last_bar():
        return
    
    # 初始化或递增计数器
    if not hasattr(ContextInfo, 'counter'):
        ContextInfo.counter = 0
    ContextInfo.counter += 1

QMT 避免 ContextInfo 状态回退的方法

避免状态回退的核心思路是将关键状态存储在不受回测引擎快照恢复机制影响的位置。常见方法包括:使用 Python 全局变量存储关键状态、使用文件持久化、在 handlebar 中进行状态校验和修复。

Python 全局变量存储在模块级别,不受 ContextInfo 快照恢复机制的影响,因此不会发生状态回退。但全局变量的缺点是管理不够规范。另一种方法是在每次 handlebar 执行时检查状态的一致性,如果发现状态被回退则重新计算或从备份恢复。最可靠的方法是将关键状态持久化到文件,在每次需要时从文件读取。

小建议 对于可能受状态回退影响的关键变量(如持仓记录、交易计数),使用 Python 全局变量作为备份。在 handlebar 中检测 ContextInfo 变量是否被回退,如果被回退则从全局变量备份中恢复。对于需要跨策略重启保持的状态,使用文件持久化。

# Python 全局变量作为状态备份(不受回退影响)
_position_backup = {}
_counter_backup = 0

def init(ContextInfo):
    global _position_backup, _counter_backup
    ContextInfo.position = {}
    ContextInfo.counter = 0
    _position_backup = ContextInfo.position
    _counter_backup = 0

def handlebar(ContextInfo):
    global _counter_backup
    
    if not ContextInfo.is_last_bar():
        return
    
    # 检测状态回退:如果 counter 小于备份值,说明发生了回退
    if not hasattr(ContextInfo, 'counter') or ContextInfo.counter < _counter_backup:
        print(f'检测到状态回退,从备份恢复: counter {getattr(ContextInfo, "counter", "N/A")} -> {_counter_backup}')
        ContextInfo.counter = _counter_backup
    
    # 正常逻辑
    ContextInfo.counter += 1
    _counter_backup = ContextInfo.counter  # 更新备份
    print(f'当前 counter: {ContextInfo.counter}')

QMT 推荐的状态管理方式

综合考虑状态持久性、可维护性和抗回退能力,推荐的状态管理策略是:ContextInfo 变量用于常规状态管理,Python 全局变量用于关键状态的抗回退备份,文件持久化用于跨重启状态保存。

三层状态管理各有分工:ContextInfo 变量是主要的状态载体,日常读写都在 ContextInfo 上进行;Python 全局变量作为关键状态的"安全网",在 ContextInfo 发生回退时提供恢复依据;文件持久化在策略停止和重启之间保持状态连续性。这种分层管理既保证了日常开发的便捷性,又提供了状态安全保障。

小建议 按以下原则管理策略状态:常规状态(信号标记、计算缓存)仅用 ContextInfo 变量即可;关键状态(持仓记录、订单状态)同时使用 ContextInfo 变量和 Python 全局变量备份;需要跨重启保持的状态(累计收益、历史交易记录)使用文件持久化。在 init 中统一初始化三层状态,在 handlebar 中先检测回退再执行逻辑。

import json

# === 第一层:Python 全局变量(抗回退备份)===
_position_backup = {}
_pnl_backup = 0.0

# === 常量定义 ====
MAX_POSITION = 1000
STATE_FILE = 'strategy_state.json'

def init(ContextInfo):
    global _position_backup, _pnl_backup
    
    # === 第二层:ContextInfo 变量(主要状态)===
    ContextInfo.position = {}
    ContextInfo.total_pnl = 0.0
    ContextInfo.counter = 0
    
    # === 第三层:文件持久化(跨重启状态)===
    try:
        with open(STATE_FILE, 'r') as f:
            state = json.load(f)
            ContextInfo.position = state.get('position', {})
            ContextInfo.total_pnl = state.get('total_pnl', 0.0)
            print(f'从文件恢复状态: {state}')
    except (FileNotFoundError, json.JSONDecodeError):
        print('未找到状态文件,使用默认初始化')
    
    # 同步三层状态
    _position_backup = dict(ContextInfo.position)
    _pnl_backup = ContextInfo.total_pnl

def handlebar(ContextInfo):
    global _position_backup, _pnl_backup
    
    if not ContextInfo.is_last_bar():
        return
    
    # 检测状态回退
    if ContextInfo.total_pnl < _pnl_backup:
        print(f'检测到状态回退,恢复 total_pnl: {ContextInfo.total_pnl} -> {_pnl_backup}')
        ContextInfo.total_pnl = _pnl_backup
        ContextInfo.position = dict(_position_backup)
    
    # 策略逻辑
    ContextInfo.counter += 1
    close = ContextInfo.get_close_price()
    
    # 更新备份
    _position_backup = dict(ContextInfo.position)
    _pnl_backup = ContextInfo.total_pnl

def save_state(ContextInfo):
    """在策略停止前调用,保存状态到文件"""
    state = {
        'position': ContextInfo.position,
        'total_pnl': ContextInfo.total_pnl,
    }
    with open(STATE_FILE, 'w') as f:
        json.dump(state, f)
    print(f'状态已保存: {state}')