Python基于LSTM预测股市：完整源码、模型与数据集搭建指南

简介：这份资源面向金融量化初学者与深度学习爱好者，提供一套基于LSTM长短期记忆网络的股市预测完整实现，帮助理解循环神经网络在时间序列预测中的应用。压缩包共19个文件，约3.92MB，包含9个Python脚本、2个CSV与1个Excel数据集、2个H5训练模型、2张预测对比图及说明文档，覆盖数据加载、预处理、模型构建、训练、评估与可视化全流程。代码以Keras搭建LSTM，对开盘价、收盘价、最高价、最低价和交易量等字段做清洗、归一化与序列化，并输出MSE、MAE等指标，同时涉及移动平均、指数平滑等技术指标与ARIMA对比思路。已有517人学习下载，适合作为深度学习落地金融预测的练手案例，读者可据此复现实验、调整参数并迁移到自有股票数据，同时需注意市场复杂性，模型预测不能替代风险管理与投资策略。

1. 用 LSTM 预测股市：一套能跑通的源码、模型与数据集该怎么搭

很多人第一次搜「Python基于LSTM模型实现预测股市源代码+模型+数据集」，心里想的其实是：有没有一份现成的东西，我拉下来改改就能看到预测曲线。现实是，网上散落的 lstm模型代码 大多只给一个model.fit()就结束，数据从哪来、特征怎么造、预测出来的价格为什么永远滞后一天，全都没交代。这篇就把这套东西从头搭一遍：用 Python 和 PyTorch 写一个 LSTM 时间序列预测的完整流程，包含数据获取与清洗、特征工程、模型定义、训练、回测评估，以及一套能直接复用的目录结构。适合会一点 Python、想入门 python量化交易策略代码 的从业者，也适合已经跑过 lstm预测 但结果不对劲、想搞清楚边界的人。全程不碰任何实盘接口，只做离线可复现的研究流程。

2. 数据与特征：LSTM 预测股市的输入到底怎么造

2.1 为什么原始收盘价直接喂给 LSTM 会翻车

把每日收盘价当成一维序列丢进 LSTM，是最常见也最容易失败的做法。原因有三个。第一，股价是非平稳序列，均值和方差随时间漂移，模型在训练集学到的数值范围到测试集就失效了。第二，原始价格量纲大，LSTM 的激活函数和梯度对输入尺度敏感，不归一化几乎不收敛。第三，单靠价格这一列，模型学不到成交量、波动率这些真正带信息量的维度。

常见做法是先做差分或对数收益率转换，把绝对价格变成相对变化，再叠加技术指标作为多变量输入。我一般会用「对数收益率 + 成交量变化率 + 滚动波动率 + MACD 柱」这四类特征，既保留了价格动态，又引入了量能信息。注意，所有滚动类指标都必须用shift(1)把当前 bar 的信息推到下一根，否则就是未来函数，回测会虚高得离谱。

2.2 用 pandas 造多变量特征并做窗口切分

下面这段代码负责从原始 OHLCV 数据生成特征矩阵，并切成 LSTM 需要的三维张量(样本数, 时间步, 特征数)。数据来源可以是本地 CSV，也可以是你自己用 python爬虫 抓下来存好的文件，这里只约定列名。

