基于Transformer的时间序列预测Python源码解析与调优

简介：一套基于Transformer模型的时间序列预测Python源码项目，主要面向需要完成课程期末大作业、毕业设计或入门深度学习时序建模的高校学生与开发者。项目已在本地编译运行通过，难度适中，且经过助教老师审定，能够满足课程学习、项目借鉴和二次开发的需求。压缩包大小约28千字节，共包含12个文件，其中包括11个Python脚本和1个Jupyter记事本。这些脚本覆盖了完整的时间序列预测流程：数据处理模块负责读取与批量加载，时间特征模块负责提取周期与趋势特征，掩码模块负责构造注意力机制的遮挡逻辑，评估模块负责计算预测误差；模型部分则实现了嵌入层、编码器、解码器以及多头自注意力组件。记事本文件则演示了从原始序列到模型输出的完整序列到序列预测过程；当前已有1817人学习浏览，下载后可直接运行体验，也可以替换自定义数据集进行迁移实验，是一份结构清晰、上手难度低、实用性强的Transformer时序预测参考代码。

1. 基于Transformer模型的时间序列预测Python源码：这份zip能解决什么问题

基于Transformer模型的时间序列预测，近两年几乎成了LSTM时序预测方案里最主流的新路线，尤其当序列超过100个点、存在日周期或周周期依赖时。这个标着“高分项目”的Python源码包，核心是用Transformer架构训练一套“用过去N个点预测未来M个点”的回归模型，可迁移到电力负荷、交通流量、气象温度等常见时序场景。

打开这类包之前先明确：它不是开箱即用的预测软件，而是一套可复现的完整训练管线——数据预处理、滑动窗口、模型定义、训练评估和可视化都在里面。对课程设计来说是现成的闭环方案，对工程师来说是把transformer时序预测从论文落到本地的快速路径。下文按“原理→跑通→调参→避坑→改造”的顺序，把这套代码真正吃透。

2. Transformer做时间序列预测的原理：从自注意力到预测输出

2.1 为什么选Transformer而不是LSTM：时序预测的选型逻辑

先回答一个矛盾：时间序列明明是顺序数据，为什么不用天生按顺序建模的RNN/LSTM？

LSTM做时序的痛点是循环结构本身——每个时间步依赖上一个时间步的输出，训练必须按时间展开，长序列里梯度经过几十步后很容易消失，学到的是“近邻关系”而不是“长周期关系”。要记住一周前的同一点，LSTM需要把信息在隐状态里保存很久，这个容量非常有限。你搜lstm时间序列预测python的老教程，会看到大量“预测值比真实值滞后一个时刻”的翻车案例，根因就在这里。

Transformer解决这个问题的核心是自注意力机制：序列里任意两个位置直接建立联系，位置i和位置j之间只需要一次矩阵计算就能得到注意力分数，不管隔着多远。这让模型捕捉日周期、周周期这种长距离依赖比LSTM从容得多。更实际的好处是并行——LSTM必须一步步算，Transformer一次性算完整段序列的表示。这两点就是整套transformer架构在时序任务上立足的根本理由。

但代价也明显：Transformer对位置天然不敏感，输入元素任意换序，注意力结果几乎不变，所以必须额外加位置编码；在小样本场景下，它的参数量和训练难度反而比LSTM高，容易过拟合。这也是为什么这类源码里通常会配一个LSTM基线做对比实验，用数值说话。对比表如下：

2.2 位置编码与注意力掩码：源码里的关键组件

理解transformer代码，第一个要过的是位置编码。说“Transformer架构没有顺序概念”不是夸张——把输入的第3和第50个位置对调，模型输出的差异只来自输入本身。所以必须把“这是第几个时刻”变成一个向量加进嵌入里。最常用的是正弦位置编码：把位置映射成不同频率的sin/cos组合，模型既知道绝对位置，也能推断相对位置差。

