4、滚动窗口切分法：滚动窗口原理、窗口大小选择策略、步长设置、滚动窗口在归因模型中的应用案例

4.1 滚动窗口切分原理

滚动窗口切分法（Rolling Window Split）是一种时序敏感的样本切分技术，特别适用于绩效归因模型训练。与简单的时间序列切分不同，滚动窗口通过固定大小的训练窗口在时间轴上逐步滑动，生成多个训练-验证子集，从而在保持时间顺序的同时最大化数据利用率。

4.2 窗口大小选择策略

窗口大小 W 是滚动窗口切分中最关键的参数，直接影响模型对市场周期、风格轮动和因子衰减的捕捉能力。选择不当会导致过拟合或欠拟合。

4.3 步长设置

步长 S 控制窗口滑动的间隔，决定了生成训练-验证子集的数量和重叠程度。步长设置需要在计算效率与样本覆盖之间取得平衡。

常见步长模式

• S = 1（逐期滚动）： 每期都生成一个子集，子集数量最多，但计算开销最大，且相邻子集高度重叠（重叠率 ≈ (W-1)/W）。适用于模型稳定性要求极高的场景。
• S = W/2（半窗口步长）： 子集重叠率约 50%，在计算成本与样本多样性之间取得良好平衡。推荐作为默认设置。
• S = W（非重叠窗口）： 子集之间无重叠，计算效率最高，但子集数量少，可能遗漏关键过渡期。适用于长期趋势分析。
• S = 验证集长度： 步长与验证窗口对齐，确保每次验证的数据不重叠。常用于回测框架。

4.4 滚动窗口在归因模型中的应用案例

案例：多因子绩效归因模型的滚动训练与验证

假设我们构建了一个包含 5 个风格因子（价值、动量、质量、低波、规模）的绩效归因模型，使用 A 股市场过去 10 年（2014-2023）的月度数据。采用滚动窗口切分法进行模型训练与验证。

实施步骤

1. 参数设定： 窗口大小 W=36 个月（3年），步长 S=6 个月（半年度滚动），验证集长度 V=12 个月（1年）。
2. 切分执行： 从 2014 年 1 月开始，第 1 个训练窗口为 2014.01~2016.12，验证集为 2017.01~2017.12；第 2 个训练窗口为 2014.07~2017.06，验证集为 2017.07~2018.06；依此类推，共生成 15 个训练-验证子集。
3. 模型训练： 在每个子集上使用加权最小二乘法估计因子暴露系数，权重采用指数衰减（半衰期 12 个月），使近期数据获得更高权重。
4. 绩效归因： 对每个验证集计算因子贡献、特质收益和累计归因结果，最终将 15 组结果进行统计聚合（中位数、置信区间）。

结果分析

代码示例：滚动窗口切分实现

import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression

def rolling_window_split(df, window=36, step=6, val_length=12):
    """
    滚动窗口切分生成器
    
    Parameters:
    -----------
    df : DataFrame, 索引为时间, 包含因子暴露和收益率
    window : int, 训练窗口长度（月）
    step : int, 步长（月）
    val_length : int, 验证集长度（月）
    
    Yields:
    -------
    train_df, val_df : 训练集和验证集DataFrame
    """
    dates = df.index.unique()
    n = len(dates)
    
    for start in range(0, n - window - val_length + 1, step):
        train_end = start + window
        val_end = train_end + val_length
        
        train_dates = dates[start:train_end]
        val_dates = dates[train_end:val_end]
        
        train_df = df.loc[train_dates]
        val_df = df.loc[val_dates]
        
        yield train_df, val_df

# 使用示例
results = []
for train, val in rolling_window_split(monthly_data, window=36, step=6, val_length=12):
    X_train = train[['value', 'momentum', 'quality', 'low_vol', 'size']].values
    y_train = train['excess_return'].values
    
    model = LinearRegression()
    model.fit(X_train, y_train)
    
    X_val = val[['value', 'momentum', 'quality', 'low_vol', 'size']].values
    y_val = val['excess_return'].values
    r2 = model.score(X_val, y_val)
    
    results.append({
        'train_period': f"{train.index[0].strftime('%Y%m')}-{train.index[-1].strftime('%Y%m')}",
        'val_period': f"{val.index[0].strftime('%Y%m')}-{val.index[-1].strftime('%Y%m')}",
        'r2': r2,
        'alpha': np.mean(y_val - model.predict(X_val)) * 12
    })