DLinear 模型滚动预测实战：5步配置实现ETTh1数据集96步预测（附完整代码）

DLinear 模型滚动预测实战5步配置实现ETTh1数据集96步预测附完整代码时间序列预测一直是数据分析领域的核心挑战之一。在众多预测方法中DLinear模型以其简洁高效的架构脱颖而出特别适合需要快速验证模型效果的中级开发者。本文将带您从零开始通过5个关键步骤完成ETTh1数据集上的96步滚动预测任务并提供可直接运行的Python代码和可视化方案。1. 环境准备与数据加载首先需要配置Python环境和安装必要的依赖库。建议使用Python 3.8版本并创建独立的虚拟环境conda create -n dlinear python3.8 conda activate dlinear pip install torch1.12.1 pandas matplotlib scikit-learnETTh1数据集是电力变压器温度监测的经典时间序列数据集包含7个特征列和1个目标列OT表示油温。我们可以直接从GitHub获取预处理好的数据import pandas as pd # 加载ETTh1数据集 data pd.read_csv(https://raw.githubusercontent.com/zhouhaoyi/ETDataset/main/ETT-small/ETTh1.csv) print(data.head()) # 可视化前1000个时间点的OT值 import matplotlib.pyplot as plt data[OT][:1000].plot(figsize(12,4)) plt.title(ETTh1 Oil Temperature (First 1000 Points)) plt.xlabel(Time Index) plt.ylabel(Temperature) plt.show()关键参数说明seq_len输入序列长度建议设置为96pred_len预测步长本文设置为96features预测模式MS表示多变量预测单变量2. 数据预处理与滚动窗口构建时间序列预测需要将原始数据转换为监督学习格式。我们实现滚动窗口生成函数import numpy as np from sklearn.preprocessing import StandardScaler def create_rolling_windows(data, seq_len, pred_len): X, Y [], [] for i in range(len(data)-seq_len-pred_len1): X.append(data[i:iseq_len]) Y.append(data[iseq_len:iseq_lenpred_len]) return np.array(X), np.array(Y) # 数据标准化 scaler StandardScaler() data_scaled scaler.fit_transform(data[[OT]]) # 生成滚动窗口 X, y create_rolling_windows(data_scaled, seq_len96, pred_len96) print(f生成窗口形状X{X.shape}, y{y.shape}) # 划分训练集和测试集8:2比例 split int(0.8*len(X)) X_train, X_test X[:split], X[split:] y_train, y_test y[:split], y[split:]提示滚动预测的关键在于保持时间顺序切勿在划分数据集前进行随机打乱操作。3. DLinear模型实现DLinear的核心思想是将时间序列分解为趋势项和季节项分别用线性层处理。以下是PyTorch实现import torch import torch.nn as nn class MovingAvg(nn.Module): def __init__(self, kernel_size, stride): super(MovingAvg, self).__init__() self.kernel_size kernel_size self.avg nn.AvgPool1d(kernel_sizekernel_size, stridestride, padding0) def forward(self, x): # 序列两端填充 front x[:, 0:1, :].repeat(1, (self.kernel_size-1)//2, 1) end x[:, -1:, :].repeat(1, (self.kernel_size-1)//2, 1) x torch.cat([front, x, end], dim1) x self.avg(x.permute(0,2,1)) x x.permute(0,2,1) return x class DLinear(nn.Module): def __init__(self, seq_len, pred_len, individualFalse): super(DLinear, self).__init__() self.seq_len seq_len self.pred_len pred_len # 序列分解模块 self.decomposition MovingAvg(kernel_size25, stride1) # 趋势和季节线性层 self.linear_trend nn.Linear(seq_len, pred_len) self.linear_seasonal nn.Linear(seq_len, pred_len) def forward(self, x): # x形状: [Batch, SeqLen, Channel] seasonal, trend self.decomposition(x) seasonal seasonal.permute(0,2,1) trend trend.permute(0,2,1) seasonal_out self.linear_seasonal(seasonal) trend_out self.linear_trend(trend) return (seasonal_out trend_out).permute(0,2,1)模型关键组件说明MovingAvg通过平均池化提取序列趋势线性层分别处理季节和趋势成分参数共享所有特征通道共享相同的线性变换4. 模型训练与验证配置训练参数并启动训练过程device torch.device(cuda if torch.cuda.is_available() else cpu) model DLinear(seq_len96, pred_len96).to(device) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lr1e-3) # 转换数据为PyTorch张量 train_x torch.FloatTensor(X_train).to(device) train_y torch.FloatTensor(y_train).to(device) # 训练循环 for epoch in range(100): model.train() optimizer.zero_grad() outputs model(train_x) loss criterion(outputs, train_y) loss.backward() optimizer.step() if (epoch1) % 10 0: print(fEpoch [{epoch1}/100], Loss: {loss.item():.4f}) # 测试集评估 model.eval() with torch.no_grad(): test_x torch.FloatTensor(X_test).to(device) preds model(test_x).cpu().numpy()性能优化技巧使用学习率调度器如ReduceLROnPlateau增加早停机制EarlyStopping尝试更大的batch_size如32或645. 结果可视化与分析将预测结果与真实值对比展示# 反标准化数据 preds scaler.inverse_transform(preds.reshape(-1,1)) y_true scaler.inverse_transform(y_test.reshape(-1,1)) # 绘制最后5个预测窗口 plt.figure(figsize(15,6)) for i in range(5): idx -5i plt.plot(range(i*96, (i1)*96), y_true[idx*96:(idx1)*96], labelTrue if i0 else None, colorblue) plt.plot(range(i*96, (i1)*96), preds[idx*96:(idx1)*96], labelPred if i0 else None, colorred, linestyle--) plt.title(DLinear 96-step Rolling Prediction on ETTh1) plt.xlabel(Time Steps) plt.ylabel(Oil Temperature) plt.legend() plt.show()评估指标计算from sklearn.metrics import mean_absolute_error, mean_squared_error def calculate_metrics(y_true, y_pred): mae mean_absolute_error(y_true, y_pred) mse mean_squared_error(y_true, y_pred) return {MAE: mae, MSE: mse} metrics calculate_metrics(y_true, preds) print(预测性能指标:) for k, v in metrics.items(): print(f{k}: {v:.4f})实际项目中我发现当预测长度超过192步时DLinear的性能会明显下降。这时可以考虑以下改进方案使用层级预测策略先预测96步再用预测值作为输入预测后续96步尝试调整分解核大小kernel_size参数增加特征工程如添加移动平均、差分等特征