import torch import math class PositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int = 5000): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float32).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) if d_model % 2 == 0: pe[:, 1::2] = torch.cos(position * div_term) else: # d_model为奇数时cos列少一列，用[:-1]对齐维度 pe[:, 1::2] = torch.cos(position * div_term[:-1]) self.register_buffer('pe', pe) def forward(self, x: torch.Tensor) -> torch.Tensor: # x: (B, L, d_model)，L为序列长度 return x + self.pe[:x.size(1)]

逻辑说明：div_term是对应sin/cos频率的指数衰减因子，d_model维向量里前半段频率高、对应相邻时刻的关系，后半段频率低、对应远距离关系。register_buffer把pe注册成模型持久缓冲区，不参与梯度更新但会随模型一起保存。改造源码时如果换了更长的序列，max_len填太小会直接在切片处越界报错，建议保持5000不动。

位置编码之后是注意力掩码。如果源码用的是encoder-decoder结构做逐步预测，解码器里第t个位置的输出不能看到未来时刻的真实值，否则训练时等于抄答案。实现方式是生成一个上三角全为负无穷的矩阵，加到注意力分数上，softmax之后这些位置的权重归零。

def generate_square_subsequent_mask(sz: int) -> torch.Tensor: """生成因果掩码：sz为解码序列长度。True的位置代表“不允许看”。""" mask = torch.triu(torch.ones(sz, sz), diagonal=1).bool() return mask # 用法：在计算注意力分数后 # scores = scores.masked_fill(mask, float('-inf'))

逻辑说明：triu生成上三角为True的布尔矩阵，第i行第j列为True表示i<j，也就是“后面的位置”。masked_fill把这些位置填成负无穷，softmax后权重为0，第t步只能看到0到t-1步的信息。很多老代码用“上三角乘-1e9”实现同样效果。不少简化版高分项目其实只用encoder，把整段历史丢进去再接输出层，那种结构不需要mask，但理解这段对读全尺寸代码仍然重要。

2.3 从编码器输出到预测值：任务头怎么接

模型主体是若干层TransformerEncoderLayer，输入和输出形状一致。这类源码里最常见的接法：输入(B, L, in_features)经过线性嵌入到d_model维度，加位置编码，过N层编码器，取最后一个时间步的向量，再接全连接层输出pred_len个数。

class TimeSeriesTransformer(nn.Module): def __init__(self, in_features: int = 1, d_model: int = 64, nhead: int = 4, num_layers: int = 2, pred_len: int = 12, dropout: float = 0.1): super().__init__() self.input_proj = nn.Linear(in_features, d_model) self.pos_encoder = PositionalEncoding(d_model) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=d_model * 4, dropout=dropout, batch_first=True # 输入按 (B, L, E) 而不是 (L, B, E) ) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.head = nn.Sequential( nn.Linear(d_model, d_model // 2), nn.ReLU(), nn.Linear(d_model // 2, pred_len), ) def forward(self, x: torch.Tensor) -> torch.Tensor: # x: (B, L, in_features)，B是批大小，L是输入序列长度 x = self.input_proj(x) # (B, L, d_model) x = self.pos_encoder(x) # 加位置编码 x = self.encoder(x) # 多层自注意力，形状不变 return self.head(x[:, -1, :]) # (B, pred_len)

逻辑说明：batch_first=True是PyTorch后来推荐的参数，让输入形状符合(B, L, E)这种直觉写法，避免在序列维和批量维之间反复transpose。取x[:, -1, :]的意思是“用最后一个历史时刻的聚合表示”去预测未来一整段，这是最简单也最稳定的接法。想更强一点，可以对所有时间步做全局平均池化再接head，能多利用一点历史信息。head里套一个隐藏层，是为了给“历史→未来曲线”这段映射增加非线性，比单层线性稳。

3. 跑通源码：从环境配置到第一个预测结果

3.1 环境版本与依赖安装

拿到一份源码包，第一件事不是直接python main.py，而是先看它依赖哪个PyTorch版本。Transformer项目最怕环境版本不对，尤其TransformerEncoderLayer的接口在PyTorch 1.7到2.x之间变过几次，旧代码要求输入形状是(L, B, E)，新代码加了batch_first参数，版本不对轻则告警重则直接跑不起来。先建一个干净环境是最稳的做法。

