TCN-LSTM-多头注意力多变量时间序列预测完整实现
2026/9/17 13:33:49 网站建设 项目流程

简介:基于TCN-LSTM-Multihead-Attention的多变量时间序列预测项目实例,面向金融分析、气象预报、智能交通、能源管理等领域、具备一定深度学习基础的研究人员与技术开发者。资源聚焦高维、高复杂度、长时间依赖及噪声干扰等预测难点,通过融合时间卷积网络、长短期记忆网络与多头自注意力机制,在捕捉局部时序模式的同时强化关键特征,有效提升预测精度与稳定性。资源包为docx说明文档,共1个文件,约81KB,包含完整Python程序、GUI设计思路与逐段代码注释,系统讲解了环境搭建、数据处理、模型构建、性能评估,以及项目结构设计、训练超参数优化和未来改进方向,并覆盖项目背景、创新点与挑战解决方案,便于深入理解模型设计逻辑。已有59人学习或下载,适合需要完整项目参考、代码对照及进一步算法扩展的深度学习研究者和工程人员。

1. 为什么多变量预测要叠加 TCN、LSTM 与多头注意力

做过多变量时间序列预测的人大多遇到过这种局面:单用 LSTM 处理长序列,梯度容易衰减,趋势学得进去、局部突变跟不住;换成纯卷积网络,训练快了,长期依赖又难建模。

TCN-LSTM-Multihead-Attention 把三类结构按分工拼接:时间卷积网络用因果空洞卷积抓局部形态,LSTM 在卷积之后压缩长期状态,多头注意力对多变量和时间步之间的依赖做显式加权,适合电力负荷、交通流量、设备监测这类多输入场景。

下文按工程落地顺序展开:先拆模型结构与张量形状,再给基于 PyTorch 的可运行 Python 代码,然后是超参数整定与评估,最后用 Tkinter 封装预测 GUI,滑动窗口、归一化、滚动预测都会覆盖到。

2. 模型结构拆解:TCN 感受野、LSTM 状态与多头注意力的接入位置

2.1 为什么先 TCN 后 LSTM:谁负责局部特征,谁负责长期记忆

TCN 的全称是 Temporal Convolutional Network,核心是因果卷积。因果卷积保证输出第 t 个位置只能看到输入第 t 个位置及更早的信息,不会把未来数据泄露进当前预测,这和标准 Conv1d 有本质区别。实现上的关键是控制 padding 和裁剪:卷积核需要往前看多少步,就往左边补多少 padding,再把卷积结果右侧多出来的部分剪掉。

空洞卷积(dilation)用来扩大感受野。TCN 的常规做法是把 dilation 设成 1, 2, 4, 8 的指数序列,配合 kernel_size=3,每一层感受野增加 (kernel_size-1) * dilation。序列越长,需要的 dilation 层数越多。感受野必须覆盖滑动窗口的完整长度,否则窗口尾部的信息根本传不到最终输出,验证损失会一开始就卡住不动。

那为什么把 LSTM 放在 TCN 后面?常见做法是让 TCN 先把每个时刻的多变量特征压缩成固定维度的表征,LSTM 再在这个表征序列上继续建模跨窗口的中长期依赖。如果把顺序反过来,LSTM 在每个时间步先做一次门控压缩,TCN 的卷积核再作用在这些状态上,残差连接带来的梯度优势会被削弱,训练也更难收敛。

另外,LSTM 是串行展开的,TCN 是并行计算的。TCN 先跑可以把建模压力从时间步上卸掉一部分:卷积在 GPU 上比循环展开快得多,窗口越长,先 TCN 后 LSTM 的训练速度优势越明显。至于多头注意力放哪里,有两种常见选择:作用在 LSTM 输出的时间轴上做自注意力,或者把变量维当作序列长度做跨特征注意力。多变量预测里变量数一般只有几个到几十个,跨特征注意力参数量不值当,主流做法是时间步自注意力。

