如何针对样本量随年份变化的面板数据使用TimeSeriesSplit做交叉验证？

我尝试用sklearn的TimeSeriesSplit处理面板数据（逐年人口快照数据），希望按年份拆分数据，但因为人口随时间演变，每年样本量不同，无法直接使用TimeSeriesSplit。

我想要实现的交叉验证方案为：按时间顺序递进拆分，每次训练集包含当前测试年份之前的所有年份数据，测试集仅包含下一个年份的数据（例如：训练集2010，测试集2011；训练集2010-2011，测试集2012；直到训练集2010-2019，测试集2020）。

我已经通过拆分年份的方式实现了该方案（代码如下），想知道是否有标准方法，能通过sklearn的CV对象来拆分数据，以便直接用于cross_val_score函数中。

import pandas as pd, numpy as np
import seaborn as sns, matplotlib.pyplot as plt

from sklearn.datasets import make_regression
from sklearn.dummy import DummyRegressor
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import TimeSeriesSplit

X_test, y_test = [], []

start_year = 2010
end_year = 2020

for year in np.arange(start_year, end_year+1):
    X_year, y_year = make_regression(n_samples=5+year-start_year, n_features=2, bias=100, noise=1, random_state=year)
    X_year = pd.DataFrame(X_year).rename(columns={0:'X1', 1:'X2'})
    X_year['year'] = year
    y_year = pd.Series(y_year)
    X_test.append(X_year)
    y_test.append(y_year)
    
X_test, y_test = pd.concat(X_test), pd.concat(y_test)

# modelling

X = X_test
y = y_test
years = np.unique(X_test['year'])

# modelisation
model = DummyRegressor(strategy="mean")
metric = mean_squared_error
cv = TimeSeriesSplit(n_splits=len(years)-1)

years_folds = []
res = []

for i, (train_year, test_year) in enumerate(cv.split(years)):
    
    print(f"Fold {i}:")
    print(f"  Train: index={years[train_year]}")
    print(f"  Test:  index={years[test_year]}")
    
    years_folds.append((years[train_year], years[test_year]))
    
    train_filter = X['year'].isin(years[train_year])
    test_filter = X['year'].isin(years[test_year])
    
    X_train, y_train = X.loc[train_filter.values], y[train_filter.values]
    X_test, y_test = X.loc[test_filter.values], y[test_filter.values]
    
    model.fit(X_train, y_train)
    score = metric(model.predict(X_test), y_test)
    print(f' {score=:.3}')
    res.append((years[test_year][0], score))

# plot_year_folds(years_folds)
    
folds_res = pd.DataFrame(res,columns=['test_year', metric.__name__])
folds_res.plot.scatter(x='test_year', y=metric.__name__, title=f'{metric.__name__} over test_year');

注：为提供可运行示例，我使用了虚拟数据集和虚拟模型，这并非核心内容。

有两种标准方法可以实现你的需求，让交叉验证逻辑能直接适配cross_val_score：

方法1：使用官方GroupTimeSeriesSplit（推荐） #

sklearn从0.24版本开始提供了GroupTimeSeriesSplit，专门用于按分组（此处分组为年份）进行时间序列交叉验证，完全匹配你的按年份递进拆分的需求：

from sklearn.model_selection import GroupTimeSeriesSplit
from sklearn.model_selection import cross_val_score

# 初始化交叉验证器，n_splits设为年份总数-1
cv = GroupTimeSeriesSplit(n_splits=len(years)-1)

# 调用cross_val_score时，将年份列作为groups参数传入
scores = cross_val_score(
    model, 
    X.drop('year', axis=1),  # 训练时排除年份列
    y, 
    cv=cv, 
    groups=X['year'], 
    scoring='neg_mean_squared_error'
)

# 转换为MSE（sklearn默认用负向分数优化）
mse_scores = -scores
print("各折MSE分数:", mse_scores.round(3))

方法2：自定义交叉验证迭代器 #

如果你的sklearn版本较低，可继承BaseCrossValidator类实现自定义的年份拆分逻辑，同样能适配cross_val_score：

from sklearn.model_selection import BaseCrossValidator
from sklearn.model_selection import cross_val_score

class YearTimeSeriesSplit(BaseCrossValidator):
    def __init__(self, n_splits=None):
        self.n_splits = n_splits
    
    def get_n_splits(self, X=None, y=None, groups=None):
        # 未指定n_splits时，默认用年份数量-1
        if self.n_splits is None:
            return len(np.unique(groups)) - 1
        return self.n_splits
    
    def split(self, X, y=None, groups=None):
        if groups is None:
            raise ValueError("必须传入年份作为groups参数")
        
        years = np.unique(groups)
        ts_split = TimeSeriesSplit(n_splits=self.get_n_splits(X, y, groups))
        
        for train_year_idx, test_year_idx in ts_split.split(years):
            train_years = years[train_year_idx]
            test_years = years[test_year_idx]
            
            # 获取对应年份的样本索引
            train_idx = np.where(groups.isin(train_years))[0]
            test_idx = np.where(groups.isin(test_years))[0]
            
            yield train_idx, test_idx

# 使用自定义验证器
cv = YearTimeSeriesSplit()
scores = cross_val_score(
    model, 
    X.drop('year', axis=1), 
    y, 
    cv=cv, 
    groups=X['year'], 
    scoring='neg_mean_squared_error'
)

mse_scores = -scores
print("各折MSE分数:", mse_scores.round(3))

这两种方法都能替代你手动拆分年份的逻辑，直接与sklearn的cross_val_score、GridSearchCV等工具兼容，其中GroupTimeSeriesSplit是官方提供的标准方案，更简洁可靠。

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