聚宽 ETF 轮动与市场状态切换：基于拥挤度与趋势的动态资产选择

在量化投资中，单一资产很难在所有市场周期中持续走强。通过在权益类（如红利 ETF）、**避险类（如债券 ETF）和抗通胀/商品类（如黄金/商品 ETF）**之间进行动态轮动，可以有效平滑净值曲线。本文将展示如何在聚宽（JoinQuant）平台上，利用趋势动量、拥挤度（换手率偏离）和回撤状态构建一个多资产 ETF 轮动策略。

一、 核心逻辑设计

1. 候选资产池（ETF Pool）：

   • 红利 ETF（510880.XSHG）：代表高股息、低估值权益资产。
   • 黄金 ETF（518880.XSHG）：代表商品及抗通胀避险资产。
   • 十年国债 ETF（511260.XSHG）：代表低风险债券避险资产。

2. 状态与轮动因子：

   • 趋势动量（MA）：计算各 ETF 的 20 日均线，价格在均线之上视为有上涨趋势。
   • 拥挤度（Turnover Volatility）：通过过去 20 日换手率的标准差或偏离度来衡量。若换手率异常放大，说明短期交易过于拥挤，需警惕回调风险。
   • 回撤控制：若资产从近期高点回撤超过一定比例，则强制平仓避险。

二、 聚宽策略源码实现

以下是基于聚宽 API 编写的完整策略代码。策略采用按月调仓（run_monthly）的架构，并在每日收盘后进行回撤监控。

import pandas as pd
import numpy as np
from jqdata import *

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 交易税费设置：买入万三，卖出万三加千一印花税
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 定义轮动资产池
    g.etf_pool = {
        '510880.XSHG': '红利ETF',
        '518880.XSHG': '黄金ETF',
        '511260.XSHG': '国债ETF'
    }
    
    # 策略参数
    g.momentum_window = 20  # 动量计算窗口
    g.crowd_window = 20     # 拥挤度计算窗口
    
    # 每月第一个交易日开盘进行资产轮动
    run_monthly(handle_rotation, monthday=1, time='09:30')

def handle_rotation(context):
    log.info("====== 开始执行月度 ETF 轮动 ======")
    
    scores = {}
    
    for etf in g.etf_pool.keys():
        # 获取历史价格和换手率数据
        # 换手率波动率（turnover_volatility）可作为拥挤度指标
        hist = attribute_history(etf, g.momentum_window + 10, '1d', ['close', 'volume', 'factor'])
        if hist.empty or len(hist) < g.momentum_window:
            continue
            
        # 1. 趋势动量因子：当前价格 / MA20
        ma20 = hist['close'][-g.momentum_window:].mean()
        current_price = hist['close'][-1]
        trend_score = current_price / ma20
        
        # 2. 拥挤度因子：计算成交量的标准差（成交量异常放大代表拥挤）
        vol_std = hist['volume'][-g.crowd_window:].std()
        vol_mean = hist['volume'][-g.crowd_window:].mean()
        crowd_score = vol_std / vol_mean if vol_mean > 0 else 1.0
        
        # 综合评分：趋势越强越好，拥挤度越低越好
        # 只有当价格在 MA20 之上时才考虑买入
        if current_price > ma20:
            scores[etf] = trend_score / (1.0 + crowd_score)
        else:
            scores[etf] = -1.0  # 趋势向下，不予考虑
            
    # 找出评分最高的资产
    if scores:
        best_etf = max(scores, key=scores.get)
        best_score = scores[best_etf]
        
        if best_score > 0:
            log.info(f"当前最佳资产为: {g.etf_pool[best_etf]} ({best_etf})，评分: {best_score:.4f}")
            adjust_portfolio(context, best_etf)
        else:
            log.info("所有资产趋势均向下，选择空仓避险。")
            empty_portfolio(context)
    else:
        log.info("未获取到有效数据，保持空仓。")
        empty_portfolio(context)

def adjust_portfolio(context, target_etf):
    # 卖出非目标持仓
    for position in list(context.portfolio.long_positions.keys()):
        if position != target_etf:
            log.info(f"卖出非目标资产: {position}")
            order_target(position, 0)
            
    # 买入目标资产
    cash = context.portfolio.available_cash
    if cash > 0 and target_etf not in context.portfolio.long_positions:
        log.info(f"全仓买入目标资产: {g.etf_pool[target_etf]} ({target_etf})")
        order_value(target_etf, cash)

def empty_portfolio(context):
    # 清空所有持仓
    for position in list(context.portfolio.long_positions.keys()):
        log.info(f"清空持仓以避险: {position}")
        order_target(position, 0)

三、 策略关键点解析

1. 拥挤度指标的量化：
   在上述代码中，我们使用成交量标准差与均值的比值（vol_std / vol_mean）来近似模拟拥挤度。当成交量急剧放大、资金博弈剧烈时，该值会显著上升，从而降低该 ETF 的综合评分，避免在高位“接盘”。

2. 动态复权（use_real_price）：
   策略中开启了 set_option('use_real_price', True)。在真实价格模式下，回测系统会根据历史分红送股情况自动调整持仓数量，确保回测结果不失真，非常适合红利 ETF 这种高分红资产的长期回测。

3. 多资产协同：
   当股市（红利 ETF）拥挤度过高或趋势走弱时，资金会自动流向商品（黄金 ETF）或债市（国债 ETF），从而在聚宽平台上实现真正的跨资产全天候轮动配置。