聚宽 ETF 动量轮动策略：参数网格搜索与 Walk-Forward 滚动窗口验证指南

问题描述

聚宽 ETF 动量轮动如何做参数网格搜索与 Walk-Forward 验证？
如何在聚宽中对动量窗口、调仓周期、持仓数量和候选 ETF 组合进行网格搜索，并用样本外和 Walk-Forward 验证降低过拟合？

解决方案

聚宽 ETF 动量轮动策略：参数网格搜索与 Walk-Forward 滚动窗口验证指南

在量化投资中，ETF 动量轮动策略因其交易成本低、不易踩雷、趋势跟踪效果好而广受欢迎。然而，动量策略高度依赖参数设置（如动量计算窗口、调仓周期、持仓标的数量）。如果仅在单一历史区间进行回测并挑选最优参数，极易陷入**过拟合（Overfitting）**陷阱。

为了提高策略的鲁棒性，本文将介绍如何在聚宽（JoinQuant）平台中，对 ETF 动量轮动策略进行多维参数网格搜索，并引入Walk-Forward（前向滚动）验证机制，确保策略在样本外（Out-of-Sample）依然有效。

一、核心概念与验证逻辑

1. 网格搜索（Grid Search）参数空间

我们需要优化的核心参数包括：

动量窗口 ($W$)：计算过去 $N$ 个交易日的累计收益率（如 10, 20, 40, 60 日）。
调仓周期 ($P$)：每隔 $M$ 个交易日调仓一次（如 5, 10, 20 日）。
持仓数量 ($K$)：动量排名前 $K$ 名的 ETF 进入持仓（如 1, 2, 3 只）。
候选 ETF 组合：如宽基组合（沪深300、中证500、创业板指、科创50）或行业/跨境组合（黄金、纳指、标普500、芯片等）。

2. Walk-Forward 滚动窗口验证

Walk-Forward 验证是一种动态的样本外测试方法，比传统的简单单次分割（Train-Test Split）更贴近实盘：

划分区间：将历史数据划分为多个重叠的“训练期（样本内）”和紧随其后的“测试期（样本外）”。例如，训练期 2 年，测试期 0.5 年。
滚动优化：
- 在第 1 阶段训练期（如 2018-2019）运行网格搜索，找出夏普比率最高的参数组合 $P^*_1$。
- 将 $P^*_1$ 应用于第 1 阶段测试期（如 2020 上半年），记录样本外净值。
- 窗口向前滚动 0.5 年，在第 2 阶段训练期（如 2018.5-2020.5）重新搜索最优参数 $P^*_2$。
- 将 $P^*_2$ 应用于第 2 阶段测试期（如 2020 下半年）。
拼接业绩：将所有测试期的样本外收益率拼接起来，形成一条无未来函数的Walk-Forward 样本外净值曲线。

二、聚宽研究环境：多维网格搜索与 Walk-Forward 验证实现

由于 Walk-Forward 涉及多次参数重估，最适合在聚宽的**研究模块（Notebook）**中通过 API 获取数据并进行离线模拟计算，最后将最优参数输入回测引擎。

以下是完整的 Python 实现代码：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from jqdata import *

# 1. 定义候选 ETF 组合
etf_pool = [
    '510300.XSHG', # 沪深300 ETF
    '510500.XSHG', # 中证500 ETF
    '159915.XSHE', # 创业板 ETF
    '513100.XSHG', # 纳指 ETF
    '518880.XSHG', # 黄金 ETF
    '511010.XSHG'  # 国债 ETF (避险资产)
]

# 2. 获取历史日线数据
start_date = '2016-01-01'
end_date = '2023-12-31'
df_close = get_price(etf_pool, start_date=start_date, end_date=end_date, frequency='daily', fields=['close'])['close']

# 3. 定义单次动量轮动模拟函数
def backtest_momentum(df_sub, w, p, k):
    """
    在指定子区间上模拟动量轮动
    w: 动量窗口
    p: 调仓周期
    k: 持仓数量
    """
    returns = df_sub.pct_change()
    # 计算动量得分（过去 w 天的累计收益）
    momentum_scores = df_sub.pct_change(w)
    
    portfolio_returns = []
    dates = df_sub.index[w:]
    
    current_holdings = []
    
    for i, date in enumerate(dates):
        # 满足调仓周期
        if i % p == 0:
            # 获取当前时刻的动量排名，剔除 NaN
            scores = momentum_scores.loc[date].dropna()
            if not scores.empty:
                # 按动量降序排列，选择前 K 个
                current_holdings = scores.sort_values(ascending=False).head(k).index.tolist()
        
        if current_holdings:
            # 假设等权重持有
            daily_ret = returns.loc[date, current_holdings].mean()
            portfolio_returns.append(daily_ret)
        else:
            portfolio_returns.append(0.0)
            
    # 计算夏普比率
    port_df = pd.Series(portfolio_returns, index=dates)
    annual_ret = port_df.mean() * 242
    annual_std = port_df.std() * np.sqrt(242)
    sharpe = (annual_ret - 0.03) / annual_std if annual_std > 0 else -99
    return sharpe, port_df