conda create -n ts_transformer python=3.9 -y conda activate ts_transformer # 无NVIDIA显卡或只想先跑通，先装CPU版，确定模型能forward再换GPU版 pip install torch --index-url https://download.pytorch.org/whl/cpu pip install numpy pandas matplotlib scikit-learn python -c "import torch; print(torch.__version__, torch.cuda.is_available())"

参数说明：python=3.9是兼容性最稳的选择，3.10以上部分老代码里的numpy接口会告警；torch先装CPU版可以把CUDA问题排到后面去，如果源码里有informer、sktime这类辅助库，用pip install -r requirements.txt按需安装，但建议逐条装，避免自动解析依赖时把torch版本顶掉。

如果你是用VSCode写python，记得把解释器切到ts_transformer这个conda环境。终端里装好的包在编辑器里import不到，多半是解释器路径不一致，这是python环境配置里最常出现的ModuleNotFoundError来源，和源码本身没关系。

3.2 数据格式、滑动窗口与数据加载器改造

这类源码的数据一般是一个CSV：第一列是时间，后面若干列是特征。时间列主要用来画图，模型本身不读时间，只读数值矩阵。先把数据读进来，按“滑动窗口”切成样本对。每个样本是两段：输入窗口和标签窗口。

import pandas as pd import numpy as np def create_sequences(data: np.ndarray, seq_len: int = 48, pred_len: int = 12): """把时间序列切成(x, y)样本对。 data: (N, features)的数值矩阵 返回 xs: (样本数, seq_len, features)，ys: (样本数, pred_len) """ xs, ys = [], [] for i in range(len(data) - seq_len - pred_len + 1): xs.append(data[i:i + seq_len]) ys.append(data[i + seq_len:i + seq_len + pred_len][:, 0]) # 只预测第0列 return np.array(xs), np.array(ys) raw = pd.read_csv('dataset.csv') cols = ['load', 'temp', 'humidity'] # 按你的CSV表头替换 values = raw[cols].values.astype(np.float32) xs, ys = create_sequences(values, seq_len=48, pred_len=12) print(xs.shape, ys.shape) # 例如 (8740, 48, 3) 和 (8740, 12)

逻辑说明：切片范围用len(data) - seq_len - pred_len + 1，保证第i个样本的输入窗口[i, i+seq_len)和标签窗口[i+seq_len, i+seq_len+pred_len)都完整落在数据内。ys只取[:, 0]表示预测第一个特征列的未来值，这是“多变量输入、单变量预测”的常见设定；如果源码要求多变量输出，把ys改成保存二维切片即可。

注意：切分训练集和测试集必须先于切窗口，且scaler只在训练集上fit。很多课程设计项目在这里埋了数据泄漏的雷，后面避坑章节会展开。

然后包成PyTorch标准的Dataset和DataLoader：

from torch.utils.data import Dataset, DataLoader class TimeSeriesDataset(Dataset): def __init__(self, xs: np.ndarray, ys: np.ndarray): self.xs = torch.FloatTensor(xs) self.ys = torch.FloatTensor(ys) def __len__(self): return len(self.xs) def __getitem__(self, idx): return self.xs[idx], self.ys[idx] train_dataset = TimeSeriesDataset(xs_train, ys_train) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, drop_last=True)

参数说明：训练loader的shuffle=True可以让每个batch来自不同时间段的片段，提升稳定性；验证集loader要shuffle=False，确保评估顺序和真实时间一致。drop_last=True避免最后一个batch形状不齐，对Transformer这种固定形状输入的模型省去不少边界判断。

3.3 训练命令与评估指标

模型实例化、损失函数、优化器定下来就能跑。训练循环本身不大，重点在两个地方：梯度裁剪防止梯度爆炸，以及每几个epoch打印一次训练损失和验证损失，方便判断是欠拟合还是过拟合。