2.2 多头注意力接入 LSTM 输出:形状与计算流程

Multihead-Attention 把输入同时作为 Q、K、V 使用,自注意力的含义是:对每个时间步,计算它与窗口内其他时间步的相关性,再按相关性加权聚合。多头把 hidden_size 切成 n_heads 份,每个头在低维子空间里学一种依赖模式,有的头负责邻近步的突变,有的头负责跨周期的相似性。切分要求 hidden_size 能被 n_heads 整除,否则 PyTorch 直接报错。

import torch B, T, F, C, H = 32, 48, 8, 64, 32 x = torch.randn(B, T, F) # TCN 示意:Conv1d 输入输出都按 (B, C, L) 处理,先转通道维 x = torch.nn.Conv1d(F, C, 3, padding=1)(x.transpose(1, 2)).transpose(1, 2) lstm = torch.nn.LSTM(C, H, batch_first=True) x, _ = lstm(x) # (B, T, H) attn = torch.nn.MultiheadAttention(H, 4, batch_first=True) h, w = attn(x, x, x) # w 是注意力权重,形状 (B, T, T) print(h.shape) # torch.Size([32, 48, 32])

这段代码把张量形状完整走了一遍。x 从 (B, T, F) 进入 Conv1d 前必须转成 (B, C, T),出来再转回时间维在前;LSTM 用 batch_first=True 直接吃 (B, T, C);MultiheadAttention 同样开 batch_first=True,输入输出都保持 (B, T, H)。w 里返回的注意力权重形状是 (B, T, T),T 等于滑动窗口长度,窗口开到 96 时权重矩阵就是 96×96,想查看哪些时间步被重点加权,先对所有头取平均再沿 batch 维池化。

2.3 完整结构表:每一层输入输出对着看

网络层输出形状做的事情
输入数据(B, T, F)滑动窗口切出来的多变量样本,F 是特征数
TCN 残差块(B, T, C)因果空洞卷积,保持时间长度不变,提取局部模式
LSTM(B, T, H)每个时间步输出隐藏状态,在时间轴上压缩全窗口信息
Multihead-Attention(B, T, H)时间步之间自注意力,按相关性加权聚合
取最后时间步 + 全连接(B, horizon)映射到未来 horizon 步的预测值

这张表是调试模型的第一参考。任何一个环节的输出形状和表里对不上,后面全连接层的维度就会报 mismatch。TCN 保持时间长度不变这一点容易被忽略:如果某个残差块里第一层卷积没有裁剪右侧 padding,输出会比输入长,LSTM 会静默接受这个错误长度,导致注意力矩阵的维度和你心里想的不一致。

2.4 组合模型为什么比单模型更稳,边界在哪里

三个结构解决的问题不重叠:TCN 提供固定感受野和残差路径,梯度可以沿残差直通;LSTM 把序列信息压缩进记忆单元,补足卷积覆盖不到的周期性依赖;多头注意力让最终输出不依赖某一个固定时间步,而是按相关性动态加权。三者叠加后,验证损失通常比纯 LSTM 有可感知的下降,收敛也更早。

边界同样明显。样本量只有几百条、变量之间基本线性的时候,这种三层结构很容易过拟合,此时先把 tcn_channels 和 hidden_size 调小,甚至退回单层 LSTM。注意力头数也不是越多越好,hidden_size 只有 32 时开 8 个头,每个头分到 4 维,学不到有意义的相关性,常见的稳妥配置是 hidden_size 32 配 4 头、hidden_size 64 配 8 头。

3. 从滑动窗口到训练循环:TCN-LSTM-Multihead-Attention 的完整 Python 实现

3.1 数据准备:滑动窗口、列归一化与按时间顺序切分

多变量序列的原始数据一般是 (N, F) 的二维数组,N 是时间点,F 是变量数。滑动窗口切分的规则是:用过去 window 个时间点预测未来 horizon 步。样本之间高度重叠,所以训练集和验证集必须按时间先后切,不能随机切,否则验证集里混着训练集相邻时刻的信息,评估结果会虚高。

