Python端到端流水线中TimeSeriesSplit时序交叉验证最佳实践

前置数据准备 #

先确保时序数据按时间排序，处理非标准时间戳：

import pandas as pd
import numpy as np
from sklearn.model_selection import TimeSeriesSplit, cross_val_score, GridSearchCV
from sklearn.ensemble import RandomForestRegressor
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_absolute_error, r2_score
import matplotlib.pyplot as plt

# 加载并处理数据（根据实际时间格式调整parse规则）
df = pd.read_csv("your_dataset.csv")
df['TS_24hrs'] = pd.to_datetime(df['TS_24hrs'], format="%Y%m%d%H")  # 适配非标准时间格式
df = df.sort_values('TS_24hrs').reset_index(drop=True)

# 拆分保留测试集与CV训练池
cv_data = df.iloc[:200].copy()
holdout_test = df.iloc[200:].copy()  # 对应你提到的最后74条数据

# 定义特征与目标列（根据业务场景调整）
X = cv_data.drop(['TS_24hrs', 'target'], axis=1)
y = cv_data['target']
X_holdout = holdout_test.drop(['TS_24hrs', 'target'], axis=1)
y_holdout = holdout_test['target']

1. TimeSeriesSplit参数设置与拆分可视化 #

针对200条CV数据，设置合理参数：

• n_splits=5: 保证拆分次数足够，同时每个fold有足够样本
• test_size=30: 固定测试集长度，保证评估一致性
• max_train_size=100: 限制训练集最大长度，避免过旧数据干扰模型

可视化拆分结果验证合理性：

# 初始化时序拆分器
tscv = TimeSeriesSplit(n_splits=5, test_size=30, max_train_size=100)

# 绘制拆分区间
plt.figure(figsize=(12, 6))
for fold_idx, (train_idx, test_idx) in enumerate(tscv.split(X)):
    # 标记训练集
    plt.plot(cv_data['TS_24hrs'].iloc[train_idx], [fold_idx]*len(train_idx), 
             marker='_', color='blue', linewidth=2, label='训练集' if fold_idx==0 else "")
    # 标记测试集
    plt.plot(cv_data['TS_24hrs'].iloc[test_idx], [fold_idx]*len(test_idx), 
             marker='_', color='red', linewidth=2, label='测试集' if fold_idx==0 else "")

plt.yticks(range(tscv.n_splits), [f'折数 {i+1}' for i in range(tscv.n_splits)])
plt.xlabel('时间戳')
plt.title('TimeSeriesSplit 拆分结果可视化')
plt.legend()
plt.grid(axis='y', linestyle='--')
plt.show()

参数合理性说明：

• 训练集逐步扩大但不超过100条，平衡数据量与时效性
• 测试集固定30条，确保每个fold的评估标准一致
• 严格遵循时序顺序，无未来数据泄露风险

2. 集成时序交叉验证的Pipeline实现（解决效果下降问题） #

之前Pipeline效果下滑的核心原因是预处理阶段的数据泄露（比如用全量数据计算标准化统计量）。正确做法是将预处理与模型封装进Pipeline，结合TimeSeriesSplit确保每个fold的预处理仅基于该fold的训练数据：

# 构建端到端Pipeline（替换为你的实际特征工程步骤）
pipeline = Pipeline([
    ('scaler', StandardScaler()),  # 示例标准化步骤
    ('rf', RandomForestRegressor(random_state=42))
])

# 用时序交叉验证评估模型
cv_mae_scores = cross_val_score(pipeline, X, y, cv=tscv, scoring='neg_mean_absolute_error')
cv_r2_scores = cross_val_score(pipeline, X, y, cv=tscv, scoring='r2')

print(f"CV平均MAE: {-np.mean(cv_mae_scores):.2f}")
print(f"CV平均R²: {np.mean(cv_r2_scores):.2f}")

# 可选：结合GridSearchCV做参数调优
param_grid = {
    'rf__n_estimators': [100, 200],
    'rf__max_depth': [5, 10, None]
}

grid_search = GridSearchCV(pipeline, param_grid, cv=tscv, 
                           scoring='neg_mean_absolute_error', n_jobs=-1)
grid_search.fit(X, y)

print(f"最优参数组合: {grid_search.best_params_}")
best_model = grid_search.best_estimator_

关键注意事项：

• 所有预处理步骤必须放在Pipeline内部，禁止在外部对全量CV数据做预处理
• 交叉验证全程使用TimeSeriesSplit，避免随机拆分导致的时序逻辑破坏

3. 有无交叉验证的模型效果对比与可视化 #

3.1 无CV基线模型 #

# 训练无交叉验证的基线模型
baseline_rf = RandomForestRegressor(random_state=42)
baseline_rf.fit(X, y)

# 预测保留测试集
y_pred_baseline = baseline_rf.predict(X_holdout)
mae_baseline = mean_absolute_error(y_holdout, y_pred_baseline)
r2_baseline = r2_score(y_holdout, y_pred_baseline)

print(f"基线模型MAE: {mae_baseline:.2f}")
print(f"基线模型R²: {r2_baseline:.2f}")

3.2 CV调优模型 #

# 用最优模型预测保留测试集
y_pred_cv = best_model.predict(X_holdout)
mae_cv = mean_absolute_error(y_holdout, y_pred_cv)
r2_cv = r2_score(y_holdout, y_pred_cv)

print(f"CV调优模型MAE: {mae_cv:.2f}")
print(f"CV调优模型R²: {r2_cv:.2f}")

3.3 结果可视化 #

# 预测值与真实值对比图
plt.figure(figsize=(14, 7))
plt.plot(holdout_test['TS_24hrs'], y_holdout, label='真实值', color='green', linewidth=2)
plt.plot(holdout_test['TS_24hrs'], y_pred_baseline, label='无CV基线预测', color='orange', linestyle='--')
plt.plot(holdout_test['TS_24hrs'], y_pred_cv, label='CV调优预测', color='blue', linestyle=':')

plt.xlabel('时间戳')
plt.ylabel('目标值')
plt.title('有无交叉验证的模型预测效果对比')
plt.legend()
plt.grid(linestyle='--')
plt.show()

# 指标对比柱状图
metrics = ['MAE', 'R²']
baseline_scores = [mae_baseline, r2_baseline]
cv_scores = [mae_cv, r2_cv]

x = np.arange(len(metrics))
width = 0.35

fig, ax = plt.subplots(figsize=(8, 5))
rects1 = ax.bar(x - width/2, baseline_scores, width, label='无CV基线')
rects2 = ax.bar(x + width/2, cv_scores, width, label='CV调优')

ax.set_xticks(x)
ax.set_xticklabels(metrics)
ax.set_title('模型指标对比')
ax.legend()

# 添加数值标签
def autolabel(rects):
    for rect in rects:
        height = rect.get_height()
        ax.annotate(f'{height:.2f}',
                    xy=(rect.get_x() + rect.get_width() / 2, height),
                    xytext=(0, 3),
                    textcoords="offset points",
                    ha='center', va='bottom')

autolabel(rects1)
autolabel(rects2)

plt.show()

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