model = TimeSeriesTransformer(in_features=3, d_model=64, nhead=4, num_layers=2, pred_len=12) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) criterion = torch.nn.MSELoss() for epoch in range(50): model.train() epoch_loss = 0.0 for x, y in train_loader: optimizer.zero_grad() pred = model(x) # (B, pred_len) loss = criterion(pred, y) # y: (B, pred_len) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) # 防梯度爆炸 optimizer.step() epoch_loss += loss.item() if (epoch + 1) % 5 == 0: val_loss = evaluate(model, val_loader, criterion) print(f"Epoch {epoch+1:3d} train_loss={epoch_loss/len(train_loader):.5f} " f"val_loss={val_loss:.5f}")

评估时除了MSE，还要看一个百分比指标：

def evaluate(model, loader, criterion): model.eval() total = 0.0 with torch.no_grad(): for x, y in loader: pred = model(x) total += criterion(pred, y).item() * len(x) return total / len(loader.dataset) def mape(y_true: np.ndarray, y_pred: np.ndarray) -> float: """平均绝对百分比误差，越小越好，用于向业务方解释模型效果。""" eps = 1e-8 return float(np.mean(np.abs((y_true - y_pred) / (y_true + eps))) * 100)

注意顺序：必须先把预测值和真实值反归一化回原始量纲，再算MAPE。直接对标准化后的值算MAPE没有现实意义。源码里的metric.py做的事基本就是这一套，唯一要小心的是MAPE在真实值接近0的序列上会爆炸，那种场景改用SMAPE或直接看MAE。

4. Transformer时间序列预测的调优：五个必调参数

4.1 输入序列长度与预测步长：窗口怎么定

seq_len是第一个要动的参数。经验法则是：输入长度至少覆盖预测长度的2到4倍，最好覆盖一个完整周期。以电力负荷为例，日周期24小时，seq_len最好不小于48，这样能看到两个完整日周期；如果数据有时段规律，取96或168更稳。但seq_len不是越长越好——注意力矩阵按序列长度平方增长，训练速度和显存都变差，还可能把注意力分散到无关历史噪声上。

pred_len直接决定任务难度。预测未来1点很容易，预测未来24点误差会随步长累积。这类项目评分时通常分别报告短步长和长步长，比如pred_len=12和pred_len=24。实操里不要一上来挑战最长步长，先把短步长跑通，确认模型结构没问题，再加长。

4.2 d_model与nhead：模型容量分配

d_model是每个时间步被编码成的向量长度。64是能出成果的起点，数据量大、特征多时可以调128。16太小学习能力不足，256以上在小数据集上很容易过拟合。nhead是注意力头数，每个头在d_model里取d_model/nhead维的子空间，去关注不同角度，比如一个头关注周期性、一个头关注突变点。经验上取4或8，且d_model能被nhead整除。

一个反直觉的点：nhead不是越多越好。头过多会把d_model切得很碎，每个头的表示能力变弱，训练时间却线性增加。建议让它和d_model配套调：d_model=64配nhead=4，d_model=128配nhead=8，通常最稳。这两个参数改动前先跑一次基线，之后一次只动一个，否则分不清是哪个参数生效。

4.3 学习率与Dropout：Transformer的敏感地带

Transformer对学习率的敏感度比LSTM高很多，直接拿默认lr=1e-3训练，loss很容易变成nan或高位震荡。常见做法是Adam加warmup加线性衰减，warmup阶段让学习率从0爬升到峰值，让模型在“小步幅”下先稳定内部统计量，再大步调整。

def warmup_linear_schedule(step, d_model=64, warmup_steps=2000): """warmup + 线性衰减学习率缩放系数，出自Attention is All You Need原论文。""" step = max(step, 1) if step < warmup_steps: return float(step) / float(warmup_steps) decay = float(warmup_steps) ** 1.5 / float(step) ** 0.5 return decay scheduler = torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda=warmup_linear_schedule) # 训练循环里每个batch之后调用 scheduler.step()，而不是每个epoch

逻辑说明：lambda函数返回当前学习率相对初始学习率的缩放系数。warmup阶段比率从0线性涨到1，之后按step的平方根倒数衰减。很多源码会简化成“前10个epoch用固定小学习率”，效果接近。如果不想动调度器，把初始学习率固定在1e-4到3e-4之间，也比默认的1e-3稳得多。

