简介:Transformer时间序列预测项目面向具备PyTorch基础的数据分析者、算法工程师及相关专业学生,针对长时间序列预测难、依赖捕捉弱的问题,利用自注意力机制建模全局依赖,配合位置编码确保顺序信息,训练过程较RNN/LSTM更高效,且可调整参数适应不同复杂度的业务数据。压缩包共38个文件,以13个Python源码为骨架,覆盖Transformer核心模块、数据加载、模型训练、评估与可视化等环节;另含3个电力负荷CSV数据集、1个训练好的权重文件、预测结果对比图片及环境依赖清单,整包约26.48MB。项目已有984人学习,适合复现长期预测实验、对比传统序列模型效果,也可作为课堂项目或科研预研的基线代码。资源提供可直接执行的入口脚本、保存好的模型状态与可视化输出,并附原理说明,能一键启动并查看预测曲线。代码按模型、工具、数据等模块组织,边界清晰,便于定向修改嵌入维度、多头数量、层数等超参数,快速迁移到气象、交通或金融等其他时序数据集中,实现定制化训练。
1. 用Transformer做长期预测,先别急着上模型
花了一个周末,用Transformer跑电力负荷的长期预测:输入过去168个小时的序列,一次输出未来96个小时的结果。Loss降得很快,验证集MSE也算体面,可把曲线画出来才发现,预测从第20步开始就不动脑子了——直接回归到均值线,后面一大段全是平的。后来换数据、改位置编码、调整评估方式,才把这套流程跑明白。这篇要写的是Transformer做长期预测的完整落地路径:最小可跑代码、数据集怎么挑、训练和可视化怎么配合,以及我亲测踩过的五个坑。适合已经会PyTorch基础、想拿Transformer做多步预测、又不想被论文里花哨结构带偏的工程师和算法同学。
2. 手写Transformer做长期预测:位置编码、多头注意力与参数量
2.1 时间序列里的Q、K、V:Transformer架构到底在看什么
先把Transformer架构和长期预测的关系说清楚。时间序列的每个时间步,会被当成一个token,原始的特征维度映射成d_model维向量。Self-Attention在时间维上做计算:每个时间步通过Q(Query)去和其他时间步的K(Key)做相似度打分,再用softmax归一化成权重,最后对V(Value)做加权汇总。这个过程本质上是在回答一个问题:预测未来某个点的时候,过去哪个时间步更应该被参考。
对长期预测来说,注意力机制的价值在于它不像RNN那样只能按顺序传递信息,而是可以直接把第1步的信息传给第100步。比如电力负荷数据里,今天的负荷和一周前同一时刻的负荷高度相关,自注意力可以直接建立这种跨步依赖,不用靠隐状态慢慢递推。也正因如此,Informer、Autoformer这些改进模型才会从“怎么让注意力更高效”或“怎么让注意力更适合序列分解”入手。
基础Transformer做长期预测的常见结构是encoder-only加回归头:输入past_len个时间步,编码后直接输出pred_len个未来时间步,而不是像机器翻译那样用decoder逐个生成。原因有两个:第一,时间序列预测的目标是连续数值,不是离散token;第二,多步预测一次输出,训练目标和推理目标一致,误差不会在步与步之间滚雪球。我在实际项目中用的一直是这个结构,省心且容易调参。
2.2 位置编码:sin/cos编码为什么经常翻车
Transformer本身没有顺序概念,所以需要位置编码。原版用的正弦位置编码,给每个时间步注入一个绝对位置向量。这个编码在文本里表现不错,但在时间序列上并不稳定,我甚至见过去掉位置编码反而涨点的情况。
原因在于:时间序列里真正重要的不是“第几个时间步”,而是“这个时间步对应什么时刻”。同样是第100个时间步,如果是凌晨2点,电力负荷处在低谷;如果是晚上8点,负荷处在高峰。正弦位置编码表达不了“凌晨2点”这个语义。常见的做法有两种:一是把外部时间特征作为额外输入拼到序列特征里,比如小时、星期几、是否节假日,其中小时和星期几可以编码成sin/cos或one-hot;二是干脆不加位置编码,让注意力自己通过数据学顺序。我在做小时级数据时一般会把小时和星期几拼进去,效果比纯正弦位置编码稳定得多。
位置编码的另一个坑是训练和推理长度不一致。正弦编码理论上可外推,但注意力层没有见过更长的序列,照样会漂移。所以要么固定输入窗口长度不变,要么在训练时做随机截断,让模型见过多种长度。
2.3 一个最小Transformer预测器代码与参数量计算
下面是一个能直接跑的最小Transformer长期预测模型,输入形状是(batch, past_len, input_dim),输出形状是(batch, pred_len),默认预测单变量。
import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=2048): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp( torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model) ) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) self.register_buffer('pe', pe.unsqueeze(0)) def forward(self, x): # x: (batch, seq_len, d_model) return x + self.pe[:, :x.size(1), :] class TransformerForecaster(nn.Module): def __init__(self, input_dim, d_model=64, nhead=4, num_layers=2, pred_len=96): super().__init__() self.embed = nn.Conv1d(input_dim, d_model, kernel_size=1) self.pos = PositionalEncoding(d_model) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, batch_first=True, dropout=0.1 ) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.head = nn.Linear(d_model, pred_len) def forward(self, x): # x: (batch, past_len, input_dim) x = x.permute(0, 2, 1) # (batch, input_dim, past_len) x = self.embed(x) # 逐点映射到d_model x = x.permute(0, 2, 1) # (batch, past_len, d_model) x = self.pos(x) x = self.encoder(x) x = x.mean(dim=1) # 全局平均池化 out = self.head(x) # (batch, pred_len) return out这个代码里,embed用的是Conv1d加kernel_size=1,等价于对每个时间步做了一次线性映射,把原始特征维转成d_model维。pos加的是正弦位置编码,如果去掉它,直接把x = self.pos(x)这行注释掉即可。encoder输出的是每个时间步的隐状态,这里没有只取最后一个时间步,而是做了全局平均池化。原因是长期预测里,最后一步的信息量不足以支撑未来96步的输出,平均池化能让模型把整个历史窗口的信息都用上。head是一个线性层,把池化后的向量映射成pred_len个未来时间步。
参数设置上,d_model=64、nhead=4意味着每个注意力头分到16维,能满足64能被4整除这个硬约束。num_layers=2适合中等规模数据,如果样本量只有几千,层数再深就容易过拟合。pred_len=96表示输出未来96个点,对应小时级数据就是未来4天。
参数量可以大概估算:每个TransformerEncoderLayer里,QKV三个投影加输出投影约4 * d_model^2,FFN部分约2 * d_model * ffn_dim,其中ffn_dim默认是2048。以d_model=64、一层为例,注意力投影约16384个参数,FFN约262144个参数,LayerNorm约128个参数,单层合计约27万参数。两层就是55万左右,大头在FFN。所以如果数据量不大,建议把TransformerEncoderLayer里的dim_feedforward从2048调到512或256,能显著降低过拟合风险,训练速度也会快很多。
3. 数据准备与滑窗样本:把CSV变成Transformer能吃的样本
3.1 选数据集:连续时间序列长什么样才适合长期预测
不是所有CSV都适合做长期预测。Transformer擅长捕捉长距离依赖,但如果数据本身没有周期性、没有趋势,或者采样频率乱跳,模型再复杂也白搭。我一般优先选公开数据集里采样稳定、周期清晰的序列来跑通流程。
长期预测领域常用的是ETTh1和Exchange Rate这一类学术公开数据集。ETTh1是电力变压器油温数据,7个变量,每小时采样一次,数据里明确包含日周期和周周期,非常适合检验模型能不能跨96步甚至168步预测。Exchange Rate是多个国家货币汇率的日频数据,频率低,趋势性强,适合看模型对慢变量的预测能力。这两个数据集不用自己去爬,学术公开数据页面搜索名称就能找到,格式基本都是CSV,第一列是时间戳,后面是数值特征。
拿到数据先做三件事:统一时间索引、处理缺失值、确认无重复采样点。时间索引最好用pd.to_datetime转成标准格式,然后set_index。缺失值数量少就用前后线性插值;如果缺失段太长,直接删掉那一段,别硬补,否则模型会学到假的平滑过渡。
数据集划分上,时间序列不能用随机拆分,必须按时间顺序切。我一般按7:1:2切训练集、验证集、测试集,保证测试集在时间上严格晚于训练集。这样模型没见过的未来片段才能真实反映泛化能力。
3.2 滑窗构造样本:past_len、pred_len和batch
模型输入是一段连续历史窗口,输出是对应未来窗口,所以要把长序列切成很多个“过去-未来”样本对。滑窗是这里的核心操作。
import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler # 读取数据,df为多变量时间序列,index为时间 df = pd.read_csv('ETTh1.csv', index_col=0) data = df.values.astype(np.float32) # 按时间顺序划分 train_size = int(len(data) * 0.7) val_size = int(len(data) * 0.1) train_data = data[:train_size] val_data = data[train_size:train_size + val_size] test_data = data[train_size + val_size:] # 只对训练集fit归一化 scaler = StandardScaler() scaled_train = scaler.fit_transform(train_data) def sliding_window(data_array, past_len, pred_len, step=1): X, Y = [], [] for i in range(0, len(data_array) - past_len - pred_len + 1, step): X.append(data_array[i: i + past_len]) Y.append(data_array[i + past_len: i + past_len + pred_len]) return np.array(X), np.array(Y) X_train, Y_train = sliding_window(scaled_train, past_len=168, pred_len=96) print(X_train.shape, Y_train.shape)这段代码里,past_len=168是输入窗口长度,pred_len=96是预测窗口长度,step=1表示每次往后滑动一个采样点生成一个新样本。小时级数据里,168是7天,正好覆盖一个完整的周周期;96是4天。窗口长度至少要比最大业务周期长一点,不然模型永远看不见“上周同期”这种关键模式。
step=1会生成大量重叠样本,训练效果好但内存占用大。如果内存吃紧,可以改成step=pred_len,让样本之间完全不重叠。采样频率高的数据用step=1问题不大,日频数据建议直接用step=1,因为样本总量本来就少。还要注意sliding_window函数里的循环在大数据量下比较慢,几百万行序列建议改用np.lib.stride_tricks.sliding_window_view,思路一样,内存换速度。
3.3 归一化与逆归一化:统计量只允许从训练集学
归一化是很容易翻车的一步。很多人在验证集或测试集上直接调用fit_transform,等于让模型提前看见了未来的均值和方差,这属于典型的数据泄露。验证集和测试集只能调用transform,使用训练集统计好的参数。
# 验证集和测试集只能用训练集的scaler scaled_val = scaler.transform(val_data) scaled_test = scaler.transform(test_data) X_val, Y_val = sliding_window(scaled_val, past_len=168, pred_len=96) X_test, Y_test = sliding_window(scaled_test, past_len=168, pred_len=96)为什么要强调这个细节?长期预测的评估阶段要把预测结果还原成原始量纲,还原用的就是scaler.inverse_transform。如果测试集用了自己的均值和方差去标准化,还原出来的数值整体会偏移,画图和算误差的时候看着不对,但MSE可能还很漂亮。我自己踩过这个坑,最后发现是归一化统计量用错了地方,重新跑一遍之后结果完全不同。
另外,如果数据里有明显趋势,比如某个变量逐年上涨,纯StandardScaler是不够的。可以先做一阶差分再归一化,或者用训练集整段做一次趋势拟合,把趋势项先减掉。Transformer对非平稳输入很敏感,输入分布稍微漂移,注意力权重就会乱。
4. 训练循环、评估指标与预测结果可视化
4.1 训练循环:Loss、梯度裁剪和模型保存
训练长期预测模型和训练分类模型的套路类似,但有两个地方要特别注意:Loss量级大容易梯度爆炸,以及必须按验证集表现保存模型,而不是看最后一个epoch。
import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset model = TransformerForecaster( input_dim=X_train.shape[-1], d_model=64, nhead=4, num_layers=2, pred_len=96 ) criterion = nn.MSELoss() optimizer = optim.Adam(model.parameters(), lr=1e-3) scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=20, gamma=0.5) train_dataset = TensorDataset( torch.from_numpy(X_train), torch.from_numpy(Y_train) ) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) X_val_tensor = torch.from_numpy(X_val) Y_val_tensor = torch.from_numpy(Y_val) best_val_loss = float('inf') for epoch in range(80): model.train() train_loss = 0.0 for xb, yb in train_loader: optimizer.zero_grad() pred = model(xb) loss = criterion(pred, yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() train_loss += loss.item() * xb.size(0) train_loss /= len(train_dataset) model.eval() with torch.no_grad(): val_pred = model(X_val_tensor) val_loss = criterion(val_pred, Y_val_tensor).item() scheduler.step() if val_loss < best_val_loss: best_val_loss = val_loss torch.save(model.state_dict(), 'best_model.pt') if (epoch + 1) % 10 == 0: print(f'epoch {epoch+1:03d} | train {train_loss:.4f} | val {val_loss:.4f}')这里Loss直接对pred和yb算MSE,pred形状是(batch, 96),yb形状也是(batch, 96),PyTorch自动按元素求平均。多步预测的Loss是对未来96个点求平均,所以单看这个数值很难知道模型是在前10步准还是全程都准,后面可视化章节会专门讲这个问题。
梯度裁剪阈值设成1.0,是因为长期预测的MSE数值通常比分类任务的CrossEntropy大好几个量级,早期阶段反向传播的梯度容易把权重推出正常范围。如果训练过程中Loss变成NaN,优先把学习率降到1e-4,看看是不是输入里有NaN或无穷大。StepLR每20个epoch把学习率减半,算是一个稳定的基础节奏。
4.2 评估指标:MSE/MAE之外,长期预测要看分步误差
测试的时候,很多人只看一个总MSE或者总MAE就下结论,这在长期预测里会骗人。一个模型可能前10步预测得非常好,后面86步全是均值线,但总MSE看起来依然不错。所以我会把分步误差拆出来看。
def step_wise_rmse(pred, true): # pred和true形状一致:(N, pred_len) return np.sqrt(np.mean((pred - true) ** 2, axis=0)) model.eval() with torch.no_grad(): pred_test = model(torch.from_numpy(X_test)).numpy() true_test = Y_test rmse_per_step = step_wise_rmse(pred_test, true_test) # 输出前10步和后10步的对比 print(rmse_per_step[:10].mean(), rmse_per_step[-10:].mean())这段代码算出每个预测步长上的RMSE,然后对比前10步和后10步。正常模型的分步误差会随着预测步数增加缓慢上升;如果是前10步均值远小于后10步均值,说明模型只学会了短期依赖。如果所有步长的RMSE都很接近,且数值很低,才说明Transformer真的学到了长程模式。
评估方式上,基础Transformer做长期预测一般用direct策略,也就是一次输出全部pred_len个点,不需要递归。递归预测是把模型输出当成下一次输入,每步误差都会累积,到第96步预测值很可能已经漂到完全不合理的量级。direct策略虽然一步到位更难学,但至少训练和推理目标一致。
4.3 可视化:真实值、预测值和误差带一张图看清
画图不是为了好看,是为了让模型翻车方式暴露出来。数据可视化这块,核心不只是把两条线画出来,而是要看误差带的变化趋势。
import matplotlib.pyplot as plt def restore_series(arr, scaler, target_idx=0): # arr: (N, pred_len),是模型预测的单变量结果 dummy = np.zeros((arr.shape[0], arr.shape[1], scaler.n_features_in_)) dummy[:, :, target_idx] = arr flat = dummy.reshape(-1, scaler.n_features_in_) back = scaler.inverse_transform(flat).reshape(dummy.shape) return back[:, :, target_idx] # 还原成原始量纲 pred_restored = restore_series(pred_test, scaler, target_idx=0) true_restored = restore_series(true_test, scaler, target_idx=0) sample_idx = 0 pred_len = pred_restored.shape[1] fig, ax = plt.subplots(figsize=(12, 4)) ax.plot(range(pred_len), true_restored[sample_idx], label='actual', linewidth=2) ax.plot(range(pred_len), pred_restored[sample_idx], label='predicted', linestyle='--') err = np.abs(pred_restored[sample_idx] - true_restored[sample_idx]) ax.fill_between( range(pred_len), pred_restored[sample_idx] - err, pred_restored[sample_idx] + err, alpha=0.3, color='orange', label='abs error' ) ax.set_xlabel('forecast horizon (steps)') ax.set_ylabel('value') ax.legend()restore_series这个函数比较绕,但必须这么做。因为scaler是在全部变量上训练的,inverse_transform要求输入变量数一致,不能直接把(N, 96)的目标列塞进去。函数先把待还原的列放到一个dummy数组的目标位置,其他列补零,还原后再把目标列抠出来。如果直接对单列做inverse_transform,会报维度错误或者还原出错误量纲。
可视化之后要重点观察三件事:预测曲线是否整体滞后;后段是否变成直线;误差带是否在后段明显张开。这三条分别对应三种模型病态,后面避坑章节会逐个展开。
5. 长期预测常见的翻车现场:五个坑的排障记录
5.1 预测变成一条水平线
现象:训练Loss正常下降,验证集MSE也不算差,但把预测曲线画出来,从第30步开始就是一条几乎水平的直线,数值接近训练集均值。
原因:MSE作为损失函数时,模型的最优策略是输出条件均值。如果序列里的可预测成分较弱,或者输入窗口信息不足,Transformer会学会“预测一个保守的中间值”来降低整体MSE。长期预测步数越多,后期不确定性越大,均值回归现象越明显。
解决:先看分步误差,确认是后期才水平还是从一开始就水平。如果是后期水平,说明模型有短程记忆但没有长程结构,把past_len加长、把时间特征拼进输入会有效果。如果从一开始就水平,说明数据本身可预测性弱或模型欠拟合,需要增加层数或换用带序列分解的模型思路。还可以对预测目标做差分,让模型去预测未来变化量,而不是原始值,能减轻回归均值问题。
5.2 曲线整体滞后一拍但MSE不高
现象:预测曲线和真实曲线形状几乎一样,但整体向右平移了一个采样点。总MSE看着挺低,画图时一眼就能发现线条对不上。
原因:这是典型的“短时记忆”模式。模型发现前一个时刻的值和当前时刻的值强相关,于是学会把上一个时刻的观测值直接复制到下一步,成本最低。这在自回归任务里非常常见,尤其是输入和输出高度平滑的数据。
解决:检查是不是数据泄露或预处理顺序错误,比如滑窗时不小心把Y包含了X的最后一步。排除数据问题后,在评估指标里加入对滞后敏感的一阶差分误差,或者画出分步误差曲线看第1步误差是否远小于第2步。我一般会要求模型第1步误差不能比第2步小太多,否则就认为模型在偷懒复制。
5.3 短预测正常,长预测直接散掉
现象:pred_len=24时预测曲线跟着真实值走,换成pred_len=96后,后段预测值直接从量级上飞掉,甚至出现比训练集最大值还大的异常数值。
原因:预测长度超出模型有效记忆范围。Transformer的注意力理论上可以看任意远的距离,但实际训练时,数据里的长距离相关性可能并不强,模型学不到能支撑96步输出的有效特征。另外,如果head直接输出96个点,相当于让一个线性层承担所有长程映射,输出方差容易失控。
解决:不要把一次性输出长度定得太长。常见做法是分块预测:把96步拆成4段24步,每段由一个独立输出头预测,段与段之间用上一段的预测结果做条件输入。另一种办法是把输出头改成低频函数形式,比如让模型预测傅里叶系数或一组基函数权重,再还原成时序曲线,能天然抑制后段发散。
5.4 训练和推理时窗口长度不一致,输出乱飞
现象:训练时输入past_len=168,推理时为了省计算只喂最后24个点,预测结果完全不靠谱,甚至出现NaN。
原因:Transformer的输入分布和位置编码都和序列长度绑在一起。长度一变,Embedding和注意力层看到的特征分布和训练时不一致,输出自然漂移。
解决:推理侧严格使用训练时相同的past_len。如果只拿到最后24个实时数据,可以回填历史数据补足168个点。进阶做法是在训练时对输入窗口做随机截断,past_len在一定范围内随机取样,让模型对不同长度都鲁棒。但这个做法会增加训练时间,小数据量下不如直接固定长度。
5.5 逆归一化之后,整体偏小一截
现象:模型输出的标准化预测值看着没问题,用inverse_transform还原后,所有预测值都比真实值小一个固定比例或固定偏移。
原因:几乎都是归一化统计量用错了。要么验证集和测试集重新fit_transform了,要么inverse_transform时变量列数和训练时不一致。还有一种情况是模型预测的是标准化后的目标,但目标在训练前做过差分或其它变换,还原时漏掉了逆变换步骤。
解决:写代码时强制规定,全流程里scaler.fit只出现一次,且必须作用在训练集上;验证集和测试集只允许transform。逆还原时,如果scaler.n_features_in_大于1,用占位数组补全列数。这两条写成代码注释,能避免大多数低级的量纲错误。
6. 想让Transformer预测更可信:滚动评估和多seed对比
6.1 滚动窗口评估:误差随预测步数的曲线才是重点
上面第4章画的是某一个测试样本的分步曲线,这一步把它扩展成滚动评估。做法是把测试集切成多个起点,每次都用模型预测一段未来,然后计算每个预测步长上的平均误差,最后画一条误差随步数变化的曲线。相比只看一个样本,滚动评估覆盖了更多预测起点,能看出模型在不同时间段的稳定性。
def rolling_forecast(model, test_data, past_len, pred_len, step=24): preds = [] trues = [] for start in range(0, len(test_data) - past_len - pred_len, step): x = test_data[start: start + past_len] y = test_data[start + past_len: start + past_len + pred_len] with torch.no_grad(): pred = model(torch.from_numpy(x).unsqueeze(0)).squeeze(0).numpy() preds.append(pred) trues.append(y) return np.array(preds), np.array(trues)这个函数在测试集上每隔step个点做一次预测,返回所有预测块和真实块。之后把preds和trues代入step_wise_rmse,就能得到一条分步误差曲线。如果曲线在第10步就快速上翘,说明这个模型的可靠预测范围只有前10步。我现在的习惯是:总MSE只看一眼,真正决定模型能不能上线的是这条分步误差曲线。
6.2 多seed对比:什么时候果断放弃Transformer
Transformer不是长期预测的唯一答案,甚至不总是好答案。我会固定训练集和测试集,用同一个模型跑三个不同随机种子,取MSE中位数作为最终结果。同时用一个简单的线性基线做同样评估,比如DLinear或直接对历史窗口做线性映射。如果Transformer的MSE只比线性模型好不到5%,我会上线性模型,因为调参成本低、部署简单、预测更稳。这套判断标准帮我筛掉过很多看似高大上但实际不挣钱的方案。
长期预测这件事,模型结构只是起点,数据、窗口、评估和可视化才是真正决定上线效果的部分。现在我每跑完一组实验,第一件事不是盯着总MSE,而是把分步误差曲线和预测图画出来,再决定要不要继续调。这个习惯让我少走了很多弯路,也希望帮到你。
本文还有配套的精品资源,点击获取