如何在Pandas中对多时间序列执行扩展窗口线性回归并一次性返回参数

问题 #

我有一个包含多时间序列的Pandas DataFrame，结构如下：

>>> df
                  S1        S2        S3
Date
2019-01-02  0.019552  0.021958  0.001586
2019-01-03  0.000000  0.020325  0.000000
...              ...       ...       ...
2024-07-05  1.000000  0.601924  0.926365

我需要对每个时间序列执行扩展窗口（expanding）的线性回归：从第一个时间点开始逐步扩大窗口直至最后一个点，返回每个窗口的斜率（slope）和截距（intercept）。

目前的实现需要两次拟合模型分别获取斜率和截距，浪费计算资源：

minimum_data_samples = 10

def fit_linear_regression(signal_column):
    if signal_column.empty or len(signal_column) < minimum_data_samples:
        return np.nan, np.nan
 
    X = np.arange(len(signal_column)).reshape(-1, 1)
    reg = LinearRegression(n_jobs=-1).fit(X, signal_column)
    return reg.coef_[0], reg.intercept_

# 第一次拟合获取斜率
slopes = df.expanding().apply(lambda x: fit_linear_regression(x)[0])

# 第二次拟合获取截距
intercepts = df.expanding().apply(lambda x: fit_linear_regression(x)[1])

# 合并结果
slopes_and_intercepts = pd.concat([intercepts, slopes], keys=['intercepts', 'slopes'], axis=1)

输出示例（数值与上述数据不对应）：

intercepts                        slopes
                   S1        S2        S3        S1        S2        S3
Date
2014-01-02        NaN       NaN       NaN       NaN       NaN       NaN
...              ...       ...       ...       ...       ...       ...
2018-07-03   0.104788 -0.127579  0.062910  0.000143  0.001359  0.000163

我想知道是否有方法通过单次调用拟合函数完成上述操作。

注：该场景与其他类似问题有三点差异：

• 自变量X由窗口内的序列长度生成（而非原数据列）
• 每列是独立的待拟合时间序列
• 输出为独立的新DataFrame，无需合并到原数据

解决方案 #

可以通过修改expanding().apply()的返回形式，让拟合函数一次性返回斜率和截距，再将结果重塑为多级列的DataFrame，只需一次拟合即可完成。

优化代码 #

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

minimum_data_samples = 10

def fit_expanding_regression(signal_column):
    n = len(signal_column)
    if n < minimum_data_samples:
        # 返回对应数量的NaN，匹配斜率+截距的结构
        return pd.Series([np.nan, np.nan], index=['slope', 'intercept'])
    
    X = np.arange(n).reshape(-1, 1)
    reg = LinearRegression(n_jobs=-1).fit(X, signal_column)
    # 返回包含斜率和截距的Series，指定索引方便后续结构整理
    return pd.Series([reg.coef_[0], reg.intercept_], index=['slope', 'intercept'])

# 对每列执行扩展窗口拟合，result_type='expand'将返回的Series展开为多列
result = df.expanding().apply(fit_expanding_regression, result_type='expand')

# 调整列结构，整理为和原示例一致的多级索引格式
result = result.T.unstack(0).swaplevel(0, 1, axis=1).sort_index(axis=1)
# 重命名一级索引，匹配原示例的命名
result.columns = result.columns.set_levels(['slopes', 'intercepts'], level=0)

代码说明 #

1. 拟合函数调整：让fit_expanding_regression返回包含斜率和截距的pd.Series，指定索引确保返回值结构清晰，不足样本量时返回对应NaN值。
2. result_type='expand'参数：告诉apply将每个窗口的返回值（Series）展开为多列，这样每个时间点会生成对应每列的斜率和截距列。
3. 列结构整理：通过转置、解栈和交换索引层级，将结果调整为和原示例一致的多级列结构（一级为intercepts/slopes，二级为序列名）。

这种方式仅需一次扩展窗口拟合，避免了重复计算，效率提升明显。

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