简介这是一套多输入单输出LSTM模型的完整Python实现面向有深度学习基础、希望上手时间序列预测或金融数据建模的开发者。项目以股票收盘价预测为例使用开盘价、最高价、最低价等多维特征串联数据预处理、模型搭建、训练验证与结果可视化等环节便于理解LSTM在多元序列预测中的典型应用方式。压缩包共10个文件主要包括Jupyter Notebook代码文件、CSV训练数据集以及项目说明YAML文档可快速还原环境并复现完整流程资源整体约1.05MB结构紧凑、易于部署。已有618人学习浏览适合作为课程设计、毕业设计或入门实战参考。通过该资源可掌握滑动窗口构造、多特征输入格式转换、LSTM层与全连接层组合设计等关键步骤并可直接修改数据集与参数以迁移到其他单输出回归预测任务。1. 多输入单输出LSTM这个 zip 里装的是一种很通用的预测范式拿到「多输入单输出 LSTM 模型 python 实现源代码数据集项目使用说明」这个压缩包多数人第一反应是这不就是一个 LSTM 预测模型吗但「多输入单输出」这六个字决定它不是课设里那种单变量时间序列预测的玩具。它的典型场景是你有多个传感器或多个来源的时间序列想用一个模型把它们联合起来预测一个目标量。比如设备健康管理里用振动、温度、电流预判剩余寿命量化交易里用成交量、MACD、RSI 联合预测下一根 K 线的涨跌方向环境监测里用多个气象站数据预测某个站点的 PM2.5。这类问题在工业界和竞赛里非常多也是毕设和简历项目的高频选题。这个 zip 的核心价值不在于 LSTM 本身——LSTM 的经典结构到处都有——而在于把「多路输入如何组织、如何切窗、如何喂给 Keras 的 Functional API、如何反归一化」这一整条链路打通。新手拿到它能照着跑通熟手能直接改输入路数和输出口径去套自己的场景。下面我按自己实际做这类项目的顺序把这套方案拆开讲清楚。2. 把数据切成多路输入滑窗、归一化与数据集格式约定不管是做剩余寿命预测还是行情预测第一步都是把原始表格式数据变成模型能吃的东西。多输入单输出和单变量预测最大的区别就在这一层你不再是拿一列数切窗而是要把多列数据按同一个时间步对齐切窗。这个环节出问题后面模型再漂亮也是白搭。2.1 先想清楚你的「多输入」是哪种多我见过很多人在「多输入」这个词上栽跟头。严格说有两种多输入多特征单序列一个样本是一个时间窗窗内有多列特征比如某只股票过去 24 天的收盘价和成交量shape 是(samples, timesteps, features)。这种写法在 Keras 里用Sequential或Functional都能做输入层一个就够了。多路独立序列同时存在多个来源的数据比如设备上有 4 个传感器希望每个传感器单独对应一路输入shape 是多个(samples, timesteps, 1)的组合最后在模型中间拼起来。这种必须用Functional API的多个Input层。标题写的是「多输入单输出」我按常见项目方案优先讲第二种因为第一种本质上是多特征回归难度低很多网上代码一抓一大把而第二种才是真正体现「多输入」价值的做法也是做 PHM2012 这类多传感器寿命预测、量化里多指标联合预测的常见套路。如果你拿到的是第一种数据下面的代码也能改造成适用成本很低。2.2 滑窗切分的标准写法假设你有一份干净的 DataFrame索引是时间每一列是一个输入特征或目标值。多路输入切窗的核心思路是对每一个输入特征列单独切窗但所有列的切窗起点严格对齐否则模型学到的关系是错位的。import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler def create_multi_input_windows(data, input_cols, target_col, seq_len24, pred_len1): 把多列输入按同一时间轴切成滑窗返回适合 Keras Functional API 的输入列表 参数: data: DataFrame索引为时间 input_cols: 参与输入的列名列表每一列会变成一路输入 target_col: 目标列名只输出该列的预测 seq_len: 回看窗口长度 pred_len: 预测未来几个点默认 1 返回: X_list: list of arrays每个 shape 为 (样本数, seq_len, 1) y: shape 为 (样本数, pred_len) 的目标数组 scalers: 每个输入的 MinMaxScaler之后反归一化要用 total_len len(data) - seq_len - pred_len 1 if total_len 0: raise ValueError(数据长度不足seq_len pred_len 必须小于总样本数) X_list [] scalers {} for col in input_cols: scaler MinMaxScaler() # 对该列单独归一化注意这里只是拿到列数据先不做 fit col_values data[col].values.reshape(-1, 1) scaler.fit(col_values) # 实际项目中 fit 应该只在训练集上做这里为演示简化 col_scaled scaler.transform(col_values).reshape(-1) scalers[col] scaler # 对该列切窗 windows [] for i in range(total_len): windows.append(col_scaled[i : i seq_len]) X_list.append(np.array(windows).reshape(total_len, seq_len, 1)) # 目标列默认取未来第 pred_len 个时刻的值 y_values data[target_col].values.reshape(-1, 1) y_scaler MinMaxScaler() y_scaler.fit(y_values) y_scaled y_scaler.transform(y_values).reshape(-1) y [] for i in range(seq_len, seq_len total_len): y.append(y_scaled[i : i pred_len]) y np.array(y) return X_list, y, scalers, y_scaler这段代码的切窗逻辑是对于每一路输入滑窗起点i从 0 到total_len - 1每个窗口取该列从i到i seq_len的数值目标值则取窗口结束之后pred_len个时刻的值。这样每个样本的每一路输入在时间轴上严格对齐模型才有机会学到跨特征的时序依赖关系。参数上有三个值得注意的地方。第一seq_len的选择要看数据本身的周期性做日线行情预测24 或 30 是常见起点做设备状态监测如果采样频率是 1kHz那一个窗口至少要包含几个完整的工作循环不能拍脑袋定。第二pred_len大于 1 时目标数组的 shape 会变成(样本数, pred_len)输出层也要相应改成Dense(pred_len)这点后面会再讲。第三我这里为了让演示完整把归一化 fit 写在了函数里但实际项目里scaler.fit应只在训练集上执行验证集和测试集只用transform这是防止数据泄漏的关键前提第 5 章会单独展开。2.3 归一化只能 fit 一次数据泄漏是从这里开始的很多人在 Jupyter 里顺手就对全量数据做了MinMaxScaler().fit_transform(df)短期看不出问题等到做验证集评测时指标异常地好才意识到泄漏了未来信息。正确做法是先把原始数据按时间切出训练集、验证集、测试集再分别对每一路特征 fit 训练集。下面这段是推荐写法train_df, val_df, test_df split_by_time(df, train_ratio0.7, val_ratio0.15) # 对每一列单独 fit 训练集 feature_scalers {} for col in INPUT_COLS: scaler MinMaxScaler() scaler.fit(train_df[[col]]) feature_scalers[col] scaler # 然后统一 transform for df_part in (train_df, val_df, test_df): for col in INPUT_COLS: df_part[col] feature_scalers[col].transform(df_part[[col]])这里有一个容易被忽略的细节如果数据有明显的趋势或周期性且测试集时间范围超出训练集的最大值MinMaxScaler 在 transform 时可能出现超出 [0,1] 区间的值这是正常的不需要报错。但如果你想让输出稳定在 [0,1] 内可以考虑用 RobustScaler 或提前留出余量。总之归一化边界在处理新数据时永远是个不可知量接受这一点比强行修正更省心。3. 用 Keras Functional API 搭多输入单输出 LSTM网络结构与代码实现数据准备好之后下一步是搭网络。多输入单输出的模型结构用Sequential写不了因为Sequential只允许一条线串下去而多路输入需要在某个地方合并。Keras 的Functional API就是为了这种非线性的网络拓扑设计的。3.1 为什么必须是 Functional APISequential的写法是model.add(...)一层一层往下加每层只有一个输入一个输出。多输入模型需要的是若干个独立的输入层各自经过自己的 LSTM 或全连接层然后在某一个节点把特征拼接起来最后接输出层。这种「多路汇合」的结构只能用函数式写法表达。两者的差距不只是语法。用Functional API时你拥有每个张量的句柄可以在合并前对每一路单独做处理比如一路用 LSTM另一路用一维卷积甚至可以给不同的输入配不同的seq_len。这种灵活性在做真实项目时非常实用因为不同来源的数据往往有不一样的采样频率或特征形态。3.2 基础结构两路输入拼一路输出假设输入有两路信号每路都是一个长度为seq_len的序列目标是回归一个数值。下面这段是完整的模型定义from tensorflow.keras.layers import Input, LSTM, Dense, Concatenate, Dropout from tensorflow.keras.models import Model from tensorflow.keras.optimizers import Adam def build_multi_input_lstm(seq_len, n_features_per_input1, lstm_units64, pred_len1): 构建多输入单输出 LSTM 模型 # 定义两个输入层shape 都是 (seq_len, 1) input_a Input(shape(seq_len, n_features_per_input), nameinput_a) input_b Input(shape(seq_len, n_features_per_input), nameinput_b) # 每一路各过一个 LSTM 层 lstm_a LSTM(lstm_units, return_sequencesFalse, namelstm_a)(input_a) lstm_b LSTM(int(lstm_units / 2), return_sequencesFalse, namelstm_b)(input_b) # 拼接两个 LSTM 的输出 merged Concatenate(nameconcat)([lstm_a, lstm_b]) # 接一个全连接层做非线性变换 dense Dense(32, activationrelu, namedense_hidden)(merged) dropout Dropout(0.2, namedropout)(dense) # 输出层pred_len1 时输出一个数 output Dense(pred_len, activationlinear, nameoutput)(dropout) model Model(inputs[input_a, input_b], outputsoutput) return model model build_multi_input_lstm(seq_len24, lstm_units64, pred_len1) model.compile(optimizerAdam(learning_rate1e-3), lossmse, metrics[mae]) model.summary()这段代码有几个点需要说明。每一路输入都单独经过一个 LSTM然后通过Concatenate在特征维度上拼接。这里我故意给两路分配了不同的 LSTM 单元数——input_a是 64input_b是 32——原因是不同输入的信息量不一定相同给更重要的那一路更多容量是常见工程做法。如果你不确定哪路更重要可以先用相同单元数跑一遍基线看哪路的贡献更大。return_sequencesFalse在这里至关重要。如果你误设成TrueLSTM 会输出完整的序列shape 变成(batch_size, seq_len, units)和另一路输出的三维张量拼接就会报维度错误。初学者在这个参数上翻车的频率极高后面避坑章会再提。另外注意Concatenate拼接的是最后一维也就是特征维度。两路 LSTM 输出的 shape 分别是(None, 64)和(None, 32)拼接后变成(None, 96)。这 96 维特征再经过一个全连接层压缩到 32 维最后输出 1 维预测值。全连接层的relu激活是可选项但如果你发现训练后期 loss 震荡把relu换成tanh有时能改善稳定性。3.3 扩展到三路、四路甚至 N 路输入真实项目很少只有两路输入。比如做设备寿命预测振动、温度、转速、电流四路传感器很常见。把上面的代码改成 N 路其实非常机械def build_n_input_lstm(seq_len, n_inputs, lstm_units64, pred_len1): inputs [] lstm_branches [] for i in range(n_inputs): inp Input(shape(seq_len, 1), namefinput_{i}) inputs.append(inp) branch LSTM(lstm_units, return_sequencesFalse, nameflstm_{i})(inp) lstm_branches.append(branch) merged Concatenate(nameconcat)(lstm_branches) if n_inputs 1 else lstm_branches[0] dense Dense(32, activationrelu)(merged) output Dense(pred_len, activationlinear)(dense) model Model(inputsinputs, outputsoutput) return model动态构建多路输入的模型inputs列表的顺序必须与create_multi_input_windows返回的X_list顺序一致这是使用Functional API时最容易忽略的隐性约定。数据是[风速, 温度, 气压]的顺序切片模型输入就必须是[input_风速, input_温度, input_气压]一旦错位模型照样能训练但学到的关系完全错了而且这种错误非常隐蔽。4. 在 Jupyter Notebook 里跑通训练关键参数与调优顺序模型结构定好了接下来就是在 Jupyter Notebook 里把训练流程跑通。Jupyter 的交互特性适合逐步调试但很多人把训练循环写得一团乱——所有代码堆在一个 cell 里报错了从头重跑。下面这套流程是经过实践拆分的每个 cell 干一件事。4.1 跑通训练的最小代码流程# cell 1: 加载数据 import pandas as pd df pd.read_csv(sensor_data.csv, parse_dates[time], index_coltime) INPUT_COLS [vibration, temperature, current] TARGET_COL rul # cell 2: 按时间切分 train_df df.iloc[:int(len(df) * 0.7)] val_df df.iloc[int(len(df) * 0.7):int(len(df) * 0.85)] test_df df.iloc[int(len(df) * 0.85):] # cell 3: 切窗用第2章的 create_multi_input_windows X_train, y_train, _, _ create_multi_input_windows(train_df, INPUT_COLS, TARGET_COL, seq_len24, pred_len1) X_val, y_val, _, _ create_multi_input_windows(val_df, INPUT_COLS, TARGET_COL, seq_len24, pred_len1) # cell 4: 构建模型 model build_multi_input_lstm(seq_len24, lstm_units64, pred_len1) model.compile(optimizerAdam(learning_rate1e-3), lossmse, metrics[mae]) # cell 5: 训练 history model.fit( X_train, y_train, validation_data(X_val, y_val), epochs50, batch_size32, verbose1 )整个流程五个 cell思路是数据加载 → 时间切分 → 切窗 → 建模 → 训练。注意在切窗时train_df和val_df是分别调用create_multi_input_windows的因为滑动窗口不能跨数据集边界——如果在训练集窗口里包含了验证集的数据验证就没意义了。4.2 三个必调参数units、batch_size、学习率模型能跑通和模型能用中间隔着参数调优这道坎。根据我的经验多输入 LSTM 的调参优先级应该是学习率 batch_size 神经元数量。参数常见范围现象与调整learning_rate1e-4 ~ 1e-2loss 震荡不降优先降 lrloss 下降极慢可以升 lrbatch_size16 ~ 128太小导致 loss 抖动剧烈太大会显存溢出或泛化变差lstm_units32 ~ 128过拟合时降 units 或加 Dropout欠拟合时先升 unitsseq_len视数据周期而定窗口太短学不到长期依赖太长信息冗余且训练变慢学习率是最容易出问题的一个。很多人第一次跑Adam默认lr1e-3对于小数据集通常没问题但多输入模型因为拼接后的特征维度较高loss 曲面更复杂1e-3有可能让 loss 前期下降快、后期在最优值附近来回震荡。我一般会在训练中加一个学习率衰减调度器这是 Jupyter 里非常好用的「后悔药」from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau early_stop EarlyStopping( monitorval_loss, patience10, restore_best_weightsTrue ) reduce_lr ReduceLROnPlateau( monitorval_loss, factor0.5, patience5, min_lr1e-6, verbose1 ) history model.fit( X_train, y_train, validation_data(X_val, y_val), epochs100, batch_size32, callbacks[early_stop, reduce_lr], verbose1 )EarlyStopping的作用是当val_loss连续 10 个 epoch 不下降时停止训练并自动回滚到验证集上表现最好的权重。这里restore_best_weightsTrue很关键默认情况下EarlyStopping只是停止训练不会恢复最优权重模型最后保存的是最后一轮的权重而不是最优的那一轮。ReduceLROnPlateau则是在val_loss连续 5 轮不降时把学习率减半给模型一次跳出局部极小的机会。这两个回调组合是 LSTM 项目最省心的配置能省掉你大量手动调参的时间。4.3 训练完怎么判断好坏不要只看训练集 loss训练结束后第一件事是画 loss 曲线而不是急着看预测结果。如果train_loss持续下降但val_loss先降后升说明过拟合了需要增加 Dropout、减小lstm_units或增加数据量如果两者都居高不下优先怀疑学习率或数据预处理。画曲线是可以在 Jupyter 里直接复制粘贴的常用代码import matplotlib.pyplot as plt plt.figure(figsize(10, 4)) plt.plot(history.history[loss], labeltrain_loss) plt.plot(history.history[val_loss], labelval_loss) plt.legend() plt.xlabel(epoch) plt.ylabel(loss) plt.grid(True) plt.show()一个常见的误区是在多输入 LSTM 的训练中val_loss在训练初期可能异常高因为随机初始化的权重会让拼接后的特征尺度失衡。这不代表模型有问题等前 5~10 个 epoch 过去后会回落。真正需要警惕的是val_loss在中间某个点突然跳高再也没回来那大概率是数据泄漏或某一列输入里有异常值。5. 避坑章节多输入单输出 LSTM 的 5 个常见问题排查多输入 LSTM 比单变量复杂在数据组织和张量形状我把自己和同事踩过、也帮别人排查过的坑做了个清单每一条都按「现象 → 原因 → 解决」写。5.1 归一化时泄漏未来数据验证集指标好到不真实现象训练集 loss 正常验证集 loss 极低但拿到新数据上一测预测值基本是一条直线。原因在切分训练集和验证集之前就对整个数据集做了MinMaxScaler.fit()导致验证集的统计信息混入了训练集的归一化参数中——重点是验证数据本身的取值范围被「剧透」给了模型。解决严格按第 2.3 节的做法先用训练集fit再对验证集和测试集只做transform。我建议把归一化封装成一个函数输入已经是切分好的三段数据从流程上杜绝这种操作顺序错误。5.2 出现ValueError: Shapes (None, 24, 32) and (None, 32) are incompatible现象训练时报错提示两个张量的 shape 无法匹配通常发生在Concatenate层或Dense层附近。原因某个 LSTM 层的return_sequences误设为True导致该路输出是三维张量(batch, seq_len, units)而另一路是二维(batch, units)。解决检查所有 LSTM 层的return_sequences。多输入拼接前通常都设False只在最后一层 LSTM 需要输出完整序列给后续TimeDistributed层时才设True。5.3 loss 变成 NaN训练直接崩掉现象前几个 epoch 正常突然 loss 变成 NaN之后所有指标全部变成 NaN。原因多输入场景下这一般不是模型结构问题而是数据里有Inf或极端值。传感器数据经常出现掉线时的默认值 9999.0 或除以零产生的无穷大归一化后依然很大导致梯度爆炸。解决在切窗之前先做数据清洗用np.isfinite过滤或对异常值做截断同时把学习率降到 1e-4 重试。另外检查归一化时是否有零方差列——某一路信号如果是常数MinMaxScaler会除零报错或产生 NaN这类列应该直接删掉。5.4 预测结果滞后真实值一拍看起来像把 t-1 的值搬到了 t现象预测曲线和真实曲线高度相似但整体向右偏移一个时间步相关性很高、误差却很大。原因这是时间序列预测最常见的「假成功」本质是模型学到的是「把上一个时刻的值复制过来」而不是学到真正的动态规律。常见触发条件是滑窗时把当前时刻的值放进了输入窗口而目标恰好是下一时刻的值信息泄漏。解决检查切窗代码确认输入窗口是i : i seq_len目标是i seq_len : i seq_len pred_len两者时间上完全没有重叠。另外如果目标序列本身是强自相关的比如价格序列这种滞后几乎无法完全避免这也是为什么很多量化项目不做价格预测而做收益预测的原因。5.5 多路输入顺序对不上训练正常但预测全错现象模型能训练loss 也能降但换一批数据预测完全不对或者验证集上指标波动极大。原因create_multi_input_windows返回的X_list顺序和build_n_input_lstm里Input层的顺序不一致。我在第 3.3 节特别强调过这一点——这类错误模型不会报错因为张量 shape 完全匹配但语义对应是错的。解决用一个可复现的约定来组织输入建议把输入列名列表作为全局配置切窗和建模都从这个列表生成避免在两个函数里手写不同的顺序。更稳妥的做法是在model.fit时给输入传字典history model.fit( {input_a: X_train[0], input_b: X_train[1]}, y_train, validation_data({input_a: X_val[0], input_b: X_val[1]}, y_val), epochs50 )这样输入和网络层的对应关系由name显式指定从根上消除了顺序错位的问题。6. 从单步到多步预测一个能直接拿去改的验证框架如果单步预测已经稳定自然会想往前走一步——预测未来多个时刻。多步预测有两种主流做法迭代式预测和直接式预测。迭代式是把第 t1 步的预测值当作输入的一部分去预测 t2 步实现简单但误差会累积直接式是修改输出层一次输出未来多个时刻的值误差不累积但输出维度变高训练难度增加。对多输入 LSTM 来说直接式更容易实现因为你不需要把预测值重新组织成多路输入格式。直接式多步预测的核心改动只有两处。第一处切窗时指定pred_len我第 2 章的代码已经支持这个参数目标数组 shape 会变成(样本数, pred_len)。第二处模型输出层需要改成output Dense(pred_len, activationlinear, nameoutput)(dense)损失函数依然用mse评价指标可以用mae但在实际项目里我更推荐按业务口径评估。比如做寿命预测关心的是预测值和真实值的绝对误差在一段时间内的分布做行情预测关心的是方向准确率。模型训练时用的指标和业务关心的指标可以是两码事这个不要混。多步预测的验证方式和单步不同。单步可以直接在测试集上算误差但多步预测应该用「滚动验证」从测试集开头逐步推进每次只预测未来一个窗口然后让真实数据进入历史窗口。这是最接近真实使用场景的做法代码也不复杂def rolling_predict(model, X_test_list, seq_len, n_steps): 滚动预测每次预测 n_steps 步然后真实值滑动进入历史 preds [] # 假设 X_test_list 是已经切好的多路输入 for i in range(0, len(X_test_list[0]) - n_steps, n_steps): window_inputs [x[i:i1] for x in X_test_list] # 取一个样本 pred model.predict(window_inputs, verbose0) preds.append(pred[0]) return np.array(preds)注意这里window_inputs是一个列表推导生成的列表每个元素 shape 是(1, seq_len, 1)正好对应model.predict对多输入模型的预期格式。实现的细节之外请记住一个教训我在第一个多输入项目里没有做滚动验证直接在固定测试集上算了一次预测误差当时指标很漂亮上线后才发现模型在长周期上的误差比预期大得多。滚动验证多花半小时能省掉上线后一个月的返工。希望这份拆解能帮你少走这些弯路。本文还有配套的精品资源点击获取