简介:这份资源面向高校学生与Python初学者,提供一套可直接运行的LSTM时间序列预测完整项目,适用于期末大作业、课程设计或入门深度学习实践。项目以空气质量等真实序列数据为样本,覆盖数据读取、预处理、模型搭建、训练与预测全流程,代码注释详尽,新手也能看懂并快速部署。压缩包共129个文件,约5.42MB,其中78个py脚本承担模型定义与训练逻辑,26个csv与若干txt提供原始数据及说明,另含checkpoint、h5权重与TensorFlow数据分片,便于直接复现结果。目前已有169人学习下载。整体目录结构清晰,读者可据此掌握LSTM做时序预测的完整思路,并在此基础上替换数据集完成自己的课题,具备较高的参考与复用价值。
1. 从一份期末大作业说起:LSTM 时间序列预测到底能跑出什么结果
如果你手头正拿着一份「基于 LSTM 进行时间序列数据预测」的期末大作业,大概率会遇到两个尴尬:代码能跑,但预测曲线像一条滞后三拍的影子;数据能读,但换个数据集就报形状不匹配。LSTM 时间序列预测这件事,真正卡人的从来不是model.fit()那一行,而是窗口怎么切、归一化在哪一步做、反归一化有没有对齐、评估指标是不是在自欺欺人。
这篇笔记面向三类人:正在赶期末大作业、需要一份能直接复现的 Python 源码结构的学生;想把 LSTM 用到设备寿命预测、销量预测、传感器趋势预测上的工程师;以及刚配好 PyCharm 或 VS Code Python 环境、想找一个完整项目练手的新手。我会按「数据长什么样 → 窗口怎么造 → 模型怎么搭 → 训练怎么盯 → 坑怎么躲」的顺序,把一份可运行的 LSTM 预测方案拆开讲清楚,参数给到能改的程度,失败时该看哪里也一并说明。读完你应该能自己判断:这份源码值不值得改,以及怎么改成能写进简历的版本。
2. 数据准备与窗口构造:LSTM 吃进去的到底是什么
2.1 时间序列预测的任务定义与数据格式
先把问题说清楚。时间序列预测,本质是用过去一段时间的观测值,去推断未来一个或多个时间点的值。单变量场景下,输入是一列按时间排序的数值,比如每天销量、每小时温度、每周期设备振动幅值;多变量场景下,输入是多列数值,外加时间戳对齐。
LSTM 不会直接吃原始序列,它吃的是「样本」。一个样本由两部分组成:一段长度为look_back的历史窗口,和一个长度为horizon的预测目标。假设原始序列是[10, 12, 11, 13, 14, 15],look_back=3、horizon=1,那么第一个样本是输入[10,12,11]、标签13;第二个样本是输入[12,11,13]、标签14。这一步叫滑动窗口构造,是 LSTM 时间序列预测里最容易写错、也最容易被忽略的地方。
常见的数据格式有三种:CSV 单列时间序列、CSV 多列带时间戳、以及数据库导出的宽表。期末大作业里最常见的是第一种,只有一列数值,没有时间列。这种数据反而好处理,因为不需要考虑时间间隔不规律的问题。但如果你拿到的是带时间戳的数据,第一件事是检查时间列是否等间隔,不等间隔要先重采样。
提示:不要一上来就
pd.read_csv然后直接丢进模型。先df.head()、df.info()、df.describe()三连,确认缺失值、异常值、量纲范围。LSTM 对量纲敏感,没归一化的数据会让训练直接翻车。
2.2 用 Pandas 和 NumPy 构造滑动窗口
下面这段代码是我一般会用的窗口构造方式,单变量多变量都能兼容。核心思路是先把数据归一化到[0,1],再切窗口,最后 reshape 成 LSTM 需要的三维张量(样本数, 时间步, 特征数)。
import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler # 读取数据,假设只有一列数值,列名为 'value' df = pd.read_csv('data.csv') values = df['value'].values.reshape(-1, 1) # 归一化:只在训练集上 fit,避免数据泄漏 scaler = MinMaxScaler(feature_range=(0, 1)) scaled = scaler.fit_transform(values) def create_dataset(data, look_back=24, horizon=1): """ data: 归一化后的二维数组 (n, 1) look_back: 历史窗口长度 horizon: 预测未来第几个点 返回: X (样本数, look_back, 1), y (样本数,) """ X, y = [], [] for i in range(len(data) - look_back - horizon + 1): X.append(data[i:i + look_back, 0]) y.append(data[i + look_back + horizon - 1, 0]) X = np.array(X) y = np.array(y) # reshape 成 LSTM 输入格式 X = X.reshape(X.shape[0], X.shape[1], 1) return X, y look_back = 24 # 用过去 24 个点预测 horizon = 1 # 预测未来第 1 个点 X, y = create_dataset(scaled, look_back, horizon) print(X.shape, y.shape) # 例如 (1000, 24, 1) (1000,)这段代码有三个关键参数。look_back决定模型能看到多长的历史,太小会欠拟合,太大会引入噪声且训练变慢;horizon决定预测步长,设为 1 是单步预测,设为 7 就是预测未来第 7 个点;feature_range默认(0,1),如果数据里有明显异常值,可以改成(-1,1)配合tanh激活。
逻辑说明:create_dataset里循环的终止条件是len(data) - look_back - horizon + 1,这个+1很容易漏,漏了会少一个样本。X.reshape那一步是必须的,因为 Keras/TensorFlow 的 LSTM 层要求输入是三维的,PyTorch 的nn.LSTM默认也要求(batch, seq_len, input_size),除非你设了batch_first=False。
参数说明:look_back的经验值是取一个完整周期,比如日数据取 7 或 30,小时数据取 24 或 168。horizon根据业务定,期末大作业一般取 1 就够了。归一化器一定要保存,预测完要用它反归一化,否则你得到的是一堆 0 到 1 之间的小数,没法解释。
2.3 训练集、验证集、测试集的切分顺序
时间序列不能随机打乱切分,这是血泪经验。随机切分会让未来信息泄漏到训练集,模型在测试集上表现虚高,实际部署就崩。正确做法是按时间顺序切:前 70% 训练,中间 15% 验证,最后 15% 测试。如果数据量小,可以 80/20 两段切。
n = len(X) train_end = int(n * 0.7) val_end = int(n * 0.85) X_train, y_train = X[:train_end], y[:train_end] X_val, y_val = X[train_end:val_end], y[train_end:val_end] X_test, y_test = X[val_end:], y[val_end:] print(X_train.shape, X_val.shape, X_test.shape)切完之后检查一下三个集合的均值和方差,如果差异很大,说明数据分布不稳定,可能需要做差分或者滚动归一化。这一步很多人跳过,结果训练 loss 降得很漂亮,测试集一塌糊涂,回头找原因找半天。
3. 模型搭建与训练:Keras 和 PyTorch 两条路怎么选
3.1 用 Keras 搭一个最小可用的 LSTM 预测模型
Keras 适合期末大作业和快速验证,代码短,报错友好。下面是一个两层 LSTM 加全连接输出的结构,输入形状是(look_back, 1)。
from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from tensorflow.keras.optimizers import Adam model = Sequential() # 第一层 LSTM,return_sequences=True 才能接第二层 LSTM model.add(LSTM(64, return_sequences=True, input_shape=(look_back, 1))) model.add(Dropout(0.2)) # 第二层 LSTM,只返回最后一个时间步的输出 model.add(LSTM(32, return_sequences=False)) model.add(Dropout(0.2)) # 输出层,单值预测 model.add(Dense(1)) model.compile(optimizer=Adam(learning_rate=0.001), loss='mse', metrics=['mae']) model.summary() history = model.fit( X_train, y_train, epochs=50, batch_size=32, validation_data=(X_val, y_val), verbose=1 )参数说明:第一层LSTM(64)的 64 是隐藏单元数,数据量大可以加到 128,数据量小就降到 32。return_sequences=True是接第二层 LSTM 的必要条件,忘了会报形状错误。Dropout(0.2)是防过拟合的,如果训练 loss 和验证 loss 差距大,可以加到 0.3 或 0.4。learning_rate=0.001是 Adam 的常用起点,loss 震荡就降到 0.0005,收敛太慢就升到 0.002。
逻辑说明:model.fit返回的history里存了每轮的 loss 和 mae,后面画学习曲线要用。batch_size=32是默认值,数据量小于 1000 可以改成 16 或 8,梯度更新更频繁,收敛更稳。epochs=50不是越多越好,要看验证 loss 什么时候不再下降,配合EarlyStopping更省事。
3.2 PyTorch 版本的 LSTM 模型与训练循环
如果你用 PyTorch,结构要自己写,但控制力更强。下面是一个等价的单层 LSTM 模型。
import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset class LSTMPredictor(nn.Module): def __init__(self, input_size=1, hidden_size=64, num_layers=2, output_size=1): super().__init__() self.hidden_size = hidden_size self.num_layers = num_layers self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True) self.fc = nn.Linear(hidden_size, output_size) def forward(self, x): # x: (batch, seq_len, input_size) h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size) c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size) out, _ = self.lstm(x, (h0, c0)) # 取最后一个时间步 out = self.fc(out[:, -1, :]) return out # 转成 Tensor X_train_t = torch.tensor(X_train, dtype=torch.float32) y_train_t = torch.tensor(y_train, dtype=torch.float32).unsqueeze(1) train_loader = DataLoader(TensorDataset(X_train_t, y_train_t), batch_size=32, shuffle=False) model = LSTMPredictor() criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=0.001) for epoch in range(50): model.train() total_loss = 0 for xb, yb in train_loader: optimizer.zero_grad() pred = model(xb) loss = criterion(pred, yb) loss.backward() optimizer.step() total_loss += loss.item() print(f'Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.6f}')参数说明:hidden_size=64和 Keras 版对应,num_layers=2表示两层 LSTM。batch_first=True让输入格式变成(batch, seq_len, feature),和 Keras 一致,不设的话要自己转置。shuffle=False在时间序列训练里通常要设 False,因为 DataLoader 的 shuffle 会打乱样本顺序,虽然对 LSTM 的梯度传播影响不大,但保持时间顺序更符合直觉。
逻辑说明:h0和c0是 LSTM 的初始隐藏状态和细胞状态,全零初始化是常见做法。out[:, -1, :]取最后一个时间步的输出,因为我们要用整个窗口的信息预测下一个点。训练循环里optimizer.zero_grad()不能漏,漏了梯度会累加,loss 直接爆炸。
3.3 训练过程要盯哪几个信号
训练不是设完参数就等结果。我一般会盯三个信号:训练 loss 和验证 loss 的差距、loss 下降的平滑度、以及验证 loss 的最低点出现在第几轮。
如果训练 loss 持续下降但验证 loss 先降后升,说明过拟合,加 Dropout 或减层数。如果两个 loss 都震荡,说明学习率太大或 batch_size 太小。如果两个 loss 都不降,先检查数据归一化和窗口构造,再检查模型输入形状。
import matplotlib.pyplot as plt plt.plot(history.history['loss'], label='train_loss') plt.plot(history.history['val_loss'], label='val_loss') plt.xlabel('Epoch') plt.ylabel('MSE Loss') plt.legend() plt.show()这张图是判断模型状态最直接的工具。验证 loss 最低点对应的轮数,就是你应该停下来的轮数。Keras 里可以用EarlyStopping(patience=10)自动停,PyTorch 里要自己写判断逻辑。
4. 预测、反归一化与评估:别让曲线骗了你
4.1 反归一化与预测结果对齐
模型输出的是归一化后的值,必须用同一个 scaler 反变换回原始量纲。这一步看起来简单,但对齐容易错。
# 预测 y_pred_scaled = model.predict(X_test) # 反归一化 y_pred = scaler.inverse_transform(y_pred_scaled) y_true = scaler.inverse_transform(y_test.reshape(-1, 1)) # 对齐时间轴 test_start = look_back + val_end time_index = df.index[test_start:test_start + len(y_pred)] result = pd.DataFrame({ 'true': y_true.flatten(), 'pred': y_pred.flatten() }, index=time_index) print(result.head())逻辑说明:scaler.inverse_transform要求输入是二维数组,所以y_test要先reshape(-1,1)。test_start的计算要跟窗口构造时的偏移对齐,否则时间轴会错位。这个错位在图上表现为预测曲线整体平移,很多人以为是模型滞后,其实是索引算错了。
参数说明:如果归一化时用的是多列数据,inverse_transform要传入相同列数的数组,单变量预测时只传预测列即可。如果训练时用了差分,反变换还要把差分加回去。
4.2 评估指标:MAE、RMSE、MAPE 各自说明什么
单看一条曲线不够,要有数值指标。常用的三个:MAE 反映平均绝对误差,RMSE 对大误差更敏感,MAPE 是百分比误差,适合跨量纲比较。
from sklearn.metrics import mean_absolute_error, mean_squared_error mae = mean_absolute_error(y_true, y_pred) rmse = np.sqrt(mean_squared_error(y_true, y_pred)) mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100 print(f'MAE: {mae:.4f}') print(f'RMSE: {rmse:.4f}') print(f'MAPE: {mape:.2f}%')参数说明:MAPE 在y_true接近 0 时会爆炸,如果数据里有零值,改用 SMAPE 或直接看 MAE。RMSE 和 MAE 的比值能反映误差分布,比值接近 1 说明误差均匀,比值大说明存在个别大误差点。
注意:不要只报一个指标。期末大作业里常见只写 MSE,但 MSE 的量纲是原始值的平方,不好解释。至少报 MAE 和 MAPE,答辩时能说清楚。
4.3 画一张能放进报告的预测对比图
图要能看出三件事:真实值的趋势、预测值的跟随程度、误差最大的区间。
plt.figure(figsize=(12, 5)) plt.plot(result.index, result['true'], label='True', linewidth=1.5) plt.plot(result.index, result['pred'], label='Pred', linewidth=1.5, linestyle='--') plt.fill_between(result.index, result['true'], result['pred'], alpha=0.2, color='gray') plt.xlabel('Time') plt.ylabel('Value') plt.legend() plt.title('LSTM Prediction vs True') plt.tight_layout() plt.show()fill_between把误差区域填灰,一眼能看出哪里预测得差。如果误差集中在峰值附近,说明模型对突变捕捉不够,可以加特征或改用注意力机制。如果误差整体平移,检查反归一化和索引对齐。
5. 避坑与排查:LSTM 时间序列预测最常见的 5 个翻车现场
5.1 现象:训练 loss 正常但预测是一条直线
原因:模型学到了均值,没有学到序列的动态。常见于look_back太小、归一化范围不对、或者输出层激活函数用错。
解决:把look_back增大到一个完整周期以上;检查归一化是不是在训练集上 fit 的;输出层不要加sigmoid或tanh,回归任务用线性输出。
5.2 现象:验证 loss 比训练 loss 低很多
原因:验证集太小或分布和训练集差异大,也可能是 Dropout 在验证时没关闭。
解决:检查切分比例,验证集至少占总样本 10%;Keras 的 Dropout 在evaluate和predict时自动关闭,PyTorch 要手动model.eval()。
5.3 现象:预测结果整体滞后一个窗口
原因:窗口构造时标签偏移算错,或者反归一化时索引没对齐。
解决:重新检查create_dataset里的i + look_back + horizon - 1,确认标签取的是窗口之后第horizon个点;检查test_start的计算是否和窗口偏移一致。
5.4 现象:换数据集后报形状错误
原因:新数据集的列数或时间步长和旧的不一致,input_shape没改。
解决:把input_shape=(look_back, n_features)里的n_features改成实际特征数;多变量时确认create_dataset返回的X是三维的。
5.5 现象:GPU 显存不够或训练速度极慢
原因:batch_size太大、look_back太长、或者模型层数太多。
解决:先把batch_size降到 16 或 8;look_back超过 200 时考虑用卷积层先降维;层数从 2 层开始加,不要一上来就堆 4 层。
6. 把这份源码改成能写进简历的项目:三个进阶技巧
6.1 用多变量输入提升预测精度
单变量预测只看历史值,多变量可以加入时间特征、周期特征、外部变量。比如销量预测可以加入星期几、是否节假日、促销标记。做法是把这些列一起归一化,然后create_dataset返回的X形状变成(样本数, look_back, 特征数)。
# 假设 df 有多列:value, dayofweek, is_holiday features = df[['value', 'dayofweek', 'is_holiday']].values scaler = MinMaxScaler() scaled = scaler.fit_transform(features) def create_multivariate_dataset(data, look_back=24, horizon=1, target_col=0): X, y = [], [] for i in range(len(data) - look_back - horizon + 1): X.append(data[i:i + look_back, :]) y.append(data[i + look_back + horizon - 1, target_col]) return np.array(X), np.array(y) X, y = create_multivariate_dataset(scaled, look_back=24, horizon=1, target_col=0) print(X.shape) # (样本数, 24, 3)target_col=0表示预测第一列,其他列作为辅助特征。模型结构不用大改,只把input_shape改成(look_back, 3)。
6.2 用早停和模型保存避免白跑
训练轮数设大一点,配合早停,验证 loss 不降就停,同时保存最佳模型。
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint callbacks = [ EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True), ModelCheckpoint('best_lstm.h5', monitor='val_loss', save_best_only=True) ] history = model.fit( X_train, y_train, epochs=200, batch_size=32, validation_data=(X_val, y_val), callbacks=callbacks, verbose=1 )patience=10表示验证 loss 连续 10 轮不降就停,restore_best_weights=True把权重恢复到最佳轮次。ModelCheckpoint保存最佳模型,后面可以直接加载做预测,不用重新训练。
6.3 一个我常用的验证习惯
每次改完窗口长度或模型结构,我不会直接跑全量数据。先取前 500 个点跑一遍,看 loss 曲线和预测图,确认没有形状错误和明显滞后,再上全量。这个习惯帮我省了很多次白等一小时的训练。
另外,我会把每次实验的look_back、hidden_size、learning_rate、MAE记在一个表格里,跑过五六组之后就能看出哪组参数最稳。这个表格也是期末大作业报告里「参数调优」部分最好的素材。
希望帮到你。
本文还有配套的精品资源,点击获取