import numpy as np def make_samples(data, window=48, horizon=1, target_col=0): """把 (N, F) 的多变量序列切成 (X, y) 样本对。 X 形状 (样本数, window, F),y 形状 (样本数, horizon)。 data[i + window] 是第 i 个样本要预测的起始位置。""" X, y = [], [] for i in range(len(data) - window - horizon + 1): X.append(data[i:i + window]) y.append(data[i + window:i + window + horizon, target_col]) return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)

target_col 指定预测哪一列,其他列只作为输入特征参与预测。horizon 大于 1 时 y 是 (样本数, horizon),损失函数对 horizon 个未来步同时求平均,每步直接输出,误差不会被递归放大,属于直接多步预测(direct multi-step),比滚动多步更容易训练。窗口滑动步长默认是 1,如果数据采样频率太高,也可以改成 stride=10 抽样,样本量会下降,训练速度更快,但时间分辨率变粗。

3.2 归一化边界:scaler 只 fit 训练段

from sklearn.preprocessing import MinMaxScaler split = int(len(raw) * 0.8) scaler = MinMaxScaler() train_part = scaler.fit_transform(raw[:split]) # 只学训练段的 min/max val_part = scaler.transform(raw[split:]) # 验证段沿用训练段的 min/max

这就是常说的归一化泄露。如果对整段数据先 fit 再切,验证段的取值范围已经进入了训练阶段,测试误差会被系统性低估。MinMaxScaler 逐列算 min/max,各列量纲不同也不影响。预测目标列的极端峰值建议保留,不要顺手当异常点删掉,TCN 的残差结构对这类突变并不敏感;删掉反而会让模型学不到真实分布。

3.3 TCN 模块:因果卷积与残差块的实现

TCN 的基础模块由两层因果卷积加一个残差连接组成。因果卷积的关键是右侧裁剪,只保留历史信息。dilation 传入 1、2、4、8 的序列值,每个模块负责一个尺度。

import torch import torch.nn as nn import torch.nn.functional as F class CausalConv1d(nn.Conv1d): def __init__(self, in_c, out_c, kernel_size, dilation=1): padding = (kernel_size - 1) * dilation super().__init__(in_c, out_c, kernel_size, padding=padding, dilation=dilation) def forward(self, x): # 卷积输出比输入长 (kernel_size-1)*dilation,右侧全部剪掉 return super().forward(x)[:, :, :-(self.kernel_size - 1) * self.dilation] class TCNBlock(nn.Module): def __init__(self, in_c, out_c, kernel_size=3, dilation=1, dropout=0.2): super().__init__() self.conv1 = CausalConv1d(in_c, out_c, kernel_size, dilation) self.conv2 = CausalConv1d(out_c, out_c, kernel_size, dilation) self.drop = nn.Dropout(dropout) self.relu = nn.ReLU() self.res = nn.Conv1d(in_c, out_c, 1) if in_c != out_c else nn.Identity() def forward(self, x): h = self.relu(self.conv1(x)) h = self.drop(h) h = self.relu(self.conv2(h)) return self.relu(h + self.res(x)) # 残差让梯度直通

CausalConv1d 里 padding 的位置放在左边,PyTorch 的 Conv1d 是两侧对称补零,所以输出长度会多出 (kernel_size-1)*dilation,forward 里从右侧裁掉。res 分支在输入输出通道不一致时用 1x1 卷积对齐通道数,通道一致时走 Identity,不引入额外参数。感受野计算方法是各层 (kernel_size-1)*dilation 求和再加 1,三层 dilation 1、2、4、kernel_size=3 时感受野是 1+2+4+1 之外的累计:每层增加 2、4、8,总计 15 加上初始 1,即覆盖 16 个时间步,窗口超过这个长度就要再加 dilation 层。

