聚宽 ETF 轮动策略：如何实现日内回撤保护与弱势市场切换？

在量化交易中，ETF 轮动策略因其交易成本低、覆盖面广而备受青睐。然而，在面临系统性风险或多市场同时走弱时，单纯的轮动往往会失效。本文将介绍如何在聚宽（JoinQuant）平台上，构建一个融合了多市场弱势判断、日内回撤守卫、冷却期机制与防御资产切换的高回撤控制 ETF 轮动策略。

一、 核心设计思路

1. 多市场弱势状态判断：
   • A 股市场：以沪深300指数（000300.XSHG）为基准，当其价格低于 20 日均线时，判定 A 股处于弱势。
   • 海外市场：以纳指 ETF（513100.XSHG）或标普500 ETF（513500.XSHG）为基准，当其价格低于 20 日均线时，判定海外市场处于弱势。
2. 防御资产切换：
   • 当 A 股和海外市场同时判定为弱势时，策略不再持有权益类 ETF，而是全仓切换至防御资产（如国债 ETF 511010.XSHG 或货币 ETF 511880.XSHG）。
3. 日内回撤守卫（回撤保护）：
   • 实时监控持仓标的的日内价格。若某只 ETF 相比于日内最高价回撤超过设定的阈值（如 3%），则立即触发日内止损，强制平仓。
4. 冷却期机制：
   • 一旦某只标的触发日内止损，该标的将进入为期 $N$ 个交易日的“冷却期”，在冷却期内，策略将不再买入该标的，避免频繁止损带来的摩擦成本。

二、 聚宽 Python 策略源码实现

以下是基于聚宽 API 实现的完整策略代码。策略采用分钟频运行，以支持日内回撤守卫的实时监控。

import jqdata
import numpy as np
import pandas as pd

def initialize(context):
    # 开启真实价格模式
    set_option('use_real_price', True)
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    
    # 交易费率设置
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 策略参数配置
    g.asset_pool = [
        '510300.XSHG', # 沪深300 ETF
        '510500.XSHG', # 中证500 ETF
        '159915.XSHE', # 创业板 ETF
        '513100.XSHG', # 纳指 ETF
    ]
    g.defense_asset = '511880.XSHG' # 防御资产：银华日利货币 ETF
    
    g.market_a = '000300.XSHG'      # A股基准
    g.market_global = '513100.XSHG' # 海外基准
    
    g.drawdown_threshold = 0.03     # 日内回撤止损阈值：3%
    g.cool_down_days = 5            # 止损后的冷却期：5天
    
    # 记录状态的全局变量
    g.high_prices = {}              # 记录持仓标的的日内最高价
    g.cooldown_dict = {}            # 记录标的的冷却截止日期
    
    # 定时运行：每天开盘前重置日内最高价，更新冷却期
    run_daily(before_market_open, time='09:00')
    # 定时运行：每天收盘前 10 分钟进行轮动决策
    run_daily(rebalance, time='14:50')
    # 定时运行：每分钟监控日内回撤
    run_daily(monitor_drawdown, time='every_bar')

def before_market_open(context):
    # 重置日内最高价记录
    g.high_prices = {}
    for security in context.portfolio.positions.keys():
        current_data = get_current_data()
        if security in current_data:
            g.high_prices[security] = current_data[security].day_open
            
    # 打印当前的冷却期状态
    today = context.current_dt.date()
    g.cooldown_dict = {k: v for k, v in g.cooldown_dict.items() if v > today}
    if g.cooldown_dict:
        log.info(f"当前处于冷却期的标的: {g.cooldown_dict}")

def is_market_weak(context):
    """判断 A 股和海外市场是否同时处于弱势"""
    # 获取 A 股和海外基准的 20 日均线
    hist_a = attribute_history(g.market_a, 20, '1d', ['close'])
    hist_global = attribute_history(g.market_global, 20, '1d', ['close'])
    
    ma_a = hist_a['close'].mean()
    ma_global = hist_global['close'].mean()
    
    curr_a = hist_a['close'][-1]
    curr_global = hist_global['close'][-1]
    
    # 如果两者均低于均线，则判定为双重弱势
    return (curr_a < ma_a) and (curr_global < ma_global)

