如何快速实现滚动回归（rolling regression），替代for循环提升运算速度

你当前使用的双重Python循环慢的核心原因是每次循环都要做数据切片、初始化回归对象、单次拟合，存在大量Python层的冗余开销，以下是三种可直接落地的提速方案：

方案1：基于Numpy矩阵运算的批量求解（速度最快） #

OLS回归的Beta有解析解$(X^TX)^{-1}X^TY$，可以通过滑动窗口视图把所有窗口的因子、收益数据批量转为三维矩阵，一次性调用底层优化过的线性代数接口完成所有回归计算，完全避免Python层循环：

import numpy as np
import pandas as pd
from numpy.lib.stride_tricks import sliding_window_view

# 参数配置
window_len = 252
factor_num = len(fac_cols)
security_num = len(sec_rets.columns)
valid_time_num = len(XY) - window_len + 1

# 生成无内存拷贝的滑动窗口视图
# X窗口维度：[有效时间点数量, 窗口长度, 因子数量]
X_window = sliding_window_view(XY[fac_cols].values, window_shape=window_len, axis=0).transpose(0,2,1)
# Y窗口维度：[有效时间点数量, 窗口长度, 个股数量]
Y_window = sliding_window_view(XY[sec_rets.columns].values, window_shape=window_len, axis=0).transpose(0,2,1)

# 批量求解所有窗口的Beta，lstsq比直接求逆数值稳定性更高
beta_batch = np.linalg.lstsq(X_window, Y_window, rcond=None)[0]
# 维度：[有效时间点数量, 因子数量, 个股数量]

# 转为MultiIndex DataFrame方便后续使用
beta_df = pd.DataFrame(
    beta_batch.reshape(-1, factor_num * security_num),
    index = XY.index[window_len-1:],
    columns = pd.MultiIndex.from_product([fac_cols, sec_rets.columns], names=['factor', 'security'])
)

该方案速度是原生双重循环的100倍以上，适合样本量较大的场景。

方案2：基于Pandas滚动Apply的简化实现（平衡性最佳） #

如果需要保留scikit-learn的回归接口兼容性，可以用Pandas底层C实现的滚动逻辑替代Python层循环，仅需自定义单次窗口的回归逻辑即可：

from sklearn import linear_model

# 把因子列放在前面，个股收益列放在后面
reg_input = XY[fac_cols + sec_rets.columns.tolist()].values
factor_count = len(fac_cols)

def window_reg(window_arr):
    # 输入维度：[窗口长度, 因子数量 + 个股数量]
    X = window_arr[:, :factor_count]
    Y = window_arr[:, factor_count:]
    # 一次拟合所有个股的回归
    return linear_model.LinearRegression().fit(X, Y).coef_.flatten()

# 滚动计算所有窗口的系数
beta_result = pd.DataFrame(reg_input).rolling(window=252).apply(window_reg, raw=True).dropna()

该方案实现简单，速度是原生双重循环的10~20倍。

方案3：基于专用金融工具库的开箱实现（开发效率最高） #

可以直接使用已经做过性能优化的第三方库滚动回归接口，无需自行实现矩阵逻辑：

from pyfinance.ols import RollingOLS

beta_dict = {}
for sec in sec_rets.columns:
    # 直接调用封装好的滚动回归接口
    model = RollingOLS(Y=XY[sec], X=XY[fac_cols], window=252)
    beta_dict[sec] = model.beta

# 合并所有个股的Beta结果
all_beta = pd.concat(beta_dict, axis=1, keys=sec_rets.columns)

该方案代码量最少，适合快速验证逻辑的场景。

内容的提问来源于stack exchange，提问作者Swatantra Yadav