LSTM-Transformer融合时序预测：多特征工业数据实战

简介本资源是一套基于PyTorch实现的LSTM与Transformer融合架构的多特征时间序列预测方案面向机器学习初学者及工业时序建模从业者解决风力/光伏功率预测、设备剩余寿命评估、环境浓度趋势推演等实际场景中的长程依赖建模难题。压缩包共160个文件含31个核心Python源码如data_loader.py、MultiWaveletCorrelation.py、87个已编译pyc文件、26个备份文件zbak、4个实测数据集CSV格式含ETTm2/ETTh2等经典基准数据及配置类文件整体仅1.95MB轻量易部署。已有72人下载学习代码经系统调试并配有详尽注释完整覆盖数据加载、模型构建、训练验证与结果可视化全流程特别整合LSTM局部时序建模能力与Transformer全局注意力机制在保持结构清晰的同时显著提升复杂模式下的预测稳定性与精度适合二次开发与学术研究复现。1. 为什么单用LSTM或Transformer在多特征时间序列预测里总差一口气你手头有一份工业传感器数据温度、压力、振动幅值、电流、pH值每5秒采一次连续跑7天——共120,960条记录。你想预测未来1小时12个点的设备退化趋势。用纯LSTM训练慢、长程依赖衰减明显30步以外的周期性信号就“记不住”换纯Transformer注意力机制对短时局部模式建模粗糙小幅度突变比如冷却泵瞬时抖动直接被平均掉更糟的是当输入特征维度升到8维以上标准Transformer的QKV计算量爆炸显存占用翻倍batch_size被迫压到4收敛抖得像心电图。这不是模型能力问题是结构错配LSTM擅长捕捉时序因果链Transformer强在全局关系建模但真实工业/金融/气象场景从不只有一种模式主导。这篇笔记要解决的就是如何用PyTorch把二者拧成一股绳——不是简单拼接LSTM输出喂Transformer而是让LSTM的隐藏状态动态调控Transformer的注意力权重让模型自己决定此刻该信局部变化还是该看全局格局。附带的完整源码已实测通过PyTorch 2.0、CUDA 11.8含清洗好的电力负荷气象多源数据集CSV格式含时间戳、8维特征、目标变量所有代码可直接python train.py启动无需魔改路径。适合有PyTorch基础、正卡在多变量时序预测精度瓶颈的工程师和算法同学。2. 模型架构设计为什么LSTM-Transformer融合必须分三段解耦融合不是把两个黑匣子焊在一起。我试过5种结构LSTM→Transformer、Transformer→LSTM、并行后拼接、门控加权、以及本文采用的分阶段特征解耦架构。最终选第三种原因很实在它让每个模块各司其职且梯度流清晰可控。下面拆解三层设计逻辑并给出PyTorch实现的关键张量形状约束这是复现不翻车的第一道防线。2.1 第一阶段LSTM编码器——提取时序因果特征这里不用双向LSTM而用单向多层LSTM理由很朴素预测任务本质是因果推断未来信息不能泄露。关键参数设置input_size8你的8维特征hidden_size64太小抓不住复杂模式太大易过拟合64是工业数据实测平衡点num_layers2一层LSTM对长序列记忆有限两层足够建模跨小时级依赖dropout0.2仅在层间Dropout最后一层不Drop避免破坏时序连贯性# lstm_encoder.py import torch import torch.nn as nn class LSTMEncoder(nn.Module): def __init__(self, input_size, hidden_size, num_layers, dropout0.2): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, dropoutdropout if num_layers 1 else 0 ) # 输出投影将LSTM最后时刻的h_n映射为query向量 self.query_proj nn.Linear(hidden_size, hidden_size) def forward(self, x): # x: [B, T, F] - Bbatch, Tsequence_len, Ffeatures lstm_out, (h_n, _) self.lstm(x) # lstm_out: [B, T, H], h_n: [L, B, H] # 取最后一层的h_n作为query基底 last_h h_n[-1] # [B, H] query self.query_proj(last_h) # [B, H] return lstm_out, query逻辑说明lstm_out保留了整个时间步的隐藏状态用于后续位置编码对齐query则是LSTM对整段序列的“总结性理解”它将作为Transformer注意力的Query来源。注意h_n[-1]取最后一层而非h_n.mean(0)——均值会模糊层间语义分工。2.2 第二阶段Transformer解码器——注入LSTM语义的动态注意力标准Transformer Decoder需要memory来自Encoder和tgt目标序列。但这里我们不提供传统tgt而是用LSTM生成的query去动态调制Self-Attention的权重。核心改动在MultiheadAttention的forward中插入一个bias项# transformer_decoder.py from torch.nn import MultiheadAttention class DynamicAttentionDecoder(nn.Module): def __init__(self, d_model, nhead, dim_feedforward, dropout0.1): super().__init__() self.self_attn MultiheadAttention(d_model, nhead, dropoutdropout, batch_firstTrue) self.linear1 nn.Linear(d_model, dim_feedforward) self.dropout nn.Dropout(dropout) self.linear2 nn.Linear(dim_feedforward, d_model) self.norm1 nn.LayerNorm(d_model) self.norm2 nn.LayerNorm(d_model) self.dropout1 nn.Dropout(dropout) self.dropout2 nn.Dropout(dropout) # 动态偏置生成器将LSTM query映射为attention bias self.bias_proj nn.Sequential( nn.Linear(d_model, d_model // 2), nn.ReLU(), nn.Linear(d_model // 2, nhead) # 输出nhead个标量每个头一个bias ) def forward(self, tgt, memory, query): # tgt: [B, T, D] —— 这里tgt是positionally encoded的输入序列 # memory: [B, T, D] —— LSTM输出的lstm_out # query: [B, D] —— LSTM生成的summary vector # Step 1: 计算动态bias (B, nhead) bias_per_head self.bias_proj(query) # [B, nhead] # 扩展为(B, nhead, T, T)的attention bias # 这里简化对每个头应用相同bias实际可扩展为矩阵 bias_matrix bias_per_head.unsqueeze(-1).unsqueeze(-1) # [B, nhead, 1, 1] # Step 2: 自注意力注入bias tgt2 self.self_attn(tgt, tgt, tgt, attn_maskNone, key_padding_maskNone, need_weightsFalse)[0] # 注意PyTorch原生MultiheadAttention不支持外部bias # 我们重写attn计算见下方自定义函数 tgt tgt self.dropout1(tgt2) tgt self.norm1(tgt) # Step 3: FFN tgt2 self.linear2(self.dropout(torch.relu(self.linear1(tgt)))) tgt tgt self.dropout2(tgt2) tgt self.norm2(tgt) return tgt # 自定义带bias的attention关键 def dynamic_multi_head_attention(query, key, value, embed_dim_to_check, num_heads, in_proj_weight, in_proj_bias, bias_k, bias_v, add_zero_attn, dropout_p, out_proj_weight, out_proj_bias, trainingTrue, key_padding_maskNone, need_weightsTrue, attn_maskNone, use_separate_proj_weightFalse, q_proj_weightNone, k_proj_weightNone, v_proj_weightNone, static_kNone, static_vNone, dynamic_biasNone): # 此处省略冗长的qkv计算聚焦bias注入点 # 在计算attn_output_weights后 # attn_output_weights attn_output_weights dynamic_bias # [B, nhead, T, T] # 然后softmax pass # 实际代码见源码包中的attention_utils.py参数说明dynamic_bias是核心创新点——它由LSTM的query实时生成值域在[-1,1]作用于Attention Score矩阵。当LSTM检测到强周期性如每24小时负荷峰值bias会增强对应时间步的注意力权重当检测到突变如电流骤降bias则抑制平滑区域聚焦异常窗口。这比固定位置编码更适应真实数据漂移。2.3 第三阶段特征解耦头——分离趋势、周期、残差三路预测很多失败案例源于“一把抓”用同一套权重预测所有目标。我们把预测头拆成三路预测分支输入特征激活函数适用场景趋势分支LSTM输出的h_nLinear长期单调变化如设备老化周期分支Transformer输出Sin固定周期波动如日负荷曲线残差分支LSTMTransformer拼接Tanh短时随机扰动如传感器噪声# prediction_head.py class DecoupledHead(nn.Module): def __init__(self, hidden_size, pred_len, num_features1): super().__init__() self.trend_head nn.Linear(hidden_size, pred_len * num_features) self.cycle_head nn.Linear(hidden_size, pred_len * num_features) self.resid_head nn.Linear(hidden_size * 2, pred_len * num_features) # 周期分支强制正弦输出物理意义约束 self.cycle_freq nn.Parameter(torch.tensor([2 * 3.1415926 / 24])) # 日周期 def forward(self, lstm_h, trans_out): # lstm_h: [B, H], trans_out: [B, T, H] trend self.trend_head(lstm_h).view(-1, pred_len, 1) # [B, P, 1] # 周期分支用sin拟合主频避免过拟合高频噪声 cycle_base self.cycle_head(trans_out.mean(1)) # [B, P*1] cycle torch.sin(cycle_base.view(-1, pred_len, 1) * self.cycle_freq) # 残差分支拼接后压缩 resid_input torch.cat([lstm_h, trans_out.mean(1)], dim-1) # [B, 2H] resid torch.tanh(self.resid_head(resid_input)).view(-1, pred_len, 1) # 加权融合可学习权重 alpha torch.sigmoid(nn.Parameter(torch.tensor(0.5))) beta torch.sigmoid(nn.Parameter(torch.tensor(0.3))) gamma 1 - alpha - beta final_pred alpha * trend beta * cycle gamma * resid return final_pred为什么用Sin激活工业数据中温度、负荷等天然具周期性强制正弦输出比ReLU/Linear更符合物理规律实测MAE降低12%。cycle_freq设为可学习参数初始值按业务常识设定如24小时模型会微调它。3. 数据预处理与加载多特征时间序列的四个隐形陷阱多特征时间序列不是把CSV读进来就能训。我踩过的坑里80%出在数据准备阶段。以下四点不处理模型再 fancy 也白搭。3.1 陷阱一特征尺度差异导致梯度淹没非标准化温度0~100℃和电流0~500A数值量级差5倍若直接输入LSTM的梯度更新会被大尺度特征主导。不能只用MinMaxScaler——它压缩范围但不解决分布偏斜。正确做法对每个特征单独做RobustScaler用中位数和四分位距抗异常值再对整体序列做Z-score归一化均值为0标准差为1适配Transformer的LayerNorm# data_preprocess.py from sklearn.preprocessing import RobustScaler import numpy as np def robust_normalize(features): # features: [T, F] scaler RobustScaler() scaled scaler.fit_transform(features) # 每列独立缩放 # 再做Z-score按时间维度非特征维度 mean scaled.mean(axis0, keepdimsTrue) # [1, F] std scaled.std(axis0, keepdimsTrue) # [1, F] normalized (scaled - mean) / (std 1e-8) return normalized, scaler, mean, std # 使用示例 data np.loadtxt(sensor_data.csv, delimiter,) norm_data, scaler, mean, std robust_normalize(data)血泪经验曾用MinMaxScaler训了3天验证集loss卡在0.8不动换成RobustScalerZ-score后首epoch就降到0.3。因为电流突增时MinMaxScaler把正常值全压到[0,0.1]LSTM根本学不到细节。3.2 陷阱二时间戳未对齐引发的相位错乱你的CSV可能有缺失时间戳如某分钟设备离线若直接插值会引入虚假周期。必须重建严格等间隔索引import pandas as pd def align_timestamps(df, freq5S): # df.index 应为datetime full_index pd.date_range(startdf.index.min(), enddf.index.max(), freqfreq) df_aligned df.reindex(full_index, methodnearest) # 对缺失值用前向填充线性插值组合 df_aligned df_aligned.fillna(methodffill).interpolate() return df_aligned # 加载时强制解析时间戳 df pd.read_csv(data.csv, parse_dates[timestamp], index_coltimestamp) df align_timestamps(df, freq5S) # 5秒间隔提示methodnearest比pad更鲁棒避免离线期间的长距离填充。interpolate()只用于少量缺失3%超限则标记为异常段剔除。3.3 陷阱三多步预测的标签构造方式错误想预测未来12步常见错误是❌y X[t1:t13]用t时刻输入预测t1到t13✅ 正确应为y X[thorizon:thorizon12]其中horizon是预测起点偏移为什么因为模型需要看到足够的历史来建立上下文。若horizon1模型只见过1步历史无法判断趋势方向。工业场景推荐horizon24即用过去24步预测未来12步实测提升趋势捕捉能力。# dataset.py class TimeSeriesDataset(torch.utils.data.Dataset): def __init__(self, data, seq_len96, pred_len12, horizon24): self.seq_len seq_len self.pred_len pred_len self.horizon horizon self.data data # [T, F] def __getitem__(self, index): # 输入从index开始取seq_len长度 s_begin index s_end s_begin self.seq_len # 标签从s_end horizon开始取pred_len长度 r_begin s_end self.horizon r_end r_begin self.pred_len seq_x self.data[s_begin:s_end] # [L, F] seq_y self.data[r_begin:r_end] # [P, F] return seq_x, seq_y def __len__(self): return len(self.data) - self.seq_len - self.horizon - self.pred_len3.4 陷阱四特征相关性未显式建模导致Transformer注意力失效Transformer的Self-Attention默认假设所有特征地位平等。但温度和湿度强相关电流和振动弱相关——若不告诉模型它会浪费注意力在无关特征对上。解决方案构造特征协方差矩阵作为额外输入# feature_correlation.py def build_feature_corr_matrix(data, window168): # 168小时1周 # data: [T, F] corr_mat np.corrcoef(data.T) # [F, F] # 归一化到[0,1]便于嵌入 corr_mat (corr_mat 1) / 2 return torch.tensor(corr_mat, dtypetorch.float32) # 在模型中注入 corr_embed self.corr_embedding(corr_matrix) # [F, F] - [F, D] # 与原始特征相乘通道注意力雏形 x_weighted x * corr_embed.unsqueeze(0) # [B, T, F] * [1, F, D] - [B, T, D]避坑总结这四点不处理模型会学出“看起来不错但实际不可用”的结果——验证集MAE低但部署后遇到真实突变就崩。尤其第三点horizon是多数开源项目忽略的致命细节。4. 训练与调优避开LSTM-Transformer融合的五个典型翻车现场融合模型的训练比单模型更脆弱。以下是我在3个工业客户项目中总结的5个高频翻车点每条都附现象、根因、解决命令。4.1 现象Loss在0.5附近震荡不下降梯度norm爆表1000原因LSTM和Transformer的学习率冲突。LSTM需要较大学习率1e-3快速收敛时序模式Transformer需较小学习率5e-4稳定注意力矩阵。统一用1e-3会导致Transformer权重剧烈震荡。解决分层学习率Layer-wise LR# 使用torch.optim.AdamW为不同模块设不同lr optimizer torch.optim.AdamW([ {params: model.lstm_encoder.parameters(), lr: 1e-3}, {params: model.transformer_decoder.parameters(), lr: 5e-4}, {params: model.prediction_head.parameters(), lr: 1e-3} ], weight_decay1e-5)参数说明weight_decay1e-5对所有层统一防止过拟合不要给Transformer加更大weight_decay——它本身就有Dropout和LayerNorm双重正则会欠拟合。4.2 现象验证集Loss持续下降但预测曲线严重滞后相位延迟1-2步原因位置编码Positional Encoding与LSTM的隐状态时序信息冲突。LSTM已编码了绝对位置再加sin/cos位置编码模型学会“抄近路”——只依赖位置忽略内容。解决禁用Transformer的位置编码改用相对位置编码Rotary Position Embedding, RoPE它将位置信息融入Q/K计算不干扰value。# rope.py精简版 def apply_rope(q, k, positions): # q,k: [B, H, T, D/H] # positions: [T]如torch.arange(T) sin_pos torch.sin(positions * 10000**(-torch.arange(0, q.size(-1), 2)/q.size(-1))) cos_pos torch.cos(positions * 10000**(-torch.arange(0, q.size(-1), 2)/q.size(-1))) # RoPE公式q_rot q * cos q_shift * sin ... # 完整实现见源码包rope_layer.py return q_rot, k_rot # 在MultiheadAttention前调用 q_rot, k_rot apply_rope(q, k, torch.arange(seq_len))为什么RoPE有效它让模型关注“第5步相对于第3步是什么”而非“第5步绝对位置是5”。这与LSTM的递归特性天然兼容实测相位误差降低70%。4.3 现象训练初期Loss极低0.01但验证集MAE高达2.5量纲未统一原因损失函数用MSE但目标变量如负荷MW和特征如温度℃量纲不同模型学会“只拟合大数值特征”忽略小数值目标。解决多任务损失加权 目标归一化# loss.py class MultiTaskLoss(nn.Module): def __init__(self, weights{trend: 0.4, cycle: 0.3, resid: 0.3}): super().__init__() self.weights weights self.mse nn.MSELoss() def forward(self, pred_trend, pred_cycle, pred_resid, target): # target已按特征维度归一化见3.1节 loss_trend self.mse(pred_trend, target) loss_cycle self.mse(pred_cycle, target) loss_resid self.mse(pred_resid, target) total_loss ( self.weights[trend] * loss_trend self.weights[cycle] * loss_cycle self.weights[resid] * loss_resid ) return total_loss # 训练时 criterion MultiTaskLoss(weights{trend: 0.5, cycle: 0.3, resid: 0.2})参数说明trend权重最高因为趋势是预测主干cycle次之resid最低它本就是修正项。权重需根据验证集MAE手动微调不要用自动加权。4.4 现象GPU显存OOMbatch_size1仍报错原因Transformer的Attention矩阵占显存O(T^2)当seq_len96时96^29216尚可但若误设seq_len512512^2262144显存直接爆。解决分块注意力Block Attention 梯度检查点# model.py from torch.utils.checkpoint import checkpoint class BlockAttention(nn.Module): def __init__(self, block_size32): super().__init__() self.block_size block_size def forward(self, q, k, v): # 将序列分块T - (N, B)Nceil(T/B) T q.size(1) N (T self.block_size - 1) // self.block_size # 分块计算attention内存O(B^2) ... # 在Transformer层中启用梯度检查点 def custom_forward(*inputs): return self.transformer_layer(*inputs) # 替代原forward output checkpoint(custom_forward, x, memory, query)提示block_size32是平衡速度与显存的黄金值。小于16速度骤降大于64显存风险回升。配合checkpoint显存占用可降60%。4.5 现象预测结果出现“阶梯状”伪影连续多步值完全相同原因解耦头中cycle分支的sin输出未加噪声约束模型学会输出恒定值sin(0)0来最小化loss。解决在sin输出后加高斯噪声正则项# prediction_head.py def forward(self, lstm_h, trans_out): # ... 前面同上 cycle torch.sin(cycle_base.view(-1, pred_len, 1) * self.cycle_freq) # 添加微小噪声标准差0.01打破对称性 cycle cycle torch.randn_like(cycle) * 0.01 # 残差分支同理 resid torch.tanh(self.resid_head(resid_input)).view(-1, pred_len, 1) resid resid torch.randn_like(resid) * 0.005 # ... 后续融合玄学参数0.01和0.005是经验值。太大引入真噪声太小不起作用。在验证集上观察预测曲线平滑度调整。5. 效果验证与工业落地技巧用三个硬指标判断是否真可用模型训完不是终点工业场景要回答“它能不能扛住真实数据”我用三个不可妥协的硬指标验证缺一不可。5.1 指标一滚动预测一致性Rolling Consistency单次预测准不算本事连续滚动预测100次每次用最新数据重训结果必须稳定。计算滚动标准差Rolling Std# eval_rolling.py def rolling_consistency(model, data_loader, steps100, pred_len12): predictions [] for i, (x, y) in enumerate(data_loader): if i steps: break with torch.no_grad(): pred model(x) # [B, P, 1] predictions.append(pred[0].cpu().numpy()) # 取第一个样本 # predictions: [steps, pred_len] pred_array np.array(predictions) # [100, 12] # 计算每步的std第0步预测的100个值的标准差 step_stds pred_array.std(axis0) # [12] return step_stds.mean() # 单一指标 # 要求step_stds.mean() 0.05归一化后 consistency_score rolling_consistency(model, test_loader) print(fRolling Consistency: {consistency_score:.4f}) # ✅ 达标0.05 # ❌ 翻车0.15模型对微小数据扰动敏感为什么重要真实系统每5秒进一条新数据模型需高频重训。若一致性差运维人员会失去信任——“上次说要报警这次又说没事”。5.2 指标二突变点捕获率Spike Detection Rate工业数据最怕漏报突变如轴承温度骤升。定义突变点当前值 前10步均值 3倍标准差。计算模型预测值是否在突变发生后3步内给出预警# spike_eval.py def spike_detection_rate(model, data, window10, threshold3, alert_window3): spikes [] alerts [] for t in range(window, len(data)): window_mean data[t-window:t].mean() window_std data[t-window:t].std() if data[t] window_mean threshold * window_std: spikes.append(t) # 对每个spike检查预测是否在[t, talert_window]内越阈值 for spike_t in spikes: # 获取模型在spike_t前的输入需构造seq x_input data[spike_t-96:spike_t] # seq_len96 with torch.no_grad(): pred model(torch.tensor(x_input).float().unsqueeze(0)) # pred: [1, 12, 1] pred_flat pred.squeeze().numpy() # 检查pred_flat[0:3]是否有值 当前真实值的0.9倍预警阈值 if any(pred_flat[:alert_window] data[spike_t] * 0.9): alerts.append(1) else: alerts.append(0) return np.mean(alerts) if alerts else 0 # 要求Spike Detection Rate 0.85 sdr spike_detection_rate(model, raw_data) print(fSpike Detection Rate: {sdr:.3f})落地价值这个指标直接关联设备停机损失。85%捕获率意味着每100次突变模型能提前拦下85次剩下15次靠人工巡检兜底。5.3 指标三冷启动适应性Cold-start Adaptation新产线/新设备上线只有7天数据约12,000条模型能否在2小时内完成微调并达标测试方法用源数据预训练模型100 epoch冻结LSTM和Transformer主干只微调prediction_head用新数据训练5 epochbatch_size32# 微调脚本 python finetune.py \ --model_path ./pretrained/model.pth \ --data_path ./new_line/data.csv \ --freeze_backbone \ --epochs 5 \ --lr 1e-3 \ --output_dir ./finetuned/验收标准微调后验证集MAE ≤ 预训练模型MAE × 1.2允许轻微上升且推理延迟 50ms满足实时监控。我实测在T4 GPU上7天数据微调耗时1分42秒MAE上升仅8%完全达标。5.4 一个反直觉但救命的技巧用“伪标签”扩充小样本客户常抱怨“只有2周数据怎么训”我的做法是先用2周数据训一个粗糙模型用它预测未来30天生成伪标签pseudo-labels将伪标签与真实数据混合按置信度加权高置信度伪标签权重0.7真实数据权重1.0# pseudo_label.py def generate_pseudo_labels(model, data, confidence_threshold0.8): model.eval() pseudo_labels [] confidences [] for i in range(0, len(data)-96, 12): # 每12步滑动一次 x data[i:i96] with torch.no_grad(): pred model(torch.tensor(x).float().unsqueeze(0)) # 置信度 预测值标准差的倒数越平滑越可信 std_pred pred.std().item() conf 1.0 / (std_pred 1e-5) if conf confidence_threshold: pseudo_labels.append(pred.squeeze().cpu().numpy()) confidences.append(conf) return np.array(pseudo_labels), np.array(confidences) # 训练时加权 weights torch.tensor([1.0]*len(real_data) list(confidences)) sampler torch.utils.data.WeightedRandomSampler(weights, len(weights)) train_loader DataLoader(dataset, samplersampler, ...)效果在仅有10天数据的风电功率预测任务中伪标签使MAE从0.21降至0.16逼近30天数据的水平。关键是confidence_threshold——设太高伪标签少设太低引入噪声0.75是多数场景的甜点。我坚持在每个工业项目交付前跑完这三项验证。不是为了炫技而是确保模型不会在客户凌晨三点的告警电话里掉链子。希望帮到你。本文还有配套的精品资源点击获取