简介：这份资源面向时间序列预测方向的研究者、算法工程师与高年级学生，聚焦用PyTorch复现并改进DeepAR、Informer、Transformer、RNN等经典模型，解决多变量时序分析与预测问题，覆盖金融、气象、工业等场景。压缩包共61个文件，约55.25MB，以49个Python实现文件为主，另含数据压缩包、结果示意图、说明文档与配置json，代码按data、models、exp、test等模块组织，便于按实验脚本逐项复现。目前已有95人学习下载。读者可拿到各模型的完整实现代码与实验脚本，理解DeepAR的概率分布建模、Informer的稀疏注意力、Transformer的自注意力机制以及RNN的时序依赖处理思路，并借助统一数据加载、特征工程、评价指标与可视化脚本，快速开展多变量预测实验与二次开发，同时参考文档中的实验指导与结果分析，排查训练与调参中的常见问题。

1. 从一份 PyTorch 时间序列预测代码包说起：DeepAR、Informer、Transformer、RNN 到底怎么选

手里这份资源，标题很长，核心其实就一句话：用 PyTorch 把时间序列预测里几个主流模型——DeepAR、Informer、Transformer、RNN——全部复现一遍，并且支持多变量输入和多步预测。如果你正在做销量预测、设备传感器趋势分析、电力负荷预估这类任务，大概率已经在论文里见过这几个名字，但真到动手时，往往卡在“用哪个模型”“数据怎么组织”“多变量怎么喂进去”这三步上。这份代码包的价值不在于它写了多少行，而在于它把概率预测（DeepAR）、长序列注意力（Informer）、通用序列建模（Transformer）和经典循环结构（RNN）放在同一个数据管道下，让你能横向对比，而不是每个模型都重新搭一遍脚手架。适合已经会写 PyTorch 训练循环、但没时间从零复现论文的从业者，也适合想快速验证“我的数据到底适合哪类模型”的算法工程师。下面我按实际拆包和跑通的顺序，把选型、数据组织、训练配置和踩坑点逐一讲清楚。

2. 模型选型与数据管道：为什么多变量预测不能直接套单变量代码

2.1 DeepAR、Informer、Transformer、RNN 各自吃什么样的数据

多变量时间序列预测和单变量最大的区别在于：你不再只预测一个数值，而是可能同时预测多个相关指标，或者用多个外生变量辅助预测一个目标。这份代码包里，四个模型的输入输出结构并不完全一样，选错模型或者喂错形状，轻则 loss 不降，重则直接报维度错误。

RNN（这里通常指 LSTM 或 GRU）是最容易上手的。它接受(batch, seq_len, input_size)的输入，input_size就是特征数。如果你有 5 个变量参与预测，input_size=5，输出可以是单步也可以是多步。它的优势是训练稳定、显存占用小，缺点是长序列容易遗忘，预测跨度超过 50 步后效果衰减明显。

Transformer 和 Informer 都基于注意力机制。标准 Transformer 的编码器-解码器结构对序列长度比较敏感，seq_len超过 200 后显存和计算量上升很快。Informer 的核心改进是 ProbSparse 注意力，把复杂度从 O(L²) 降到 O(L log L)，所以它更适合长序列——比如 500 步以上的历史窗口。代码包里 Informer 的配置通常会带distil=True来启用蒸馏层，减少编码器堆叠带来的冗余。

DeepAR 是概率预测模型，输出不是单个值，而是高斯分布或负二项分布的参数。它要求数据按时间序列分组，每个组有独立的cat特征（比如不同门店 ID）。如果你的数据是“多个实体 + 时间”的面板结构，DeepAR 比前三个更合适，因为它能学习跨实体的共享模式。但要注意，DeepAR 对缺失值敏感，输入前必须做填充或掩码。

选型上我一般这样判断：序列长度小于 100、变量少于 10 个，先跑 RNN 做 baseline；序列长度 100 到 500、需要捕捉长期依赖，上 Transformer；超过 500 步或者显存吃紧，换 Informer；如果业务需要预测区间而不是点估计，直接上 DeepAR。

2.2 统一数据管道：滑动窗口、标准化和数据集划分

这份代码包能同时跑四个模型，关键在于它抽象了一个统一的数据管道。不管后面接哪个模型，数据先经过滑动窗口切分，再按特征维度标准化，最后划分训练集和验证集。下面这段代码是我从包里抽出来的核心逻辑，你可以直接抄到自己项目里。