import numpy as np import pandas as pd def build_features(df: pd.DataFrame, window: int = 30) -> tuple: """ df 需包含列: open, high, low, close, volume，索引为日期 window: 回看时间步，即用过去多少天预测下一天 """ df = df.copy() # 对数收益率，平稳化处理 df['log_ret'] = np.log(df['close'] / df['close'].shift(1)) # 成交量变化率，加 1 防止除零 df['vol_chg'] = df['volume'].pct_change() # 10 日滚动波动率 df['volatility'] = df['log_ret'].rolling(10).std() # 简易 MACD 柱: 12 日 EMA 减 26 日 EMA，再取 9 日 EMA ema12 = df['close'].ewm(span=12, adjust=False).mean() ema26 = df['close'].ewm(span=26, adjust=False).mean() dif = ema12 - ema26 dea = dif.ewm(span=9, adjust=False).mean() df['macd_hist'] = (dif - dea) * 2 feat_cols = ['log_ret', 'vol_chg', 'volatility', 'macd_hist'] df = df[feat_cols + ['close']].dropna() # 标准化，只用训练段统计量，避免穿越 split = int(len(df) * 0.8) train_mean = df[feat_cols].iloc[:split].mean() train_std = df[feat_cols].iloc[:split].std() df[feat_cols] = (df[feat_cols] - train_mean) / train_std # 构造滑窗样本，标签是下一天的对数收益率 X, y, dates = [], [], [] values = df[feat_cols].values target = df['log_ret'].values for i in range(window, len(df) - 1): X.append(values[i - window:i]) y.append(target[i + 1]) dates.append(df.index[i + 1]) return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32), dates, df

逻辑说明：log_ret是预测目标也是特征之一，vol_chg、volatility、macd_hist提供额外维度。标准化用训练段前 80% 的均值和方差，这是防止数据泄漏的关键一步，很多人图省事用全量统计量，回测收益能好看一倍，实盘直接打回原形。window=30是回看天数，A 股一个月约 22 个交易日，30 天能覆盖一个多月的信息，参数可以按标的调整，日内高频数据要相应缩小。

参数说明：window太小（如 5）模型记不住趋势，太大（如 120）样本数骤减且梯度容易消失。volatility的滚动窗口 10 是经验值，波动剧烈的品种可以调到 20。MACD 参数用经典的 12/26/9，不建议在没做参数敏感性分析前乱改。

2.3 数据集切分与 DataLoader 的坑

时间序列不能随机打乱切分，必须按时间顺序切。常见做法是前 70% 训练、15% 验证、15% 测试，或者用滚动窗口做 walk-forward 验证。下面用 PyTorch 的TensorDataset和DataLoader封装，注意shuffle=False。

import torch from torch.utils.data import TensorDataset, DataLoader def make_loaders(X, y, batch_size=64): n = len(X) n_train = int(n * 0.7) n_val = int(n * 0.85) splits = { 'train': (0, n_train), 'val': (n_train, n_val), 'test': (n_val, n) } loaders = {} for name, (s, e) in splits.items(): ds = TensorDataset(torch.from_numpy(X[s:e]), torch.from_numpy(y[s:e])) loaders[name] = DataLoader(ds, batch_size=batch_size, shuffle=False) return loaders

逻辑说明：shuffle=False是时间序列的铁律，打乱会破坏时序依赖，模型学到的就是噪声。batch_size 设 64 是折中，太小梯度震荡，太大显存吃紧且泛化变差。切分比例不是死的，样本量少于 2000 时建议用 walk-forward，每次训练窗口向前滚动，评估更贴近实盘。

3. LSTM 模型定义与训练：从 nn.LSTM 到可收敛的损失曲线

3.1 PyTorch LSTM 层的输入输出维度怎么对齐

PyTorch 的nn.LSTM输入要求是(seq_len, batch, input_size)，除非设batch_first=True。很多人第一次写 lstm模型代码 就卡在维度上，报错expected input to have 3 dimensions。我一般统一用batch_first=True，输入变成(batch, seq_len, input_size)，和上面造的X形状一致。

模型结构上，单层 LSTM 加一个全连接输出层就够做基线。隐藏维度hidden_size设 64 或 128，层数num_layers从 1 开始试，超过 2 层在小数据集上基本过拟合。别忘了dropout，但 PyTorch 的 LSTM 只有在num_layers > 1时 dropout 才生效，单层要自己在输出后加nn.Dropout。

