简介基于Transformer架构实现长期预测的Python代码包面向NLP与时序预测方向的学习者适合希望从零搭建序列模型、进行多步预测并输出可视化结果的开发者。压缩包内共39个文件以13个Python脚本为核心涵盖数据加载、时间特征处理、模型构建与训练等环节另包含预训练权重pth、结果图png、预测结果csv及项目配置文件整包大小26.49MB。目前已有274人学习下载。代码基于ETTh1数据集演示电力负荷长期预测通过自注意力、多头注意力、位置编码及残差连接等模块的完整实现帮助读者理解Transformer内部机制并可直接运行main.py查看预测序列与真实值的对比图还可利用现成评估指标衡量效果为迁移到其他序列预测任务提供可复用的工程参考。1. Transformer做长期预测为什么这个方向越来越值得投入如果你手里有一条以小时或天为粒度的业务指标例如设备温度、仓库销量、集群负载想一次性往后预测几十甚至上百个时间步传统的LSTM和ARIMA往往会在预测窗口拉长之后快速失效——误差累积、趋势滞后、峰值被抹平。这个场景正是Transformer模型切入长期预测的价值点用注意力机制直接建模序列中任意两个时间点之间的关系而不是像循环网络那样按时间步顺序传递信息从而让“未来的某个值受二十步之前某个尖峰影响”这种依赖成为可能。本文要拆解的是一个非常具体的落地路径拿到一套python代码.zip里面是Transformer模型实现长期预测并可视化结果的完整流程。我按自己实际做过的方案把任务定义、数据切分、模型搭建、训练调参、结果可视化到最后的坑踩一遍保证你照着能跑通也明白每一步为什么这么做。适合刚接触Transformer时间序列应用、或者已经在用LSTM做预测但效果受限的从业者。2. 长期预测的任务定义与Transformer选型从自回归到滑窗训练2.1 长期预测到底在预测什么单步与多步的数学定义长期预测Long-term Forecasting和普通预测的区别不在模型而在输出长度。假设历史序列为X [x_1, x_2, ..., x_T]单步预测只输出x_{T1}多步预测要一次输出[x_{T1}, x_{T2}, ..., x_{TH}]这里H就是预测长度也叫horizon。工程上H大于等于24就可以算长期预测按“天”粒度的数据H30就是预测未来一个月。有两种做法来实现多步输出。第一种叫递归预测先用模型预测x_{T1}把它拼回输入序列再预测x_{T2}依次滚下去。缺点很明显每一步都会把上一步的误差带进新输入随着H加长误差迅速膨胀最后预测结果经常退化成一条平线。第二种叫直接多步预测一次性输出H个值让模型自己学习未来窗口内部的相关性。我的建议是既然用了Transformer直接做第二种——Transformer的结构优势本来就是并行处理整个序列你让它一次把未来H个点全吐出来比让它自回归滚动要稳得多。2.2 Transformer比LSTM更适合长序列的机制注意力与位置编码LSTM处理长序列时信息要经过每个时间步的门控单元逐级传递路径长度等于时间步数这会导致远端信息在传递过程中被稀释。Transformer把这种传递路径缩短到了常数级任意两个时间步之间只隔了一次注意力计算。无论第i步和第j步距离多远注意力权重都能直接计算出来。这就是Transformer能捕获长程依赖的核心原因。但代价是Transformer本身是“无序”的它不像RNN天然有先后顺序。所以必须把位置信息编码进输入。经典做法是用不同频率的正弦余弦函数做位置编码也叫绝对位置编码。后来Informer、PatchTST这些模型证明对时间序列预测来说位置编码不一定用正弦直接用可学习的embedding甚至截断的时间戳特征小时、星期、月份效果更好。原因在于时间序列的位置有语义——第23点和第0点之间的距离不只是“差了23个位置”而是“跨了一天”这种周期性语义正弦编码表达得不够直接。我一般会在代码里把两套都实现出来默认用可学习位置编码后面接一个开关切换方便在特定数据集上对比效果。你拿到代码包后先看它的PositionalEncoding类用的是正弦还是可学习参数这点直接决定你的模型有没有利用到时间周期性。2.3 两类Transformer预测架构编码器全输出与自回归解码抛开各种改进变体Transformer做时间序列预测的主流框架就两种。一种是编码器-解码器结构模仿NLP翻译任务编码器读历史序列解码器以teacher forcing方式生成未来序列。Informer的ProbSparse注意力就是在这一框架下做的改进。这种结构训练稳定但实现复杂解码器的mask机制、动态position embedding处理不好就容易训崩。另一种是纯编码器结构也是我拿到这类代码包后最推荐优先跑通的方案把历史窗口整段丢进TransformerEncoder取编码器输出的最后一个或全部时间步的表示过一个全连接层直接输出H个未来值。这样省略了解码器训练更快代码能压缩到一百行以内。代价是表达能力相对弱一些但对大多数业务指标预测足够用。选哪条路取决于你手里数据的复杂度。单变量、周期明显、噪声不大的序列纯编码器足矣多变量强耦合、有长周期有突发事件的序列先跑纯编码器做基线再切换到编码器-解码器对比提升幅度。接下来我按纯编码器结构作为主线讲因为这套代码最简单、最好排查问题你理解了它再看任何Transformer预测代码都不会懵。3. 数据预处理长期预测最容易翻车的环节3.1 滑窗切分与数据集划分训练/验证/测试集的比例与顺序问题时间序列数据和图像数据最大的区别是不能随机打乱。你如果像分类任务那样把样本shuffle了模型学到的是“记忆答案”而不是“预测未来”——训练集里包含测试集时间段的数据验证集就失去了意义。正确做法是按时间顺序依次切分。滑窗切分的逻辑是给定原始序列、输入窗口长度L和预测窗口长度H从第0个位置开始取[0:L]作为输入、[L:LH]作为标签然后向后滑动步长s重复这个过程。代码包里的预处理脚本核心就是这个函数import numpy as np def create_sliding_windows(data, input_len, pred_len, stride1): 将一维或多维时间序列切分为滑窗样本。 data: shape (T, N)T为时间步数N为特征数 input_len: 输入窗口长度即用过去多少个点做预测 pred_len: 预测窗口长度即要预测未来多少个点 stride: 滑窗步长步长越大样本越少 返回: X: shape (num_samples, input_len, N) Y: shape (num_samples, pred_len, N) X, Y [], [] for start in range(0, len(data) - input_len - pred_len 1, stride): end_input start input_len X.append(data[start:end_input]) Y.append(data[end_input:end_input pred_len]) return np.array(X), np.array(Y)这段代码做了三件事第一窗口起点从0开始以stride为间隔滑动第二每个样本的输入段是[start:end_input]标签段紧跟其后第三样本总数大约等于(T - input_len - pred_len) / stride数据量不够时优先调小stride而不是调小input_len因为输入窗口长度影响模型能看到的历史范围。划分数据集时我有一个不成文的习惯按时间顺序切出前70%做训练中间15%做验证最后15%做测试。为什么留最后一段做测试因为预测任务要模拟真实场景——用过去的数据训练在尚未发生的时间段上验证只有最后15%才是模型“没看过”的时段。3.2 归一化的坑用MinMax还是Standard为什么要防止数据泄露归一化在LSTM时代就是个头疼问题Transformer更是如此。Transformer里的注意力机制计算的是Q和K的点积数值范围直接影响softmax的锐度。输入数据如果量纲差距大比如温度20度和压力5000注意力分数会被大数值特征主导模型学不到真正的依赖关系。MinMax归一化把数据压缩到[0,1]或[-1,1]适合分布比较平稳的序列Standard标准化z-score把数据变成均值0、方差1适合有离群值或分布厚尾的序列。长期预测场景我优先推荐Standard原因有二一是预测目标值有时会超出历史数据的[min, max]范围MinMax会把超出部分截断在边界上导致模型预测永远“顶格”二是Standard能保留离群值的信息让注意力机制有机会关注到异常点。看代码包里preprocess.py时重点检查一个细节归一化器的fit操作有没有在划分数据集之后、并且只在训练集上执行。from sklearn.preprocessing import StandardScaler # 正确做法先分割再fitscaler只用在训练集上 # 假设data已是按时间排序的原始数据 train_size int(len(data) * 0.7) val_size int(len(data) * 0.15) train_data data[:train_size] val_data data[train_size:train_sizeval_size] test_data data[train_sizeval_size:] scaler StandardScaler() # 只对训练序列fit否则验证集/测试集的统计量会泄进训练过程 train_data_scaled scaler.fit_transform(train_data) val_data_scaled scaler.transform(val_data) test_data_scaled scaler.transform(test_data)这里的坑藏得很深如果你的滑窗是在原始数据上切、然后统一normalize就造成数据泄露——测试集的均值、方差参与了训练输入的变换模型在训练时“偷看”了测试集的分布信息测试结果虚高。正确顺序是先划分时间区间再fit scaler于训练段最后用同一个scaler去transform验证段和测试段。你判断一份代码规不规范第一眼看这个位置就行。3.3 多变量输入输出设计target列的选择与特征拼接多数业务场景不是单变量预测。以设备温度预测为例目标变量是温度但输入还包括环境湿度、负载功率、风速等外部变量。这些外部特征不能直接丢进Transformer要分清楚哪些是“可预知未来值”的——比如天气预报值、计划停机时段哪些是“未知未来值”的——比如负载功率本身就是预测目标。我的处理策略是把目标变量和外部特征拼成一个二维数组时间步, 特征数作为输入滑窗切分时输出窗口只保留目标变量列。模型输入整段历史的所有特征输出只有未来H步的目标变量。这样Transformer可以用注意力机制自行决定历史哪个时刻的外部特征对预测最有帮助。如果外部特征在未来时段不可知一个尽量简洁的做法是未来时段用零填充或重复最近值让模型学会对这些特征降权。复杂做法是把输入分成“历史全特征”和“未来已知特征”分别编码再融合这个在代码包里通常没有现成实现不做第一步尝试。4. 用PyTorch手写Transformer预测模型编码输出与训练循环4.1 位置编码与多头注意力的代码实现先写出位置编码和TransformerEncoder的基础组件。虽然PyTorch的nn.TransformerEncoder是现成的但位置编码需要自己做。以可学习位置编码为例import torch import torch.nn as nn import math class LearnablePositionalEncoding(nn.Module): 可学习位置编码为每个时间步分配一个可训练的位置向量 def __init__(self, d_model, max_len2000): super().__init__() # 位置embedding矩阵形状 (max_len, d_model) self.position_emb nn.Parameter(torch.randn(max_len, d_model) * 0.1) def forward(self, x): # x: (batch_size, seq_len, d_model) seq_len x.size(1) # 截取当前序列长度对应的位置向量并叠加 return x self.position_emb[:seq_len, :].unsqueeze(0)这里把位置向量初始化为均值为0、标准差0.1的正态分布而不是全零。如果初始化全零前几个训练步模型很难打破对称性收敛变慢。参数max_len对应能处理的最长序列输入长度超过它就会索引越界所以代码包里的fff MAX_LEN建议设置成你在预处理脚本里input_len的1.5倍以上。多头注意力不用自己写nn.MultiheadAttention或nn.TransformerEncoderLayer内部已经做了Q/K/V投影和head拼接。但你要理解两个参数的作用nhead是注意力头数d_model必须能被nhead整除dropout默认0.1在数据量少时建议降到0.05或0否则训练损失降不下去。4.2 编码器堆叠与全连接预测头把序列映射到未来窗口有了位置编码核心的预测模块组装如下。这里选择纯编码器结构把TransformerEncoder最后一层的输出取出来先做一个flatten再接全连接层输出直接是pred_len个值。注意flatten之前要把特征维度压缩掉否则全连接层参数量爆炸。class TransformerForecaster(nn.Module): def __init__(self, d_model64, nhead4, num_layers3, input_len168, pred_len24, num_features3, dropout0.05): super().__init__() # 输入特征维度映射到d_model self.input_proj nn.Linear(num_features, d_model) self.pos_enc LearnablePositionalEncoding(d_model, max_leninput_len 50) # 编码器层batch_firstTrue方便输入格式为 (batch, seq, feature) encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, dim_feedforwardd_model * 4, dropoutdropout, batch_firstTrue, activationgelu ) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_layers) # 展平后接两层MLP输出预测窗口 self.head nn.Sequential( nn.Linear(input_len * d_model, 128), nn.GELU(), nn.Dropout(dropout), nn.Linear(128, pred_len) ) def forward(self, x): # x: (batch_size, input_len, num_features) x self.input_proj(x) # (batch, input_len, d_model) x self.pos_enc(x) # 叠加位置编码 x self.encoder(x) # (batch, input_len, d_model) # 取全部时间步输出而不是最后一个时间步 x x.reshape(x.size(0), -1) # (batch, input_len * d_model) return self.head(x) # (batch, pred_len)这段代码的关键决策有三个。第一取全部时间步输出做flatten而不是只取最后一步因为长期预测中未来某个点可能依赖历史任意位置的信息全量保留能让注意力机制充分发挥代价是input_len*d_model的中间向量维度可能很大所以我把预测头设计成先降到128再输出。第二激活函数用了GELU而不是ReLU它在负半轴不是硬截断梯度更平滑对回归任务更友好。第三dim_feedforward设置成d_model的4倍这是Transformer论文里的默认比例太小会让前馈网络表达能力不足太大会拖慢训练。如果你拿到代码包发现预测头只有一层线性层也正常响应式工业预测里有人为了省显存这么干。但如果你的训练集样本量不大两层MLP反而容易过拟合可以用dropout控制。4.3 训练循环与早停Loss曲线怎么判读训练部分的核心是损失函数。长期预测是回归任务均方误差MSE是最常用的选择它对大误差点的惩罚更重能迫使模型把尖峰也拟合到。如果你的业务更关注趋势方向判对没判对可以在MSE基础上加一个方向惩罚项但先从纯MSE开始不要一上来就设计复杂Loss。训练循环里有一个参数对Transformer特别敏感学习率。Transformer对学习率非常挑剔太大直接梯度爆炸太小收敛极慢很多时候表现为“怎么训练Loss都不降”。我习惯先跑一个3到5个epoch的烟雾测试观察一下Loss曲线是下降还是震荡import torch import torch.nn as nn from torch.optim import AdamW # 初始化模型、优化器 model TransformerForecaster(d_model64, nhead4, num_layers3, input_len168, pred_len24, num_featurestrain_loader.dataset.X.shape[2]) optimizer AdamW(model.parameters(), lr1e-3, weight_decay1e-5) criterion nn.MSELoss() scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max30) best_val_loss float(inf) patience 8 counter 0 for epoch in range(60): model.train() train_loss_sum 0.0 for batch_x, batch_y in train_loader: pred model(batch_x) loss criterion(pred, batch_y) optimizer.zero_grad() loss.backward() # 梯度裁剪Transformer训练的关键稳定器 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() train_loss_sum loss.item() * batch_x.size(0) scheduler.step() # 验证集评估 model.eval() val_loss 0.0 with torch.no_grad(): for val_x, val_y in val_loader: val_pred model(val_x) val_loss criterion(val_pred, val_y).item() * val_x.size(0) val_loss / len(val_loader.dataset) train_loss train_loss_sum / len(train_loader.dataset) print(fEpoch {epoch1}: train_loss{train_loss:.6f}, val_loss{val_loss:.6f}) # 早停验证集Loss连续patience轮不下降就停止 if val_loss best_val_loss: best_val_loss val_loss counter 0 torch.save(model.state_dict(), best_model.pth) else: counter 1 if counter patience: print(fEarly stop at epoch {epoch1}) break几个参数的选值逻辑lr1e-3是d_model64时的安全起点d_model越大lr应适当降低d_model128时建议1e-4weight_decay1e-5是给全连接权重做轻微约束防止过拟合梯度裁剪max_norm1.0解决注意力机制偶尔产生大梯度的问题——如果没有这行训练到某个epoch Loss突然变成nan是常事早停的patience设8轮配合CosineAnnealing先快后慢的学习率衰减基本能保证在20-30个epoch内收敛。5. 长期预测的常见问题排查现象、原因、解决5.1 预测结果是一条直线滞后效应的三个来源现象验证集上预测值和真实值几乎平行像把前一段历史平移了H步误差看起来不大但完全没预测出峰值和拐点。原因分三种。第一数据里存在强自相关模型学到的最优策略就是“照抄近期值”这在损失为MSE时很常见因为均值比峰值出现的概率高得多预测均值损失最小。第二输入窗口太长而训练数据不足模型没有足够的样本学到真正的动态模式于是退化成平滑策略。第三Loss函数对峰值不敏感如果数据分布极不均衡尖峰带来的MSE增量在整个训练集上占比太小。解决顺序先看数据的自相关系数如果lag1的自相关超过0.9先做一阶差分或季节差分把趋势和周期移除后再训练其次是缩短输入窗口比如把168个点缩到48个点观察验证集是否出现波动最后可以给损失函数加权重真实值偏离均值越大权重越高迫使模型关注峰值。5.2 训练Loss下降但验证集预测全盘崩溃现象训练Loss稳定降到0.01以下画出预测图却完全对不上有时还出现数值爆炸的毛刺。原因数据泄露的一种隐蔽形式——滑窗切分时没有先分离标准化导致均值方差泄露是一种另一种是滑窗步长太小时训练集和验证集样本高度重叠。比如stride1时相邻两个训练样本的输入窗口只差1个时间点验证集里如果混入了和训练样本重叠的区间验证Loss表面好看切换到一个完全没见过的连续时间段预测就崩。解决检查数据划分代码按时间顺序切割后再做滑窗两个集合的原始时间区间必须完全不重叠同时把滑窗stride调大训练集内部样本重叠度高本身没问题但验证集必须每隔H个步长采一次确保验证样本之间也没有时间重叠。5.3 收敛很慢或Loss值一直震荡不降现象训练前几个epoch Loss就在一个区间反复横跳像是随机猜测或者Loss在某个值附近长期横盘怎么都降不下去。原因学习率设置不当要么太大导致震荡要么太小导致长期横盘另一个常见原因是归一化没有生效——模型输入的数值范围还是几千几万Transformer里的注意力点积直接溢出softmax饱和梯度消失。解决先打印模型输入的均值、方差确认StandardScaler真正生效了然后把学习率按数量级递减搜索1e-3不行就试1e-41e-4不行就试3e-5最后检查d_model和输入数据特征维度是否匹配——如果d_model太小比如8注意力头之间的信息交互容量不够也会表现为难收敛把d_model调到32或64重试。5.4 反归一化后结果超出物理范围现象模型预测值的分布和真实值完全一致但某个时间点的预测突然冲到超过正常范围几十倍比如温度预测出500度。原因模型输出的是归一化空间里的值反归一化时把scaler的inverse_transform传入了错误形状张量或者预测值里有离群值而标准化的逆变换会把离群值放大回原始量纲。解决反归一化前先对预测值做一次裁剪把归一化空间里的值限制在训练集最小/最大值附近比如训练集标准化后范围约[-3, 3]就把预测结果clip到[-4, 4]再反归一化。这个技巧不优雅但有效对付偶发数值溢出很实用。真正的原因通常还是训练过程中某个batch产生了异常大的激活值配合梯度裁剪和更小的学习率能根治。6. 可视化结果与进阶验证一张图判断模型是否真的学到了趋势可视化不是最后画个曲线图就完事。我拿到任何代码包第一件事是跑完训练后画三张图训练集上的预测vs真实、验证集上的预测vs真实、测试集上连续时间段的多窗口预测。只有第三张图能说明模型在真实场景下可用。测试集的可视化要这样画从测试集起点开始以滑窗方式向前滚动但每次都用真实历史值做输入预测未来H步然后把预测值和真实值画在同一条时间轴上。这样能看出模型在多个连续窗口下是否稳定——如果窗口1预测出的趋势和窗口2预测的趋势出现明显跳变说明模型捕捉到的是噪声而非规律。import matplotlib.pyplot as plt # 假设test_loader中每条样本都是 (输入窗口, 真实未来窗口) # 从测试集起点开始连续预测5个窗口 model.load_state_dict(torch.load(best_model.pth)) model.eval() future_horizon 24 predictions [] actuals [] past_window None # 用真实历史滚动预测模拟真实在线预测的输入获取方式 inputs_list, targets_list test_loader.dataset.X, test_loader.dataset.Y for i in range(0, min(5, len(inputs_list)), 1): x torch.FloatTensor(inputs_list[i]).unsqueeze(0) y_true targets_list[i] with torch.no_grad(): y_pred model(x).squeeze(0).numpy() predictions.append(y_pred) actuals.append(y_true) predictions np.array(predictions) actuals np.array(actuals) # 绘制时间轴把所有预测窗口按顺序拼接 plt.figure(figsize(14, 4)) for i in range(len(predictions)): # 每个窗口内的x轴位置是窗口内部的相对时间 x_axis range(i * future_horizon, (i 1) * future_horizon) plt.plot(x_axis, actuals[i], b-, linewidth1.2, labelActual if i 0 else ) plt.plot(x_axis, predictions[i], r--, linewidth1.0, labelPredicted if i 0 else ) plt.legend() plt.xlabel(Time steps) plt.ylabel(Scaled value) plt.grid(alpha0.3) plt.savefig(forecast_visualization.png, dpi150, bbox_inchestight)这段代码的核心逻辑是逐窗口展开预测值然后再拼接而不是把每个窗口画成重叠的对比图。重叠图只能看出单窗口误差展开放置能看出误差是否随窗口推进而累积。如果连续5个窗口的预测线都在真实线附近小幅波动且没有出现越来越大的偏移这个模型才算达到部署标准。进阶方向我建议有余力时尝试两块。一是概率预测让模型输出分布的均值和方差而不是单点值对业务决策更有价值二是PatchTST这类模型先把序列切成patch再进Transformer把输入长度缩小一个数量级训练速度和稳定性都更好。这套Transformer代码就是理解那些进阶模型的基础。做这类项目的习惯我一直保留跑通第一版后先固定模型结构调数据预处理再固定数据调结构一次只动一个变量。别看Transformer很强大它同样需要你给它干净的输入。希望这篇文章帮你少踩几个坑把长期预测这条链路真正跑起来。本文还有配套的精品资源点击获取