import numpy as np import torch from torch.utils.data import Dataset, DataLoader class TimeSeriesDataset(Dataset): def __init__(self, data, input_len=96, pred_len=24, stride=1, mode='train'): """ data: shape (total_len, num_features) 的二维数组 input_len: 历史窗口长度 pred_len: 预测窗口长度 stride: 滑动步长，训练集通常为1，验证集可以大一些 """ self.data = data self.input_len = input_len self.pred_len = pred_len self.stride = stride self.mode = mode # 计算样本数量 self.n_samples = (len(data) - input_len - pred_len) // stride + 1 def __len__(self): return self.n_samples def __getitem__(self, idx): start = idx * self.stride x = self.data[start : start + self.input_len] # (input_len, F) y = self.data[start + self.input_len : start + self.input_len + self.pred_len] # (pred_len, F) # 如果只预测目标列，y 可以只取最后一列 return torch.FloatTensor(x), torch.FloatTensor(y) def build_dataloader(data, input_len, pred_len, batch_size, train_ratio=0.7, val_ratio=0.1): # 按时间顺序划分，不能打乱 n = len(data) train_end = int(n * train_ratio) val_end = int(n * (train_ratio + val_ratio)) train_data = data[:train_end] val_data = data[train_end:val_end] test_data = data[val_end:] # 标准化：用训练集的均值和方差，避免未来信息泄露 mean = train_data.mean(axis=0) std = train_data.std(axis=0) + 1e-8 train_data = (train_data - mean) / std val_data = (val_data - mean) / std test_data = (test_data - mean) / std train_set = TimeSeriesDataset(train_data, input_len, pred_len, stride=1) val_set = TimeSeriesDataset(val_data, input_len, pred_len, stride=1) test_set = TimeSeriesDataset(test_data, input_len, pred_len, stride=1) train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True, drop_last=True) val_loader = DataLoader(val_set, batch_size=batch_size, shuffle=False) test_loader = DataLoader(test_set, batch_size=batch_size, shuffle=False) return train_loader, val_loader, test_loader, mean, std

这段代码有三个关键点。第一，标准化必须用训练集的均值和方差，验证集和测试集只能做变换，不能重新计算统计量，否则就是典型的数据泄露，线下指标好看线上直接崩。第二，stride在训练集设为 1 是为了最大化样本量，验证集和测试集也可以设为 1，但如果数据量太大，可以适当增大来减少评估时间。第三，drop_last=True只在训练集用，避免最后一个不完整的 batch 影响 BatchNorm 的统计，验证集和测试集保留全部样本。

参数上，input_len和pred_len的比值建议控制在 2:1 到 4:1 之间。比如预测未来 24 步，历史窗口取 96 步比较合理。如果input_len太短，模型看不到足够上下文；太长则训练慢且容易过拟合。batch_size在显存允许的情况下尽量大，32 到 128 之间比较常见，太小会导致梯度噪声大，太大则泛化变差。

提示：多变量预测时，如果只想预测其中某一列，y可以只取目标列，但输入x仍然保留全部特征。代码包里默认是全部预测，你可以在__getitem__里改一行y = y[:, -1:]来切换。

3. 四个模型的 PyTorch 实现要点：从 RNN 到 Informer 的代码拆解

3.1 RNN/LSTM 基线：别小看它，调好参数能打平 Transformer

RNN 类模型在这份代码包里通常作为 baseline 存在，但实际项目中我见过不少场景 LSTM 调好了比 Transformer 还稳。原因很简单：数据量不够大时，注意力机制容易过拟合，而 LSTM 的归纳偏置更适合短序列。

代码包里 LSTM 的实现大致如下：

import torch.nn as nn class LSTMForecaster(nn.Module): def __init__(self, input_size, hidden_size=128, num_layers=2, pred_len=24, dropout=0.2): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0 ) self.fc = nn.Linear(hidden_size, pred_len) # 直接输出多步 def forward(self, x): # x: (batch, input_len, input_size) out, (h_n, c_n) = self.lstm(x) # 取最后一个时间步的隐藏状态 last_hidden = out[:, -1, :] # (batch, hidden_size) pred = self.fc(last_hidden) # (batch, pred_len) return pred

