JoinQuant 策略风控：如何实时监控并控制最大回撤？

这是一个非常实用的风控需求。在聚宽（JoinQuant）中实现最大回撤止损，核心逻辑是**“高水位法”（High Water Mark）**。

我们需要在策略运行过程中，实时记录账户的历史最高权益（total_value），并计算当前权益相对于历史最高权益的跌幅。一旦跌幅超过设定的阈值（20%），则触发清仓操作并停止后续交易。

策略实现逻辑

1. 初始化：定义全局变量 g.max_value 记录历史最高资产，g.drawdown_limit 设定为 0.2（20%），g.stop_trading 标记是否已触发止损。
2. 实时监控：在 handle_data（分钟级或日级）中，不断更新 g.max_value。
3. 计算回撤：公式为 (历史最高资产 - 当前总资产) / 历史最高资产。
4. 触发风控：如果回撤 > 20%，执行清仓函数，并将 g.stop_trading 设为 True，阻止后续开仓。

完整策略代码

为了演示效果，我在代码中加入了一个简单的买入逻辑（全仓买入平安银行），以便在回测中产生持仓和波动。

# -*- coding: utf-8 -*-
import jqdata

def initialize(context):
    """
    初始化函数
    """
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    ### 股票相关设定 ###
    # 设定手续费
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    ### 风控参数设置 ###
    g.max_value = 0             # 记录账户历史最高权益
    g.drawdown_limit = 0.20     # 最大回撤阈值 (20%)
    g.stop_trading = False      # 止损触发标记，True表示已触发止损，停止交易
    
    ### 演示用变量 ###
    g.security = '000001.XSHE'  # 演示标的：平安银行

    # 每天开盘时运行交易逻辑
    run_daily(my_trade, 'every_bar')
    # 每天收盘后打印当日风控状态
    run_daily(print_risk_status, 'after_trading_end')

def handle_data(context, data):
    """
    按分钟回测时，每分钟都会调用；按天回测时，每天调用一次。
    这里用于实时监控账户净值和回撤。
    """
    # 1. 如果已经触发了止损，不再进行任何计算，直接返回
    if g.stop_trading:
        return

    # 2. 获取当前账户总资产
    current_value = context.portfolio.total_value
    
    # 3. 更新历史最高权益 (高水位)
    if current_value > g.max_value:
        g.max_value = current_value
        
    # 4. 计算当前回撤
    # 避免除以0的错误（虽然初始资金通常不为0）
    if g.max_value > 0:
        current_drawdown = (g.max_value - current_value) / g.max_value
    else:
        current_drawdown = 0
        
    # 5. 检查是否超过阈值
    if current_drawdown > g.drawdown_limit:
        log.error("【风控触发】当前回撤: {:.2%}，超过阈值: {:.2%}，执行清仓止损！".format(
            current_drawdown, g.drawdown_limit))
        
        # 执行清仓
        clear_all_positions(context)
        
        # 标记停止交易
        g.stop_trading = True

def my_trade(context):
    """
    演示用的交易函数
    """
    # 如果触发了止损，禁止开仓
    if g.stop_trading:
        return
        
    # 简单的买入逻辑：没持仓就全仓买入
    if len(context.portfolio.positions) == 0:
        order_value(g.security, context.portfolio.available_cash)
        log.info("买入标的: {}".format(g.security))

def clear_all_positions(context):
    """
    清仓函数：卖出所有持仓
    """
    positions = context.portfolio.positions
    for security in list(positions.keys()):
        order_target(security, 0)
        log.info("止损卖出: {}".format(security))

def print_risk_status(context):
    """
    盘后打印当前的回撤情况
    """
    if g.stop_trading:
        log.info("当前策略处于【止损停机】状态。")
    else:
        current_value = context.portfolio.total_value
        if g.max_value > 0:
            dd = (g.max_value - current_value) / g.max_value
            log.info("今日收盘总资产: {:.2f}, 历史最高: {:.2f}, 当前回撤: {:.2%}".format(
                current_value, g.max_value, dd))

代码关键点解析

1. g.max_value (高水位线):
   这是计算回撤的基准。我们在 handle_data 中实时比较 context.portfolio.total_value 和 g.max_value，始终保持 g.max_value 为历史最大值。

2. handle_data vs run_daily:

   • 风控逻辑写在 handle_data 中是最安全的。因为在分钟级回测中，盘中剧烈波动可能瞬间触发20%回撤。如果写在收盘后检查，可能会因为盘中跌幅过大而错过最佳止损点。
   • 即使是按天回测，handle_data 也会在每天执行一次，逻辑依然通用。

3. g.stop_trading (熔断机制):
   一旦触发止损，我们将此变量设为 True。在具体的交易函数（如 my_trade）中，第一步就是检查这个变量。如果为 True，则不再执行任何买入操作，防止清仓后策略又自动买入。

4. clear_all_positions:
   遍历 context.portfolio.positions 并使用 order_target(security, 0) 是标准的清仓写法。

常见问答 (Q&A)

Q1: 这个策略触发止损后，第二天会恢复交易吗？
A1: 不会。代码中使用了 g.stop_trading 变量。一旦变为 True，除非你手动修改代码或重启策略，否则策略将永久停止开仓。如果你希望在回撤恢复后（例如回撤缩小到10%以内）重新交易，需要修改逻辑，将 g.stop_trading 在满足特定条件时重置为 False。

Q2: 为什么回测结果里的最大回撤可能略大于20%？
A2: 这是正常的。

1. 滑点和手续费：卖出时会有交易成本。
2. 跌停板：如果股票跌停，无法卖出，资产会继续缩水。
3. 分钟级延迟：handle_data 在分钟回测中是每分钟结束时运行，如果某一分钟内瞬间暴跌，止损单会在下一分钟发出，成交价可能更低。

Q3: 如何只对单只股票止损，而不是全账户清仓？
A3: 本代码是针对账户总权益（Net Value）的止损。如果是针对单只股票，需要记录每只股票的 avg_cost（持仓成本）或该股票持仓期间的最高价，对比 current_price 来决定是否卖出该特定股票。