def rebalance(context):
    """每日轮动决策"""
    today = context.current_dt.date()
    
    # 1. 判断市场是否双重走弱
    if is_market_weak(context):
        log.info("检测到 A 股与海外市场同时走弱，触发防御机制，切换至防御资产。")
        # 卖出所有非防御资产
        for security in list(context.portfolio.positions.keys()):
            if security != g.defense_asset:
                order_target(security, 0)
        # 买入防御资产
        cash = context.portfolio.available_cash
        if cash > 10000:
            order_value(g.defense_asset, cash)
        return

    # 2. 正常轮动逻辑（动量评分：过去 20 天涨幅）
    scores = {}
    for security in g.asset_pool:
        # 排除处于冷却期的标的
        if security in g.cooldown_dict and g.cooldown_dict[security] > today:
            continue
        hist = attribute_history(security, 20, '1d', ['close'])
        increase = (hist['close'][-1] - hist['close'][0]) / hist['close'][0]
        scores[security] = increase
        
    if not scores:
        return
        
    # 选出动量最强的标的
    best_target = max(scores, key=scores.get)
    
    # 如果最强标的动量为负，也切换至防御资产
    if scores[best_target] <= 0:
        best_target = g.defense_asset
        
    # 执行调仓
    for security in list(context.portfolio.positions.keys()):
        if security != best_target:
            order_target(security, 0)
            
    cash = context.portfolio.available_cash
    if best_target not in context.portfolio.positions and cash > 10000:
        order_value(best_target, cash)
        log.info(f"轮动调仓：买入最强动量标的 {best_target}")

def monitor_drawdown(context):
    """日内回撤守卫：每分钟监控持仓回撤"""
    current_data = get_current_data()
    today = context.current_dt.date()
    
    for security in list(context.portfolio.positions.keys()):
        # 防御资产不参与日内止损
        if security == g.defense_asset:
            continue
            
        if security in current_data:
            price = current_data[security].last_price
            
            # 更新日内最高价
            if security not in g.high_prices:
                g.high_prices[security] = price
            else:
                g.high_prices[security] = max(g.high_prices[security], price)
                
            # 计算自日内最高点的回撤
            high_price = g.high_prices[security]
            drawdown = (high_price - price) / high_price
            
            # 触发日内回撤保护
            if drawdown >= g.drawdown_threshold:
                log.warning(f"标的 {security} 日内回撤达 {drawdown:.2%}，触发回撤守卫！立即平仓并进入冷却期。")
                order_target(security, 0)
                # 设定冷却期截止日期
                g.cooldown_dict[security] = today + datetime.timedelta(days=g.cool_down_days)

三、 策略风控机制深度解析

1. 双重弱势过滤器：
   策略通过 is_market_weak 函数，同时监控 A 股（沪深300）与海外（纳指）的趋势。当全球核心权益资产均跌破 20 日均线时，表明系统性风险极高，策略果断放弃轮动，全仓买入 511880（货币 ETF）避险。这能有效避免在“泥沙俱下”的熊市中反复割肉。
2. 日内回撤守卫（High-Water Mark）：
   在 monitor_drawdown 中，策略利用 get_current_data() 获取分钟级最新价，并动态维护一个日内最高价 g.high_prices。一旦价格从日内高点回撤超 3%，立即市价单平仓。这能有效抵御日内“天地针”或突发利空暴跌。
3. 冷却期（Cool-down Period）：
   止损是痛苦的，而频繁在震荡市中“止损后又买回”会导致资金曲线迅速失血。引入 g.cooldown_dict 后，被止损的 ETF 将被强制“禁赛” 5 个交易日，给市场情绪一个缓冲期，从而大幅提升了风控的实用性。