当前位置：首页 > news >正文

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

news 2026/7/8 22:22:12

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

时间序列预测一直是数据分析领域的核心挑战之一。在众多预测方法中，DLinear模型以其简洁高效的架构脱颖而出，特别适合需要快速验证模型效果的中级开发者。本文将带您从零开始，通过5个关键步骤完成ETTh1数据集上的96步滚动预测任务，并提供可直接运行的Python代码和可视化方案。

1. 环境准备与数据加载

首先需要配置Python环境和安装必要的依赖库。建议使用Python 3.8+版本，并创建独立的虚拟环境：

conda create -n dlinear python=3.8 conda activate dlinear pip install torch==1.12.1 pandas matplotlib scikit-learn

ETTh1数据集是电力变压器温度监测的经典时间序列数据集，包含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：输入序列长度（建议设置为96）
pred_len：预测步长（本文设置为96）
features：预测模式（'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_len+1): X.append(data[i:i+seq_len]) Y.append(data[i+seq_len:i+seq_len+pred_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_len=96, pred_len=96) 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_size=kernel_size, stride=stride, padding=0) 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], dim=1) 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, individual=False): super(DLinear, self).__init__() self.seq_len = seq_len self.pred_len = pred_len # 序列分解模块 self.decomposition = MovingAvg(kernel_size=25, stride=1) # 趋势和季节线性层 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_len=96, pred_len=96).to(device) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=1e-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 (epoch+1) % 10 == 0: print(f'Epoch [{epoch+1}/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或64）

5. 结果可视化与分析

将预测结果与真实值对比展示：

# 反标准化数据 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 = -5+i plt.plot(range(i*96, (i+1)*96), y_true[idx*96:(idx+1)*96], label='True' if i==0 else None, color='blue') plt.plot(range(i*96, (i+1)*96), preds[idx*96:(idx+1)*96], label='Pred' if i==0 else None, color='red', 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的性能会明显下降。这时可以考虑以下改进方案：