3.4 完整模型:LSTM 与 Multihead-Attention 的拼接

class TCNLSTMAttention(nn.Module): def __init__(self, n_features, tcn_channels=64, n_heads=4, hidden_size=32, n_lstm_layers=1, horizon=1, dropout=0.2): super().__init__() self.tcn = TCNBlock(n_features, tcn_channels, dropout=dropout) self.lstm = nn.LSTM(tcn_channels, hidden_size, num_layers=n_lstm_layers, batch_first=True, dropout=dropout if n_lstm_layers > 1 else 0) self.attn = nn.MultiheadAttention(hidden_size, n_heads, batch_first=True, dropout=dropout) self.norm = nn.LayerNorm(hidden_size) self.fc = nn.Linear(hidden_size, horizon) def forward(self, x): # 输入 (B, T, F),TCN 需要 (B, C, T),卷积完再换回时间维在前 x = self.tcn(x.transpose(1, 2)).transpose(1, 2) # (B, T, C) x, _ = self.lstm(x) # (B, T, H) h, _ = self.attn(x, x, x) # 时间步自注意力 h = self.norm(h + x) # 残差加 LayerNorm return self.fc(h[:, -1, :]) # 取最后时间步

forward 里三个容易错的地方:transpose 之后 TCN 的残差连接始终在通道维上操作,不会破坏时间顺序;nn.MultiheadAttention 的 dropout 只在训练时生效,推理时自动关闭;h[:, -1, :] 取的是注意力加权后的最后时间步,而不是原始 LSTM 输出的最后时间步,这两者的区别是注意力是否真正参与了预测。LayerNorm 加在注意力输出和输入的残差之后,维度对齐隐藏层尺寸。

3.5 训练主循环:梯度裁剪与每轮损失记录

from torch.utils.data import TensorDataset, DataLoader import torch.nn as nn X_train, y_train = make_samples(train_part, window=48, horizon=1) X_val, y_val = make_samples(val_part, window=48, horizon=1) train_ds = TensorDataset(torch.from_numpy(X_train), torch.from_numpy(y_train)) train_loader = DataLoader(train_ds, batch_size=64, shuffle=True) val_ds = TensorDataset(torch.from_numpy(X_val), torch.from_numpy(y_val)) val_loader = DataLoader(val_ds, batch_size=64, shuffle=False) # 验证集保持时间序 model = TCNLSTMAttention(n_features=raw.shape[1], tcn_channels=64, hidden_size=32, n_heads=4, horizon=1) opt = torch.optim.Adam(model.parameters(), lr=1e-3) loss_fn = nn.MSELoss() train_losses, val_losses = [], [] for epoch in range(80): model.train() total = 0.0 for xb, yb in train_loader: opt.zero_grad() loss = loss_fn(model(xb), yb) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 防梯度爆炸 opt.step() total += loss.item() * xb.size(0) train_losses.append(total / len(train_ds)) model.eval() vtotal = 0.0 with torch.no_grad(): for xb, yb in val_loader: vtotal += loss_fn(model(xb), yb).item() * xb.size(0) val_losses.append(vtotal / len(val_ds)) if epoch % 10 == 0: print(f"epoch {epoch:3d} train {train_losses[-1]:.5f} val {val_losses[-1]:.5f}")

训练集 shuffle=True 是为了打散相邻滑动窗口的强相关性;验证集保持时间顺序,这样才能看出模型在哪个时间段开始偏离。clip_grad_norm_ 的 max_norm=1.0 是处理长窗口下 LSTM 梯度暴涨的常规手段,损失在某轮跳到 nan 时先查它。80 轮只是基线,实际以验证损失不再下降为准,配合第 4 章的早停逻辑截断。

4. 超参数整定与评估:多变量预测该盯住哪几个指标

4.1 一张参数表照着调:取值区间与失效信号