Dropout建议在0.1到0.3之间。序列短、数据量大时0.1足够；小样本防过拟合可以调到0.2或0.3，超过0.4会让模型欠拟合，连训练集都拟合不动。dim_feedforward一般取d_model×4，这个参数很少动，动错了容易直接拉爆显存。

4.4 批量大小与梯度累积：显存与稳定性的平衡

batch_size影响两个东西：梯度估计的稳定性和显存占用。batch越大梯度噪声越小，但显存线性上涨。单卡12GB跑d_model=64、seq_len=168时，batch_size通常放到32到64就到头了。显存不够又不想降batch时，用梯度累积等效一个大batch：

accumulation_steps = 4 optimizer.zero_grad() for step, (x, y) in enumerate(train_loader): loss = criterion(model(x), y) / accumulation_steps loss.backward() if (step + 1) % accumulation_steps == 0: optimizer.step() optimizer.zero_grad()

逻辑说明：loss先除以accumulation_steps，累积4个batch的梯度之和就等于直接用一个4倍大小batch算出的梯度。这样在单卡上也能跑出大batch的效果，代价是总训练时间不变，只是峰值显存降下来了。注意梯度裁剪要在optimizer.step()之前做，放在循环里会影响累积的梯度。

把这几个参数放到一起看，推荐的调整顺序是：先定seq_len和pred_len，再定d_model与nhead，最后调学习率、dropout、batch_size。不要一次改三四个，否则训练曲线一乱就无从排查。

5. 避坑：Transformer时序预测的常见问题与排查

5.1 数据泄漏（Data Leakage）

现象：验证集和测试集指标非常好，测试集MAPE能到个位数，但一换新数据预测能力断崖下降。

原因：最常见的是对整段序列先做MinMaxScaler再切分训练/测试，scaler在归一化时已经看到了测试段的全局最大最小值，测试集的分布信息提前流进了训练过程。第二个来源是滑动窗口跨过了训练测试边界，某个样本先用到了训练段尾部的真实值，又用它去预测测试段开头，边界处互相咬合。

解决：严格按时间顺序切分后再fit scaler，或只在训练集上fit，测试集用同一套参数transform。我的习惯是写一个类把“切分”和“归一化”绑在一起，保证调用顺序不被打乱。

from sklearn.preprocessing import StandardScaler split_idx = int(len(df) * 0.8) train_df, test_df = df.iloc[:split_idx], df.iloc[split_idx:] # 先按时间切 scaler = StandardScaler() train_values = scaler.fit_transform(train_df.values) # 只在训练集上fit test_values = scaler.transform(test_df.values) # 测试集用同一套参数

5.2 训练不收敛或损失震荡

现象：训练几轮后loss变成nan，或者一直在高位震荡不下降。

原因：最常见是输入数值没归一化，数量级相差太大，梯度一乘就爆炸。其次是学习率太高，Transformer对学习率敏感，默认1e-3对深层模型偏大。还有可能是数据里有NaN或Inf，预处理没清干净。

解决：先打印训练集的数值范围确认没有NaN和Inf；用StandardScaler把输入标准化到均值0方差1附近；学习率降到1e-4或加warmup。如果显存和内存都够，先把num_layers和batch_size调小，确认能快速出一个结果，再逐步放大配置。“先小规模跑通、再大规模调参”可以避免在配置错误上浪费排查时间。

5.3 预测结果滞后或变成水平线

现象：预测曲线和真实曲线形状相似，但整体向右平移了一段；或者预测值几乎落在某个固定值附近，完全看不到波动。

原因：平移通常是因为输入输出高度自相关，模型学到的近似是“下一个值约等于上一个值”，尤其是差值很小、噪声很低的序列。水平线则常见于归一化后标签方差太小，模型把输出压到均值上；也可能是dropout太大或head过深，把信号都抹平了。

解决：先对原序列做一阶差分（differencing）再喂模型，让模型学“增量”而不是“绝对值”，能去掉大部分自相关。或者把损失从纯MSE改成MSE加0.1倍MAE，用MAE惩罚偏差。对有周期性的数据，把“同时刻的历史值”作为额外特征拼进输入，也能有效消除滞后。

