尧图网页设计 
简介时间序列预测是数据科学和工业场景中的核心任务传统模型如LSTM虽能捕捉一定时序依赖但面临长序列训练效率低、记忆容量有限等瓶颈。Transformer通过自注意力机制实现任意位置间的直接交互支持并行计算显著提升长序列建模能力。在实际应用中将Transformer应用于时序预测需关注数据构造、位置编码、归一化等关键环节避免信息泄漏与训练不稳定。滑动窗口生成样本、基于训练集统计的归一化、以及Encoder-only结构设计是搭建高效预测系统的基础。本文基于PyTorch给出从数据预处理到模型训练评估的完整代码以ETTh2电力数据集为例演示如何构建可复现的时间序列预测基线并总结了常见问题与调优方向帮助读者快速落地Transformer时序预测项目。 这段时间我收到不少和Transformer时间序列预测有关的私信。很多人从NLP那边转过来看到Transformer的论文就想拿它跑时序结果一动手就卡住数据格式怎么构造位置编码加在哪为什么别人代码跑出来指标那么低自己一跑就是nan我自己也完整走过一遍这条路所以把一套能直接跑的Pytorch完整源码和数据整理出来了。这份代码不是竞赛刷分版本而是把标准的Encoder-only Transformer从头到尾接起来你换成自己的CSV也能用项目里有数据加载、滑动窗口切分、归一化、模型定义、训练循环、评估指标和断点保存。如果你正准备从LSTM迁移到Transformer或者只是想快速拿到一份能复现的时间序列预测基线这份记录应该能帮你省下不少时间。我会先把模型原理讲清楚再贴完整代码最后把我实测中踩过的坑和几个改造方向一起列出来避免你在同一个地方再翻车。1. 为什么把Transformer搬进时间序列预测而不是继续用LSTM1.1 LSTM处理长序列时的三个老问题LSTM在时间序列预测里的地位不用多说很多做工业预测的老代码都是LSTM打底。它靠门控机制把信息通过隐状态一步一步往后传理论上能记住长期依赖但实际用下来有几个绕不开的痛。第一个问题是顺序计算。LSTM必须按时间步依次计算t时刻的隐状态要等t-1算完训练时没法在序列维度上并行。序列一长训练速度就慢。第二个问题是记忆容量。门控能缓解梯度消失但隐状态向量容量有限几百步之前的信息经过多轮门控之后已经衰减得很厉害。第三个问题是梯度传播路径太长如果梯度裁剪和初始化没做好训练很容易不稳定。我在项目里用LSTM做对比时同样预测未来24个点LSTM在一个epoch上的训练时间大概是Transformer的1.5倍多而且序列越长这个差距越明显。这不是说LSTM不行而是它天生是顺序建模处理长序列的代价就摆在那。1.2 自注意力机制到底改了什么Transformer的核心是自注意力。每一个时间步都可以直接和序列里所有其他时间步计算相关性不再依赖上一个时间步把信息“接力”传过来。这样的好处是任意两个位置之间的交互路径长度为1长距离依赖问题从机制上被绕开了。同时自注意力在计算上是可以并行的。所有位置的Query、Key、Value可以一次算出来再用矩阵乘法得到注意力权重多个时间步互不依赖。这就是为什么Transformer在大规模数据上能训练得动LSTM却很难做超长序列。放在时间序列这个场景里自注意力还有个很直观的解释它等于自动帮你找出“历史上哪几个时刻和当前趋势最相关”。温度预测里可能是24小时前的同一个时段用电负荷预测里可能是昨天同一时刻加上未来几小时的气温特征。LSTM靠隐状态隐式记忆这些关系Transformer直接显式建模这种跨时间步的相关性。1.3 时序预测和NLP任务的差异为什么用Encoder-only很多新手一上来就套用NLP的Transformer seq2seq结构非得写一个Encoder加一个Decoder还配置teacher forcing训练复杂度一下上去了。但实际上标准时间序列预测大多不需要Decoder。NLP里的Transformer是生成式任务输入一句话要一个词一个词地生成目标句子。而常见的时间序列预测任务是回归式任务给你过去96个时间点的值让你预测未来24个时间点的值。这是从一段连续数值到另一段连续数值的映射完全可以由一个Encoder提取历史模式再接一个回归头一次性输出所有预测点。我这份代码就是Encoder-only结构。模型把历史窗口编码成一组向量取最后一个时间步的输出过几个全连接层直接输出未来24个值。这么做的好处是训练稳定、代码简单、不容易出现自回归误差累积。如果后面你想做更长周期的滚动预测再考虑引入Decoder不迟这个我在最后一部分会讲。2. 数据先行从ETTh2到滑动窗口样本的完整工序2.1 数据选择与项目目录项目里用的数据是ETT数据集中的ETTh2这是时序预测论文里非常常见的benchmark内容是电力变压器的一系列指标和油温粒度是1小时一条。你在网上搜ETDataset的公开仓库就能下载到CSV文件里包含date、HUFL、HULL、MUFL、MULL、LUFL、OT这些列。本项目中只用OT列做单变量预测。下载之后把文件放到项目data目录下整体结构长这样transformer_ts/ ├── data/ │ └── ETTh2.csv ├── dataset.py ├── model.py ├── train.py └── README.md环境方面Python 3.9以上PyTorch 2.0以上再加pandas、numpy、matplotlib就够了。PyTorch直接从官网根据自己的系统和CUDA版本生成安装命令不要用默认的旧版教程硬抄版本差异会带来一堆奇怪问题。2.2 滑动窗口如何把一维序列变成训练样本原始时间序列就是一个一维数组长度是T。要把序列变成监督学习格式最常用的办法是滑动窗口。举个例子seq_len96pred_len24步长step1那么从序列开头开始取0到96作为x96到120作为y然后往后移一步取1到97作为x97到121作为y一直滑到序列末尾。假设原始序列有1000个点经过这种方式切分后样本数量是1000 - 96 - 24 1 881个。这里有个细节需要注意窗口长度一定要覆盖至少一个周期。ETTh2是小时粒度一天24个点seq_len取96就是4天足够让模型看到完整的日周期。如果序列是7天周期那窗口至少取168会比较稳。2.3 归一化先fit训练集再transform验证集归一化是时序预测里最容易出问题的一步。很多刚入门的同学会先对整个数据求min和max把全部序列归一化之后再切分数据集然后发现验证集指标好到离谱。这不是模型厉害而是数据泄漏了验证集的极大值和极小值已经参与了训练前的统计计算模型提前“见过”了验证集的范围。正确做法是只用训练集部分计算min和max然后用训练集算出的参数去归一化训练集、验证集和测试集。预测完成之后再用同一组参数反归一化把结果还原到原始温度尺度。下面是我项目里的dataset.py包含归一化和滑动窗口数据集import numpy as np import pandas as pd import torch from torch.utils.data import Dataset def normalize(values, train_len): 只用训练集部分计算 min/max避免信息泄漏 train_part values[:train_len] min_val float(train_part.min()) max_val float(train_part.max()) normed (values - min_val) / (max_val - min_val 1e-8) return normed, min_val, max_val class TimeSeriesDataset(Dataset): def __init__(self, values, seq_len96, pred_len24, step1): self.values values self.seq_len seq_len self.pred_len pred_len self.step step self.indices range(0, len(values) - seq_len - pred_len 1, step) def __len__(self): return len(self.indices) def __getitem__(self, index): start self.indices[index] x self.values[start: start self.seq_len] y self.values[start self.seq_len: start self.seq_len self.pred_len] return torch.FloatTensor(x).unsqueeze(-1), torch.FloatTensor(y)调用方式如下df pd.read_csv(data/ETTh2.csv) values df[OT].values.astype(float) train_len int(len(values) * 0.7) normed, min_v, max_v normalize(values, train_len) train_ds TimeSeriesDataset(normed[:train_len], seq_len96, pred_len24)这里有一个值得注意的点验证集和测试集的起点不能直接从分割点开始而是要往前回退seq_len个点。比如从train_len开始构造验证集那么验证集第一个样本的x就只包含从train_len到train_len96的历史但其中train_len以后的部分已经是验证集数据了这会破坏“历史窗口完全来自训练段”的干净状态。我的写法是从train_len - seq_len开始构造验证集这样第一个验证样本的x刚好落在训练段末尾预测目标是验证段开头逻辑上才说得通。测试集同理。数据增强方面时间序列和图像不一样直接给输入加高斯噪声容易破坏自相关结构小数据集上不一定有效。我的建议是先不要加增强把干净的基线跑出来后面调优阶段再考虑时间扭曲、随机mask之类的方案。3. 模型源码拆解位置编码、Transformer Encoder与预测头3.1 整个模型的数据流先明确数据形状。输入x的shape是(B, seq_len, 1)B是batch_sizeseq_len是历史窗口长度1是特征数。在单变量温度预测里每个时间点只有温度一个数值。模型内部的数据流是输入经过一个Linear层把维度从1映射到d_model得到(B, seq_len, d_model)。加上位置编码得到带顺序信息的向量。经过若干层TransformerEncoderLayer。取最后一个时间步的输出向量。经过全连接预测头输出(B, pred_len)。这里我特意取最后一个时间步而不是把所有时间步做平均。原因是最后一个时间步前面已经通过自注意力聚合了所有历史位置的信息而且它天然带有时序上的“最近状态”。如果取平均反而会把较早时刻的无关信息混进来削弱最近的趋势信号。3.2 位置编码的实现逻辑自注意力本身是“无序”的。你把序列倒过来喂给模型注意力计算结果和正序几乎一样只有对应位置对调。这在NLP里也不行在时间序列里更不行因为时序的意义就藏在先后顺序里。位置编码的作用就是给每个时间步注入一个独立的顺序标记。常见的做法是用不同频率的正余弦函数偶数维度用sin计算奇数维度用cos计算不同频率的组合让每个位置都有一条独特的编码轨迹。这个编码不需要训练还能外推到比训练时更长的序列。代码里用register_buffer保存这样它会跟着模型一起搬进GPU但不参与梯度更新。3.3 完整模型代码model.py下面是完整的模型定义直接存成model.py就能用。import math import torch import torch.nn as nn class PositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int 5000, dropout: float 0.1): super().__init__() self.dropout nn.Dropout(dropout) pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1).float() 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) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x: torch.Tensor) - torch.Tensor: x x self.pe[:, : x.size(1)] return self.dropout(x) class TransformerTimeSeries(nn.Module): def __init__( self, d_model: int 64, nhead: int 4, num_encoder_layers: int 2, dim_feedforward: int 128, seq_len: int 96, pred_len: int 24, dropout: float 0.1, ): super().__init__() self.input_proj nn.Linear(1, d_model) self.pos_encoder PositionalEncoding(d_model, dropoutdropout) encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, dim_feedforwarddim_feedforward, dropoutdropout, batch_firstTrue, activationgelu, ) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_encoder_layers) self.head nn.Sequential( nn.Linear(d_model, d_model // 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model // 2, pred_len), ) def forward(self, x: torch.Tensor) - torch.Tensor: x self.input_proj(x) x self.pos_encoder(x) x self.encoder(x) x x[:, -1, :] out self.head(x) return out几个值得说明的细节batch_firstTrue让输入维度直接是(B, seq_len, d_model)不用手动转置可读性好很多。这个参数在PyTorch 1.9之后才支持所以强烈建议用PyTorch 2.x。激活函数我选择了GELU它在Transformer类模型里比ReLU更顺滑收敛也更稳。改成ReLU也不会报错但效果会略有波动。预测头第一层把d_model压缩到一半再加GELU和dropout最后输出pred_len。这种“先降维再升维”的结构其实是一个简单瓶颈层能强迫模型把信息压缩到更有用的表示里。为什么这里没有Decoder前面说过因为任务是回归式预测不是生成式预测。Encoder把历史信息压缩最后一步向量就够了。先跑通这个版本再考虑Decoder会轻松得多。4. 训练与评估优化器、损失函数和指标怎么配才靠谱4.1 训练配置与超参Transformer的收敛特性和LSTM不太一样。LSTM用Adam配一个固定学习率通常没问题但Transformer对学习率更敏感学习率太大会出现loss突然变成nan的情况。我的初始配置是优化器AdamWlr1e-3weight_decay1e-5学习率调度CosineAnnealingLRT_max50batch_size256epochs50梯度裁剪clip_grad_norm_1.0如果你在训练中遇到loss震荡或nan先把学习率降到3e-4再把TransformerEncoderLayer的norm_first设为True试试。PyTorch官方实现默认是post-normalization也就是先计算子层再LayerNorm改成pre-normalization之后训练会稳定不少代价是收敛速度可能会略慢。4.2 损失函数与评估指标预测任务最常用的损失函数是MSE它和L2范数对应对大的预测误差惩罚更重梯度推导也简单。如果你更关心预测曲线的整体形状而不是极值点可以改用Huber Loss它对离群点更鲁棒。评估指标我建议同时看MSE、MAE、RMSE和MAPE。前三者收敛趋势基本一致MAPE能反映相对误差但在真实值接近0的时候会爆炸所以计算时加一个epsilon。指标名称说明MSE均方误差预测值与真实值差的平方的平均放大较大误差MAE平均绝对误差对误差取绝对值平均更贴近实际偏差RMSE均方根误差对MSE开根号量纲和原始数据一致MAPE平均绝对百分比误差相对误差越小说明预测精度越高指标计算一定要在反归一化之后做否则得到的是归一化尺度下的数字不方便和论文或业务目标对比。4.3 训练循环完整代码train.py下面是训练主程序我特意把数据加载到评估的完整链路都写在一起。你只需要把前面两个文件准备好直接运行train.py即可。import numpy as np import pandas as pd import torch import torch.nn as nn from torch.utils.data import DataLoader from torch.optim import AdamW from torch.optim.lr_scheduler import CosineAnnealingLR from dataset import normalize, TimeSeriesDataset from model import TransformerTimeSeries def compute_metrics(pred, true): pred np.asarray(pred) true np.asarray(true) mse np.mean((pred - true) ** 2) mae np.mean(np.abs(pred - true)) rmse np.sqrt(mse) mape np.mean(np.abs((true - pred) / (true 1e-8))) * 100 return {MSE: mse, MAE: mae, RMSE: rmse, MAPE: mape} def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss 0.0 for x, y in loader: x, y x.to(device), y.to(device) optimizer.zero_grad() pred model(x) loss criterion(pred, y.squeeze(-1)) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() total_loss loss.item() * x.size(0) return total_loss / len(loader.dataset) torch.no_grad() def evaluate(model, loader, criterion, device): model.eval() total_loss 0.0 preds, trues [], [] for x, y in loader: x, y x.to(device), y.to(device) pred model(x) loss criterion(pred, y.squeeze(-1)) total_loss loss.item() * x.size(0) preds.append(pred.cpu().numpy()) trues.append(y.squeeze(-1).cpu().numpy()) preds np.concatenate(preds, axis0) trues np.concatenate(trues, axis0) return total_loss / len(loader.dataset), preds, trues if __name__ __main__: device torch.device(cuda if torch.cuda.is_available() else cpu) df pd.read_csv(data/ETTh2.csv) values df[OT].values.astype(float) total_len len(values) train_len int(total_len * 0.7) val_len int(total_len * 0.15) normed, min_v, max_v normalize(values, train_len) seq_len, pred_len 96, 24 train_ds TimeSeriesDataset(normed[:train_len], seq_len, pred_len, step1) val_ds TimeSeriesDataset(normed[train_len - seq_len: train_len val_len], seq_len, pred_len, step1) test_ds TimeSeriesDataset(normed[train_len val_len - seq_len:], seq_len, pred_len, step1) train_loader DataLoader(train_ds, batch_size256, shuffleTrue) val_loader DataLoader(val_ds, batch_size256, shuffleFalse) test_loader DataLoader(test_ds, batch_size256, shuffleFalse) model TransformerTimeSeries( d_model64, nhead4, num_encoder_layers2, dim_feedforward128, seq_lenseq_len, pred_lenpred_len, dropout0.1, ).to(device) criterion nn.MSELoss() optimizer AdamW(model.parameters(), lr1e-3, weight_decay1e-5) scheduler CosineAnnealingLR(optimizer, T_max50) best_val_loss float(inf) 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() 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(fEpoch {epoch 1:3d} | Train Loss {train_loss:.5f} | Val Loss {val_loss:.5f}) model.load_state_dict(torch.load(best_model.pt)) test_loss, test_preds, test_trues evaluate(model, test_loader, criterion, device) test_preds test_preds * (max_v - min_v 1e-8) min_v test_trues test_trues * (max_v - min_v p a hrefhttps://download.csdn.net/download/kjm13182345320/89995016 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p Transformer时间序列预测实战：PyTorch完整实现与踩坑指南
简介时间序列预测是数据科学和工业场景中的核心任务传统模型如LSTM虽能捕捉一定时序依赖但面临长序列训练效率低、记忆容量有限等瓶颈。Transformer通过自注意力机制实现任意位置间的直接交互支持并行计算显著提升长序列建模能力。在实际应用中将Transformer应用于时序预测需关注数据构造、位置编码、归一化等关键环节避免信息泄漏与训练不稳定。滑动窗口生成样本、基于训练集统计的归一化、以及Encoder-only结构设计是搭建高效预测系统的基础。本文基于PyTorch给出从数据预处理到模型训练评估的完整代码以ETTh2电力数据集为例演示如何构建可复现的时间序列预测基线并总结了常见问题与调优方向帮助读者快速落地Transformer时序预测项目。 这段时间我收到不少和Transformer时间序列预测有关的私信。很多人从NLP那边转过来看到Transformer的论文就想拿它跑时序结果一动手就卡住数据格式怎么构造位置编码加在哪为什么别人代码跑出来指标那么低自己一跑就是nan我自己也完整走过一遍这条路所以把一套能直接跑的Pytorch完整源码和数据整理出来了。这份代码不是竞赛刷分版本而是把标准的Encoder-only Transformer从头到尾接起来你换成自己的CSV也能用项目里有数据加载、滑动窗口切分、归一化、模型定义、训练循环、评估指标和断点保存。如果你正准备从LSTM迁移到Transformer或者只是想快速拿到一份能复现的时间序列预测基线这份记录应该能帮你省下不少时间。我会先把模型原理讲清楚再贴完整代码最后把我实测中踩过的坑和几个改造方向一起列出来避免你在同一个地方再翻车。1. 为什么把Transformer搬进时间序列预测而不是继续用LSTM1.1 LSTM处理长序列时的三个老问题LSTM在时间序列预测里的地位不用多说很多做工业预测的老代码都是LSTM打底。它靠门控机制把信息通过隐状态一步一步往后传理论上能记住长期依赖但实际用下来有几个绕不开的痛。第一个问题是顺序计算。LSTM必须按时间步依次计算t时刻的隐状态要等t-1算完训练时没法在序列维度上并行。序列一长训练速度就慢。第二个问题是记忆容量。门控能缓解梯度消失但隐状态向量容量有限几百步之前的信息经过多轮门控之后已经衰减得很厉害。第三个问题是梯度传播路径太长如果梯度裁剪和初始化没做好训练很容易不稳定。我在项目里用LSTM做对比时同样预测未来24个点LSTM在一个epoch上的训练时间大概是Transformer的1.5倍多而且序列越长这个差距越明显。这不是说LSTM不行而是它天生是顺序建模处理长序列的代价就摆在那。1.2 自注意力机制到底改了什么Transformer的核心是自注意力。每一个时间步都可以直接和序列里所有其他时间步计算相关性不再依赖上一个时间步把信息“接力”传过来。这样的好处是任意两个位置之间的交互路径长度为1长距离依赖问题从机制上被绕开了。同时自注意力在计算上是可以并行的。所有位置的Query、Key、Value可以一次算出来再用矩阵乘法得到注意力权重多个时间步互不依赖。这就是为什么Transformer在大规模数据上能训练得动LSTM却很难做超长序列。放在时间序列这个场景里自注意力还有个很直观的解释它等于自动帮你找出“历史上哪几个时刻和当前趋势最相关”。温度预测里可能是24小时前的同一个时段用电负荷预测里可能是昨天同一时刻加上未来几小时的气温特征。LSTM靠隐状态隐式记忆这些关系Transformer直接显式建模这种跨时间步的相关性。1.3 时序预测和NLP任务的差异为什么用Encoder-only很多新手一上来就套用NLP的Transformer seq2seq结构非得写一个Encoder加一个Decoder还配置teacher forcing训练复杂度一下上去了。但实际上标准时间序列预测大多不需要Decoder。NLP里的Transformer是生成式任务输入一句话要一个词一个词地生成目标句子。而常见的时间序列预测任务是回归式任务给你过去96个时间点的值让你预测未来24个时间点的值。这是从一段连续数值到另一段连续数值的映射完全可以由一个Encoder提取历史模式再接一个回归头一次性输出所有预测点。我这份代码就是Encoder-only结构。模型把历史窗口编码成一组向量取最后一个时间步的输出过几个全连接层直接输出未来24个值。这么做的好处是训练稳定、代码简单、不容易出现自回归误差累积。如果后面你想做更长周期的滚动预测再考虑引入Decoder不迟这个我在最后一部分会讲。2. 数据先行从ETTh2到滑动窗口样本的完整工序2.1 数据选择与项目目录项目里用的数据是ETT数据集中的ETTh2这是时序预测论文里非常常见的benchmark内容是电力变压器的一系列指标和油温粒度是1小时一条。你在网上搜ETDataset的公开仓库就能下载到CSV文件里包含date、HUFL、HULL、MUFL、MULL、LUFL、OT这些列。本项目中只用OT列做单变量预测。下载之后把文件放到项目data目录下整体结构长这样transformer_ts/ ├── data/ │ └── ETTh2.csv ├── dataset.py ├── model.py ├── train.py └── README.md环境方面Python 3.9以上PyTorch 2.0以上再加pandas、numpy、matplotlib就够了。PyTorch直接从官网根据自己的系统和CUDA版本生成安装命令不要用默认的旧版教程硬抄版本差异会带来一堆奇怪问题。2.2 滑动窗口如何把一维序列变成训练样本原始时间序列就是一个一维数组长度是T。要把序列变成监督学习格式最常用的办法是滑动窗口。举个例子seq_len96pred_len24步长step1那么从序列开头开始取0到96作为x96到120作为y然后往后移一步取1到97作为x97到121作为y一直滑到序列末尾。假设原始序列有1000个点经过这种方式切分后样本数量是1000 - 96 - 24 1 881个。这里有个细节需要注意窗口长度一定要覆盖至少一个周期。ETTh2是小时粒度一天24个点seq_len取96就是4天足够让模型看到完整的日周期。如果序列是7天周期那窗口至少取168会比较稳。2.3 归一化先fit训练集再transform验证集归一化是时序预测里最容易出问题的一步。很多刚入门的同学会先对整个数据求min和max把全部序列归一化之后再切分数据集然后发现验证集指标好到离谱。这不是模型厉害而是数据泄漏了验证集的极大值和极小值已经参与了训练前的统计计算模型提前“见过”了验证集的范围。正确做法是只用训练集部分计算min和max然后用训练集算出的参数去归一化训练集、验证集和测试集。预测完成之后再用同一组参数反归一化把结果还原到原始温度尺度。下面是我项目里的dataset.py包含归一化和滑动窗口数据集import numpy as np import pandas as pd import torch from torch.utils.data import Dataset def normalize(values, train_len): 只用训练集部分计算 min/max避免信息泄漏 train_part values[:train_len] min_val float(train_part.min()) max_val float(train_part.max()) normed (values - min_val) / (max_val - min_val 1e-8) return normed, min_val, max_val class TimeSeriesDataset(Dataset): def __init__(self, values, seq_len96, pred_len24, step1): self.values values self.seq_len seq_len self.pred_len pred_len self.step step self.indices range(0, len(values) - seq_len - pred_len 1, step) def __len__(self): return len(self.indices) def __getitem__(self, index): start self.indices[index] x self.values[start: start self.seq_len] y self.values[start self.seq_len: start self.seq_len self.pred_len] return torch.FloatTensor(x).unsqueeze(-1), torch.FloatTensor(y)调用方式如下df pd.read_csv(data/ETTh2.csv) values df[OT].values.astype(float) train_len int(len(values) * 0.7) normed, min_v, max_v normalize(values, train_len) train_ds TimeSeriesDataset(normed[:train_len], seq_len96, pred_len24)这里有一个值得注意的点验证集和测试集的起点不能直接从分割点开始而是要往前回退seq_len个点。比如从train_len开始构造验证集那么验证集第一个样本的x就只包含从train_len到train_len96的历史但其中train_len以后的部分已经是验证集数据了这会破坏“历史窗口完全来自训练段”的干净状态。我的写法是从train_len - seq_len开始构造验证集这样第一个验证样本的x刚好落在训练段末尾预测目标是验证段开头逻辑上才说得通。测试集同理。数据增强方面时间序列和图像不一样直接给输入加高斯噪声容易破坏自相关结构小数据集上不一定有效。我的建议是先不要加增强把干净的基线跑出来后面调优阶段再考虑时间扭曲、随机mask之类的方案。3. 模型源码拆解位置编码、Transformer Encoder与预测头3.1 整个模型的数据流先明确数据形状。输入x的shape是(B, seq_len, 1)B是batch_sizeseq_len是历史窗口长度1是特征数。在单变量温度预测里每个时间点只有温度一个数值。模型内部的数据流是输入经过一个Linear层把维度从1映射到d_model得到(B, seq_len, d_model)。加上位置编码得到带顺序信息的向量。经过若干层TransformerEncoderLayer。取最后一个时间步的输出向量。经过全连接预测头输出(B, pred_len)。这里我特意取最后一个时间步而不是把所有时间步做平均。原因是最后一个时间步前面已经通过自注意力聚合了所有历史位置的信息而且它天然带有时序上的“最近状态”。如果取平均反而会把较早时刻的无关信息混进来削弱最近的趋势信号。3.2 位置编码的实现逻辑自注意力本身是“无序”的。你把序列倒过来喂给模型注意力计算结果和正序几乎一样只有对应位置对调。这在NLP里也不行在时间序列里更不行因为时序的意义就藏在先后顺序里。位置编码的作用就是给每个时间步注入一个独立的顺序标记。常见的做法是用不同频率的正余弦函数偶数维度用sin计算奇数维度用cos计算不同频率的组合让每个位置都有一条独特的编码轨迹。这个编码不需要训练还能外推到比训练时更长的序列。代码里用register_buffer保存这样它会跟着模型一起搬进GPU但不参与梯度更新。3.3 完整模型代码model.py下面是完整的模型定义直接存成model.py就能用。import math import torch import torch.nn as nn class PositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int 5000, dropout: float 0.1): super().__init__() self.dropout nn.Dropout(dropout) pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1).float() 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) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x: torch.Tensor) - torch.Tensor: x x self.pe[:, : x.size(1)] return self.dropout(x) class TransformerTimeSeries(nn.Module): def __init__( self, d_model: int 64, nhead: int 4, num_encoder_layers: int 2, dim_feedforward: int 128, seq_len: int 96, pred_len: int 24, dropout: float 0.1, ): super().__init__() self.input_proj nn.Linear(1, d_model) self.pos_encoder PositionalEncoding(d_model, dropoutdropout) encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, dim_feedforwarddim_feedforward, dropoutdropout, batch_firstTrue, activationgelu, ) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_encoder_layers) self.head nn.Sequential( nn.Linear(d_model, d_model // 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model // 2, pred_len), ) def forward(self, x: torch.Tensor) - torch.Tensor: x self.input_proj(x) x self.pos_encoder(x) x self.encoder(x) x x[:, -1, :] out self.head(x) return out几个值得说明的细节batch_firstTrue让输入维度直接是(B, seq_len, d_model)不用手动转置可读性好很多。这个参数在PyTorch 1.9之后才支持所以强烈建议用PyTorch 2.x。激活函数我选择了GELU它在Transformer类模型里比ReLU更顺滑收敛也更稳。改成ReLU也不会报错但效果会略有波动。预测头第一层把d_model压缩到一半再加GELU和dropout最后输出pred_len。这种“先降维再升维”的结构其实是一个简单瓶颈层能强迫模型把信息压缩到更有用的表示里。为什么这里没有Decoder前面说过因为任务是回归式预测不是生成式预测。Encoder把历史信息压缩最后一步向量就够了。先跑通这个版本再考虑Decoder会轻松得多。4. 训练与评估优化器、损失函数和指标怎么配才靠谱4.1 训练配置与超参Transformer的收敛特性和LSTM不太一样。LSTM用Adam配一个固定学习率通常没问题但Transformer对学习率更敏感学习率太大会出现loss突然变成nan的情况。我的初始配置是优化器AdamWlr1e-3weight_decay1e-5学习率调度CosineAnnealingLRT_max50batch_size256epochs50梯度裁剪clip_grad_norm_1.0如果你在训练中遇到loss震荡或nan先把学习率降到3e-4再把TransformerEncoderLayer的norm_first设为True试试。PyTorch官方实现默认是post-normalization也就是先计算子层再LayerNorm改成pre-normalization之后训练会稳定不少代价是收敛速度可能会略慢。4.2 损失函数与评估指标预测任务最常用的损失函数是MSE它和L2范数对应对大的预测误差惩罚更重梯度推导也简单。如果你更关心预测曲线的整体形状而不是极值点可以改用Huber Loss它对离群点更鲁棒。评估指标我建议同时看MSE、MAE、RMSE和MAPE。前三者收敛趋势基本一致MAPE能反映相对误差但在真实值接近0的时候会爆炸所以计算时加一个epsilon。指标名称说明MSE均方误差预测值与真实值差的平方的平均放大较大误差MAE平均绝对误差对误差取绝对值平均更贴近实际偏差RMSE均方根误差对MSE开根号量纲和原始数据一致MAPE平均绝对百分比误差相对误差越小说明预测精度越高指标计算一定要在反归一化之后做否则得到的是归一化尺度下的数字不方便和论文或业务目标对比。4.3 训练循环完整代码train.py下面是训练主程序我特意把数据加载到评估的完整链路都写在一起。你只需要把前面两个文件准备好直接运行train.py即可。import numpy as np import pandas as pd import torch import torch.nn as nn from torch.utils.data import DataLoader from torch.optim import AdamW from torch.optim.lr_scheduler import CosineAnnealingLR from dataset import normalize, TimeSeriesDataset from model import TransformerTimeSeries def compute_metrics(pred, true): pred np.asarray(pred) true np.asarray(true) mse np.mean((pred - true) ** 2) mae np.mean(np.abs(pred - true)) rmse np.sqrt(mse) mape np.mean(np.abs((true - pred) / (true 1e-8))) * 100 return {MSE: mse, MAE: mae, RMSE: rmse, MAPE: mape} def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss 0.0 for x, y in loader: x, y x.to(device), y.to(device) optimizer.zero_grad() pred model(x) loss criterion(pred, y.squeeze(-1)) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() total_loss loss.item() * x.size(0) return total_loss / len(loader.dataset) torch.no_grad() def evaluate(model, loader, criterion, device): model.eval() total_loss 0.0 preds, trues [], [] for x, y in loader: x, y x.to(device), y.to(device) pred model(x) loss criterion(pred, y.squeeze(-1)) total_loss loss.item() * x.size(0) preds.append(pred.cpu().numpy()) trues.append(y.squeeze(-1).cpu().numpy()) preds np.concatenate(preds, axis0) trues np.concatenate(trues, axis0) return total_loss / len(loader.dataset), preds, trues if __name__ __main__: device torch.device(cuda if torch.cuda.is_available() else cpu) df pd.read_csv(data/ETTh2.csv) values df[OT].values.astype(float) total_len len(values) train_len int(total_len * 0.7) val_len int(total_len * 0.15) normed, min_v, max_v normalize(values, train_len) seq_len, pred_len 96, 24 train_ds TimeSeriesDataset(normed[:train_len], seq_len, pred_len, step1) val_ds TimeSeriesDataset(normed[train_len - seq_len: train_len val_len], seq_len, pred_len, step1) test_ds TimeSeriesDataset(normed[train_len val_len - seq_len:], seq_len, pred_len, step1) train_loader DataLoader(train_ds, batch_size256, shuffleTrue) val_loader DataLoader(val_ds, batch_size256, shuffleFalse) test_loader DataLoader(test_ds, batch_size256, shuffleFalse) model TransformerTimeSeries( d_model64, nhead4, num_encoder_layers2, dim_feedforward128, seq_lenseq_len, pred_lenpred_len, dropout0.1, ).to(device) criterion nn.MSELoss() optimizer AdamW(model.parameters(), lr1e-3, weight_decay1e-5) scheduler CosineAnnealingLR(optimizer, T_max50) best_val_loss float(inf) 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() 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(fEpoch {epoch 1:3d} | Train Loss {train_loss:.5f} | Val Loss {val_loss:.5f}) model.load_state_dict(torch.load(best_model.pt)) test_loss, test_preds, test_trues evaluate(model, test_loader, criterion, device) test_preds test_preds * (max_v - min_v 1e-8) min_v test_trues test_trues * (max_v - min_v p a hrefhttps://download.csdn.net/download/kjm13182345320/89995016 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p