import torch.nn as nn class StockLSTM(nn.Module): def __init__(self, input_size=4, hidden_size=64, num_layers=1, dropout=0.2): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0.0 ) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_size, 1) def forward(self, x): # x: (batch, seq_len, input_size) out, (h_n, c_n) = self.lstm(x) # 取最后一个时间步的输出 last = out[:, -1, :] return self.fc(self.dropout(last)).squeeze(-1)

逻辑说明：out[:, -1, :]取序列最后一步的隐藏状态，代表模型对整段历史的压缩表示。squeeze(-1)把输出从(batch, 1)压成(batch,)，和标签形状对齐。input_size=4对应前面四个特征，特征增减时这个参数必须同步改，否则第一层矩阵乘法直接报错。

参数说明：hidden_size从 64 起步，数据量上万可以试 128；num_layers=1是基线，加到 2 层时记得 dropout 才会生效；dropout=0.2是常规正则，过拟合严重可以提到 0.3~0.5。学习率用 1e-3 配 Adam，损失函数用 MSE 或 Huber，Huber 对收益率里的极端值更稳。

3.2 训练循环与早停：损失不降时该看什么

训练循环本身不复杂，关键是加验证集监控和早停。下面这段是可直接复用的训练骨架，包含梯度裁剪，LSTM 不加裁剪容易梯度爆炸。

