聚宽全天候自适应轮动：如何用 RSRS 过滤进攻与防守 ETF？

聚宽全天候自适应轮动策略设计方案

在量化交易中，将资金分配到不同的逻辑账户（如进攻账户和防守账户）独立运行，可以有效降低策略间的干扰。本方案基于聚宽（JoinQuant）平台，通过 set_subportfolios 初始化两个独立的子账户，并结合 RSRS（阻力支撑相对强度）斜率指标 进行大盘择时，配合日内止损、回撤止损和交易冷却机制，实现全天候自适应的 ETF 轮动。

一、 核心技术要点与聚宽 API 实现

1. 独立子账户配置 (SubPortfolio)

在 initialize 中，我们使用 set_subportfolios 创建两个子账户：

• 子账户 0（进攻账户）：专注于高 Beta 的宽基或行业 ETF（如创业板 ETF、半导体 ETF）。
• 子账户 1（防守账户）：专注于低 Beta、避险资产（如黄金 ETF、国债 ETF）。

def initialize(context):
    # 设定初始资金
    init_cash = context.portfolio.starting_cash
    # 进攻账户分配 60% 资金，防守账户分配 40% 资金
    set_subportfolios([
        SubPortfolioConfig(cash=init_cash * 0.6, type='stock'),
        SubPortfolioConfig(cash=init_cash * 0.4, type='stock')
    ])

2. RSRS 趋势择时指标

RSRS（Resistance Support Relative Strength）通过对过去 $N$ 天的最高价和最低价进行线性回归，利用斜率（Slope）的标准化值（Z-Score）来衡量阻力和支撑的相对强度。当 $Z > 0.7$ 时，进攻账户满仓进攻；当 $Z < -0.7$ 时，进攻账户空仓，防守账户启动避险 ETF 轮动。

3. 多重止损与冷却机制

• 日内止损：在 handle_data（分钟频）中监控，若单只 ETF 日内跌幅超过 3%，立即市价平仓。
• 回撤止损：监控子账户的 total_value，若从历史最高净值回撤超过 8%，触发清仓，并进入 3 个交易日的冷却期（冷却期内不买入）。

二、 完整策略源码（Python 3）

import numpy as np
import pandas as pd
from jqlib.technical_analysis import *

def initialize(context):
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    
    # 1. 初始化双子账户：子账户0(进攻), 子账户1(防守)
    init_cash = context.portfolio.starting_cash
    set_subportfolios([
        SubPortfolioConfig(cash=init_cash * 0.6, type='stock'),
        SubPortfolioConfig(cash=init_cash * 0.4, type='stock')
    ])
    
    # 2. 策略参数配置
    g.attack_etfs = ['510300.XSHG', '159915.XSHE', '512480.XSHG'] # 沪深300, 创业板, 半导体
    g.defense_etfs = ['511010.XSHG', '518880.XSHG'] # 国债ETF, 黄金ETF
    
    g.rsrs_index = '000300.XSHG' # RSRS 择时标的
    g.rsrs_N = 18 # 回归窗口
    g.rsrs_M = 600 # 标准化窗口
    
    # 止损与冷却变量
    g.cooling_days = 3
    g.cool_timer = [0, 0] # 两个账户的冷却计时器
    g.max_net_value = [0.0, 0.0] # 记录子账户历史最高净值
    
    # 定时运行
    run_daily(market_open, time='09:30')
    run_daily(stop_loss_monitor, time='every_bar') # 分钟级监控止损

def get_rsrs_zscore(context):
    # 获取历史高低价进行线性回归
    h = attribute_history(g.rsrs_index, g.rsrs_N + g.rsrs_M, '1d', ['high', 'low'])
    slopes = []
    for i in range(len(h) - g.rsrs_N + 1):
        sub_h = h.iloc[i:i+g.rsrs_N]
        X = sub_h['low'].values
        Y = sub_h['high'].values
        slope = np.polyfit(X, Y, 1)[0]
        slopes.append(slope)
    
    # 计算当前斜率的 Z-Score
    current_slope = slopes[-1]
    mean_slope = np.mean(slopes[:-1])
    std_slope = np.std(slopes[:-1])
    z_score = (current_slope - mean_slope) / std_slope
    return z_score

def market_open(context):
    # 每日更新最高净值与冷却计时
    for i in range(2):
        sub = context.subportfolios[i]
        if sub.total_value > g.max_net_value[i]:
            g.max_net_value[i] = sub.total_value
        if g.cool_timer[i] > 0:
            g.cool_timer[i] -= 1
            
    # 获取 RSRS 择时信号
    z_score = get_rsrs_zscore(context)
    
    # ---------------- 进攻账户逻辑 (pindex=0) ----------------
    sub_att = context.subportfolios[0]
    if g.cool_timer[0] == 0:
        if z_score > 0.7:
            # 轮动买入动量最强的进攻ETF
            target_etf = get_strongest_etf(g.attack_etfs, 20)
            adjust_portfolio(target_etf, pindex=0, context=context)
        elif z_score < -0.5:
            # 趋势转弱，清空进攻仓位
            clear_portfolio(pindex=0, context=context)
            
    # ---------------- 防守账户逻辑 (pindex=1) ----------------
    sub_def = context.subportfolios[1]
    if g.cool_timer[1] == 0:
        if z_score <= 0.7:
            # 避险时期，轮动防守资产
            target_etf = get_strongest_etf(g.defense_etfs, 10)
            adjust_portfolio(target_etf, pindex=1, context=context)
        else:
            # 市场极好时，防守账户亦可轻仓或空仓
            clear_portfolio(pindex=1, context=context)

def get_strongest_etf(etf_list, window):
    # 简单动量：计算过去 window 天涨幅最大的 ETF
    returns = {}
    for etf in etf_list:
        prices = attribute_history(etf, window, '1d', ['close'])
        returns[etf] = (prices['close'][-1] - prices['close'][0]) / prices['close'][0]
    return max(returns, key=returns.get)

def adjust_portfolio(target_etf, pindex, context):
    sub = context.subportfolios[pindex]
    # 卖出非目标ETF
    for stock in list(sub.long_positions.keys()):
        if stock != target_etf:
            order_target(stock, 0, pindex=pindex)
    # 买入目标ETF
    if target_etf not in sub.long_positions:
        order_value(target_etf, sub.available_cash, pindex=pindex)

def clear_portfolio(pindex, context):
    sub = context.subportfolios[pindex]
    for stock in list(sub.long_positions.keys()):
        order_target(stock, 0, pindex=pindex)

def stop_loss_monitor(context):
    # 分钟级多重止损监控
    for i in range(2):
        sub = context.subportfolios[i]
        
        # 1. 账户最大回撤止损 (从最高点回撤 8%)
        if g.max_net_value[i] > 0 and (sub.total_value / g.max_net_value[i] - 1.0) < -0.08:
            log.warning(f"子账户 {i} 触发最大回撤止损，清仓并冷却 {g.cooling_days} 天")
            clear_portfolio(pindex=i, context=context)
            g.cool_timer[i] = g.cooling_days
            g.max_net_value[i] = sub.total_value # 重置最高净值
            continue
            
        # 2. 单标的日内止损 (日内跌幅超过 3%)
        for stock, pos in sub.long_positions.items():
            current_data = get_current_data()
            day_open = current_data[stock].day_open
            last_price = current_data[stock].last_price
            if day_open > 0 and (last_price / day_open - 1.0) < -0.03:
                log.warning(f"标的 {stock} 触发日内止损，立即平仓")
                order_target(stock, 0, pindex=i)