# 4. Walk-Forward 滚动验证主循环
def walk_forward_validation(df_all, train_len_years=2, test_len_years=0.5):
    # 参数网格空间
    w_space = [10, 20, 40, 60]
    p_space = [5, 10, 20]
    k_space = [1, 2]
    
    all_dates = df_all.index
    start_dt = all_dates[0]
    end_dt = all_dates[-1]
    
    current_train_start = start_dt
    wf_test_returns = pd.Series(dtype=float)
    
    while True:
        # 计算当前阶段的训练期和测试期边界
        train_end = current_train_start + timedelta(days=train_len_years * 365)
        test_end = train_end + timedelta(days=test_len_years * 365)
        
        if test_end > end_dt:
            break
            
        # 截取训练集和测试集数据
        df_train = df_all.loc[current_train_start:train_end]
        df_test = df_all.loc[train_end:test_end]
        
        # 样本内网格搜索最优参数
        best_sharpe = -99
        best_params = (20, 10, 1) # 默认初始参数
        
        for w in w_space:
            for p in p_space:
                for k in k_space:
                    # 确保训练集长度大于动量窗口
                    if len(df_train) > w:
                        sharpe, _ = backtest_momentum(df_train, w, p, k)
                        if sharpe > best_sharpe:
                            best_sharpe = sharpe
                            best_params = (w, p, k)
                            
        # 提取最优参数并在样本外（测试集）运行
        opt_w, opt_p, opt_k = best_params
        print(f"训练期: {current_train_start.date()} ~ {train_end.date()} | 最优参数: W={opt_w}, P={opt_p}, K={opt_k} | 样本内夏普: {best_sharpe:.2f}")
        
        # 用最优参数跑测试集
        _, test_rets = backtest_momentum(df_all.loc[train_end - timedelta(days=opt_w*1.5):test_end], opt_w, opt_p, opt_k)
        # 只保留纯样本外测试期的数据
        test_rets_pure = test_rets.loc[train_end:test_end]
        wf_test_returns = pd.concat([wf_test_returns, test_rets_pure])
        
        # 窗口向前滚动（步长为测试期长度）
        current_train_start = current_train_start + timedelta(days=test_len_years * 365)
        
    return wf_test_returns

# 运行前向滚动验证
wf_returns = walk_forward_validation(df_close)

# 5. 计算并展示样本外 Walk-Forward 业绩指标
if not wf_returns.empty:
    cum_returns = (1 + wf_returns).cumprod() - 1
    total_ret = cum_returns.iloc[-1]
    annual_ret = wf_returns.mean() * 242
    annual_std = wf_returns.std() * np.sqrt(242)
    sharpe = (annual_ret - 0.03) / annual_std
    print(f"\n=== Walk-Forward 样本外总业绩 ===")
    print(f"测试区间: {wf_returns.index[0].date()} 至 {wf_returns.index[-1].date()}")
    print(f"样本外年化收益率: {annual_ret*100:.2f}%")
    print(f"样本外夏普比率: {sharpe:.2f}")

三、如何在聚宽回测引擎中部署该策略？

在聚宽的策略回测/模拟盘中，我们不能直接运行耗时极长的多维网格搜索。标准的做法是：

离线计算参数：在研究环境中运行上述 Walk-Forward 代码，获取最新的最优参数组合（例如当前最新阶段的最优参数为 $W=40, P=10, K=2$）。
回测/实盘部署：在策略代码中，开启真实价格模式 set_option('use_real_price', True)，并使用 run_daily 或 run_weekly 定时触发轮动逻辑。

聚宽回测代码示例：

import pandas as pd
from jqdata import *

def initialize(context):
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True) # 开启真实价格模式
    
    # 设定离线 Walk-Forward 验证得到的最优参数
    g.W = 40  # 动量窗口
    g.P = 10  # 调仓周期
    g.K = 2   # 持仓数量
    
    g.etf_pool = ['510300.XSHG', '510500.XSHG', '159915.XSHE', '513100.XSHG', '518880.XSHG']
    g.timer = 0
    
    # 每天运行
    run_daily(market_open, time='09:30')

def market_open(context):
    if g.timer % g.P == 0:
        # 1. 获取历史数据计算动量
        df = attribute_history(g.etf_pool, g.W, '1d', ['close'], df=False)
        scores = {}
        for etf in g.etf_pool:
            close_prices = df[etf]['close']
            if len(close_prices) >= g.W:
                # 计算变动率
                scores[etf] = (close_prices[-1] - close_prices[0]) / close_prices[0]
        
        # 2. 排序并选择前 K 个
        sorted_etfs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        target_holdings = [x[0] for x in sorted_etfs[:g.K]]
        
        # 3. 调仓执行
        # 卖出不在目标持仓中的资产
        for etf in list(context.portfolio.long_positions.keys()):
            if etf not in target_holdings:
                order_target(etf, 0)
                
        # 等权重买入目标资产
        if target_holdings:
            target_value = context.portfolio.available_cash / len(target_holdings)
            for etf in target_holdings:
                order_target_value(etf, target_value)
                
    g.timer += 1

四、降低过拟合的实用技巧

引入惩罚项：在网格搜索时，不要单纯以“最高收益率”为目标，建议使用**夏普比率（Sharpe Ratio）或卡玛比率（Calmar Ratio，收益/最大回撤）**作为目标函数，惩罚高波动和高回撤的参数。
参数平原法则（Parameter Flatness）：选择最优参数时，避开孤立的“参数孤峰”（即周围参数表现极差，只有某一个参数极好），选择参数平原（即参数小幅变动时，策略表现依然稳定的区域）。
加入摩擦成本：在网格搜索的模拟中，务必通过 set_order_cost 加上真实的滑点和佣金。频繁调仓（如 $P=5$）在无摩擦时表现极好，但在扣除千分之一的摩擦后，其样本外表现往往会大幅退化。

问题描述

解决方案