PyTorch LSTM多变量多步股票预测完整实战：从数据预处理到模型训练

简介面向金融时序预测与深度学习初学者的PyTorch LSTM多变量多步股票预测资源包基于真实股票数据600519.xlsx演示从数据预处理、特征构建、序列到序列建模到多步预测与评估的完整流程适合希望系统掌握LSTM在金融领域应用的读者。压缩包内共11个文件包含PyTorch主脚本、数据文件、结果可视化图及工程配置说明文档涵盖py、xlsx、png、xml等整体仅222KB轻量易上手便于快速复现和二次开发。已有9240人浏览学习可见该案例对同类任务有较高参考价值。通过项目主脚本中的LSTM训练与预测代码可掌握多步预测中教师强制、MSE损失函数、Adam优化器及训练验证策略等关键细节资源还体现数据标准化、缺失值填补、特征组合等完整技术路线并附带对比图辅助评估预测效果兼顾原理理解与实战落地。此外工程配置与说明文档便于在IDE中直接打开项目复现。1. 多变量多步预测没有你想的那么简单先看清这个 PyTorch LSTM 项目用 PyTorch 的 LSTM 做股票多变量多步预测最容易翻车的地方大概率不在模型结构而在数据窗口怎么切。这个项目把开盘、最高、最低、收盘、成交量、成交额六列日线数据组织成输入序列让 LSTM 一次性输出未来 5 个交易日的收盘价曲线。它适合两类人一类是刚接触深度学习的开发者想打通“数据预处理 → 模型 → 训练 → 评估”的完整链路另一类是手头已有行情数据、想给策略加一个预测模块的量化爱好者。整份源码的核心就三块滑动窗口切数据、StockLSTM 模型、训练评估脚本下面按章节拆开讲。2. 项目定位与技术选型一场「多进多出」的序列预测2.1 输入输出先对齐模型吃进什么吐出什么先把这个项目的输入输出说得再直白一点。原始行情表是日线维度字段至少包含 date、open、high、low、close、volume、amount 这 7 列。模型一次只看过去 20 个交易日也就是一个 (20, 6) 的二维矩阵6 代表除 date 外的全部数值特征模型要回答的问题是未来 5 个交易日的收盘价会落在什么位置即 (5,) 的一维向量。所以整条 pipeline 的输出维度是 (batch, 20, 6) → (batch, 5)这也是后面所有代码的默认形状。字段含义是否参与模型输入date交易日日期否只用于排序open开盘价是high最高价是low最低价是close收盘价是同时作为标签volume成交量是amount成交额是在做数据对齐时我一般只保留真实存在于交易日历上的行而不是用自然日去补。如果某个交易日因为停牌缺少一行宁可删掉也不要 fillna 塞一个假行情进去否则模型会学到一个并不存在的价格水平。这一步排干净之后才能进入滑动窗口切分。2.2 项目目录结构几个文件撑起一个完整训练链路stock-lstm-predict/ ├── data/ │ └── stock_data.csv # 日线行情7 个字段 ├── src/ │ ├── data_preprocess.py # 清洗 滑动窗口切分 归一化 │ ├── model.py # StockLSTM 定义 │ ├── train.py # 训练、验证、保存 checkpoint │ └── predict.py # 加载模型、画预测曲线 └── requirements.txt # pytorch、pandas、numpy、sklearn、matplotlib结构与大多数单机训练 demo 一样没有用工程化的 mlflow、config 管理都是为了让人更快看明白。data_preprocess.py 只负责把 CSV 变成 TensorDatasetmodel.py 只做模型定义train.py 承担训练、评估、保存三件事predict.py 在训练结束后加载权重做可视化。后期想换数据只需要修改 data_preprocess.py 里的字段列表想换模型只需要改 model.py 的 forward。2.3 LSTM 为什么能处理股票序列门控逻辑与时间步展开在深度学习里处理序列数据LSTM 是最被熟悉的一种结构但先说清一个预期股票数据噪声非常大LSTM 不会变成印钞机。这个项目里我把它当作一个序列特征抽取器来用而不是预测神器。LSTM 的核心是三组门——遗忘门、输入门、输出门它们在每个时间步决定“上一刻的记忆保留多少、新进来的行情更新多少、当前步输出什么”。与普通 RNN 相比它在长序列梯度反传时的衰减被门控结构抑制住了所以拿 20 天窗口训练一般不会像老式 RNN 那样梯度消失到完全学不动。在 PyTorch 里你不需要自己写门控矩阵nn.LSTM 封装了全部实现但理解它按时间步展开的机制对调参有帮助batch_firstTrue 意味着输入张量的第一维是 batch第二维是时间步第三维是特征否则默认是 (seq_len, batch, input_size)。这次代码统一用 batch_firstTrue因为从 torch 的 Dataset 出来的张量更习惯把 batch 放最前面。2.4 多步预测的三种做法直接多步、递归多步和 seq2seq做法训练方式推理方式误差累积实现成本递归多步每步输出 1 个点上次输出作为下次输入明显越远越偏低直接多步一次输出 M 个点一次输出 M 个点各步独立无累积中Seq2Seq编码器 解码器解码器逐位生成相对可控但训练复杂高这个项目选了第二种直接多步模型全连接头的输出维度直接设成 output_step5训练和推理完全一致不需要 teacher forcing也不需要在推理时把预测值循环喂回去。递归多步虽然模型简单但每一步都带着上一步的误差往前走预测第 5 天时误差已经被放大很多轮seq2seq 在数据量少的股票场景容易过拟合对入门项目来说性价比反而不高。M 的值也不是越大越好5 天和 20 天的难度完全不同预测步长越长模型能依赖的中期趋势信息越少损失会显著上升。如果你第一次跑发现第 4 天、第 5 天误差大不要先加复杂模型先看输出步长是不是设得太长。提示这个项目只做技术验证输出的预测结果不构成任何投资建议。3. 数据准备与滑动窗口从 CSV 到 TensorDataset 的完整链路3.1 原始行情加载排序、去空、处理停牌import pandas as pd def load_stock_data(path): df pd.read_csv(path) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) # 数值列里如果出现空值先看是停牌还是数据缺失 required [open, high, low, close, volume, amount] df df.dropna(subsetrequired).reset_index(dropTrue) return dfdate 的格式在不同数据源里可能不一样先统一成 datetime 再排序否则后续按时间轴分隔数据时顺序会错乱。dropna 丢弃的是数据源里实打实缺的行而不是用上一日值填充如果一定要处理停牌我一般会检查缺失日期是否连续连续超过 5 个交易日就直接删掉这一段避免把“不存在”的价格学进模型。3.2 滑动窗口切分让模型理解时间上下文def make_sequences(df, feature_cols, input_len20, output_len5): feature_data df[feature_cols].to_numpy(dtypenp.float32) close_col feature_cols.index(close) x, y [], [] for i in range(len(df) - input_len - output_len 1): x.append(feature_data[i: i input_len]) y.append(feature_data[i input_len: i input_len output_len, close_col]) return np.array(x), np.array(y) # x: (N, 20, 6), y: (N, 5)这里窗口参数是 input_len20、output_len5。第 i 个样本用的是第 i 到 i19 天做输入第 i20 到 i24 天收盘价做标签特征窗口和标签窗口严格不重叠。如果你只想预测未来 1 天把 output_len 改成 1 即可窗口大小不要拍脑袋可以先看数据总量1000 个交易日大概能切成 976 个样本样本太少就把 output_len 降到 3。3.3 归一化与反变换scaler 只允许 fit 训练段from sklearn.preprocessing import MinMaxScaler def fit_train_scaler(x_all, y_all, train_ratio0.8): n_train int(len(x_all) * train_ratio) output_len y_all.shape[1] feature_scaler MinMaxScaler(feature_range(0, 1)) feature_scaler.fit(x_all[:n_train].reshape(-1, x_all.shape[-1])) def scale_samples(samples, scaler): n, seq, f samples.shape flat samples.reshape(-1, f) scaled scaler.transform(flat) return scaled.reshape(n, seq, f) x_train scale_samples(x_all[:n_train], feature_scaler) x_val scale_samples(x_all[n_train:], feature_scaler) y_scaler MinMaxScaler(feature_range(0, 1)) y_scaler.fit(y_all[:n_train].reshape(-1, 1)) y_train y_scaler.transform(y_all[:n_train].reshape(-1, 1)).reshape(-1, output_len) y_val y_scaler.transform(y_all[n_train:].reshape(-1, 1)).reshape(-1, output_len) return (x_train, y_train), (x_val, y_val), feature_scaler, y_scalerfit 只发生在训练段验证段对训练段归一化后才 transform缩放参数不接触验证数据。这样也符合实盘场景预测未来时你没有未来数据来算全量 min 和 max。为什么用 MinMax 而不是 StandardScaler股价的数值区间相对有限MinMax 可以保留每个时刻的相对位置而且反变换回真实价格时逻辑最简单。反变换是 predict.py 里必然用到的一步模型输出的是 0~1 区间值必须用 y_scaler.inverse_transform 转回真实价格才能画图和计算指标。3.4 DataLoader 组织shuffle 该开就开from torch.utils.data import TensorDataset, DataLoader train_dataset TensorDataset( torch.tensor(x_train, dtypetorch.float32), torch.tensor(y_train, dtypetorch.float32) ) val_dataset TensorDataset( torch.tensor(x_val, dtypetorch.float32), torch.tensor(y_val, dtypetorch.float32) ) train_loader DataLoader(train_dataset, batch_size64, shuffleTrue) val_loader DataLoader(val_dataset, batch_size128, shuffleFalse)训练集 shuffleTrue 是为了让每个 batch 的样本在时间上打乱帮助优化器跳出局部震荡验证集 shuffleFalse 则保证评估时样本顺序固定便于复现结果。滑动窗口生成的相邻样本会共享大量时间上下文所以评估时我不会打乱顺序并且会在验证集上计算整体指标而不是按每个 batch 分别算。4. 核心模块拆解LSTM forward 到 loss 反传的完整链路4.1 StockLSTM 定义多层 LSTM 加一个全连接头import torch.nn as nn class StockLSTM(nn.Module): def __init__(self, input_size6, hidden_size64, num_layers2, output_step5): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue ) self.fc nn.Sequential( nn.Linear(hidden_size, hidden_size // 2), nn.ReLU(), nn.Linear(hidden_size // 2, output_step), ) def forward(self, x): # x: (batch, input_len20, input_size6) _, (h_n, _) self.lstm(x) last_hidden h_n[-1] # (batch, hidden_size) out self.fc(last_hidden) # (batch, output_step) return outinput_size 直接对应 feature_cols 的特征数量 6output_step 对应未来 5 天收盘价。模型用 h_n[-1] 而不是 LSTM 每个时间步的 outputs是因为我们要一次性预测后面 5 天信息量集中在最后一个时间步的隐藏状态里。hidden_size 和 num_layers 是超参股票这种千级别样本量64 和 2 是比较稳妥的起点数据量更小可以降到 32 和 1防止过拟合。相比 TransformerLSTM 在这种几千条日线数据的体量下不容易过拟合训练也快。4.2 训练循环梯度裁剪与学习率衰减是组合拳def evaluate(model, loader, criterion): model.eval() total_loss, n 0.0, 0 with torch.no_grad(): for bx, by in loader: pred model(bx) total_loss criterion(pred, by).item() * bx.size(0) n bx.size(0) model.train() return total_loss / n optimizer torch.optim.Adam(model.parameters(), lr1e-3) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5 ) criterion nn.MSELoss() for epoch in range(50): model.train() epoch_loss 0.0 for batch_x, batch_y in train_loader: optimizer.zero_grad() pred model(batch_x) loss criterion(pred, batch_y) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() epoch_loss loss.item() * batch_x.size(0) avg_train_loss epoch_loss / len(train_loader.dataset) val_loss evaluate(model, val_loader, criterion) scheduler.step(val_loss) print(fepoch {epoch:03d} | train {avg_train_loss:.6f} | val {val_loss:.6f})clip_grad_norm 在 backward 之后、step 之前执行作用是把整个参数梯度向量按模长压缩到 1.0 以内。LSTM 梯度跨 20 个时间步累积不裁剪的话遇到异常行情 batchloss 很容易跳到 NaN。scheduler 用 val_loss 而不是 train_loss是为了让学习率在验证集不再下降时才减半避免模型已经在训练集上过拟合了还继续用大的学习率往前冲。epoch 一般先跑 50 轮重点看 val_loss 是否在 20 轮后进入平台期如果还在持续下降可以往上加到 80。4.3 评估指标MAE、MAPE 和方向准确率import numpy as np def evaluate_metrics(y_true, y_pred): y_true np.asarray(y_true) y_pred np.asarray(y_pred) mae np.mean(np.abs(y_true - y_pred)) mape np.mean(np.abs((y_true - y_pred) / (np.abs(y_true) 1e-6))) * 100 # 方向准确率比较相邻两个预测日的涨跌方向 sign_true np.sign(np.diff(y_true, axis1)) sign_pred np.sign(np.diff(y_pred, axis1)) direction_acc np.mean(sign_true sign_pred) return {mae: mae, mape: mape, direction_acc: direction_acc}为什么关注方向准确率而不是只盯着 mae。股票预测里价格误差小不一定方向对比如真实收盘 10 元涨到 11 元模型预测 10.1 元绝对误差不大但方向完全反了。direction_acc 比较的是相邻两个预测日的变化方向能直接反映模型对涨跌方向的判断能力。分母加 1e-6 是为了防止 y_true 恰好为 0 时出现除零。评估多步预测时只算总体的 mae 会掩盖第 1 天准、第 5 天飘的问题我习惯再加一步按列分别计算第 1 天到第 5 天的 mae 并打印成 5 个数字一眼就能看出误差随预测距离放大的趋势。4.4 保存与加载checkpoint 里带上配置和 scalerimport torch torch.save({ model_state: model.state_dict(), optimizer_state: optimizer.state_dict(), feature_scaler: feature_scaler, y_scaler: y_scaler, config: { input_size: 6, hidden_size: 64, num_layers: 2, input_len: 20, output_step: 5, } }, checkpoint.pt) ckpt torch.load(checkpoint.pt) model StockLSTM( input_sizeckpt[config][input_size], hidden_sizeckpt[config][hidden_size], num_layersckpt[config][num_layers], output_stepckpt[config][output_step], ) model.load_state_dict(ckpt[model_state]) model.eval()很多只存 model_state 的做法有一个问题模型结构变了就加载失败也不知道当初用的什么窗口长度。把 config 和两个 scaler 一起存进 checkpoint是让项目能够一键恢复到训练时状态的最稳做法。load 之后记得调用 model.eval()因为训练时的 dropout 与 batch norm 行为在推理时应该关闭。5. 避坑记录数据泄漏、梯度爆炸与均值回归预测的 5 个真实案例5.1 数据泄漏验证集 loss 好看换一段行情就翻车现象训练跑 30 个 epoch验证集 mae 漂亮得不像话预测曲线和真实曲线几乎贴在一起但把模型拿到后一段新行情上预测线却严重滞后。原因归一化时把训练、验证、测试全部扔进一个 scaler 去 fit测试集的统计量提前进入了模型这是时序项目最常见的数据泄漏。解决把 fit 严格限定在训练段验证段只用 transform3.3 的代码已经按这个顺序写好了。检查方法也很简单打印 scaler.data_min_ 和 data_max_看它是不是等于训练段的最小最大值而不是全量数据。5.2 标签泄漏预测曲线整体右移一天训练 loss 低到离谱现象模型看起来学到了什么但仔细看预测线整体滞后真实线约一天loss 比正常水平低好几倍。原因构造样本时把输出窗口也并进了输入窗口典型的错误写法是 feature_data[i: iinput_lenoutput_len] 作为 x再取同区间的收盘价当标签模型其实直接把标签里的信息抄了一遍。解决回看 3.2 的区间公式输入区间结束位置严格等于 iinput_len标签从 iinput_len 开始二者没有重叠。改完代码可以打印两个样本人工确认第 20 天的特征和第 21 天的标签来自不同日期。5.3 梯度爆炸训练第 10 个 epoch 后 loss 变成 NaN现象训练前几轮正常第 10 轮左右 loss 突然变成 nan后续回不来检查参数发现 LSTM 权重里出现了 inf。原因特征没归一化或归一化边界不对股价从几元到几千元不等量纲差异大LSTM 内部矩阵乘法的梯度跨时间步累积后溢出另一个常见原因是学习率设到 1e-2。解决先确认 y_scaler 对标签做了归一化再把 lr 降到 1e-3最后在 optimizer.step() 之前加 clip_grad_norm_。这组组合要同时调整不要只加裁剪。5.4 递归多步误差累积多步推理越推越偏现象模型单独预测明天很准用上一步预测值循环预测未来 5 天时第 3 天开始明显偏离第 5 天可能飘到不能看。原因训练时 LSTM 每一步都吃真实值推理时吃的是自己上一轮的输出预测误差被一次次迭代放大。解决项目默认用直接多步输出模型一步给出 5 天结果训练和推理一致从根上避免累积。想验证这个现象把 output_step 改成 1在 predict.py 里写个 for 循环对比一下直接多步和递归多步在第 5 天的误差。5.5 均值回归预测线牢牢贴着近期均值涨跌方向几乎给不出现象预测曲线像一条温水线在最近几天收盘价均值附近轻微波动连明显的单边趋势都拟合不出来。原因MSE 损失在噪声大的数据上模型发现输出区域中心的代价最小于是学会了“贴均值”这在统计上是最小均方误差的无偏估计但对交易没有意义。解决两个方向一是把原始价格换算成收益率再训练让模型预测涨跌幅度而不是绝对价格二是在损失函数里加方向约束下一章给出具体代码。另外可以把输入窗口从 20 天拉到 60 天给模型更多趋势上下文。6. 二次开发实战换数据、加方向约束、做滚动回测6.1 替换自己的行情数据拿到一份新数据先别急着改模型。打开 data_preprocess.py把 load_stock_data 的路径换成新 CSV逐一核对 feature_cols 里的列名缺一列就补一列字段值单位如果不同先统一再进归一化。同步检查输入窗口与预测步长数据量少于 500 条的把 input_len 降到 15、output_step 降到 3否则样本数量不足以支撑 LSTM 参数学习。model.py 里的 input_size 要和 feature_cols 长度一致这是最容易起步就报错的地方。6.2 损失函数加方向约束def directional_loss(pred, target, mse_weight1.0, direction_weight0.2): mse nn.functional.mse_loss(pred, target) sign_pred torch.sign(pred[:, 1:] - pred[:, :-1]) sign_true torch.sign(target[:, 1:] - target[:, :-1]) direction_penalty torch.mean((sign_pred ! sign_true).float()) return mse_weight * mse direction_weight * direction_penaltydirection_weight 从 0.1 开始试太大模型会为了方向正确牺牲价格精度预测点只有 5 个diff 之后出现 4 段方向比较计算开销可以忽略。6.3 滚动回测验证for start in range(0, total_len - train_len - test_len, step): # 每次前移 step 天重训一次模型 x_train, y_train cut_window(start, start train_len) x_test, y_test cut_window(start train_len, start train_len test_len) model retrain_on(x_train, y_train) metrics evaluate_metrics(y_test, model.predict(x_test)) print(metrics)每次只取最近 train_len 条训练在紧随其后的 test_len 上验证然后整个窗口往前挪 step 天再重来。滚动回测比一次性切分更接近真实交易中的顺序指标也更可信。这个验证循环是二次开发中性价比最高的一步。这五件事里最让我长记性的是 5.1 那条数据泄漏。从那以后我每次换数据集都会先打印一遍 scaler 的 fit 范围、检查预测曲线有没有滞后一天再跑正式训练花十分钟自检能省掉后面几十个 epoch 的返工。希望帮到你。本文还有配套的精品资源点击获取