注意：差分后的预测还原到原始量纲时，要加回差分前的最后一个真实值，即pred = pred_diff + last_actual。漏掉这一步，画出来的曲线整体错位一大截。

5.4 反归一化后误差突然变大

现象：训练时loss很稳，inverse_transform之后预测范围明显超出或低于真实值，曲线尾部翘上天。

原因：MinMaxScaler在训练集上fit的范围是[训练min, 训练max]，测试阶段一旦出现超出这个范围的数值，归一化后落到0~1之外，还原回去就是异常点。MinMax对长尾分布尤其敏感，一些极端值会把反归一化结果整个带偏。

解决：对这种场景直接用StandardScaler，反归一化只依赖均值和标准差，不依赖“是否超出训练范围”。如果必须用MinMax，反归一化后加一个裁剪np.clip(pred, train_min, train_max)，虽然会牺牲极端值精度，但曲线不会飞掉。

5.5 PyTorch版本不兼容或显存不足

现象：运行报错“Expected input to have 3 dimensions, but got 2 dimensions”，或“CUDA out of memory”，换个机器预测结果对不上。

原因：第一个报错通常是输入形状不对，模型期望(B, L, features)，数据加载器给成了(B, L)。第二个是batch_size与seq_len、d_model的组合超出显存。torch版本不一致则会导致TransformerEncoderLayer的接口签名差异，老代码在1.7能跑，新代码在2.x报错。

解决：先做最小化验证——构造一个batch_size=1的假数据tensor跑一次forward，确认形状流程没毛病再上训练。显存不足优先减小batch_size并配合梯度累积，再不行把num_layers从2降到1。版本问题直接看Traceback，定位到TransformerEncoderLayer就用batch_first=True显式指定；旧代码不支持该参数就升级到新版再改。

这些坑不管是自己写还是改造别人源码都会轮流来一遍。数据泄漏和训练不收敛高频出现，差分还原和反归一化则是那种“平时不遇到、遇到就懵”的隐蔽问题。以“现象→原因→解决”的顺序排查，比盲目改参数试错快得多。

6. 把源码改造成自己的数据集：接口改造与效果验证

6.1 从单变量到多变量：Embedding层怎么动

本文所有模型代码里，第一层都是input_proj线性嵌入，默认in_features=1是单变量。换成多变量只动两个地方：初始化时改in_features，切窗口时把标签列换成你要预测的那一列。

model = TimeSeriesTransformer(in_features=3, d_model=64, nhead=4, num_layers=2, pred_len=12) ys = data[i + seq_len:i + seq_len + pred_len][:, 0] # 标签仍然只取“负荷”列

代码改动就这么多，真正的成本在数据侧：多特征之间量纲不同，必须统一缩放；特征和目标相关性太低的要先剔除，否则模型学不到有效融合。很多人把股票因子、成交量和涨跌幅拼成多变量特征做走势预测，和python量化交易策略代码里常见的预处理套路完全一致——本质都是这套输入输出接口的适配。

6.2 滚动验证：比单次划分更可信

单次切分只能证明模型在一个时间段的表现。更可信的做法是滚动验证：把数据按时间平移切成多个窗口，反复评估并汇总。

def rolling_validate(model, data, seq_len=48, pred_len=12, step=24): preds, trues = [], [] for start in range(0, len(data) - seq_len - pred_len + 1, step): pred = model.predict(data[start:start + seq_len]) preds.append(pred) trues.append(data[start + seq_len:start + seq_len + pred_len][:, 0]) return mape(np.concatenate(trues), np.concatenate(preds))

step控制窗口滑动的间隔，间隔越密覆盖的时间段越全，当然计算量也越大。如果单次测试集指标好看但滚动验证泛不起来，先回去查数据预处理里还有没有泄漏。最后保留一个习惯：每次训练完把预测曲线和真实曲线叠加画出来，看是滞后、超前还是相位偏差——图和指标一起看才可信。希望帮到你。

本文还有配套的精品资源，点击获取