简介:这份资源面向需要完成时间序列预测课程设计、期末大作业或入门深度学习实战的学生与开发者,核心是用Python实现基于LSTM的股票收盘价预测。压缩包共30个文件,约1.83MB,包含1个可直接运行的py主程序、3个xlsx数据表与1个csv样本数据,另有16张png原理图、2份md分析报告及若干xml、iml等工程配置,兼顾代码、数据与讲解。内容从RNN节点结构、LSTM与RNN的区别、中间变量与计算过程等基础原理展开,再落到用时间序列模型学习股票收盘价并预测未来价格的完整流程,配有分析报告帮助理解建模思路与结果解读。目前已有1792人学习下载,适合希望快速拿到可复现方案、对照原理图梳理LSTM内部机制并完成高分作业的读者参考。
1. 时间序列预测配 LSTM:一份能跑通的 Python 代码到底长什么样
拿到「时间序列预测(LSTM模型)python代码实现(95分以上).zip」这个标题,多数人第一反应是找一份能直接跑的代码。但真正卡住人的从来不是模型本身,而是从原始时序数据到 LSTM 输入张量之间那段没人讲清楚的预处理链路。我见过太多人把 CSV 丢进model.fit()然后对着一条水平直线怀疑人生——问题不在 LSTM,在于窗口怎么切、归一化在哪一步做、lookback设多少。这篇笔记按一条完整落地路径拆:先讲清 LSTM 做时序预测的输入输出契约,再给可复现的 Python 代码,最后把调参和踩坑摊开。适合已经会 Python 基础语法、想把这套东西真正跑在自己数据上的人,也适合跑通过一次但结果不稳定、想搞清楚边界在哪的人。
2. LSTM 时序预测的输入契约:窗口、维度与归一化顺序
2.1 为什么原始时序不能直接喂给 LSTM
LSTM 的输入张量形状是(samples, timesteps, features)。一列按时间排列的数值,既没有samples也没有features,直接 reshape 成三维只会得到一堆无意义的切片。核心操作是滑动窗口:用前 N 个时刻的值预测第 N+1 个时刻的值。这个 N 就是lookback,也叫时间步长。
假设你有 1000 个连续观测点,lookback=24,那么能构造出的样本数是1000 - 24 = 976个。每个样本的输入是连续 24 个点,标签是第 25 个点。这一步决定了模型能"看到"多长的历史依赖,是整套流程里第一个必须显式设定的参数。
常见做法是先做差分或去趋势再切窗口,但差分会让预测目标从"原始值"变成"变化量",评估指标的含义也跟着变。我一般建议第一版不做差分,先把原始值预测跑通,看到 baseline 之后再决定要不要加。
2.2 归一化的时机比方法更重要
归一化本身不复杂,MinMaxScaler或StandardScaler一行搞定。真正容易翻车的是时机:必须在切窗口之前对整列数据做 fit,然后用同一个 scaler 做 inverse_transform 还原预测值。如果先切窗口再对每个窗口单独归一化,每个样本的缩放基准都不一样,模型学到的模式会被破坏。
还有一个更隐蔽的问题:如果用全量数据 fit scaler,测试集的极值信息会泄漏进训练过程。严谨做法是只用训练段 fit,然后 transform 验证段和测试段。下面代码里我会把这条边界标出来。
import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler def make_windows(series, lookback): """ series: 一维 numpy 数组,已归一化 lookback: 用前多少个时刻预测下一个时刻 返回 X shape=(n, lookback, 1), y shape=(n,) """ X, y = [], [] for i in range(len(series) - lookback): X.append(series[i : i + lookback]) y.append(series[i + lookback]) X = np.array(X).reshape(-1, lookback, 1) y = np.array(y) return X, y # 假设 df['value'] 是原始时序 raw = df['value'].values.reshape(-1, 1) split = int(len(raw) * 0.8) # 前 80% 做训练 scaler = MinMaxScaler() scaler.fit(raw[:split]) # 只用训练段 fit,避免泄漏 scaled = scaler.transform(raw).flatten() lookback = 24 X, y = make_windows(scaled, lookback) # 按时间顺序切分,不能 shuffle X_train, X_test = X[:split - lookback], X[split - lookback:] y_train, y_test = y[:split - lookback], y[split - lookback:]make_windows里reshape(-1, lookback, 1)的最后一个1是特征维度,单变量预测就是 1,多变量时改成对应列数。split - lookback这个偏移量容易写错:因为窗口构造消耗了前lookback个点,训练集的样本边界要相应左移,否则训练段和测试段会有重叠。
2.3 lookback 怎么选:从自相关出发而不是拍脑袋
lookback不是越大越好。设太小,模型看不到完整周期;设太大,参数量和训练时间上去,还容易过拟合。一个可操作的起点是看自相关函数(ACF):找到自相关系数第一次跌破置信区间的位置,或者业务周期的整数倍。
比如日频数据有明显周周期,lookback至少设 7;小时频数据有日周期,设 24 或 48。如果 ACF 在 lag=12 之后基本落在零附近,那lookback=24大概率是浪费。我一般会跑三组对比:lookback取周期长度、周期两倍、以及 ACF 截断点,看验证集 loss 再定。
3. 用 Keras 搭一个能收敛的 LSTM:层数、单元数与训练配置
3.1 模型结构:单层 LSTM 加全连接输出就够了
时序预测不是越深越好。单变量、数据量在几千到几万条这个量级,一层 LSTM 加一层 Dense 输出通常就能拿到合理结果。堆两层 LSTM 的收益在长序列、多变量场景才明显,而且第二层必须加return_sequences=True,否则维度对不上。
import tensorflow as tf from tensorflow.keras import layers, models, callbacks def build_lstm(lookback, units=64, dropout=0.2): model = models.Sequential([ layers.LSTM(units, input_shape=(lookback, 1)), layers.Dropout(dropout), layers.Dense(1) # 单步预测输出一个值 ]) model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3), loss='mse', metrics=['mae'] ) return model model = build_lstm(lookback=24, units=64) model.summary()units=64是 LSTM 隐藏状态的维度,不是层数。数据量小于 5000 条时我一般从 32 起步,大于 5 万条可以上 128。Dropout放在 LSTM 之后而不是 LSTM 内部,是因为 LSTM 内部的 dropout 在 Keras 里对 recurrent 连接的处理有版本差异,放外面更可控。
3.2 训练配置:EarlyStopping 是必须的后悔药
LSTM 训练最容易出现的情况是验证 loss 先降后升,而你没有在拐点停下来。EarlyStopping配合ModelCheckpoint是标准操作:
es = callbacks.EarlyStopping( monitor='val_loss', patience=10, restore_best_weights=True ) ckpt = callbacks.ModelCheckpoint( 'best_lstm.keras', monitor='val_loss', save_best_only=True ) history = model.fit( X_train, y_train, validation_split=0.1, # 从训练段再切 10% 做验证 epochs=200, batch_size=32, callbacks=[es, ckpt], verbose=1 )patience=10意味着验证 loss 连续 10 轮不下降就停。restore_best_weights=True保证模型回到验证 loss 最低的那一轮,而不是停在最后。batch_size=32是通用起点,数据量小可以降到 16,数据量大可以升到 64 或 128,但要注意学习率可能需要同步调整。
3.3 预测与还原:inverse_transform 的维度陷阱
模型输出的是归一化空间的预测值,必须用同一个 scaler 还原。这里最常见的错误是 scaler 的维度对不上:
pred_scaled = model.predict(X_test) # shape=(n, 1) pred = scaler.inverse_transform(pred_scaled) # 还原到原始量纲 true = scaler.inverse_transform(y_test.reshape(-1, 1)) from sklearn.metrics import mean_absolute_error, mean_squared_error mae = mean_absolute_error(true, pred) rmse = np.sqrt(mean_squared_error(true, pred)) print(f'MAE={mae:.4f}, RMSE={rmse:.4f}')如果训练时 scaler 是对单列 fit 的,inverse_transform的输入必须是(n, 1)形状。y_test是一维的,要先reshape(-1, 1)。这个细节不报错但会算出错误结果,属于典型的静默翻车。
4. 避坑与排查:LSTM 时序预测最常见的 5 个翻车现场
4.1 预测出来是一条直线
现象:模型输出几乎不随时间变化,MAE 看起来不大但完全没有预测能力。
原因:最常见的是归一化后数据范围太小(比如全在 0.4 到 0.6 之间),LSTM 直接学到了均值。其次是学习率过大,模型在第一步就跳到平凡解。
解决:检查归一化后数据的std,如果小于 0.1 考虑换StandardScaler。把学习率降到1e-4再跑一次。如果还是直线,检查窗口构造是否有误——打印X_train[0]和y_train[0]确认输入输出对应关系正确。
4.2 训练 loss 下降但验证 loss 震荡
现象:loss稳步下降,val_loss上下跳动不收敛。
原因:batch_size太小导致梯度噪声大,或者训练段和验证段的数据分布不一致(比如验证段恰好落在异常区间)。
解决:先把batch_size翻倍。如果无效,检查切分点附近的数据是否有突变。我一般会画一张训练段和验证段的叠图,肉眼确认两段分布没有系统性偏移。
4.3 多步预测误差累积爆炸
现象:单步预测还行,递归预测 10 步之后完全偏离。
原因:递归多步预测把上一步的预测值当作下一步的输入,误差指数级放大。这是 LSTM 做多步预测的固有问题,不是调参能解决的。
解决:改用直接多步策略——训练时输出维度直接设为预测步数,一次前向传播出所有步的预测值。代价是每个步数需要独立的输出头,样本构造也要相应调整。
4.4 GPU 显存够但训练极慢
现象:nvidia-smi显示显存占用很低,但每个 epoch 耗时远超预期。
原因:数据在 CPU 和 GPU 之间频繁拷贝,或者batch_size太小导致 GPU 利用率上不去。
解决:用tf.data.Dataset把数据预取到 GPU:
train_ds = tf.data.Dataset.from_tensor_slices((X_train, y_train)) \ .shuffle(1000).batch(64).prefetch(tf.data.AUTOTUNE)prefetch(tf.data.AUTOTUNE)让数据加载和模型计算重叠,通常能提速 30% 以上。
4.5 换了新数据后预测完全失效
现象:在原始数据集上表现良好,换一批同时段的新数据后 MAE 翻倍。
原因:scaler 是用旧数据 fit 的,新数据的取值范围超出了旧 scaler 的data_min_和data_max_,transform后值被截断到 [0,1] 边界。
解决:定期用滑动窗口重新 fit scaler,或者改用对异常值不敏感的RobustScaler。生产环境下我倾向于每预测一批就用最近 N 条数据重新 fit 一次 scaler,代价很小但能避免分布漂移导致的系统性偏差。
5. 把单变量 LSTM 扩到多变量与滚动预测的实操技巧
单变量跑通之后,下一步通常是加入外生变量——温度、节假日标记、滞后特征等。多变量 LSTM 的输入形状从(samples, lookback, 1)变成(samples, lookback, n_features),make_windows里那个1要改成实际特征数。但这里有个容易忽略的点:目标列和其他特征的归一化必须用各自的 scaler,不能混在一起 fit,否则量纲差异会让模型偏向数值大的特征。
from sklearn.preprocessing import MinMaxScaler feature_cols = ['value', 'temp', 'is_holiday'] scalers = {} scaled_df = pd.DataFrame() for col in feature_cols: s = MinMaxScaler() s.fit(df[col].values[:split].reshape(-1, 1)) scaled_df[col] = s.transform(df[col].values.reshape(-1, 1)).flatten() scalers[col] = s data = scaled_df.values # shape=(T, n_features) X, y = [], [] for i in range(len(data) - lookback): X.append(data[i : i + lookback, :]) y.append(data[i + lookback, 0]) # 只预测 value 列 X = np.array(X) # (n, lookback, n_features) y = np.array(y)y只取第 0 列(value),其他特征作为输入但不作为预测目标。还原时只用scalers['value']对预测结果做inverse_transform。
滚动预测是另一个实用技巧:每次只预测一步,把预测值追加到输入末尾,滑窗前进一格再预测下一步。这样可以在不重新训练的情况下做任意长度的预测,但误差会累积。我的习惯是滚动步数不超过lookback的一半,超过就重新训练一个直接多步模型。
最后说一个我自己的教训:不要在没有 baseline 的情况下直接上 LSTM。先跑一个"用昨天同时刻的值预测今天"的朴素模型,拿到 MAE 之后再对比 LSTM。如果 LSTM 没有显著优于朴素模型,问题大概率在数据质量或窗口构造上,不在模型结构。这个习惯帮我省过很多次无意义的调参时间。希望帮到你。
本文还有配套的精品资源,点击获取