3、滑动窗口技术：滚动统计量、滚动相关系数、滚动回归与性能优化

滑动窗口，说白了就是给时间序列数据加一个「移动的放大镜」。你想想看，我们做量化交易，最怕的就是用全量数据算出一个静态指标——市场都变天了，你的参数还停留在三年前。我个人习惯，几乎所有时序特征都要用滚动窗口来算。

嗯，这一节我们就把滑动窗口技术拆开揉碎了讲。从最基础的滚动统计量，到滚动回归，再到性能优化，一步到位。

3.1 滚动统计量：均值与标准差

滚动均值是最简单的。但别小看它，我在项目中遇到过一个大坑：直接用 .rolling().mean() 算出来的结果，前 N-1 个值是 NaN。很多新手直接 dropna() 扔掉，结果数据量少了一大截。

避坑指南： 我曾经因为没处理滚动窗口前期的 NaN，导致回测曲线在开头出现诡异的跳变。后来养成了习惯——要么用 min_periods 参数，要么用 expanding() 做前期填充。 
import pandas as pd
import numpy as np

# 生成模拟价格数据
np.random.seed(42)
prices = pd.Series(np.cumsum(np.random.randn(1000) * 0.01) + 100)

# 滚动均值与标准差（窗口20）
rolling_mean = prices.rolling(window=20, min_periods=5).mean()
rolling_std = prices.rolling(window=20, min_periods=5).std()

# 构建布林带
upper_band = rolling_mean + 2 * rolling_std
lower_band = rolling_mean - 2 * rolling_std

这里 min_periods=5 的意思是：窗口内至少有5个有效值就计算。这样前期的 NaN 会少很多，回测也更平滑。

3.2 滚动相关系数

滚动相关系数在配对交易里是核心工具。我见过有人用全量相关系数做配对，结果市场风格一切换，相关系数从0.9直接掉到0.3，仓位全爆了。

滚动相关系数能实时捕捉这种变化。Pandas 里直接用 .rolling().corr() 就行，但要注意——它返回的是一个 MultiIndex 的 DataFrame，很多人第一次看到会懵。

# 两个资产收益率序列
returns_a = pd.Series(np.random.randn(1000) * 0.02)
returns_b = returns_a * 0.7 + np.random.randn(1000) * 0.01

# 滚动相关系数（窗口60）
rolling_corr = returns_a.rolling(window=60).corr(returns_b)

# 当相关系数低于阈值时，开仓做配对
threshold = 0.5
signal = (rolling_corr < threshold).astype(int)
我的经验： 滚动相关系数的窗口大小很敏感。我个人习惯用60个交易日（约3个月）作为默认值。太短了噪声大，太长了反应迟钝。你可以用滚动窗口的滚动标准差来动态调整——嗯，这有点套娃，但效果不错。

3.3 滚动回归：实时计算Beta

滚动回归比滚动相关系数更强大。它能给出截距、斜率、R方等一系列统计量。在因子模型里，我们经常用滚动回归算股票的动态Beta。

Pandas 没有直接提供滚动回归函数，但我们可以用 .rolling().apply() 自己实现。不过要注意性能问题——我当年用纯Python循环跑1000只股票的滚动回归，跑了整整一个周末。

from scipy import stats

def rolling_regression(y, x, window=60):
    """滚动线性回归，返回斜率（Beta）"""
    def regress_slice(data):
        y_slice = data[:, 0]
        x_slice = data[:, 1]
        if len(y_slice) < 10:  # 最少10个样本
            return np.nan
        slope, intercept, r_value, p_value, std_err = stats.linregress(x_slice, y_slice)
        return slope
    
    # 合并数据
    combined = np.column_stack([y, x])
    result = pd.Series(
        [np.nan] * (window - 1) + 
        [regress_slice(combined[i-window+1:i+1]) for i in range(window-1, len(combined))]
    )
    return result

# 使用示例
market_returns = pd.Series(np.random.randn(1000) * 0.015)
stock_returns = market_returns * 1.2 + np.random.randn(1000) * 0.01

rolling_beta = rolling_regression(stock_returns, market_returns, window=60)
核心要点： 滚动回归的窗口选择直接影响Beta的稳定性。我个人建议：做日频策略用60天，做周频策略用24周。另外，记得检查回归的R方——如果R方太低，这个Beta其实不可靠。

3.4 窗口函数性能优化

说到性能优化，这是我最想分享的部分。你想想看，一个策略里可能有几十个滚动统计量，如果每个都跑一遍循环，回测速度会慢到让你怀疑人生。

我总结了几条实战经验：

  • 用 NumPy 向量化代替 Pandas rolling： 对于简单的滚动均值、标准差，直接用 np.convolve 或 np.cumsum 实现，速度能快10倍以上。
  • 避免在 .rolling().apply() 里用复杂函数： 每次 apply 都是 Python 级别的循环，非常慢。能用内置函数（mean, std, corr）就别自己写。
  • 用 Numba 加速自定义滚动函数： 如果必须用自定义逻辑，用 @njit 装饰器编译成机器码。
import numpy as np
from numba import njit

@njit
def rolling_mean_numba(data, window):
    """用Numba加速的滚动均值"""
    n = len(data)
    result = np.empty(n)
    result[:window-1] = np.nan
    
    # 初始窗口求和
    window_sum = np.sum(data[:window])
    result[window-1] = window_sum / window
    
    # 滑动更新
    for i in range(window, n):
        window_sum += data[i] - data[i-window]
        result[i] = window_sum / window
    
    return result

# 测试速度
data = np.random.randn(100000)
%timeit rolling_mean_numba(data, 20)  # 约0.3ms
%timeit pd.Series(data).rolling(20).mean()  # 约2.1ms
性能对比： 上面这个例子，Numba版本比Pandas原生快了7倍。如果窗口更大，差距会更明显。我建议把常用的滚动函数都写成Numba版本，存成一个工具模块，每次直接调用。

3.5 实战中的避坑指南

最后，分享几个我踩过的坑：

  1. 窗口边界处理： 滚动窗口的前期数据一定要想清楚怎么处理。是用 expanding 填充，还是直接丢弃？我建议在回测里用 expanding，在实盘里用固定窗口。
  2. 未来信息泄露： 滚动窗口一定要用「过去的数据」算「当前的指标」。我见过有人不小心用了 .shift(-1)，结果把未来的数据卷进来了——回测收益高得离谱，实盘直接亏光。
  3. 内存爆炸： 如果你同时计算几百个股票的滚动统计量，DataFrame 会变得巨大。我建议用 groupby 分批计算，或者用 dask 做分布式处理。
重要提醒： 滚动窗口的「窗口大小」本身就是一个超参数。不要用回测数据去优化窗口大小——这会导致过拟合。我一般根据交易频率来定：日频策略用20-60天，小时频策略用48-120小时。

好了，滑动窗口技术就讲到这里。下一节我们会讲「指数加权移动平均（EWMA）」，它其实是滑动窗口的一种变体，但权重设计更巧妙。到时候我会分享一个我用EWMA做趋势跟踪的实战案例。

← 上一章 📖 返回目录 下一章 →