简介:本资源是一套基于PyTorch实现的Transformer时间序列预测完整实验方案,面向机器学习初学者与时间序列建模实践者,聚焦单步与多步预测两大核心任务,适用于气象、能源、金融等领域的时序数据分析场景。压缩包共10个文件(3.21MB),包含2个核心训练脚本(transformer-singlestep.py与transformer-multistep.py)、1个真实温度数据集(daily-min-temperatures.csv)、1张预测效果可视化图(transformer-future200.png)、1个动态训练过程演示gif(result.gif)、1份环境依赖说明(requirements.txt)及README文档等,结构清晰、开箱即用。已有2909人学习下载,所有代码均经实测可直接运行,附带每日最低温度数据上的100轮训练结果与多步预测对比分析,便于读者理解模型输入构造、位置编码设计、掩码机制应用及输出解码逻辑,是深入掌握Transformer在时序领域落地的优质实践素材。 拿到一个时间序列预测需求时,很多人第一反应是先套LSTM。我一开始也是这么干的,但项目数据拉长到几百个时间步后,LSTM训练慢不说,往后预测的曲线像一个被压扁的弹簧,越来越“平”。朋友建议试试Transformer,我带着将信将疑的态度,用Pytorch从零搭了一套单步和多步预测实验。没想到的是,模型把长距离依赖抓得很稳,预测结果比LSTM明显更贴合趋势。这篇文章就是这次实验的记录,从为什么选Transformer、单步/多步怎么设计,到Pytorch完整实现和踩坑实录都会拆开讲。适合已经会上手Pytorch、但还没用Transformer处理过时间序列的同学,也适合做量化、销量预测、设备监测等场景的工程师参考。我会尽量把每个关键选择背后的原因写清楚,而不是只丢代码。
1. 整体设计与思路拆解
1.1 为什么用Transformer而不是LSTM?
在时间序列上做预测,本质上是在建模“过去的时间点如何影响未来”。LSTM和GRU的思路是递归地、逐步地把信息往后传,这种过程的优点是参数少、可解释性强,但缺点也很明显:训练是串行的,序列一长,梯度消失和长期依赖丢失的问题就会显现。Transformer的设计是直接计算序列中任意两个位置之间的关联度,通过多头注意力机制一次性看到全局信息。放在时间序列里,这就意味着模型可以自己学会“第1个时间点对第50个时间点很重要”,而不是像RNN那样靠门控一路“背”过去。
不过话说回来,Transformer也不是没有代价。它不像LSTM自带顺序概念,必须靠位置编码把顺序信息硬塞进输入;而且注意力矩阵的空间复杂度是O(n^2),窗口特别长的时候会吃显存。这决定了我们在实验里不能无脑堆窗口,需要在数据长度和模型容量之间做取舍。但总体上,对于中等长度(几十到几百个时间点)的时间序列,Transformer的精度和训练效率都很有竞争力。
我在做实验对比时,同一个数据切了两份,一份用两层LSTM,一份用两层Transformer encoder。LSTM训练了接近200个epoch才收敛,Transformer在100个epoch左右就已经更好了,尤其在后半段的预测曲线上,Transformer更贴近原始趋势。这其实不难理解:时间序列里“之前一段时间的形态”往往比“最后一个点的值”更重要,而注意力机制天生擅长捕捉这种形态。
1.2 单步预测和多步预测的差异
单步预测是经典的监督学习任务:输入过去一段序列,输出下一个时间点的值。多步预测的难度要高一个档次,因为输出的是一个连续的未来区间,不只是“下一个点”的精确值,还要考虑未来多个点之间的叠加误差。
我在实验中分别实现了两种多步方案,这里先做一个直观对比:
| 指标 | 直接多步 | 递归多步 |
|---|---|---|
| 实现复杂度 | 输出层维度设为horizon,简单 | 需要把预测值拼回输入,循环推理 |
| 推理速度 | 快,一次forward得到K个点 | 慢,要连续forward K次 |
| 误差累积 | 几乎没有,因为K个点同时输出 | 明显,越往后越偏 |
| 训练难度 | 需要同时拟合K个标签,数据量小时难度大 | 训练时仍是单步,难度低 |
| 适用场景 | horizon较大、训练数据充足 | 小规模快速实验、模型迭代期 |
- 直接多步(Direct Multi-step):Transformer输出层的维度直接设为horizon长度,让模型一次生成未来K个点。缺点是模型要同时拟合K个输出,训练难度高,但推理快。
- 递归多步(Recursive Multi-step):先用单步模型预测出下一个点,再把这个预测值当输入滑进窗口,继续预测下下个点。实现简单,但误差会随着步数累积,越往后越偏。
- 还有一种Teacher Forcing的训练方式,只在训练时用真实值替代预测值喂养模型,推理时再用预测值,这个后面在训练循环里会讲。
这两个方案我都跑了,也记录了各自的loss曲线和误差指标。直接多步在horizon比较大时会比递归更稳,但前提是训练数据要充足;递归多步在小样本上更灵活,适合快速迭代。建议你在自己的数据上都试试,对比后再选。
2. 核心细节解析与实操要点
2.1 位置编码不能省,但不要照搬NLP
Transformer本身没有顺序感。把序列顺序打乱,注意力算出来的结果是一模一样的。所以位置编码是Transformer做时序预测绕不开的基础。NLP里常见的是正弦位置编码,把位置索引用sin/cos函数映射到embedding维度上。在时间序列里,这个方式一样能用,但我测试下来,直接把时间戳(比如间隔、周几、小时)作为额外特征拼进输入,效果往往更好,尤其是数据有明显周期性的场景。
实用的做法是“加”而不是“拼”:把正弦位置编码加到输入embedding上,然后额外把标准化后的时间特征拼在输入特征维度后面。这样模型既能感知绝对位置,也能感知周期性信息。代码实现时,位置编码的维度要和d_model一致,否则维度对不上没法加。
为什么要乘sqrt(d_model)?这是Transformer原论文里的一个细节。输入经过embedding后,数值范围通常会比较小,直接往上面加位置编码,位置信息很容易盖过原始特征。乘上sqrt(d_model)相当于把输入embedding放大到和位置编码一个量级,让两者在相加时都能保留有效信息。我在实验里试过不乘这个系数,训练loss收敛明显变慢,所以这个细节不建议省。
2.2 数据预处理决定模型上限
先讲一个我踩过的坑:一开始我直接用整个序列的均值和方差做归一化,训练loss很漂亮,但测试特别差。后来才意识到,这是把未来信息透漏给了训练过程,也就是典型的数据泄漏。正确的做法是只在训练集上fit scaler,再用同一套scaler去transform验证集和测试集。
滑动窗口的构造也值得细品。单步预测时,每条样本是[look_back, feature_dim]的输入和[1]的标签;多步预测时,标签就变成[horizon]或者[horizon, feature_dim]。窗口长度look_back一般取序列长度的10%~20%,不能太短也不能太长。我常用look_back=24或48来预测未来6~12个点,效果比拍脑袋定的窗口好不少。另外,如果数据有缺失,不要用均值填充,尽量用前向填充或插值,避免引入错误的局部趋势。
关于归一化,我一般用MinMaxScaler。它在数据有明显上下界时非常稳,能让模型更快收敛。StandardScaler也不是不行,但对异常值更敏感。如果数据里有特别大的尖峰,MinMax可能把正常值压得太扁,这时候可以先做一轮异常值截断再用。不要一上来就把所有数据丢进模型,先画图看分布,这个习惯能省掉后面很多麻烦。
2.3 核心超参数怎么定,才有得跑
下面是这次实验里的核心超参数,可以直接照抄:
| 参数 | 数值 | 说明 |
|---|---|---|
| d_model | 64 | 输入embedding和注意力投影的维度 |
| nhead | 4 | 多头注意力头数,需能被d_model整除 |
| num_encoder_layers | 2 | 标准Transformer里encoder层数,我用了2层 |
| dim_feedforward | 256 | 前馈网络隐藏层维度 |
| dropout | 0.1 | 防止过拟合 |
| batch_size | 64 | 显存不大就调小,建议32~128 |
| learning_rate | 1e-3 | 初始学习率,配合cosine调度 |
| epochs | 100 | 早停耐心值建议设20 |
注意nhead一定要能整除d_model,否则Pytorch会直接报错。层数不是越多越好,我试过4层,在数据量不大的情况下反而更容易过拟合。学习率建议用AdamW + 余弦退火,Transformer对学习率比LSTM敏感,固定学习率容易震荡。dim_feedforward我习惯设成d_model的4倍左右,太大参数量涨得快,太小表达能力受限。
还有一个常被忽略的参数是dropout。时间序列训练数据一般不会特别大,dropout设太小容易过拟合,设太大又可能欠拟合。0.1到0.2之间是个安全区间。如果你的数据量超过十几万条,可以把dropout降到0.05,加快收敛。
3. 实操过程与核心环节实现
3.1 环境搭建与项目结构
动手前先把环境理顺。我习惯用conda建一个独立环境,避免跟其他项目打架。
conda create -n ts_transformer python=3.9 -y conda activate ts_transformer pip install torch==2.1.2 --index-url https://download.pytorch.org/whl/cu118 pip install numpy pandas matplotlib scikit-learn如果你用的是GPU机器,装完后第一时间检查一下CUDA是否可用:
python -c "import torch; print(torch.cuda.is_available(), torch.cuda.device_count())"输出“True 1”就说明Pytorch能用GPU了。如果输出False,不是驱动问题就是Pytorch版本没选对。我还在Jetson等嵌入式设备上装过,那又得单独挑适配版本,不建议新手一上来就碰。
项目目录我习惯这样组织:
time_series_transformer/ ├── data.csv # 原始数据 ├── dataset.py # 数据集与预处理工具 ├── model.py # Transformer模型定义 ├── train.py # 单步/多步训练脚本 ├── evaluate.py # 评估与可视化 └── requirements.txt这个结构看着简单,但足够用。实验迭代时最怕把所有代码堆在一个文件里,改一个地方牵一发动全身。把数据、模型、训练、评估拆开,后面换数据集或者换模型会轻松很多。
3.2 数据准备与Dataset类
为了把实验说透,我用了一段周期信号叠加噪声的数据,模拟传感器读数或者销量曲线。你也可以换成自己的csv,只要确保日期列按时间升序排列。
import numpy as np import pandas as pd from torch.utils.data import Dataset, DataLoader # 生成示例数据:正弦 + 线性趋势 + 噪声 np.random.seed(42) t = np.arange(0, 1200) signal = 10 * np.sin(2 * np.pi * t / 50) salary = t * 0.01 noise = np.random.normal(0, 1, size=t.shape) data = signal + salary + noise注意这个数据只是示例,真正项目里数据要复杂得多,但处理和建模逻辑是一样的。这里我故意加了一个单调趋势,是想看Transformer能不能在趋势和周期并存时学出结构。
接下来是滑动窗口。我写了一个继承Dataset的类,输入是[look_back, feature_dim]序列,标签是未来horizon个点。
class TimeSeriesDataset(Dataset): def __init__(self, data, look_back=24, horizon=6, step=1): self.x, self.y = [], [] for i in range(0, len(data) - look_back - horizon + 1, step): x = data[i:i + look_back] y = data[i + look_back:i + look_back + horizon] self.x.append(x) self.y.append(y) self.x = np.array(self.x, dtype=np.float32) self.y = np.array(self.y, dtype=np.float32) def __len__(self): return len(self.x) def __getitem__(self, idx): return self.x[idx], self.y[idx]需要注意,这里的data是一维数组,所以我构建x时每个时间点只有1个特征。如果有多维特征,沿着最后一个维度拼接即可,上一节的“拼时间戳”也是在这里做。构造好的样本可以用DataLoader加载:
look_back, horizon = 24, 6 dataset = TimeSeriesDataset(data, look_back, horizon) # 按时间顺序切分 train_size = int(0.7 * len(dataset)) val_size = int(0.15 * len(dataset)) test_size = len(dataset) - train_size - val_size train_data, val_data, test_data = torch.utils.data.random_split( dataset, [train_size, val_size, test_size], generator=torch.Generator().manual_seed(42) )这里要特别提醒:random_split会在内部做随机打乱,但因为我们构造样本时本身就是按时间顺序滑动窗口生成的,可能不太合适。更稳妥的做法是直接按索引切分,不调用random_split,具体看你自己需求。如果是严格的时序预测,建议手动切片,保证训练集在时间上完全早于验证集和测试集。
3.3 Transformer模型核心实现
标准Transformer结构里包含Encoder和Decoder。但时间序列预测不一定需要Decoder,尤其是单步预测和直接多步预测,用Encoder堆叠后接一个全连接输出层就能取得不错的效果。我在实验中选的是Encoder-only方案,结构更轻,训练更快。如果想做递归多步,同样可以复用这个Encoder,只是在推理时多绕几圈。
下面是完整的模型定义,位置编码和Encoder主体都放进去。
import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, dropout=0.1, max_len=5000): super().__init__() self.dropout = nn.Dropout(p=dropout) 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) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) def forward(self, x): x = x + self.pe[:, :x.size(1), :] return self.dropout(x) class TimeSeriesTransformer(nn.Module): def __init__(self, feature_dim=1, d_model=64, nhead=4, num_layers=2, dropout=0.1): super().__init__() self.d_model = d_model self.input_proj = nn.Linear(feature_dim, d_model) self.pos_encoder = PositionalEncoding(d_model, dropout) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=256, dropout=dropout, batch_first=True, ) self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) def forward(self, x): # x: [batch, seq_len, feature_dim] x = self.input_proj(x) * math.sqrt(self.d_model) x = self.pos_encoder(x) x = self.transformer_encoder(x) return x在单步任务里,输出层可以只取编码序列的最后一帧,再接一个Linear。在多步任务里,要么对最后一帧输出horizon维,要么把整个编码序列压平再接全连接。后者参数量大,一般用前者:
class SingleStepHead(nn.Module): def __init__(self, d_model): super().__init__() self.fc = nn.Linear(d_model, 1) def forward(self, enc_output): return self.fc(enc_output[:, -1, :]) # 取最后一个位置的编码 class MultiStepHead(nn.Module): def __init__(self, d_model, horizon): super().__init__() self.fc = nn.Linear(d_model, horizon) def forward(self, enc_output): return self.fc(enc_output[:, -1, :])为什么取“最后一个位置的编码”而不是所有位置的编码?因为TransformerEncoder每一层都会做全局注意力,也就是说每个位置的输出都已经包含了对整个序列的“总结”。最后一个位置在时间上离预测目标最近,拿它做输出头的信息瓶颈最合适。如果要保留更多细节,也可以把最后几帧拼起来,但代价是参数量增加,收益不一定明显。
3.4 单步预测训练流程
训练循环本身不复杂,但有几个地方我会特别注意。第一个是loss,单步用MSE,这是回归任务的标配。第二个是优化器,AdamW配合余弦退火几乎是当前Transformer训练的标准配置,比单纯用Adam稳定。
def train_model(model, train_loader, val_loader, epochs=50, lr=1e-3): device = 'cuda' if torch.cuda.is_available() else 'cpu' model = model.to(device) criterion = nn.MSELoss() optimizer = torch.optim.AdamW(model.parameters(), lr=lr) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs) for epoch in range(epochs): model.train() train_loss = 0.0 for x, y in train_loader: x, y = x.to(device), y.to(device) optimizer.zero_grad() enc_out = model(x) pred = model.output_head(enc_out) loss = criterion(pred, y) loss.backward() optimizer.step() train_loss += loss.item() scheduler.step() val_loss = evaluate(model, val_loader, criterion, device) if (epoch + 1) % 10 == 0: print(f'Epoch {epoch+1}, Train Loss: {train_loss/len(train_loader):.6f}, Val Loss: {val_loss:.6f}') return model写这段代码时要注意,model.output_head是我们在外部挂上去的。一个更方便的做法是在TimeSeriesTransformer里增加一个head_type参数,直接在构造函数里决定使用单步头还是多步头,这样训练代码可以共用。
训练时如果loss长时间不下降,先查两件事:学习率是不是太大,数据有没有标准化。绝大多数奇怪现象都出在这两处。
3.5 多步预测实现方式
直接多步训练起来和单步几乎一样,只是y的维度从1变成horizon,输出头换成MultiStepHead,loss仍然用MSE。推理的时候一次拿到的就是未来horizon个点,非常直接。
递归多步需要额外写一个推理函数,把模型预测出的下一个值拼到输入序列尾部,同时丢掉最前面的点,让窗口“滑”起来。
def recursive_forecast(model, history, steps, device='cpu'): # history: [look_back, feature_dim] model.eval() history = torch.tensor(history, dtype=torch.float32).unsqueeze(0).to(device) preds = [] with torch.no_grad(): for _ in range(steps): enc_out = model(history) pred = model.output_head(enc_out) # [1, 1] preds.append(pred.item()) pred = pred.unsqueeze(-1) # [1, 1, 1] history = torch.cat([history[:, 1:, :], pred], dim=1) return preds这个思路很直观,但坑也在这:输入用的预测值本身就是不准的,误差会随着步数增多不断放大。我实测下来,horizon=6时还能看,horizon=24时后面基本变成一条平滑直线。
缓解误差累积有几个实用技巧:
- 训练时用Teacher Forcing,即以一定概率用真实值替代预测值去喂下一步,让模型适应自己的预测;
- 对输出做clip,限制预测值的合理范围,避免极端值把窗口带偏;
- 推理阶段用确定性策略(比如取中位数)而不是单次路径。
3.6 评估与可视化
评估指标我用三个:MAE、RMSE和R2。MAE好解释,RMSE对大误差更敏感,R2则能反应模型对真实趋势的拟合程度。
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score def evaluate_metrics(y_true, y_pred): mae = mean_absolute_error(y_true, y_pred) rmse = mean_squared_error(y_true, y_pred, squared=False) r2 = r2_score(y_true, y_pred) return {'MAE': mae, 'RMSE': rmse, 'R2': r2}画图建议用matplotlib,把测试集真实值和预测值画在同一张图里。多步预测时,我会把每个预测窗口的起点和终点连起来画成阶梯状,能更清楚地看到误差累积发生在哪个阶段。
import matplotlib.pyplot as plt plt.figure(figsize=(12, 5)) plt.plot(y_true, label='True') plt.plot(y_pred, label='Pred') plt.legend() plt.title('Time Series Prediction') plt.show()这张图是判断模型有没有“学到东西”的最快方式。如果pred和true的相位都对不上,先检查窗口长度是否合理;如果趋势对但局部抖动不对,考虑加更多特征或调d_model。
4. 常见问题与排查技巧实录
4.1 数据泄漏:最隐蔽也最致命
数据泄漏是时间序列预测里出现频率最高的“幽灵问题”。除了归一化泄漏,还有一种是数据集划分时不注意时间顺序,把未来样本混进训练集,结果模型在测试集上表现“超神”,一到线上就崩。处理办法是严格按时间顺序划分,且在构造滑动窗口时让训练集只使用该时间点之前的数据。另外,如果数据来自多个实体,比如多个城市或设备的序列,不能把它们混在一起打乱划分,应该按实体分组切分,否则也会出现实体级别的泄漏。
我自己的判断标准很简单:如果训练集和测试集的MAE差距小得离谱,或者测试集比训练集还好,那大概率是泄漏了。正常情况训练误差应该略低于测试误差,这个经验虽然不绝对,但能帮你快速发现问题。
4.2 训练Loss不降或变成NaN
我遇到过一次Loss变成NaN的情况,查了半天发现是没有对输入数据做归一化,数值稍微大一点,经过多头注意力里的softmax后就爆炸了。解决方法是先把数据缩放到[0, 1]或标准化到均值0方差1,再开始训练。如果标准化后还是NaN,检查一下学习率,Transformer的训练不太吃固定的大学习率,建议从1e-4或者3e-4开始试。还有一种可能是位置编码的max_len不够,序列长度超过上限导致pe切片出问题,这种一般会直接报IndexError,不太会静默NaN。
另外一个容易忽视的点是batch_first。Pytorch的TransformerEncoderLayer默认batch_first=False,如果输入维度是[batch, seq_len, feature],必须显式设置batch_first=True,否则会对seq_len维度做batch维处理,虽然不报错,但结果完全不对。这种错误很难排查,因为它表现成loss一直在下降但验证曲线不对劲。
4.3 多步预测后期曲线“平”了
多步预测的误差累积问题前面提过,这是所有递归式预测的通病。处理思路有两个方向:一是把模型换成直接多步输出,一次性预测完整未来区间,从根源上避免递归累积;二是引入机器学习里的鲁棒性技巧,比如训练时对输入加小噪声、预测时对输出做平滑。我自己的经验是,如果horizon不超过10,递归多步能省显存且效果不差;如果horizon到了24以上,直接多步会更稳定。建议在实验设置里同时保留这两种模式,用验证集决定到底用哪个。
有时“曲线平了”也不全是误差累积的问题,可能是数据本身的信噪比太低。我画过频谱图之后发现,数据里真正可预测的成分只占很小一部分,模型很容易倾向于输出均值附近的值。这时候考虑对数据做差分或季节分解,把趋势和周期拆开建模,再对残差做预测。
4.4 显存和训练速度问题
注意力矩阵的大小是平方级增长的,窗口长度从24改成96,显存占用可能直接翻好几倍。如果显存不够,优先调小batch_size,或者用梯度累积(gradient accumulation)模拟更大的batch。Pytorch的自动混合精度(AMP)也值得开,torch.cuda.amp里封装得已经很成熟,能省不少显存,还能加速训练。下面是一个简化的混合精度训练片段:
scaler = torch.cuda.amp.GradScaler() for x, y in train_loader: optimizer.zero_grad() with torch.cuda.amp.autocast(): pred = model(x) loss = criterion(pred, y) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()AMP在NVIDIA GPU上用Pytorch 2.x已经非常稳定,CPU上不用开。还有一个建议:尽量用batch_first=True的TransformerEncoderLayer,Pytorch默认是False,如果搞错了维度,虽然报错你会知道,但排查起来很烦。
4.5 长序列窗口的改进空间
标准Transformer处理几百甚至上千的时间步时,显存和效率都会遇到瓶颈。我这次实验主要用的是中等长度窗口,但如果你遇到超长序列,建议看看Transformer的改进变体,比如Informer的稀疏注意力、Autoformer的分解机制、PatchTST把时间序列切成patch再编码。这些都在标准Transformer的思路上做了针对性优化,很适合高频金融数据、长时间传感器监测这类场景。我后续也在计划把PatchTST并进这套实验框架里做对比。
另外提醒一句,模型调参不是全部。数据质量、特征工程、异常值处理对最终预测效果的影响往往比换模型更明显。我在做这套实验时,花了将近一半时间在清洗和构建特征上,模型本身虽然重要,但绝不是唯一变量。
做完整套单步、多步实验后,我最深的体会是:Transformer不是一个“万能开关”,它适合的是长依赖和多变量交互明显的时间序列。在这套实验中,我用Pytorch从数据预处理、模型搭建到评估可视化完整走了一遍,也踩了不少坑。如果你正打算在自己的数据上复现,建议先把数据好好看一眼,再按文中的代码结构一点点搭,别急着直接套大模型。先把单步跑通,再多步,最后再回头调超参,这样每一步都能定位到问题。希望这篇记录能帮你少走点弯路。
本文还有配套的精品资源,点击获取