import torch import numpy as np def train_model(model, loaders, epochs=50, lr=1e-3, patience=8, device='cpu'): model = model.to(device) optimizer = torch.optim.Adam(model.parameters(), lr=lr) criterion = torch.nn.HuberLoss() best_val = float('inf') wait = 0 history = {'train': [], 'val': []} for epoch in range(epochs): model.train() train_loss = [] for xb, yb in loaders['train']: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() pred = model(xb) loss = criterion(pred, yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() train_loss.append(loss.item()) model.eval() val_loss = [] with torch.no_grad(): for xb, yb in loaders['val']: xb, yb = xb.to(device), yb.to(device) val_loss.append(criterion(model(xb), yb).item()) tr, va = np.mean(train_loss), np.mean(val_loss) history['train'].append(tr) history['val'].append(va) print(f"epoch {epoch+1:03d} train {tr:.6f} val {va:.6f}") if va < best_val: best_val = va wait = 0 torch.save(model.state_dict(), 'best_lstm.pt') else: wait += 1 if wait >= patience: print(f"early stop at epoch {epoch+1}") break return history

逻辑说明：clip_grad_norm_把梯度范数限制在 1.0，防止 LSTM 在长序列上梯度爆炸。早停的patience=8表示验证损失连续 8 轮不降就停，保存验证集最优的权重而不是最后一轮，这是避免过拟合的标准操作。HuberLoss在收益率预测上比 MSE 稳，因为收益率分布尖峰厚尾，MSE 会被极端值带偏。

参数说明：epochs=50是上限，实际常被早停截断；lr=1e-3是 Adam 的常用起点，损失震荡可以降到 3e-4；patience设 5~10，太小会误停，太大浪费算力。训练时盯验证损失，如果训练损失降、验证损失升，就是过拟合，该加 dropout 或减 hidden_size。

3.3 预测结果怎么还原成价格并评估

模型输出的是对数收益率，要还原成价格才能和真实收盘价对比。还原公式是pred_price = last_close * exp(pred_ret)。评估指标别只看 MSE，方向准确率（涨跌方向对不对）对交易更有意义。

def evaluate(model, loaders, df, device='cpu'): model.eval() preds, trues = [], [] with torch.no_grad(): for xb, yb in loaders['test']: preds.append(model(xb.to(device)).cpu().numpy()) trues.append(yb.numpy()) pred_ret = np.concatenate(preds) true_ret = np.concatenate(trues) # 方向准确率 dir_acc = np.mean(np.sign(pred_ret) == np.sign(true_ret)) # 收益率 MSE mse = np.mean((pred_ret - true_ret) ** 2) print(f"方向准确率: {dir_acc:.4f} MSE: {mse:.8f}") return pred_ret, true_ret

逻辑说明：方向准确率能到 55% 以上就算有信息量，纯随机是 50%。MSE 在收益率尺度上数值很小，看相对大小即可。注意测试集必须是最末尾那段，不能参与任何训练或标准化统计。

4. 避坑与排查：LSTM 预测股市最常见的五个翻车点

4.1 预测曲线永远滞后一天

现象：把预测价格和真实价格画在一起，预测线像是真实线右移了一格，形状几乎一样。原因：模型学到的最优策略就是「预测值等于昨天收盘价」，因为收益率接近白噪声，MSE 损失下这是最稳的解。解决：改预测目标，从绝对价格改成收益率或涨跌方向；损失函数换成方向敏感的，或者在特征里加入领先性指标。滞后本身说明模型没学到额外信息，不是代码 bug。

4.2 验证损失比训练损失还低

现象：训练损失 0.001，验证损失 0.0008，看起来模型泛化很好。原因：时间序列切分后，验证段恰好波动更小，或者标准化用了全量统计量导致泄漏。解决：检查标准化是否只用训练段统计量；用 walk-forward 多折验证，看损失是否稳定；如果验证段是低波动区间，换一段再测。

4.3 换了只股票模型就完全失效

现象：在沪深300上训练好的模型，直接预测创业板个股，方向准确率掉到 50% 以下。原因：不同标的的波动率、量价关系差异大，标准化参数和模型权重都不通用。解决：要么每只标的单独训练，要么在特征里加入横截面标准化（如相对行业指数的超额收益），让输入分布对齐。跨标的迁移不是不能做，但需要更大的数据集和更复杂的结构。

4.4 训练损失不下降，一直卡在初始值附近

现象：跑了几十轮，损失几乎不动。原因：学习率太大导致震荡，或者输入没标准化、量纲差异悬殊，或者input_size和实际特征数不匹配但没报错（某些写法会静默广播）。解决：先打印一个 batch 的输入统计量，确认均值和方差在合理范围；学习率降到 1e-4 试；检查input_size是否等于X.shape[2]。

4.5 回测收益高得离谱

现象：测试集上的策略年化收益 200%，夏普比率 5 以上。原因：几乎可以肯定是未来函数。常见来源包括：标准化用了全量数据、滚动指标没 shift、用当天收盘价预测当天收盘价、测试集参与了早停选择。解决：逐行检查特征生成代码，所有用到close的地方确认是否shift(1)；标准化统计量严格限定在训练段；早停只用验证集，测试集只在最后评估一次。

5. 把 LSTM 预测做成可复用的研究流程：目录、参数与验证习惯

一套能长期迭代的代码，比一次跑通的结果更重要。我一般会把项目组织成下面这个结构，数据、特征、模型、评估各自独立，换标的或换特征时只动对应模块。

参数管理上，我习惯把window、hidden_size、lr这些写进 yaml，训练脚本读配置，这样跑对比实验时不用改代码。验证习惯上，至少做三件事：一是用 walk-forward 而不是单次切分，看指标在不同时间段的稳定性；二是把方向准确率和收益率 MSE 一起看，单看 MSE 会被滞后解骗；三是每次改特征或超参，都在同一测试段上对比，避免「这次好像好一点」的玄学判断。

最后一个具体技巧：把预测目标从「下一天收益率」改成「未来 N 天累计收益率」，N 取 3 到 5。这样模型不用纠结单日噪声，学的是短期趋势，方向准确率通常能提升几个点，也更贴近实际调仓频率。代价是样本重叠，评估时要用 purged walk-forward 去掉重叠区间，否则指标会虚高。这个改动我踩过坑，一开始没做 purging，回测夏普直接翻倍，后来老老实实按时间隔离才拿到可信数字。希望帮到你。

本文还有配套的精品资源，点击获取