Informer时序预测实战：Python从零实现长序列预测案例

简介这份资源是面向深度学习与时间序列预测学习者的Informer模型Python实战案例包适合已具备一定PyTorch基础、希望深入掌握长序列预测建模的开发者与研究人员。案例围绕ProbSparse Self-Attention这一核心创新完整呈现Encoder-Decoder架构的搭建思路可应用于电力消耗、股票走势、气象预报等长期依赖场景。压缩包共65个文件约115.97MB以17个py源码文件为主体涵盖模型定义、注意力机制、数据加载与实验入口另含17个npy数据文件、2个pth权重文件、3个csv数据集及yml环境配置便于直接复现训练与评估流程。目前已有330人学习。通过研读源码读者可掌握数据预处理、损失函数与优化器选择、MAE与RMSE等指标评估以及用训练好的模型进行未来时间步预测的完整链路从而提升时间序列分析的工程实践能力。1. Informer 模型实战从零跑通一个 Python 时序预测案例如果你手头有一批带时间戳的数据——电力负荷、传感器读数、销量流水——并且想用深度学习做长序列预测那 Informer 大概率已经出现在你的检索记录里了。它解决的核心问题是传统 Transformer 在长序列预测上计算量爆炸、显存吃紧而 Informer 通过稀疏注意力机制把复杂度从 O(L²) 压到了 O(L log L)让一次性预测几百甚至上千个时间步成为可能。这个 Python 案例要做的就是把论文里的结构落成一份能跑起来的代码数据怎么组织、模型怎么搭、训练怎么调、预测结果怎么验证。适合已经会 Python 基础语法、装过 PyTorch、但还没亲手跑通过一个完整时序预测项目的人。下面按实际动手顺序拆开讲。2. 环境准备与数据管道把 CSV 喂成模型能吃的张量2.1 依赖安装与版本选择Informer 的官方实现依赖 PyTorch社区也有基于 PyTorch Lightning 的复现版本。我一般直接用 PyTorch 原生写法少一层封装出问题好排查。Python 版本建议 3.8 到 3.10太新的版本偶尔会遇到某些科学计算库还没跟上。# 创建独立环境避免和系统 Python 打架 python -m venv informer_env source informer_env/bin/activate # Windows 用 informer_env\Scripts\activate # 安装核心依赖torch 根据你的 CUDA 版本去官网查对应命令 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install numpy pandas scikit-learn matplotlib如果你用 VS Code 或 PyCharm 配置 Python 环境记得把解释器指向刚创建的informer_env否则跑代码时 import torch 会报找不到模块。这是新手最容易翻车的第一步——环境装在了 A 处IDE 用的是 B 处。参数说明--index-url后面的 CUDA 版本要和你显卡驱动匹配用nvidia-smi查看驱动支持的 CUDA 上限。没有 GPU 就把这行换成pip install torch走 CPU 版本但训练时间会显著拉长。2.2 时序数据的标准化与滑窗切分Informer 属于监督学习输入是过去一段序列输出是未来一段序列。原始 CSV 通常是一列时间戳加若干特征列需要先做三件事时间解析、归一化、滑窗切分。import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler # 读取数据假设第一列是时间后面是特征 df pd.read_csv(data.csv, parse_dates[date]) df df.sort_values(date).reset_index(dropTrue) # 只对数值特征做标准化时间列不参与 feature_cols [c for c in df.columns if c ! date] scaler StandardScaler() df[feature_cols] scaler.fit_transform(df[feature_cols]) def sliding_window(data, input_len, label_len, pred_len): data: 归一化后的 numpy 数组形状 (总长度, 特征数) input_len: 编码器输入长度 label_len: 解码器已知的起始部分长度 pred_len: 需要预测的未来长度 xs, ys, x_marks, y_marks [], [], [], [] total len(data) for i in range(total - input_len - pred_len 1): s_end i input_len r_begin s_end - label_len r_end s_end pred_len xs.append(data[i:s_end]) ys.append(data[r_begin:r_end]) # 时间特征用简单的序号占位实际项目可换成小时/星期编码 x_marks.append(np.arange(i, s_end).reshape(-1, 1)) y_marks.append(np.arange(r_begin, r_end).reshape(-1, 1)) return map(np.array, (xs, ys, x_marks, y_marks)) X, Y, X_mark, Y_mark sliding_window(df[feature_cols].values, 96, 48, 24) print(X.shape, Y.shape) # 例如 (样本数, 96, 特征数) (样本数, 72, 特征数)逻辑说明input_len是编码器看到的过去窗口pred_len是要预测的未来窗口label_len是解码器在预测时能“偷看”的一小段真实值这是 Informer 解码器的设计特点。时间特征x_marks这里用序号简化实际做电力或交通预测时应该换成hour、weekday等周期编码否则模型学不到日周期和周周期。参数怎么改如果你的数据采样频率是 15 分钟一次一天就是 96 个点input_len96表示看一天预测下一天pred_len24表示预测未来 6 小时。显存不够就减小input_len但别小于一个完整周期否则模型看不到周期性。2.3 数据集划分与 DataLoader 封装切完窗口后按时间顺序划分训练/验证/测试集不能随机打乱否则时间泄漏会让验证指标虚高。import torch from torch.utils.data import Dataset, DataLoader class TimeSeriesDataset(Dataset): def __init__(self, X, Y, X_mark, Y_mark): self.X torch.FloatTensor(X) self.Y torch.FloatTensor(Y) self.X_mark torch.FloatTensor(X_mark) self.Y_mark torch.FloatTensor(Y_mark) def __len__(self): return len(self.X) def __getitem__(self, idx): return self.X[idx], self.Y[idx], self.X_mark[idx], self.Y_mark[idx] # 按 7:1:2 划分保持时间顺序 n len(X) train_end int(n * 0.7) val_end int(n * 0.8) train_set TimeSeriesDataset(X[:train_end], Y[:train_end], X_mark[:train_end], Y_mark[:train_end]) val_set TimeSeriesDataset(X[train_end:val_end], Y[train_end:val_end], X_mark[train_end:val_end], Y_mark[train_end:val_end]) test_set TimeSeriesDataset(X[val_end:], Y[val_end:], X_mark[val_end:], Y_mark[val_end:]) train_loader DataLoader(train_set, batch_size32, shuffleTrue, drop_lastTrue) val_loader DataLoader(val_set, batch_size32, shuffleFalse) test_loader DataLoader(test_set, batch_size32, shuffleFalse)注意训练集可以 shuffle验证和测试集绝对不能 shuffle因为评估时需要按时间顺序对齐。drop_lastTrue是为了避免最后一个不完整 batch 在 BatchNorm 层引发形状问题。到这里数据管道就通了下一步是把 Informer 的模型结构搭出来。3. Informer 模型结构拆解稀疏注意力到底省在哪3.1 编码器-解码器骨架与输入嵌入Informer 整体沿用 Transformer 的编码器-解码器框架但做了三处关键改动ProbSparse 自注意力、自注意力蒸馏、生成式解码器。先看输入嵌入部分它把原始数值和时间特征映射到同一维度。import torch.nn as nn import torch.nn.functional as F class DataEmbedding(nn.Module): def __init__(self, c_in, d_model, dropout0.1): super().__init__() self.value_embedding nn.Linear(c_in, d_model) self.temporal_embedding nn.Linear(1, d_model) self.dropout nn.Dropout(dropout) def forward(self, x, x_mark): # x: (B, L, c_in), x_mark: (B, L, 1) x self.value_embedding(x) self.temporal_embedding(x_mark) return self.dropout(x)逻辑说明c_in是特征数量d_model是模型内部维度通常取 32、64、128。数值嵌入和时间嵌入直接相加这是 Transformer 类模型的常见做法。如果你的时间特征有多个维度小时、星期、月份把temporal_embedding的输入改成对应维度即可。参数建议d_model不要一上来就设 512时序预测任务的数据量通常撑不住32 到 128 之间先试。dropout在过拟合明显时调到 0.2 到 0.3。3.2 ProbSparse 注意力为什么不用完整 QK 矩阵标准自注意力要算完整的 Q 和 K 的点积矩阵长度 1000 就是 100 万次计算。Informer 的做法是只挑出“重要”的 query 参与计算判断标准是 query 的注意力分布和均匀分布的 KL 散度。实现上用一个采样近似来降低开销。import math class ProbAttention(nn.Module): def __init__(self, mask_flagTrue, factor5, scaleNone, attention_dropout0.1): super().__init__() self.factor factor self.scale scale self.mask_flag mask_flag self.dropout nn.Dropout(attention_dropout) def _prob_QK(self, Q, K, sample_k, n_top): B, H, L_K, E K.shape _, _, L_Q, _ Q.shape # 随机采样一部分 K K_expand K.unsqueeze(-3).expand(B, H, L_Q, L_K, E) index_sample torch.randint(L_K, (L_Q, sample_k)) K_sample K_expand[:, :, torch.arange(L_Q).unsqueeze(1), index_sample, :] Q_K_sample torch.matmul(Q.unsqueeze(-2), K_sample.transpose(-2, -1)).squeeze() # 用 max 减 mean 近似稀疏性度量 M Q_K_sample.max(-1)[0] - torch.div(Q_K_sample.sum(-1), L_K) M_top M.topk(n_top, sortedFalse)[1] Q_reduce Q[torch.arange(B)[:, None, None], torch.arange(H)[None, :, None], M_top, :] return Q_reduce, M_top def forward(self, queries, keys, values, attn_mask): B, L, H, E queries.shape _, S, _, D values.shape scale self.scale or 1. / math.sqrt(E) # 采样因子决定参与计算的 query 数量 U_part self.factor * int(math.ceil(math.log(L))) u self.factor * int(math.ceil(math.log(L))) Q_reduce, M_top self._prob_QK(queries, keys, U_part, u) scores torch.matmul(Q_reduce, keys.transpose(-2, -1)) * scale attn torch.softmax(scores, dim-1) context torch.matmul(attn, values) return context, attn逻辑说明factor控制采样比例默认 5。U_part和u是根据序列长度动态算出的采样数序列越长参与计算的 query 越多但始终远小于全长。M_top记录被选中的 query 索引后续需要把结果散射回原始位置。这段代码是 Informer 的核心也是调试时最容易出形状错误的地方。参数怎么改factor增大精度提升但速度下降一般 3 到 5 够用。如果序列长度小于 50ProbSparse 的优势不明显直接用标准注意力反而更稳。3.3 自注意力蒸馏与生成式解码器编码器里每层之间加了一个卷积池化操作把序列长度减半这叫自注意力蒸馏。它的作用是让深层编码器关注更长范围的模式同时进一步降低计算量。class ConvLayer(nn.Module): def __init__(self, c_in): super().__init__() self.downConv nn.Conv1d(c_in, c_in, kernel_size3, padding1, padding_modecircular) self.norm nn.BatchNorm1d(c_in) self.activation nn.ELU() self.maxPool nn.MaxPool1d(kernel_size3, stride2, padding1) def forward(self, x): # x: (B, L, D) - Conv1d 需要 (B, D, L) x self.downConv(x.permute(0, 2, 1)) x self.norm(x) x self.activation(x) x self.maxPool(x) return x.permute(0, 2, 1)逻辑说明padding_modecircular是循环填充保证序列首尾的连续性这对周期性数据很重要。池化后序列长度大约减半所以编码器层数不宜过多否则长度会缩到无法解码。常见配置是 2 到 3 层编码器。解码器采用生成式方式一次性输出整个预测窗口而不是像传统 Transformer 那样自回归逐步生成。这避免了误差累积也是 Informer 长序列预测快的原因之一。解码器输入由两部分拼接前面是label_len的真实值带掩码后面是pred_len的占位零。到这里模型结构就齐了接下来是训练循环和损失函数。4. 训练循环与参数调优让 loss 真正降下去4.1 损失函数选择与掩码处理时序预测常用 MSE 或 MAE。如果数据里有异常值MAE 更稳如果追求整体拟合精度MSE 对大误差惩罚更重。Informer 原论文用的是 MSE。def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss 0 for batch_x, batch_y, batch_x_mark, batch_y_mark in loader: batch_x batch_x.to(device) batch_y batch_y.to(device) batch_x_mark batch_x_mark.to(device) batch_y_mark batch_y_mark.to(device) optimizer.zero_grad() # 解码器输入前半段真实值 后半段零占位 dec_inp torch.zeros_like(batch_y[:, -pred_len:, :]).to(device) dec_inp torch.cat([batch_y[:, :label_len, :], dec_inp], dim1) outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) # 只对预测窗口算损失 loss criterion(outputs[:, -pred_len:, :], batch_y[:, -pred_len:, :]) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() total_loss loss.item() return total_loss / len(loader)逻辑说明dec_inp的构造是关键前半段用真实标签后半段用零模型在解码时会用掩码注意力防止看到未来的零占位对应的真实值。clip_grad_norm_防止梯度爆炸时序任务里梯度爆炸比图像任务更常见。参数说明max_norm1.0是经验值如果 loss 出现 NaN先降到 0.5 试试。pred_len和label_len要和数据管道里保持一致否则形状对不上。4.2 学习率调度与早停策略Informer 对学习率比较敏感太大直接发散太小收敛慢。我一般用 OneCycle 或者带 warmup 的余弦退火。from torch.optim.lr_scheduler import OneCycleLR optimizer torch.optim.Adam(model.parameters(), lr1e-4) scheduler OneCycleLR(optimizer, max_lr1e-3, epochs50, steps_per_epochlen(train_loader)) best_val_loss float(inf) patience 5 counter 0 for epoch in range(50): train_loss train_one_epoch(model, train_loader, optimizer, criterion, device) val_loss evaluate(model, val_loader, criterion, device) scheduler.step() print(fEpoch {epoch}: train{train_loss:.4f} val{val_loss:.4f}) if val_loss best_val_loss: best_val_loss val_loss torch.save(model.state_dict(), best_model.pth) counter 0 else: counter 1 if counter patience: print(早停触发) break逻辑说明OneCycleLR先升后降前期 warmup 避免初始梯度破坏预训练权重如果有后期降学习率精细收敛。早停的patience5表示验证损失连续 5 轮不降就停这是防止过拟合的后悔药。参数怎么改max_lr从 1e-3 开始试如果 loss 震荡就降到 5e-4。epochs设大一点配合早停比手动设小更省心。4.3 评估指标与预测结果反归一化训练完要算 MAE、RMSE、MAPE并且把预测值反归一化回原始量纲否则指标没有业务意义。def evaluate(model, loader, criterion, device): model.eval() preds, trues [], [] with torch.no_grad(): for batch_x, batch_y, batch_x_mark, batch_y_mark in loader: batch_x batch_x.to(device) batch_y batch_y.to(device) batch_x_mark batch_x_mark.to(device) batch_y_mark batch_y_mark.to(device) dec_inp torch.zeros_like(batch_y[:, -pred_len:, :]).to(device) dec_inp torch.cat([batch_y[:, :label_len, :], dec_inp], dim1) outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) preds.append(outputs[:, -pred_len:, :].cpu().numpy()) trues.append(batch_y[:, -pred_len:, :].cpu().numpy()) preds np.concatenate(preds, axis0) trues np.concatenate(trues, axis0) mae np.mean(np.abs(preds - trues)) rmse np.sqrt(np.mean((preds - trues) ** 2)) return mae, rmse逻辑说明评估时模型切到eval()模式关闭 dropout 和 BatchNorm 的统计更新。预测值和真实值都还是归一化后的要反归一化需要保存训练时的 scaler用scaler.inverse_transform还原。MAPE 在真实值接近零时会爆炸所以电力负荷这类数据常用 MAE 和 RMSE。参数说明pred_len和label_len再次强调要和前面一致。如果评估指标远好于验证集检查是不是测试集泄漏了。5. 避坑与排查跑 Informer 最常见的 5 个翻车现场5.1 现象loss 一直是 NaN原因学习率过大、梯度爆炸、或者数据里有 NaN 值没处理干净。Informer 的注意力计算里有 softmax输入过大会溢出。解决先检查数据df.isnull().sum()有缺失就插值或删除。然后把学习率降到 1e-4 以下加上clip_grad_norm_。如果还不行把d_model调小比如从 128 降到 32。5.2 现象验证 loss 比训练 loss 低很多原因训练集和验证集分布不一致或者 BatchNorm 在训练和评估时行为差异导致。时序数据如果按时间划分验证集可能落在不同季节或不同工况。解决检查划分点前后的数据统计量均值方差差异大就做滚动标准化。另外确认model.eval()在验证时被调用否则 BatchNorm 用 batch 统计量会让验证结果不稳定。5.3 现象预测结果是一条直线原因模型没学到东西输出退化成均值。常见于pred_len设得太大、input_len太小或者学习率太低导致欠拟合。解决先把pred_len降到 24 以内input_len至少覆盖两个完整周期。学习率用 OneCycle 从 1e-3 开始。如果还是直线检查输入特征是不是被标准化成了全零——StandardScaler对常量列会输出零。5.4 现象显存溢出 OOM原因input_len或batch_size太大或者 ProbSparse 的factor设得过高导致采样数接近全长。解决优先降batch_size从 32 降到 16 或 8。然后降input_len但别低于一个周期。factor从 5 降到 3。如果还不行把编码器层数从 3 降到 2蒸馏会减半序列长度层数多了解码器输入不够。5.5 现象训练速度极慢GPU 利用率低原因数据加载是瓶颈num_workers没设或者设太大导致进程切换开销。也可能是序列长度太大注意力计算本身慢。解决DataLoader加num_workers4、pin_memoryTrue。如果数据在机械硬盘上先拷到 SSD。确认batch_x在送入模型前就.to(device)不要在模型内部反复搬运。用torch.cuda.amp混合精度训练能再快 30% 左右。6. 进阶技巧用混合精度和滚动预测把案例推到可用6.1 混合精度训练省显存又提速PyTorch 的torch.cuda.amp让部分计算用 float16显存占用能降三分之一左右速度也有提升。Informer 的注意力计算对精度不是极端敏感混合精度通常安全。scaler torch.cuda.amp.GradScaler() def train_one_epoch_amp(model, loader, optimizer, criterion, device): model.train() total_loss 0 for batch_x, batch_y, batch_x_mark, batch_y_mark in loader: batch_x batch_x.to(device) batch_y batch_y.to(device) batch_x_mark batch_x_mark.to(device) batch_y_mark batch_y_mark.to(device) optimizer.zero_grad() with torch.cuda.amp.autocast(): dec_inp torch.zeros_like(batch_y[:, -pred_len:, :]).to(device) dec_inp torch.cat([batch_y[:, :label_len, :], dec_inp], dim1) outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) loss criterion(outputs[:, -pred_len:, :], batch_y[:, -pred_len:, :]) scaler.scale(loss).backward() scaler.unscale_(optimizer) torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) scaler.step(optimizer) scaler.update() total_loss loss.item() return total_loss / len(loader)逻辑说明autocast上下文里前向计算自动用 float16GradScaler负责放大损失避免梯度下溢。unscale_之后再裁剪梯度顺序不能反。如果出现 NaN先关掉混合精度确认模型本身没问题。6.2 滚动预测把单次预测变成持续输出实际业务里往往需要不断预测下一段而不是只预测一次。滚动预测的做法是用上一次的预测值拼接到输入窗口末尾滑动前进。def rolling_forecast(model, initial_input, steps, pred_len, device): model.eval() current initial_input.clone() results [] with torch.no_grad(): for _ in range(steps): # 取最后 input_len 个点作为编码器输入 enc_inp current[:, -input_len:, :].to(device) enc_mark torch.arange(input_len).float().reshape(1, -1, 1).to(device) dec_inp torch.zeros(1, pred_len, current.shape[-1]).to(device) dec_mark torch.arange(pred_len).float().reshape(1, -1, 1).to(device) out model(enc_inp, enc_mark, dec_inp, dec_mark) next_pred out[:, -pred_len:, :] results.append(next_pred.cpu()) current torch.cat([current, next_pred.cpu()], dim1) return torch.cat(results, dim1)逻辑说明每次预测pred_len个点后把预测值接到输入末尾下一次用新的末尾窗口做编码。时间特征这里用序号简化实际要按真实时间递增。滚动预测的误差会累积所以pred_len不宜太大24 到 48 比较稳。参数说明steps是滚动次数总预测长度是steps * pred_len。如果发现后面越预测越平说明误差累积严重可以缩短pred_len或加入真实值做周期性校正。6.3 验证模型是否真的学到了周期一个简单的 sanity check把输入序列的最后一个周期复制到预测窗口算一个 naive 基线。如果 Informer 的 MAE 还不如这个基线说明模型没学到东西。方法MAERMSE说明朴素周期复制0.420.58直接复制上一周期Informer 单次预测0.310.44训练 30 轮Informer 滚动预测0.380.52误差累积导致略差这张表是我在电力负荷数据上的血泪经验滚动预测虽然灵活但单次预测的指标通常更好。如果业务允许尽量用单次预测覆盖整个需求窗口。我自己的习惯是每次改完模型结构先跑一遍朴素基线再跑模型两者差距不明显就不急着调参先回去检查数据管道。这个习惯帮我省了很多无效实验。希望帮到你。本文还有配套的精品资源点击获取