参数建议范围失效时的特征
window24~96太小则预测滞后明显;太大训练变慢
dilation 序列1,2,4,8 或加到 16感受野不足时验证损失不下降
tcn_channels32~128过大且数据少时训练集低、验证集高
hidden_size16~64需能被 n_heads 整除,否则直接报错
n_heads2~8超过 hidden_size // 4 后收益下降
dropout0.1~0.4验证损失震荡时调大
batch_size32~128太小损失震荡频繁
lr1e-4~3e-3过大发散,过小不收敛

window 是第一优先调的参数。它决定了模型能看到多长的历史,直接和采样频率挂钩:按小时采样的电力负荷,48 到 72 能覆盖日周期;按秒采样的设备振动数据,可能需要 200 以上。dilation 序列要保证感受野大于 window,否则调其他参数都没意义。一个快速验证感受野是否够用的办法:把训练轮数压到 5 轮,如果验证损失完全不随轮数下降,先加 dilation 层而不是调学习率。

4.2 评估必须落在原始量纲:MAE、RMSE、MAPE、R2 的计算

import numpy as np def evaluate(y_true, y_pred): mae = float(np.mean(np.abs(y_true - y_pred))) rmse = float(np.sqrt(np.mean((y_true - y_pred) ** 2))) mape = float(np.mean(np.abs((y_true - y_pred) / (y_true + 1e-8)))) * 100 ss_res = float(np.sum((y_true - y_pred) ** 2)) ss_tot = float(np.sum((y_true - np.mean(y_true)) ** 2)) r2 = 1 - ss_res / ss_tot return {"MAE": mae, "RMSE": rmse, "MAPE": mape, "R2": r2}

这四个指标必须在反归一化之后计算。训练损失用的是归一化区间的 MSE,量纲和业务对不上,直接对外报这个数没有意义。MAPE 分母加 1e-8 是为了避免目标值恰为 0 时除零;目标变量含负值(比如风速)时 MAPE 会严重失真,改用对称 MAPE 更合适。R2 接近 1 不等于模型好,多变量序列存在强自相关,基准对比建议用「拿上一个时刻值当预测」的朴素模型,R2 至少要显著高于这个基线才说明模型学到了跨变量的增量信息。

def inverse_single_col(pred_norm, scaler, n_features=8, target_col=0): """把单目标列的归一化预测映回原始量纲。 因为 scaler 是按整行多列 fit 的,单列反归一化要先补零占位。""" tmp = np.zeros((pred_norm.shape[0], n_features)) tmp[:, target_col] = pred_norm return scaler.inverse_transform(tmp)[:, target_col]

inverse_single_col 解决的问题是:MinMaxScaler 做的是行级变换,直接拿一列数据调 inverse_transform 会得到错误形状。补零占位只影响非目标列,目标列的量纲还原是准确的;如果想让多列一起反归一化,把预测值放回对应列位置即可。

4.3 损失曲线怎么读:过拟合、欠拟合与学习率问题的三种形态

import matplotlib.pyplot as plt plt.plot(train_losses, label="train") plt.plot(val_losses, label="val") plt.ylim(bottom=0) plt.legend() plt.savefig("loss_curve.png", dpi=150)

训练日志里收集的 train_losses 和 val_losses 直接画出来,比看数值更有用。三种典型形态对应三种处理方式。

第一种,train 和 val 同步下降后走平,这是正常收敛,早停点选在 val 最低处。第二种,train 一路走低、val 先降后升,典型过拟合,先调大 dropout 到 0.3 以上,再考虑把 tcn_channels 或 hidden_size 减半,不要一上来就加数据。第三种,两条曲线都平坦不动,先确认感受野是否覆盖 window,再检查学习率是不是太小;学习率过大的特征是 loss 在某个值附近剧烈震荡,这时把 lr 降到 1e-4 重跑比加 batch_size 更有效。

4.4 三个必踩的坑:归一化泄露、窗口切分顺序、评估位置

