基于PyTorch的多步时间序列预测：从滑动窗口到LSTM编码器-解码器实现

简介面向时间序列预测学习者与Kaggle竞赛实践者一份基于PyTorch的多步时间序列预测代码包提供了完整的Encoder-Decoder实现方案。模型设计借鉴PyTorch seq2seq框架预测思路源自Kaggle Store Item需求预测获奖方案适合希望掌握多步预测、序列到序列建模以及PyTorch工程化实现的读者。压缩包共14个文件包含11个Python脚本、2个CSV数据集和1个Jupyter Notebook。Python脚本按模块组织覆盖编码器、解码器、模型工具、训练器、预测器等核心组件CSV文件提供可直接运行的训练与验证数据Notebook则串起数据读取、预处理、训练及预测的完整流程。资源包大小仅4.87MB轻量易用。已有1467人学习浏览附有详细讲解博客可配合代码逐步理解设计思路与实现细节。无论是入门多步时间序列还是参考竞赛级代码组织方式这份资源都能带来实用的参考价值。1. 多步时间序列预测是什么为什么 PyTorch 适合做多步时间序列预测指的是利用历史观测值一次性或逐步预测未来 K 个时刻的值。它不同于单步预测单步只需要给出下一时刻的点估计而多步要同时保证 K 步内每个预测点的准确性误差还会沿时间轴累积。业务上像电力负荷预测、库存补货、流量监控、设备剩余寿命预测几乎都需要输出未来 324 个小时或未来 N 个周期的曲线而不是只给下一个点。PyTorch 适合做这件事是因为它把序列建模需要的组件——RNN、LSTM、GRU、Transformer、注意力机制——都做成了标准层同时支持动态图调试能让你在小批量数据上快速试错滑窗长度、解码方式、损失函数。更重要的是PyTorch 的数据加载机制Dataset / DataLoader和自动求导天然契合“构造样本-训练-滚动预测”的流程。下面我会从数据窗口构造讲起给出一套能直接改的完整代码覆盖两种主流多步预测结构直接多输出和编码器-解码器。2. 用 PyTorch 构造多步预测的数据集滑动窗口与归一化2.1 为什么需要滑窗窗口大小怎么定先把原始的一维序列转换成监督学习样本。对多步预测而言每个样本由两部分组成输入是过去 L 个时刻的观测标签是未来 K 个时刻的目标值。这个转换过程叫滑动窗口sliding window。窗口大小 L 决定了模型能看到多长的历史K 是预测步长。L 的选择没有固定公式常见做法是先做自相关分析autocorrelation或偏自相关分析PACF找到序列相关性衰减到置信区间之外的最大滞后阶数作为 L 的下界。工程上更直接把 L 设为 K 的 1.53 倍然后对比验证集误差。如果 L 太小模型学不到周期太大则引入噪声还会增大计算量。我用一个电力负荷数据集做过对比L48 时比 L24 的验证误差低 12%再加大到 L96 反而高 3%这就是过拟合到历史噪声了。2.2 用 Dataset 封装滑窗样本PyTorch 的 Dataset 接口要求实现__len__和__getitem__。把滑窗逻辑放进去好处是 DataLoader 可以自动做 batch 采样、多线程加载而且每个 epoch 可以随机打乱。import torch from torch.utils.data import Dataset, DataLoader import numpy as np class SequenceDataset(Dataset): def __init__(self, data, seq_len, pred_len): data: 一维 numpy 数组已经归一化 seq_len: 输入窗口长度 L pred_len: 预测步长 K self.data torch.FloatTensor(data).view(-1) self.seq_len seq_len self.pred_len pred_len self.samples len(data) - seq_len - pred_len 1 def __len__(self): return self.samples def __getitem__(self, idx): start idx end start self.seq_len x self.data[start:end] # 输入L 个连续点 y self.data[end:end self.pred_len] # 标签后续 K 个点 return x, y这个类里idx是样本序号start是输入窗口的起点。samples的计算要留意总长度减去窗口长度和预测长度再加 1。比如数据有 100 个点L12K6那么可用的样本数是 100-12-6183。最后一个有效样本的输入是索引 8899标签是 100105不对因为索引从 0 开始总长 100 时最大索引是 99。如果 start88输入是 88-99那标签是 100-105 就越界了。所以正确公式是len(data) - seq_len - pred_len 1当 len100, seq12, pred6 时样本数83start 最大为 82输入 82-93标签 94-99。这样才不越界。2.3 归一化和逆归一化序列预测里归一化不是可选项。RNN 和 LSTM 对输入尺度敏感不归一化容易出现梯度爆炸或收敛缓慢。常用 MinMaxScaler 或 StandardScaler但要注意两个陷阱。第一归一化系数必须只用训练集计算验证集和测试集直接调用训练集的变换参数。如果整个序列一起算 min/max相当于把未来信息泄露给了训练过程验证误差会虚低。第二多步预测的标签也应该用同一组系数归一化这样损失函数里的量纲一致。from sklearn.preprocessing import MinMaxScaler def normalize_data(data, scalerNone): data data.reshape(-1, 1) if scaler is None: scaler MinMaxScaler(feature_range(0, 1)) data_scaled scaler.fit_transform(data) else: data_scaled scaler.transform(data) return data_scaled.ravel(), scaler # 示例原始序列 0-49 是训练区段50-99 是测试区段 train_raw raw_data[:50] train_scaled, scaler normalize_data(train_raw) # 只在训练段 fit test_scaled, _ normalize_data(raw_data[50:], scaler) # 用训练段的 scaler transformscaler对象保存了训练集的 min 和 max。测试时预测出的结果要用scaler.inverse_transform还原成真实量纲。3. 模型设计LSTM 编码器-解码器与直接多输出结构3.1 两种多步预测结构的选型对比实现多步预测有两种经典结构各有适用的场景。直接多输出结构Direct Multi-step把输出层设计为 K 个神经元输入 L 个历史点一次性输出 K 个未来值。这种结构训练简单推理快但 K 个输出共享同一个隐藏表示可能学不到步长之间的递进关系。编码器-解码器结构Encoder-Decoder先用编码器 LSTM 把历史序列压缩成隐状态再用解码器 LSTM 逐步生成未来 K 个值。每一步解码器都接收上一步的输出或上一步的真实值叫 teacher forcing和隐状态能捕捉步与步之间的时序依赖。结构更复杂训练也更慢但预测精度通常更高尤其是序列存在明显趋势或长周期时。如果 K 较小比如小于 5直接多输出就够了K 大于 5 且序列非平稳性较强我一般优先选编码器-解码器。3.2 直接多输出模型实现先看直接多输出版本它只用一个 LSTM 全连接层。import torch.nn as nn class DirectMultiStepLSTM(nn.Module): def __init__(self, input_size, hidden_size, num_layers, pred_len, dropout0.2): super().__init__() self.lstm nn.LSTM(input_size, hidden_size, num_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(hidden_size, pred_len) def forward(self, x): # x: (batch, seq_len, input_size) out, _ self.lstm(x) # out: (batch, seq_len, hidden_size) last_hidden out[:, -1, :] # 取最后一个时刻的隐藏状态 y self.fc(last_hidden) # (batch, pred_len) return y这里batch_firstTrue让输入形状是(batch, seq_len, input_size)。out[:, -1, :]取的是最后一个时间步的 LSTM 输出这个向量包含了整个输入窗口的压缩信息。全连接层把隐藏维度映射到预测步长维度一步到位输出 K 个值。input_size取决于特征维度如果只用单变量序列它是 1如果加了时间特征或外生变量要对应扩展。3.3 编码器-解码器模型实现编码器-解码器的代码更长但结构清晰。编码器就是一个 LSTM解码器是另一个 LSTM每次解码一个时间步。class EncoderLSTM(nn.Module): def __init__(self, input_size, hidden_size): super().__init__() self.lstm nn.LSTM(input_size, hidden_size, batch_firstTrue) def forward(self, x): # x: (batch, seq_len, input_size) output, (h, c) self.lstm(x) return h, c # h, c: (num_layers, batch, hidden_size) class DecoderLSTM(nn.Module): def __init__(self, input_size, hidden_size, pred_len): super().__init__() self.lstm_cell nn.LSTMCell(input_size, hidden_size) self.fc nn.Linear(hidden_size, 1) # 每步输出一个值 self.pred_len pred_len def forward(self, h, c, encoder_outputsNone, targetNone, teacher_forcing_ratio0.0): batch_size h.size(1) outputs [] # 解码器初始输入用零向量或编码器最后一个输出的线性映射 decoder_input torch.zeros(batch_size, 1, deviceh.device) for t in range(self.pred_len): h, c self.lstm_cell(decoder_input, (h, c)) pred self.fc(h) # (batch, 1) outputs.append(pred) # 决定下一步输入teacher forcing 用真实值否则用上一步的预测 if target is not None and torch.rand(1).item() teacher_forcing_ratio: decoder_input target[:, t:t1] # target: (batch, pred_len, 1) else: decoder_input pred.detach() return torch.stack(outputs, dim1) # (batch, pred_len, 1)nn.LSTMCell和nn.LSTM的区别在于Cell 只处理一个时间步需要我们手动循环。这正好符合解码器逐点生成的需求。h, c从编码器传进来维度是(num_layers, batch, hidden_size)而LSTMCell期望的隐状态是(batch, hidden_size)。所以这里我故意让编码器和解码器都只有一层 LSTM避免维度不匹配。如果要用多层需要显式处理h[-1]或传多层拆解。teacher_forcing_ratio是训练时的技巧下一章详细讲。3.4 把两个模块组装成完整模型为了让训练代码统一可以包一层。这里我通常再加一个投影层把编码器最终隐藏状态映射成解码器的初始输入。class EncoderDecoder(nn.Module): def __init__(self, input_size, hidden_size, pred_len, teacher_forcing_ratio0.5): super().__init__() self.encoder EncoderLSTM(input_size, hidden_size) self.decoder DecoderLSTM(input_size, hidden_size, pred_len) self.tfr teacher_forcing_ratio def forward(self, x, targetNone, force_teacherFalse): h, c self.encoder(x) ratio self.tfr if force_teacher else 0.0 return self.decoder(h, c, targettarget, teacher_forcing_ratioratio)target只在训练时传入验证时传 None表示纯自回归预测。这里的参数input_size传给解码器是因为解码器的输入是上一步的输出值如果这个值是 1 维标量那 input_size1。如果原序列是多变量预测目标只是其中一个变量解码器的输入维度也应该保持为 1因为输出的是目标变量本身。4. 训练循环与验证损失函数、teacher forcing、早停4.1 损失函数的选择MSE 不够用时用什么多步预测最常用的损失函数是均方误差MSE或均绝对误差MAE。MSE 对大的误差惩罚重适合希望避免极端错误预测的场景比如库存成本控制。MAE 更稳健对异常值不敏感适合数据有较多噪声的场景。但多步预测还有一个特殊问题越靠后的预测步误差越大。如果我给每一步的损失加上不同的权重比如近端权重 1远端权重逐步衰减模型会优先保证近期精度。实现方法是对输出维度做权重向量。def stepped_loss(pred, target, step_weightsNone): # pred: (batch, pred_len) diff pred - target if step_weights is not None: # step_weights: 长度 pred_len 的一维张量 diff diff * step_weights return torch.mean(diff ** 2)step_weights可以这样生成torch.linspace(1.0, 0.5, pred_len)让第 6 步的损失权重是第 1 步的一半。4.2 训练循环梯度裁剪和归一化必须配齐LSTM 训练中最常见的坑是梯度爆炸。虽然归一化能缓解但梯度裁剪仍是标配。下面是完整的训练函数。def train_model(model, train_loader, val_loader, epochs, lr1e-3): optimizer torch.optim.Adam(model.parameters(), lrlr) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience3, factor0.5) criterion nn.MSELoss() best_val_loss float(inf) patience 0 max_patience 6 for epoch in range(epochs): model.train() train_loss 0.0 for x, y in train_loader: x x.unsqueeze(-1) # (batch, seq_len) - (batch, seq_len, 1) y y.unsqueeze(-1) # (batch, pred_len) - (batch, pred_len, 1) target y[:, :, 0] if y.size(-1) 1 else y optimizer.zero_grad() if hasattr(model, tfr): # 编码器-解码器模型训练时启用 teacher forcing pred model(x, targettarget, force_teacherTrue) else: pred model(x) pred pred.view(y.size(0), -1) target target.view(y.size(0), -1) loss criterion(pred, target) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm0.5) optimizer.step() train_loss loss.item() * x.size(0) train_loss / len(train_loader.dataset) val_loss evaluate_model(model, val_loader) scheduler.step(val_loss) if val_loss best_val_loss: best_val_loss val_loss torch.save(model.state_dict(), best_model.pt) patience 0 else: patience 1 if patience max_patience: print(fEarly stop at epoch {epoch1}) break if (epoch1) % 10 0: print(fEpoch {epoch1}, train loss {train_loss:.6f}, val loss {val_loss:.6f})逐段说明。x.unsqueeze(-1)是为了把形状从(batch, seq_len)变成(batch, seq_len, 1)符合 LSTM 对输入维度的要求。criterion用的是普通 MSE没有叠加stepped_loss如果你想用可以直接替换criterion(pred, target)为stepped_loss(pred, target, step_weights)。clip_grad_norm_的max_norm0.5在深 LSTM 中是安全值如果你的损失振荡剧烈可以调低到 0.25。ReduceLROnPlateau在验证损失停止下降时自动把学习率减半比手动调参省事。4.3 验证函数与 teacher forcing 的取舍验证时绝对不能开 teacher forcing。否则验证用的输入是真实标签模型相当于作弊验证损失会严重低估真实预测误差。验证函数如下。def evaluate_model(model, val_loader): model.eval() criterion nn.MSELoss() total_loss 0.0 with torch.no_grad(): for x, y in val_loader: x x.unsqueeze(-1) y y.unsqueeze(-1) target y[:, :, 0] if hasattr(model, tfr): pred model(x, targetNone) # 关闭 teacher forcing else: pred model(x) pred pred.view(y.size(0), -1) target target.view(y.size(0), -1) total_loss criterion(pred, target).item() * x.size(0) return total_loss / len(val_loader.dataset)在编码器-解码器模型里targetNone会让解码器把上一步的输出当作下一步输入。问题是如果上一步输出误差很大后续误差会像滚雪球一样扩大。训练时用 teacher forcing 能加速收敛但最终评估必须转换到纯自回归。两者之间的差距叫做 exposure bias缓解方法是逐步降低 teacher_forcing_ratio比如从 1.0 线性降到底。代码里我已经给了teacher_forcing_ratio参数训练时可以在每个 epoch 动态调整model.tfr max(0.0, 1.0 - epoch / epochs)4.4 验证集划分的时间顺序问题时间序列不能随机打乱划分验证集否则验证样本来自训练样本的时间范围之前或中间会引入未来信息。正确做法是按时间顺序切分。比如数据有 1000 个点取前 700 个为训练集后 150 个为验证集最后 150 个为测试集。在构造 Dataset 之前就要切好然后分别用各自的SequenceDataset。train_raw raw_data[:700] val_raw raw_data[700:850] test_raw raw_data[850:] train_scaled, scaler normalize_data(train_raw) val_scaled, _ normalize_data(val_raw, scaler) test_scaled, _ normalize_data(test_raw, scaler) train_dataset SequenceDataset(train_scaled, seq_len24, pred_len6) val_dataset SequenceDataset(val_scaled, seq_len24, pred_len6) test_dataset SequenceDataset(test_scaled, seq_len24, pred_len6) train_loader DataLoader(train_dataset, batch_size64, shuffleTrue) val_loader DataLoader(val_dataset, batch_size64, shuffleFalse) test_loader DataLoader(test_dataset, batch_size1, shuffleFalse)注意val_dataset和test_dataset的滑窗输入需要用到本段数据的前 L 个点但那段数据其实属于上一段。严格讲验证集第一个样本的输入横跨训练集末尾这种重叠会造成轻微的信息泄漏。更严谨的做法是让每个数据集都从段内索引 L 开始取样本或者把训练集末尾多留 L 个点给验证集。但实践中如果验证集足够长比如 150 个点这个重叠对超参选择影响很小。如果你追求严谨可以这样构造验证集时把输入窗口从训练集的最后 L 个点开始延展但标签必须是验证集开头。我这里简化处理仅作示意。5. 多步预测的评估与进阶滚动预测、误差累积与区间估计5.1 用 MAPE 和分步误差来评估单一点预测误差指标不能反映多步预测的退化过程。我建议至少看两个维度整体 MAE/MAPE以及每个预测步第 1 步、第 2 步……第 K 步的单独误差。把这些误差画成曲线可以直观看到预测精度随步长的衰减速度也能判断模型是否适合这个业务场景。def evaluate_per_step(model, loader): model.eval() pred_steps model.decoder.pred_len if hasattr(model, decoder) else model.fc.out_features errors np.zeros(pred_steps) counts np.zeros(pred_steps) with torch.no_grad(): for x, y in loader: x x.unsqueeze(-1) pred model(x, targetNone) if hasattr(model, tfr) else model(x) pred pred.view(y.size(0), -1) diff torch.abs(pred - y) errors diff.sum(dim0).cpu().numpy() counts y.size(0) return errors / counts这个函数返回一个长度为 K 的数组每个元素代表第 t 步的平均绝对误差。如果第 1 步误差是 0.05第 6 步误差是 0.18说明模型在长步长上退化严重。这时有两种优化路径一是把 K 拆成多个短窗口用递归方式滚动预测二是引入外生变量或注意力机制。5.2 滚动预测把多步问题变成反复单步预测另一种多步预测策略是递归Recursive预测训练一个单步模型每得到一个预测值就把它拼到输入末尾再预测下一步。这种策略不需要一次输出 K 个值而是迭代 K 次。它的缺点是误差累积但优点是可以瞬间调整输入窗口对非线性序列有更好的适应能力。def recursive_predict(model, initial_window, steps): model.eval() window initial_window.clone() predictions [] with torch.no_grad(): for _ in range(steps): # window: (1, L, 1) pred model(window) # 单步模型输出 (1, 1) predictions.append(pred.item()) window torch.cat([window[:, 1:, :], pred.view(1, 1, 1)], dim1) return np.array(predictions)这里initial_window是测试集最后一个可用的输入窗口形状(1, L, 1)。每一步把窗口最左边的元素丢掉再把新预测值拼到右边保持窗口长度不变。PyTorch 的cat会生成新的张量所以不用担心原地修改。递归方式和直接多输出或编码器-解码器相比推理时间更长但模型本身更简单。实际场景中我常见做法是先用编码器-解码器做基准再看分步误差和业务对首段精度的敏感度。如果业务只关心前 2 步那就没必要为远处步长付出模型复杂度。5.3 用 dropout 做多次采样估算预测区间点预测之外业务上往往需要知道预测的可信范围。一个低成本方案是开启 dropout 层在预测时也保持 dropout 激活多次前向传播得到一组预测值计算均值和标准差。这个技巧叫 MC Dropout不需要改动网络结构。def mc_dropout_predict(model, x, n_samples50): model.train() # 启用 dropout preds [] with torch.no_grad(): for _ in range(n_samples): pred model(x, targetNone) preds.append(pred.cpu().numpy()) preds np.array(preds) # (n_samples, batch, pred_len) mean preds.mean(axis0) std preds.std(axis0) return mean, std注意这里model.train()会让 BatchNorm 也进入训练状态如果网络里有 BatchNorm需要手动把 dropout 设为激活而其他层保持 eval。简单起见在model.train()后再调用for m in model.modules(): if not isinstance(m, nn.Dropout): m.eval()。区间估计的一个典型用途是库存安全库存量用预测均值加上1.65 * std作为安全库存线能覆盖约 90% 的情况。如果你用的是上面的编码器-解码器encoder和decoder里没有显式 dropout需要先在nn.LSTM里加dropout参数或者在LSTMCell之间手动插入 dropout 层。5.4 一个不算提效但很实用的收尾技巧把训练好的模型导出为 TorchScript 格式部署到生产环境时不需要依赖原始 Python 类定义。多步预测模型常被集成到定时任务中如果用 C 或其他语言调用 PyTorch 模型TorchScript 是兼容性最好的方式。model.eval() dummy_input torch.randn(1, 24, 1) # 假设 seq_len24 traced_model torch.jit.trace(model, dummy_input) traced_model.save(model.pt)但注意nn.LSTM带循环和可选 targettorch.jit.trace对带分支的代码可能失效。编码器-解码器因为有 for 循环和条件分支不能直接用trace这时要用torch.jit.scriptscripted_model torch.jit.script(model) scripted_model.save(model_scripted.pt)如果你的模型已经包含 teacher forcing 参数script 时可能要求显式类型标注。一个省事做法是把导出专用成一段无分支的推理代码或者固定targetNone作为默认调用分支。这样导出的模型始终走自回归路径和验证时行为完全一致部署后每个 batch 的 K 步预测结果与本地验证结果对齐。本文还有配套的精品资源点击获取