如何在聚宽（JoinQuant）中实现残差动量ETF轮动策略？（附Python源码）

什么是残差动量？

传统的价格动量（如过去N天的涨幅）容易受到市场整体走势（Beta）的影响。在牛市中，高Beta的ETF往往涨幅最大，但这种上涨主要是由市场共性驱动的，而非其自身的超额能力。

残差动量（Residual Momentum）通过CAPM模型（资本资产定价模型）对ETF的收益率进行回归，剔除市场整体表现（Market Beta）和系统性风险，提取出回归的残差项。残差项代表了该ETF独立于大盘的超额收益。对残差进行指数加权或累计求和，即可得到残差动量因子。研究表明，残差动量比传统动量具有更低的波动和更强的超额收益稳定性。

策略核心步骤

1. 确定ETF池与基准：选择代表性强、流动性好的宽基或行业ETF（如沪深300ETF、中证500ETF、创业板ETF、黄金ETF等），并以沪深300指数（000300.XSHG）作为市场基准。
2. 获取历史收益率：获取过去N天（如252个交易日）的ETF日收益率和基准日收益率。
3. 计算残差动量：
   • 将ETF日收益率 $R_i$ 对基准收益率 $R_m$ 进行线性回归：$R_{i,t} = \alpha_i + \beta_i R_{m,t} + \epsilon_{i,t}$。
   • 提取残差项 $\epsilon_{i,t}$。
   • 计算残差的标准化得分（如残差均值除以残差标准差，即残差夏普比率），作为残差动量因子值。
4. 轮动换仓：在调仓日（如每月初），选择残差动量因子值最高的Top 2 ETF进行买入持有，卖出不在Top 2的ETF。

聚宽策略源码实现

以下是基于聚宽API实现的残差动量ETF轮动策略完整代码：

import pandas as pd
import numpy as np
import statsmodels.api as sm
from jqdata import *

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 过滤掉比error级别低的log
    log.set_level('order', 'error')
    
    # 候选ETF池
    g.etf_pool = [
        '510300.XSHG', # 沪深300ETF
        '510500.XSHG', # 中证500ETF
        '159915.XSHE', # 创业板ETF
        '518880.XSHG', # 黄金ETF
        '513100.XSHG', # 纳指ETF
        '510900.XSHG', # H股ETF
    ]
    
    # 市场基准代码
    g.market_index = '000300.XSHG'
    # 回归历史窗口长度
    g.window = 120
    # 持有ETF数量
    g.top_n = 2
    
    # 每月第一个交易日开盘运行调仓
    run_monthly(handle_rotation, monthday=1, time='9:30')

def handle_rotation(context):
    # 1. 获取历史价格数据
    all_signals = {}
    end_date = context.previous_date
    
    # 获取基准指数收益率
    benchmark_df = get_bars(g.market_index, count=g.window+1, unit='1d', fields=['close'], end_dt=end_date, df=True)
    benchmark_returns = benchmark_df['close'].pct_change().dropna()
    
    # 2. 循环计算每个ETF的残差动量
    for etf in g.etf_pool:
        try:
            etf_df = get_bars(etf, count=g.window+1, unit='1d', fields=['close'], end_dt=end_date, df=True)
            if len(etf_df) < g.window + 1:
                continue
            etf_returns = etf_df['close'].pct_change().dropna()
            
            # 线性回归：ETF收益率 = alpha + beta * 基准收益率 + residual
            X = sm.add_constant(benchmark_returns.values)
            y = etf_returns.values
            model = sm.OLS(y, X).fit()
            
            # 提取残差
            residuals = model.resid
            
            # 计算残差动量因子：残差均值 / 残差标准差 (信息比率)
            residual_momentum = np.mean(residuals) / np.std(residuals)
            all_signals[etf] = residual_momentum
        except Exception as e:
            log.warn(f"计算 {etf} 因子失败: {str(e)}")
            
    # 3. 排序并确定目标持仓
    sorted_etfs = sorted(all_signals.items(), key=lambda x: x[1], reverse=True)
    target_list = [x[0] for x in sorted_etfs[:g.top_n]]
    
    log.info(f"当前调仓日：{context.current_dt.date()}，计算得到的残差动量排名：{sorted_etfs}")
    log.info(f"目标持仓：{target_list}")
    
    # 4. 执行调仓交易
    # 卖出不在目标池中的持仓
    current_positions = list(context.portfolio.positions.keys())
    for etf in current_positions:
        if etf not in target_list:
            order_target(etf, 0)
            log.info(f"卖出：{etf}")
            
    # 等权重买入目标ETF
    if len(target_list) > 0:
        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}")

策略优化建议

1. 引入动量过滤（绝对动量）：如果排名前两名的ETF残差动量虽然最高，但其自身绝对收益为负（例如处于系统性熊市），可以考虑空仓或买入货币基金避险。
2. 动态调整回归窗口：代码中使用了120天（约半年）的回归窗口，您可以尝试使用更短的窗口（如60天）捕捉短期行业轮动，或更长的窗口（如252天）进行长期资产配置。
3. 加入交易摩擦限制：ETF交易虽然免印花税，但仍有佣金。可以通过设置滑点（set_slippage）和佣金（set_order_cost）来更真实地模拟回测表现。