QMT 策略风控：如何实现移动止损（Trailing Stop）？

这是一个基于 QMT (QuantTrader) 平台的 移动止损（Trailing Stop）策略。

策略逻辑说明

移动止损（追踪止损）是一种动态风险管理机制：

1. 入场：为了演示移动止损，我设置了一个简单的 双均线金叉（短期均线上穿长期均线）作为买入信号。
2. 记录高点：买入后，策略会持续记录持仓期间的最高价格。
3. 移动止损线：止损价格 = 持仓期间最高价 * (1 - 回撤比例)。
   • 如果价格上涨，最高价更新，止损线随之上移（锁定利润）。
   • 如果价格下跌，最高价不变，止损线保持不变。
4. 出场：当当前价格跌破移动止损线时，触发卖出信号，清仓离场。

策略代码

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

def init(ContextInfo):
    """
    初始化函数，设定策略参数和变量
    """
    # 1. 设置资金账号 (请替换为您自己的实盘或模拟账号)
    ContextInfo.accID = '600000248' 
    ContextInfo.set_account(ContextInfo.accID)
    
    # 2. 策略参数设置
    ContextInfo.short_period = 5   # 短期均线周期
    ContextInfo.long_period = 20   # 长期均线周期
    ContextInfo.trailing_gap = 0.05 # 移动止损回撤比例 (5%)
    
    # 3. 全局变量，用于记录持仓股票的最高价
    # 格式: {'stock_code': highest_price}
    ContextInfo.holding_highs = {} 
    
    print("策略初始化完成，移动止损阈值: {:.1%}".format(ContextInfo.trailing_gap))

def get_holding_volume(ContextInfo, stock_code):
    """
    辅助函数：获取当前持仓的可用股数
    """
    # 获取持仓信息
    positions = get_trade_detail_data(ContextInfo.accID, 'stock', 'position')
    for pos in positions:
        if pos.m_strInstrumentID == stock_code:
            return pos.m_nCanUseVolume # 返回可用持仓
    return 0

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前K线索引
    index = ContextInfo.barpos
    # 获取当前图表的时间
    realtime = ContextInfo.get_bar_timetag(index)
    
    # 获取当前主图的股票代码列表 (如果在界面选择了多只股票)
    # 如果是回测，通常在 set_universe 中设置，这里取当前Context的股票池
    stock_list = ContextInfo.get_universe()

    for stock in stock_list:
        # --- 1. 数据获取 ---
        # 获取足够的历史数据用于计算均线 (多取一点防止数据不足)
        data_count = ContextInfo.long_period + 5
        
        # 使用 get_market_data_ex 获取数据 (效率更高)
        # 获取 收盘价(close) 和 最高价(high)
        market_data = ContextInfo.get_market_data_ex(
            ['close', 'high'], 
            [stock], 
            period=ContextInfo.period, 
            count=data_count,
            dividend_type='front' # 前复权
        )
        
        if stock not in market_data or len(market_data[stock]) < ContextInfo.long_period:
            continue
            
        df = market_data[stock]
        close_prices = df['close']
        high_prices = df['high']
        
        # 当前最新价格
        current_price = close_prices.iloc[-1]
        current_high = high_prices.iloc[-1]
        
        # --- 2. 移动止损逻辑 (卖出逻辑) ---
        # 检查是否在我们的持仓记录中
        # 注意：实盘中建议结合 get_trade_detail_data 校验真实持仓
        current_holding_vol = get_holding_volume(ContextInfo, stock)
        
        if current_holding_vol > 0:
            # 如果持仓记录中没有该股票（可能是手动买入或初始化前买入），初始化最高价为当前价
            if stock not in ContextInfo.holding_highs:
                ContextInfo.holding_highs[stock] = current_price
            
            # A. 更新最高价：如果当前最高价创出新高，则更新记录
            record_high = ContextInfo.holding_highs[stock]
            if current_high > record_high:
                ContextInfo.holding_highs[stock] = current_high
                # print(f"{stock} 创新高: {current_high}, 止损线提升")
            
            # B. 计算动态止损价
            # 止损价 = 历史最高价 * (1 - 回撤比例)
            stop_loss_price = ContextInfo.holding_highs[stock] * (1 - ContextInfo.trailing_gap)
            
            # C. 检查是否触发止损
            if current_price < stop_loss_price:
                print(f"【移动止损触发】{stock} 当前价: {current_price} < 止损线: {stop_loss_price:.2f} (最高价: {ContextInfo.holding_highs[stock]})")
                
                # 执行卖出 (24: 卖出, 1101: 单股/手, 5: 最新价)
                passorder(24, 1101, ContextInfo.accID, stock, 5, -1, current_holding_vol, ContextInfo)
                
                # 清除最高价记录
                del ContextInfo.holding_highs[stock]
                continue # 卖出后跳过后续买入判断

        # --- 3. 买入逻辑 (双均线金叉) ---
        # 只有未持仓时才开仓
        if current_holding_vol == 0:
            # 计算均线
            ma_short = close_prices.rolling(window=ContextInfo.short_period).mean()
            ma_long = close_prices.rolling(window=ContextInfo.long_period).mean()
            
            # 获取最后两个点判断金叉
            if len(ma_short) >= 2 and len(ma_long) >= 2:
                curr_s = ma_short.iloc[-1]
                prev_s = ma_short.iloc[-2]
                curr_l = ma_long.iloc[-1]
                prev_l = ma_long.iloc[-2]
                
                # 金叉判断: 短线上穿长线
                if prev_s <= prev_l and curr_s > curr_l:
                    print(f"【金叉买入】{stock} 价格: {current_price}")
                    
                    # 执行买入 (23: 买入, 1101: 单股/手, 5: 最新价, 1000股)
                    passorder(23, 1101, ContextInfo.accID, stock, 5, -1, 1000, ContextInfo)
                    
                    # 买入后，立即初始化该股票的最高价为当前价格
                    ContextInfo.holding_highs[stock] = current_price

