简介基于LSTM的时间序列分析预测Python源码包面向希望掌握深度学习时序建模的Python开发者与数据科学初学者解决从数据预处理到模型训练、评估与预测的完整链路问题。压缩包共126个文件以75个Python脚本为核心覆盖LSTM模型构建、训练与预测逻辑26个CSV文件提供空气污染等多组实验数据另有TensorFlow模型检查点、H5模型文件及说明文档便于复现与二次开发。资源包大小仅5.42MB轻量易下载。已有5095人学习使用。通过阅读与运行源码读者可理解LSTM门控机制、Keras建模流程、归一化及序列构造等关键知识点并掌握用MSE、MAE等指标评估预测效果的方法。项目包含从原始数据读取到最终预测的整套实现代码结构清晰适合结合理论教程进行动手实践是入门时序预测与人工智能应用的实用素材。1. 基于 LSTM 的时间序列分析预测源码包先别急着跑 demo看这三处拿到带「Python 源码」的 LSTM 时间序列预测压缩包我一般不会先点开训练脚本。先看三处数据从哪进、窗口怎么滑、预测完之后怎么回到原始尺度。这三处决定了这份源码是真能用于生产还是只能跑个示意图。LSTM 解决的是带时间依赖的预测问题比如设备寿命预测、水位流量预测、电力负荷预测。适合已经有 Python 基础、手头有一条条按时间排好序数据的从业者。核心思路不复杂用过去 N 个时间步去预测未来 M 个时间步。但落地时数据形状、归一化、反归一化这些环节才是真正决定成败的地方。2. 时间序列预测为什么用 LSTM选型逻辑与数据准备2.1 LSTM 在时序预测里的定位它解决了 RNN 的什么问题传统 RNN 处理序列时信息每经过一个时间步就要乘一次权重矩阵梯度在反向传播中会指数级衰减或爆炸。时间跨度过长前面几步的信息根本传不到输出的损失里这就是常说的长期依赖缺失。LSTM 引入了一个贯穿所有时间步的 cell state 通道信息可以在这个通道里近乎无损地流动配合输入门、遗忘门、输出门决定记什么、忘什么、放什么出去。因此在中等长度的时序数据上LSTM 比 RNN 稳定得多。与 GRU 相比GRU 把三个门简化成两个门参数更少、训练更快。但在工业界和学术界LSTM 在时间序列预测里的生态最成熟参考实现、调参经验都最丰富。我的习惯是数据量少于 1 万条时先试单层 LSTM数据量大再加层数。下面是选型上的直观对比模型参数数量长期依赖能力训练速度适用场景RNN少差梯度易消失快极短序列、基线对比LSTM多强cell state 通道较慢中等长度时序预测GRU中较强较快数据量小、对推理速度敏感Transformer最多最强但依赖大样本视规模而定长序列、大规模数据实际做预测时LSTM 最常见的用法是滑窗监督学习把时间序列切成一段段的窗口用窗口内的数据作为特征窗口之后的一个或几个时间点作为标签变成一个标准回归问题。所以拿到源码先看它怎么构造这个窗口等于看懂了整份代码的骨架。2.2 滑窗构造样本集一份可直接改用的数据处理代码一份合格的 LSTM 预测源码里数据准备代码占了至少一半的工作量。下面是我把单变量序列转成监督学习格式的常用写法你拿到源码后可以直接对照import numpy as np from sklearn.preprocessing import MinMaxScaler def build_dataset(series, window24, horizon1, train_ratio0.8): 将一维时间序列构造成 LSTM 训练样本 参数: series: 原始一维序列, numpy array, 形状 (n,) window: 每个样本用过去多少个时间步作为特征 horizon: 预测未来多少个时间步, 默认 1 train_ratio: 训练集占比, 按时间顺序切分 # 归一化必须在切分之前做, 避免测试集信息混入训练集 scaler MinMaxScaler(feature_range(0, 1)) scaled scaler.fit_transform(series.reshape(-1, 1)).flatten() X, y [], [] for i in range(len(scaled) - window - horizon 1): # 特征: [i, iwindow) 之间的数据 X.append(scaled[i:i window]) # 标签: [iwindow, iwindowhorizon) 之间的数据 y.append(scaled[i window:i window horizon]) X np.array(X).reshape(-1, window, 1) # (样本数, window, 特征数) y np.array(y) # 按时间顺序切分, 不能用随机打乱 split int(len(X) * train_ratio) return X[:split], X[split:], y[:split], y[split:], scaler几个关键点。归一化必须在构造样本前做MinMaxScaler只调一次fit把序列压到 0 到 1 之间这样 LSTM 的输入输出都在同一量级训练收敛快得多。X的形状是(样本数, window, 特征数)PyTorch 的 LSTM 层默认接受这种(batch, seq_len, input_size)结构。window大小取多少跟业务周期相关预测小时级电力负荷24 就是一天设备振动数据预测寿命window 可能要 100 以上。训练集和验证集只能按时间顺序连续切分千万不能随机打散——时间序列的验证集必须是未来否则模型在训练阶段偷看不到的未来验证分数就失真了。3. 用 Python 复现 LSTM 预测核心流程模型定义、训练与预测接口3.1 模型定义PyTorch 里写一个最小可用的 LSTM 类拿到源码后先找模型定义部分。一个标准做法是nn.LSTM负责提取序列特征再接一个全连接层输出预测值。下面是完整的最小实现import torch import torch.nn as nn class LSTMPredictor(nn.Module): def __init__(self, input_size1, hidden_size64, num_layers2, output_size1, dropout0.2): super(LSTMPredictor, self).__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, dropoutdropout if num_layers 1 else 0 ) self.fc nn.Linear(hidden_size, output_size) def forward(self, x): # x 形状: (batch, window, input_size) lstm_out, _ self.lstm(x) # 取最后一个时间步的隐藏状态作为整条序列的总结 last_step lstm_out[:, -1, :] pred self.fc(last_step) return pred这里三个参数最影响效果。hidden_size是 LSTM 隐藏单元数64 到 128 是中等规模时序的常见起点太小拟合不了复杂依赖太大在小数据集上容易过拟合。num_layers2表示堆叠两层 LSTM层数越多模型越深但训练更慢数据少于 1 万条时 2 层基本够用。batch_firstTrue让输入形状从(seq_len, batch, input_size)变成(batch, seq_len, input_size)不用每次手动转置代码可读性高很多。lstm_out[:, -1, :]是所有时间步的输出维度是(batch, window, hidden_size)。这里只取每个序列最后一个时间步的隐藏状态因为在这个框架下最后一步隐含了整条历史窗口的信息全连接层拿它回归出预测值。3.2 训练循环损失函数、优化器与 early stopping训练循环是源码里改动最频繁的部分。损失函数一般用均方误差 MSE因为它对大的预测偏差惩罚更重与回归任务的目标一致。优化器优先 Adam学习率 1e-3 起步。下面这段代码可以直接替换源码里的训练主循环def train_model(model, train_loader, val_loader, epochs100, lr1e-3, patience10): optimizer torch.optim.Adam(model.parameters(), lrlr) criterion nn.MSELoss() best_val_loss float(inf) bad_epochs 0 for epoch in range(epochs): model.train() train_loss 0.0 for x_batch, y_batch in train_loader: optimizer.zero_grad() pred model(x_batch) # (batch, horizon) loss criterion(pred, y_batch) loss.backward() optimizer.step() train_loss loss.item() * x_batch.size(0) # 每个 epoch 结束跑一遍验证集 model.eval() val_loss 0.0 with torch.no_grad(): for x_batch, y_batch in val_loader: pred model(x_batch) val_loss criterion(pred, y_batch).item() * x_batch.size(0) train_loss / len(train_loader.dataset) val_loss / len(val_loader.dataset) # early stopping: 验证集不再下降就停止 if val_loss best_val_loss: best_val_loss val_loss bad_epochs 0 torch.save(model.state_dict(), best_model.pt) else: bad_epochs 1 if bad_epochs patience: print(fepoch {epoch}: early stop) break if epoch % 10 0: print(fepoch {epoch}: train_loss{train_loss:.5f}, val_loss{val_loss:.5f})训练循环里有几个值得注意的细节。model.eval()配合torch.no_grad()告诉框架关闭 Dropout 和梯度计算验证阶段的输出才是稳定的。loss.item()取出的是标量损失值而loss本身还挂着计算图直接累加会导致内存不断膨胀训练到几百个 epoch 后越来越慢这是新手跑源码最容易忽略的问题。lr1e-3是 Adam 的常见默认值。如果验证集 loss 震荡优先把 lr 降到 3e-4 或 1e-4而不是调整模型结构。patience10表示连续 10 个 epoch 验证集没有改进就提前结束避免白白浪费算力。torch.save只保存模型参数不保存整个模型对象加载时要求先实例化一个结构相同的模型。3.3 预测阶段滚动预测与误差累积训练完成后进入预测阶段。这个阶段最容易踩坑因为训练时可以看到标签标注下一个值应该是什么预测时标签不存在模型得用自己的输出作为下一步的输入这就是滚动预测。def rolling_forecast(model, x_seed, scaler, n_steps): 从 x_seed(形状 1, window, input_size) 出发 滚动预测未来 n_steps 个时间步 model.eval() forecast [] with torch.no_grad(): window x_seed.clone() for _ in range(n_steps): pred model(window) # (1, horizon) pred_value pred[0, 0].item() forecast.append(pred_value) # 把预测值追加到窗口末尾, 丢掉窗口最前面的旧值 new_value torch.tensor([[[pred_value]]]) window torch.cat([window[:, 1:, :], new_value], dim1) forecast np.array(forecast).reshape(-1, 1) return scaler.inverse_transform(forecast).flatten()这段代码的核心是torch.cat那条线每一步把模型刚生成的预测值拼到窗口末尾丢掉时间最早的一个值窗口长度保持window不变。预测出来的序列会存在误差累积——后面步长的输入是前面的预测结果预测的偏差会不断被放大。预测步数越多曲线越平滑甚至趋于一个平台。这份源码如果你要拿去生产一般只会用未来 5 到 10 步的滚动预测结果。4. 评估预测结果先回归到真实尺度再谈指标4.1 反归一化与评估指标计算顺序训练时光看归一化后的 MSE 是片面的因为 0 到 1 区间的误差放到原始单位下用户根本无感。正确顺序是先反归一化、再算指标。下面给出一段能直接替换源码评估部分的代码from sklearn.metrics import mean_squared_error, mean_absolute_error def evaluate_forecast(model, X_test, y_test, scaler): model.eval() with torch.no_grad(): pred_norm model(X_test).numpy() # 归一化尺度下的预测 # 反归一化: 把预测值和真实值都还原到原始单位 y_test_np y_test.numpy() # 形状 (n, horizon) y_test_inv scaler.inverse_transform( y_test_np.reshape(-1, 1)).reshape(y_test_np.shape) pred_inv scaler.inverse_transform( pred_norm.reshape(-1, 1)).reshape(pred_norm.shape) rmse np.sqrt(mean_squared_error(y_test_inv, pred_inv)) mae mean_absolute_error(y_test_inv, pred_inv) mape np.mean(np.abs((y_test_inv - pred_inv) / y_test_inv)) * 100 return {rmse: rmse, mae: mae, mape: mape}反归一化时要特别注意scaler对象必须来自训练集。MinMaxScaler在构建数据集时已经fit过了保存了训练集的最小值和最大值测试集和预测值都只能调用同一个 scaler 的inverse_transform不能再重新 fit否则还原出来的数值完全对不上。MAPE 会有一个会被忽略的坑如果真实值接近 0这个指标会趋近无穷大看起来模型彻底失败其实是分母问题。遇到这种情况我一般直接用 RMSE 加 MAE 汇报。4.2 多步预测的三种推进策略与选择逻辑horizon大于 1 时预测未来多步有三种常见策略源码里可能只实现了其中一种你得先看清楚是哪种。递归多步预测最直观预测完第 1 步把结果拼回窗口再预测第 2 步上一节代码就是这个思路。优点是模型结构简单、输出维度固定缺点是误差随步数累积步数越长预测越不可信。直接多步预测则是把输出层改成linear(hidden_size, horizon)一次吐出未来多个时间步。损失是每一步的误差同时反传训练快但各步之间没有显式约束曲线可能出现不合理的跳变。Seq2Seq 结构在两者之间做了折中编码器压缩历史信息解码器逐步生成未来序列精度最高但训练和调试成本都上了一个台阶。策略实现复杂度误差累积适合场景递归多步低高预测步数少≤5直接多步低低步数固定、短期预测Seq2Seq高中等长序列、业务要求高选择时不用过度纠结。数据量少、步数短直接多步最省事步数多、对精度有要求再考虑 Seq2Seq。许多宣称多步预测的源码包本质只是把单步预测的结果拼在一起画了个图评估时要注意它有没有真的用滚动方式推理。5. LSTM 落地常见问题排查4 个容易翻车的坑5.1 训练 loss 一直不下来或反复震荡现象训练了 50 个 epochtrain_loss一直在 0.5 以上或者降一阵又弹回去。原因可能有三种学习率太大导致参数在最优解附近来回震荡数据没有归一化输入量级在几百到几千梯度也处于放大状态window取得太小信息量不足以支撑预测。排查时我先看归一化代码再打印每次x_batch的最大值和最小值确认在 0 到 1 区间内。如果数据没问题把学习率从 1e-3 降到 3e-4同时给梯度加个裁剪torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)。这两个操作能解决九成的训练不收敛问题。最后检查window把它调大到至少覆盖一个完整业务周期。5.2 预测曲线整体滞后一拍现象测试集上 loss 很低画出来的预测曲线和真实曲线高度重合但仔细看是真实曲线往右平移了一格——模型几乎只是复制了上一个真实值。原因是滑窗构造时窗口最后一位就是t-1时刻的真实观测值而真正要预测的t时刻往往与t-1高度相关。模型学到的最省力方案就是直接输出t-1的值整个模型退化成延迟器。这条曲线对运维没有参考价值因为真正要预测的时候t-1的值根本不存在。解决方法是调整标签构造把标签从t时刻改成thorizon时刻强迫模型跳过最近的时间步学习更本质的趋势。另一种做法是把窗口末尾的最近一两个时间步从特征中剔除训练时人为制造信息缺口。5.3 双向 LSTM 把未来信息泄漏给了训练过程现象模型用了nn.LSTM(..., bidirectionalTrue)验证集指标好到不真实但上线后效果断崖式下跌。原因是双向 LSTM 在编码一个时间步时会同时读取它后面的时间步——如果验证集是从完整序列尾部切出来的一段训练阶段双向 LSTM 的后向传播实际上接触了验证集区间内的数据这属于数据泄漏。即使验证 loss 再好看部署后也只能用过去预测未来双向结构完全失去优势。解决纯时间序列预测场景默认使用单向 LSTM把bidirectional参数去掉如果业务确实需要双向特征必须把训练、验证、测试三段数据划分后在时间上有足够的间隔确保尾部数据不参与任何一步前向计算。我在实际项目中吃过这个亏从此凡是时序任务先检查模型定义里有没有bidirectional。5.4 反归一化后指标暴涨现象归一化尺度上的 loss 是 0.001反归一化后 RMSE 算出来几千甚至几万。原因多数不是模型坏了而是scaler被重新fit过。某些源码为了省事在评估脚本里对y_test又调用了一次scaler.fit_transform这相当于把测试集的统计量也用了进来反归一化自然错位。还有一层常见错误是反归一化时把(n, 1)的数组传入inverse_transform得到的却是(n,)的原形状后续广播出错。排查时先检查评估代码里scaler对象是不是训练集那个再打印pred_inv的前五个值和真实值的前五个值当场能看出量级差距。面板数据多时可以用百分比误差来比较不同量级变量的效果但前提是真实值都远离 0。6. 进阶做法多变量输入与注意力融合如果这源码已经跑通了单变量预测下一步是把真实业务里更丰富的特征加进去。多变量输入只需要改两个地方build_dataset里把多个特征列拼成一个二维数组模型里input_size改成特征列数。例如同时输入负荷值和温度值窗口形状变成(window, 2)def build_multivariate_dataset(df, feature_cols, window24): scaler MinMaxScaler() scaled scaler.fit_transform(df[feature_cols].values) # (n, n_features) X, y [], [] for i in range(len(scaled) - window): X.append(scaled[i:i window]) # (window, n_features) y.append(scaled[i window, 0]) # 预测第一个特征列 return np.array(X), np.array(y), scalerLSTMPredictor里的input_size同步改成len(feature_cols)其余结构不变。注意力机制在 LSTM 输出之后做融合本质是对不同时间步分配权重避免最后一帧吞掉所有信息的缺陷。在lstm_out[:, -1, :]改成先对lstm_out做注意力加权再回归短期预测的提升在生产数据有限时并不总是明显但长序列场景下值得尝试。验证多步预测时我习惯做滚动回溯验证从第 N 个时间点开始每次只追加真实值、不更新窗口以外的信息连续滚动预测多个步长再分段计算误差。这个方法很朴素但比单次预测要诚实得多。我自己改源码的习惯是每个模型实验固定一个 seed记录 window、hidden_size、lr 三个参数。这样出了新数据回来看记录就知道哪个配置更稳定。希望这些经验能帮你在同样的踩坑路上少走几步。本文还有配套的精品资源点击获取