时间序列线性回归：验证集移动平均特征构建及实现疑问

核心问题解答 #

1. 验证集的移动平均怎么处理？ #

绝对不能直接用验证集的全部真实价格计算移动平均——这属于数据泄露，会让模型在验证集上的表现失真，无法反映真实预测能力。正确的做法是严格按时间顺序模拟真实预测场景：

• 先将数据集按日期从早到晚排序（时间序列处理的核心前提），再划分为前80%的训练集和后20%的验证集；
• 计算验证集每个样本的移动平均时，仅使用该样本时间点之前的真实价格（包括训练集末尾的历史数据、以及验证集里该样本之前的已发生真实价格）。比如验证集第一个样本的10日移动平均，用训练集最后9天的价格加上它前1天的真实价格计算；验证集第二个样本的10日移动平均，则用训练集最后8天的价格加上验证集前1天的真实价格计算，以此类推。

2. 是否要用预测值作为移动平均的输入？ #

分两种场景判断：

• 单步预测（比如仅预测下一天的价格）：不需要用预测值。因为每次预测时，你已经掌握了之前所有的真实历史价格，直接用这些真实值计算移动平均即可；
• 多步预测（比如一次性预测未来7天的价格）：对于未来第2天及以后的预测，由于还没有对应真实价格，这时候可以用前面的预测值来计算后续的移动平均特征，但这种方式会引入预测误差的累积，需要谨慎评估效果。

3. scikit-learn有没有现成的实现方案？ #

sklearn没有直接封装“移动平均特征”的专用API，但可以通过两种方式实现，且能彻底避免数据泄露：

方式一：自定义Transformer类

继承sklearn的BaseEstimator和TransformerMixin，实现一个专门计算移动平均的转换器，可直接嵌入到Pipeline中，确保训练集和验证集的特征计算完全隔离：

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin

class MovingAverageTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, window_size=10):
        self.window_size = window_size
        
    def fit(self, X, y=None):
        # 时间序列特征无需训练，直接返回自身
        return self
    
    def transform(self, X):
        # 确保输入数据按日期排序
        X_sorted = X.sort_values('Date').reset_index(drop=True)
        # 计算移动平均并添加为新特征
        X_sorted[f'{self.window_size}_day_ma'] = X_sorted['Price'].rolling(window=self.window_size, min_periods=1).mean()
        return X_sorted.drop('Date', axis=1)

方式二：用FunctionTransformer快速实现

如果不需要复杂逻辑，可以用sklearn.preprocessing.FunctionTransformer包装pandas的滚动窗口操作：

from sklearn.preprocessing import FunctionTransformer

def add_ma_feature(X):
    X_sorted = X.sort_values('Date').reset_index(drop=True)
    X_sorted['10_day_ma'] = X_sorted['Price'].rolling(10, min_periods=1).mean()
    return X_sorted.drop('Date', axis=1)

ma_transformer = FunctionTransformer(add_ma_feature, validate=False)

注意：无论哪种方式，都要确保数据集在处理前已按日期排序，且划分训练/验证集时严格按时间顺序，绝对不能随机划分。

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