代码关键点解析

1. ContextInfo.holding_highs 字典：

   • 这是实现移动止损的核心。它存储了我们持仓每只股票自买入以来的历史最高价。
   • ContextInfo 对象在策略运行期间是持久存在的，所以这个字典可以跨 K 线保存状态。

2. 止损线计算：

   • stop_loss_price = ContextInfo.holding_highs[stock] * (1 - ContextInfo.trailing_gap)
   • 这里 trailing_gap 设为 0.05，意味着当价格从最高点回撤 5% 时触发卖出。

3. 更新机制：

   • 在每个 handlebar 周期中，如果持有股票，首先检查 current_high（当前K线最高价）是否大于记录的 record_high。如果是，则更新记录，从而抬高止损线。

4. 交易函数 passorder：

   • opType=23 为买入，opType=24 为卖出。
   • prType=5 代表使用最新价下单。
   • 注意：在实盘中，请确保 ContextInfo.accID 填写正确。

如何在 QMT 中使用

1. 打开 QMT 客户端，进入 模型研究/策略编辑器。
2. 新建一个 Python 策略。
3. 将上述代码复制粘贴进去。
4. 修改 ContextInfo.accID 为你的实际账号。
5. 在右侧设置好回测参数（如资金、费率等）或直接在行情图上运行。
6. 注意：get_market_data_ex 获取的数据依赖于你是否下载了本地历史数据。如果是回测，请务必先在“数据管理”中补充数据。

常见问题 Q&A

Q: 为什么使用 get_market_data_ex 而不是 get_market_data?
A: get_market_data_ex 是 QMT 较新的接口，返回的是 Pandas DataFrame 格式，处理起来更方便，且性能通常更好。

Q: 这个策略在回测和实盘中都能用吗？
A: 是的。代码中使用了 get_trade_detail_data 来获取持仓，这在实盘和回测（如果开启了撮合匹配）中都是通用的。但在纯信号回测模式下，可能需要依赖模拟持仓变量。上述代码偏向于实盘或模拟盘风格。

Q: 如何修改回撤比例？
A: 修改 init 函数中的 ContextInfo.trailing_gap = 0.05。例如改为 0.10 即为 10% 移动止损。