聚宽五福 ETF 轮动策略：结合动量、R² 拟合度与多重过滤的量化实现

在量化投资中，ETF 轮动是一种非常经典的动量交易策略。传统的动量策略仅关注收益率，容易买在趋势末端或陷入高波动震荡。本文介绍的五福 ETF 轮动策略，通过年化收益率 × 趋势拟合度 $R^2$ 的复合指标进行打分，筛选出不仅涨得快、而且涨得稳（线性趋势好）的 ETF。同时，策略加入了均线过滤、成交量筛选、折溢价控制以及个股止损等多重风控机制。

一、 核心逻辑与打分机制

1. 动量与 $R^2$ 打分：
   • 收集过去 $N$ 天（如 20 天）的每日收盘价。
   • 对价格序列（或对数价格）与时间序列进行一元线性回归，计算斜率（代表收益率）和拟合优度 $R^2$（代表趋势的平滑度，越接近 1 说明趋势越稳健，噪音越小）。
   • 得分 = 年化收益率 × $R^2$。
2. 多重过滤机制：
   • 均线过滤：标的价格必须在 20 日均线之上，确保处于上升通道。
   • 成交量过滤：过去 20 日日均成交额需大于指定阈值（如 1000 万），避免流动性风险。
   • 溢价率过滤：通过 get_extras 获取基金单位净值（unit_net_value），计算溢价率 (收盘价 - 单位净值) / 单位净值。若溢价率过高（如 > 1.5%），则不予买入，防止买贵。
   • 个股止损：持仓标的若相比买入成本回撤超过一定比例（如 5%），触发主动止损。

二、 聚宽策略源码实现

以下是基于聚宽 API 编写的完整策略代码：

import numpy as np
import pandas as pd
from scipy.stats import linregress
from jqdata import *

def initialize(context):
    # 开启真实价格模式
    set_option('use_real_price', True)
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    
    # 策略参数设置
    g.etf_pool = [
        '510300.XSHG', # 沪深300ETF
        '510500.XSHG', # 中证500ETF
        '159915.XSHE', # 创业板ETF
        '518880.XSHG', # 黄金ETF
        '513100.XSHG', # 纳指ETF
    ]
    
    g.lookback_window = 20      # 动量计算回溯窗口
    g.max_hold = 2              # 最大同时持仓数
    g.volume_limit = 10000000   # 20日均成交额过滤（1000万）
    g.premium_limit = 0.015     # 溢价率上限（1.5%）
    g.stop_loss_limit = 0.05    # 个股止损阈值（5%）
    
    # 运行规划：每周第一个交易日开盘进行轮动调仓
    run_weekly(rebalance, weekday=1, time='09:30')
    # 每日盘中进行止损检查
    run_daily(check_stop_loss, time='14:50')

def get_etf_score(security, end_date):
    """计算年化收益率 * R²"""
    # 获取历史收盘价和成交额
    bars = get_bars(security, count=g.lookback_window, unit='1d', 
                    fields=['date', 'close', 'money'], include_now=True, end_dt=end_date, df=True)
    
    if len(bars) < g.lookback_window:
        return -999, False
        
    # 1. 成交量过滤
    if bars['money'].mean() < g.volume_limit:
        return -999, False
        
    close_prices = bars['close'].values
    # 2. 均线过滤：当前价格必须在20日均线之上
    ma20 = close_prices.mean()
    current_price = close_prices[-1]
    if current_price < ma20:
        return -999, False
        
    # 3. 计算 R² 和年化收益率
    x = np.arange(len(close_prices))
    y = np.log(close_prices) # 使用对数价格拟合
    slope, intercept, r_value, p_value, std_err = linregress(x, y)
    
    r_squared = r_value ** 2
    annual_return = slope * 244 # 估算年化对数收益率
    score = annual_return * r_squared
    
    return score, True

def check_premium(security, current_date):
    """溢价率过滤"""
    try:
        # 获取最新收盘价
        current_data = get_current_data()
        close_price = current_data[security].last_price
        # 获取单位净值
        net_value_df = get_extras('unit_net_value', [security], start_date=current_date, end_date=current_date)
        if not net_value_df.empty:
            net_value = net_value_df[security].iloc[-1]
            premium_rate = (close_price - net_value) / net_value
            if premium_rate > g.premium_limit:
                return False # 溢价过高，过滤
    except Exception as e:
        log.warn(f"计算 {security} 溢价率失败: {e}")
    return True

def rebalance(context):
    """轮动调仓逻辑"""
    current_date = context.current_dt.strftime('%Y-%m-%d')
    scores = {}
    
    # 评估所有候选 ETF
    for etf in g.etf_pool:
        score, passed = get_etf_score(etf, current_date)
        if passed and check_premium(etf, current_date):
            scores[etf] = score
            
    # 按得分降序排序
    sorted_etfs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    target_list = [etf for etf, score in sorted_etfs[:g.max_hold] if score > 0]
    
    # 卖出不在目标持仓中的ETF
    current_holdings = list(context.portfolio.positions.keys())
    for etf in current_holdings:
        if etf not in target_list:
            order_target_value(etf, 0)
            log.info(f"[调仓] 卖出 {etf}")
            
    # 买入目标持仓
    if target_list:
        target_value = context.portfolio.total_value / len(target_list)
        for etf in target_list:
            order_target_value(etf, target_value)
            log.info(f"[调仓] 买入/调整 {etf} 至目标价值 {target_value:.2f}")

def check_stop_loss(context):
    """盘中个股止损检查"""
    for etf, position in context.portfolio.positions.items():
        # 获取持仓成本价
        avg_cost = position.avg_cost
        # 获取当前最新价
        current_price = position.price
        if avg_cost > 0:
            loss = (current_price - avg_cost) / avg_cost
            if loss < -g.stop_loss_limit:
                order_target_value(etf, 0)
                log.info(f"[止损] {etf} 跌幅达 {loss*100:.2f}%, 触发止损卖出")

三、 策略关键 API 解析

• get_bars：用于获取高频且不包含停牌的 K 线数据。在计算均线和线性回归时，我们使用 get_bars 获取过去 20 天的收盘价和成交额。
• get_extras：通过传入 'unit_net_value' 参数，可以获取场内基金（ETF）的单位净值，这是计算折溢价率必不可少的数据。
• linregress：来自 scipy.stats 库，用于快速进行一元线性回归，返回的 r_value ** 2 即为趋势拟合度 $R^2$。
• order_target_value：目标价值下单函数。在 ETF 轮动中，使用该函数可以非常方便地将资产等权重分配到筛选出的强势 ETF 中。