这里有几个参数需要根据你的数据调整。hidden_size从 64 到 256 都常见，太小欠拟合，太大显存和过拟合风险都上升。num_layers一般 1 到 3 层，超过 3 层收益很小，而且梯度问题更明显。dropout只在多层时生效，单层 LSTM 加 dropout 需要手动在输出后加nn.Dropout。fc层直接输出pred_len是最简单的多步策略，也可以改成 seq2seq 结构逐步解码，但代码复杂度会上升。

训练时学习率建议从 1e-3 开始，用ReduceLROnPlateau在验证 loss 不降时减半。如果 loss 震荡严重，先检查标准化是否做对了，再考虑调小学习率或增大 batch_size。

3.2 Transformer 与 Informer：注意力掩码和位置编码的坑

标准 Transformer 用于时间序列预测时，通常只保留编码器，或者用编码器-解码器结构。代码包里一般会实现一个简化版：

class TransformerForecaster(nn.Module): def __init__(self, input_size, d_model=128, nhead=8, num_encoder_layers=3, dim_feedforward=256, pred_len=24, dropout=0.1): super().__init__() self.input_proj = nn.Linear(input_size, d_model) self.pos_encoder = nn.Parameter(torch.randn(1, 500, d_model)) # 可学习位置编码 encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=dim_feedforward, dropout=dropout, batch_first=True ) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_encoder_layers) self.fc = nn.Linear(d_model, pred_len) def forward(self, x): # x: (batch, input_len, input_size) x = self.input_proj(x) # (batch, input_len, d_model) x = x + self.pos_encoder[:, :x.size(1), :] x = self.encoder(x) x = x.mean(dim=1) # 全局平均池化 return self.fc(x)

这段代码里最容易翻车的是位置编码。nn.Parameter定义的可学习位置编码长度是固定的，如果input_len超过 500 就会索引越界。常见做法是改成正弦位置编码，或者把长度设得足够大。另外，batch_first=True在 PyTorch 1.9 之后才支持，老版本需要手动转置。

Informer 的实现要复杂一些，核心是 ProbSparse 注意力。代码包里通常会提供一个简化版，用nn.MultiheadAttention加稀疏掩码来近似。实际跑的时候，distil=True会插入卷积蒸馏层，把编码器输出长度减半，减少计算量。但蒸馏层对短序列反而有害，input_len小于 100 时建议关掉。

调参方面，Transformer 的d_model一般取 64、128、256，nhead要能整除d_model。num_encoder_layers从 2 到 6 都有人用，但时间序列任务里 3 层左右足够。学习率要比 LSTM 小，通常 1e-4 到 5e-4，配合 warmup 更稳。

注意：Transformer 对缺失值非常敏感，输入里如果有 NaN，注意力权重会全部变成 NaN。数据预处理阶段一定要做填充，常见做法是前向填充加均值填充兜底。

3.3 DeepAR 的概率输出：均值方差怎么接损失函数

DeepAR 和前面三个最大的不同是输出层。它不直接预测数值，而是预测分布参数。以高斯分布为例，模型输出均值mu和标准差sigma，损失函数用负对数似然：

class DeepAR(nn.Module): def __init__(self, input_size, hidden_size=128, num_layers=2, pred_len=24): super().__init__() self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True) self.mu_head = nn.Linear(hidden_size, pred_len) self.sigma_head = nn.Linear(hidden_size, pred_len) def forward(self, x): out, _ = self.lstm(x) last = out[:, -1, :] mu = self.mu_head(last) sigma = torch.exp(self.sigma_head(last)) + 1e-6 # 保证正数 return mu, sigma def gaussian_nll_loss(mu, sigma, target): # target: (batch, pred_len) var = sigma ** 2 loss = 0.5 * torch.log(2 * torch.pi * var) + (target - mu) ** 2 / (2 * var) return loss.mean()

sigma用exp激活再加一个小常数，防止训练初期方差为 0 导致 loss 爆炸。实际训练时，DeepAR 的收敛速度比点预测模型慢，学习率要更小，通常 1e-3 起步，配合梯度裁剪。如果 loss 出现 NaN，先检查sigma是否溢出，可以把exp换成softplus。

多变量场景下，DeepAR 可以对每个变量独立输出均值和方差，也可以共享 LSTM 隐藏层再分头输出。代码包里默认是共享隐藏层，这样参数量少，但变量之间相关性特别强时，独立输出可能更好。

4. 训练、验证与调参：让四个模型跑在同一套评估标准下

4.1 统一训练循环与早停策略