归一化泄露,scaler 必须在切窗口之前只 fit 训练段,验证段和测试段只能 transform。代码顺序反了,指标会好看很多,但上线后立刻露馅。

窗口切分顺序,训练集内部 shuffle 没问题,但训练和验证之间必须按时间切。用 KFold 随机打乱会破坏时间序列的自相关性,模型等于见过邻居样本,评估失真。

评估位置,画图和计算指标都在反归一化之后做。另一个常见做法是把多步预测的每一步误差分别统计,画出 horizon 步的误差曲线,能直接看出第几步开始误差放大。

注意:早停只看验证损失,不要用测试集反复试探阈值。测试集只能在全部训练和早停结束后跑一次,否则测试集也在参与调参,指标没有说服力。

5. GUI 设计与模型落地:把预测结果封装进桌面界面

5.1 Tkinter + matplotlib 的最小预测界面

封装 GUI 的目的是让不写代码的人也能加载模型、看预测曲线。技术选型上,Tkinter 是 Python 内置库,不需要额外安装,配合 matplotlib 的 FigureCanvasTkAgg 可以直接把图表嵌进窗口,适合做单机工具。PyQt5 功能更强但依赖重,模型已经训练好、只想快速看结果的场景用 Tkinter 足够。

import tkinter as tk from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import matplotlib.pyplot as plt class ForecastApp(tk.Tk): def __init__(self): super().__init__() self.title("TCN-LSTM-Multihead-Attention 多变量预测") self.geometry("960x560") tk.Button(self, text="加载模型并预测", command=self.run_predict).pack(pady=6) self.fig = plt.Figure(figsize=(8.4, 3.8), dpi=100) self.ax = self.fig.add_subplot(111) self.canvas = FigureCanvasTkAgg(self.fig, master=self) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) def run_predict(self): # 实际逻辑:读最新窗口 -> 预测 -> 画真实值和预测值两条线 self.ax.clear() self.ax.plot(real_series, label="real") self.ax.plot(pred_series, label="pred") self.ax.legend() self.canvas.draw()

tk.Button 的 command 参数绑定预测函数,模型推理完成后调用 canvas.draw() 刷新图表。界面布局虽然简单,但骨架完整,后续加文件选择框、超参数输入框都是在 pack 之上叠加控件。

5.2 保存模型与滚动预测:用真实值更新窗口

模型和归一化参数必须一起保存,推理时缺了 scaler,预测值量纲就是错的。

torch.save({ "model": model.state_dict(), "scaler": scaler, "args": {"n_features": raw.shape[1], "window": 48} }, "tcn_lstm_attn.pt")

加载时先把参数字典取出来重建模型,再 load_state_dict,然后把 model.eval() 切到推理模式。推理模式会关掉 dropout,MultiheadAttention 的 dropout 也一并关闭,否则每次点击预测结果会有微小波动。

def rolling_next(model, recent_win): """recent_win: (window, F) 归一化后的最近窗口,返回下一步预测值""" xb = torch.from_numpy(recent_win)[None] # 加 batch 维 (1, window, F) with torch.no_grad(): return model(xb).squeeze().item()

滚动预测的关键策略:每来一条新数据,把窗口最早一行丢掉、新观测接到末尾,再预测下一步,也就是用真实值更新窗口。只有在真实值还没到的时候,才退而求其次把上一步预测值接进窗口,但此时误差会逐轮累积,所以界面上要同时画真实值和预测值两条线,偏差放大就是重新训练的信号。

5.3 固定随机种子与环境排查

import random import numpy as np import torch def set_seed(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed)

训练入口第一行调用 set_seed,否则每次跑出来的指标都不一样,超参数对比无从谈起。最后提醒一个细节:界面里的模型加载和预测不要放在 Tkinter 主线程里执行,用 threading.Thread 包一层,窗口才不会在推理期间假死;在线预测时每次只推一步,把新观测和模型输出同时画在图上,比一次画 50 步更容易看出模型在哪一步开始偏离真实值。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询