四个模型结构不同，但训练循环可以统一。下面这个模板我用了很久，直接套就行：

def train_model(model, train_loader, val_loader, epochs=100, lr=1e-3, patience=10, device='cuda'): model = model.to(device) optimizer = torch.optim.Adam(model.parameters(), lr=lr) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=5) criterion = nn.MSELoss() # DeepAR 换成 gaussian_nll_loss best_val_loss = float('inf') wait = 0 for epoch in range(epochs): model.train() train_loss = 0 for x, y in train_loader: x, y = x.to(device), y.to(device) optimizer.zero_grad() pred = model(x) if isinstance(pred, tuple): # DeepAR 返回 (mu, sigma) loss = gaussian_nll_loss(pred[0], pred[1], y[:, :, -1]) else: loss = criterion(pred, y[:, :, -1]) # 只预测最后一列 loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() train_loss += loss.item() model.eval() val_loss = 0 with torch.no_grad(): for x, y in val_loader: x, y = x.to(device), y.to(device) pred = model(x) if isinstance(pred, tuple): loss = gaussian_nll_loss(pred[0], pred[1], y[:, :, -1]) else: loss = criterion(pred, y[:, :, -1]) val_loss += loss.item() val_loss /= len(val_loader) scheduler.step(val_loss) if val_loss < best_val_loss: best_val_loss = val_loss torch.save(model.state_dict(), 'best_model.pth') wait = 0 else: wait += 1 if wait >= patience: print(f'Early stop at epoch {epoch}') break print(f'Epoch {epoch}: train_loss={train_loss/len(train_loader):.4f}, val_loss={val_loss:.4f}')

几个关键参数：clip_grad_norm_的max_norm设 1.0 到 5.0 之间，RNN 和 DeepAR 建议 1.0，Transformer 可以放宽到 5.0。patience一般 10 到 20，太小容易早停，太大浪费时间。ReduceLROnPlateau的patience设 5，比早停的 patience 小，保证学习率先降再考虑停。

评估指标除了 MSE，建议加上 MAE 和 MAPE。MAPE 在目标值接近 0 时会爆炸，可以改用 sMAPE。代码包里通常只提供 MSE，你可以自己加：

def mae(pred, target): return torch.mean(torch.abs(pred - target)) def smape(pred, target): return torch.mean(2 * torch.abs(pred - target) / (torch.abs(pred) + torch.abs(target) + 1e-8))

4.2 超参数搜索：哪些参数值得调，哪些别浪费时间

四个模型加起来参数不少，但真正影响结果的没几个。按优先级排，我一般这样调：

学习率是共性关键，四个模型都建议用 1e-4、5e-4、1e-3 三个值先粗搜。hidden_size和d_model取 64、128、256 三档。num_layers超过 3 层后收益递减，不用花时间。dropout在 0.1 到 0.3 之间试两三个值就行。

如果显存不够，优先减batch_size和d_model，不要减input_len，因为历史窗口太短会直接损失信息。Informer 的distil在长序列上开，短序列关，这个开关影响比调学习率还大。

5. 避坑与排查：多变量预测里最容易翻车的五个地方

5.1 现象：训练 loss 正常下降，验证 loss 从一开始就很高

原因：数据泄露或者标准化用错。最常见的是在划分训练集之前就做了全局标准化，验证集的信息泄露到了训练阶段。另一个可能是滑动窗口切分时，训练集和验证集有重叠。

解决：严格按时间顺序划分，先切分再标准化，标准化统计量只用训练集计算。检查TimeSeriesDataset的start索引，确保验证集的窗口起点在训练集结束之后。

5.2 现象：Transformer 训练几个 epoch 后 loss 突然变成 NaN

原因：注意力权重溢出，通常是学习率太大或者输入里有异常值。位置编码长度不够也会导致索引错误，但那种是直接报错，不是 NaN。

解决：先把学习率降到 1e-4，加梯度裁剪max_norm=1.0。检查输入数据是否有极端值，标准化后超过 5 个标准差的点建议截断。位置编码改用正弦形式，或者把nn.Parameter的长度设到 1000 以上。

5.3 现象：DeepAR 的 sigma 输出全是接近 0 的值，预测变成点估计

原因：负对数似然损失里，sigma 越小 loss 越小，模型倾向于把 sigma 压到 0 来作弊。这是 DeepAR 训练初期的常见问题。

解决：在 sigma 上加一个下限，比如sigma = torch.clamp(sigma, min=0.01)。或者用 softplus 替代 exp，softplus 在 0 附近更平滑。另外，学习率不要太大，DeepAR 比点预测模型更敏感。

5.4 现象：Informer 在短序列上效果不如 LSTM，甚至差很多

原因：Informer 的 ProbSparse 注意力和蒸馏层是为长序列设计的，短序列上稀疏注意力反而丢失了关键信息，蒸馏层也会把有用的局部模式压缩掉。

解决：input_len小于 100 时直接关掉distil，把注意力换成标准nn.MultiheadAttention。如果还是不行，说明这个任务不适合 Informer，换回 LSTM 或 Transformer。

5.5 现象：多变量预测时，某些变量预测很准，某些完全不准

原因：不同变量的量纲和波动性差异大，统一标准化后，波动小的变量被波动大的变量淹没。另外，如果某些变量和目标变量相关性弱，模型会学到噪声。

解决：对每个变量单独做标准化，或者用 RobustScaler 替代 StandardScaler。检查变量之间的相关性，把相关性低于 0.1 的变量剔除。如果必须保留，可以在损失函数里给不同变量加权。

6. 进阶技巧：用 ONNX 导出和滚动预测把模型推到生产环境

训练完模型只是第一步，真正落地还要解决两个问题：推理速度和长期预测。这份代码包里的模型都是 PyTorch 原生实现，导出 ONNX 后可以用 ONNX Runtime 加速，CPU 上通常能快 2 到 3 倍。下面以 LSTM 为例：

import torch.onnx model = LSTMForecaster(input_size=5, hidden_size=128, pred_len=24) model.load_state_dict(torch.load('best_model.pth')) model.eval() dummy_input = torch.randn(1, 96, 5) # (batch, input_len, input_size) torch.onnx.export( model, dummy_input, 'lstm_forecaster.onnx', input_names=['input'], output_names=['output'], dynamic_axes={'input': {0: 'batch', 1: 'seq_len'}, 'output': {0: 'batch'}}, opset_version=14 )

导出时dynamic_axes把 batch 和 seq_len 设为动态，这样推理时可以接受不同长度的输入。opset_version建议 14 以上，对 LSTM 和注意力支持更好。导出后可以用onnxruntime验证：

import onnxruntime as ort import numpy as np sess = ort.InferenceSession('lstm_forecaster.onnx') test_input = np.random.randn(1, 96, 5).astype(np.float32) output = sess.run(None, {'input': test_input}) print(output[0].shape) # (1, 24)

滚动预测是另一个生产环境常见需求。比如你要预测未来 7 天，但模型一次只能输出 24 步，那就预测完 24 步后，把预测值拼回输入序列，再预测下一段。代码包里通常没有现成的滚动预测函数，我一般这样写：

def rolling_predict(model, initial_input, total_steps, pred_len, device='cuda'): """ initial_input: (1, input_len, F) 初始历史窗口 total_steps: 总共要预测的步数 pred_len: 模型单次输出步数 """ model.eval() predictions = [] current_input = initial_input.clone().to(device) steps_done = 0 with torch.no_grad(): while steps_done < total_steps: pred = model(current_input) # (1, pred_len) if isinstance(pred, tuple): pred = pred[0] # DeepAR 取均值 take = min(pred_len, total_steps - steps_done) predictions.append(pred[:, :take].cpu()) # 把预测值拼到输入末尾，去掉最前面的等长部分 new_values = pred[:, :take].unsqueeze(-1) # (1, take, 1) # 如果多变量，需要把预测值扩展到所有特征维度，或者只更新目标列 current_input = torch.cat([current_input[:, take:, :], new_values.repeat(1, 1, current_input.size(-1))], dim=1) steps_done += take return torch.cat(predictions, dim=1)

滚动预测最大的坑是误差累积。每一步的预测误差会传到下一步，预测越长偏差越大。缓解办法有两个：一是用 DeepAR 的概率输出，每一步采样多个轨迹取平均；二是每隔几步用真实值校正一次，但生产环境往往拿不到实时真实值。我一般会在验证集上先测滚动预测的误差曲线，如果 7 天后的误差超过 20%，就缩短预测周期或者换更稳的模型。

从那以后我每次上线时间序列模型前，都强制走一遍滚动预测验证，不再只看单窗口的 MSE。希望帮到你。

本文还有配套的精品资源，点击获取