### Python LSTM 单特征单步预测实例 以下是基于Python和Keras库实现LSTM网络用于单特征值单步预测的时间序列数据处理方法。此代码展示了如何定义、训练并评估一个简单的LSTM模型。 #### 数据准备与预处理 为了使LSTM能够有效工作，通常需要对原始时间序列数据进行标准化或归一化处理，并将其转换为适合监督学习的形式： ```python import numpy as np from sklearn.preprocessing import MinMaxScaler # 假设我们有一个简单的一维时间序列数据集 data = np.array([i for i in range(1, 101)]).reshape(-1, 1) scaler = MinMaxScaler(feature_range=(0, 1)) scaled_data = scaler.fit_transform(data) def create_dataset(dataset, look_back=1): X, Y = [], [] for i in range(len(dataset)-look_back-1): a = dataset[i:(i+look_back), 0] X.append(a) Y.append(dataset[i + look_back, 0]) return np.array(X), np.array(Y) look_back = 1 X_train, y_train = create_dataset(scaled_data[:80], look_back) # 训练集前80个样本 X_test, y_test = create_dataset(scaled_data[80:], look_back) # 测试集剩余部分 # 调整输入形状以适应LSTM的要求 (samples, time steps, features) X_train = np.reshape(X_train, (X_train.shape[0], 1, X_train.shape[1])) X_test = np.reshape(X_test, (X_test.shape[0], 1, X_test.shape[1])) ``` #### 定义与编译LSTM模型 在此阶段，我们将创建具有单一隐藏层的LSTM神经网络结构，并指定损失函数以及优化器参数。 ```python from keras.models import Sequential from keras.layers import Dense, LSTM model = Sequential() model.add(LSTM(4, input_shape=(1, look_back))) # 配置LSTM单元数量及输入维度 model.add(Dense(1)) # 输出层只有一个节点 model.compile(loss='mean_squared_error', optimizer='adam') ``` #### 模型拟合与验证 通过多次迭代周期来调整权重直至达到最佳性能表现；随后利用测试子集检验其泛化能力。 ```python history = model.fit(X_train, y_train, epochs=100, batch_size=1, verbose=2) # 对测试集做预测 y_pred = model.predict(X_test) y_pred_rescaled = scaler.inverse_transform(y_pred) y_test_rescaled = scaler.inverse_transform([y_test]) for actual, predicted in zip(y_test_rescaled[0], y_pred_rescaled[:, 0]): print(f'Actual={actual:.2f}, Predicted={predicted:.2f}') ``` 以上即是一个完整的流程展示，涵盖了从基础的数据准备工作到最终完成预测输出的过程[^1]。 ### 复杂情况下的双层Stateful LSTM扩展版本 如果希望进一步提升模型复杂度，则可以考虑采用两层堆叠的方式构建stateful模式下的LSTM架构，如下所示： ```python batch_size = 1 model = Sequential() model.add(LSTM(4, batch_input_shape=(batch_size, 1, look_back), stateful=True, return_sequences=True)) model.add(LSTM(4, batch_input_shape=(batch_size, 1, look_back), stateful=True)) model.add(Dense(1)) model.compile(loss='mean_squared_error', optimizer='adam') for epoch in range(100): model.fit(X_train, y_train, epochs=1, batch_size=batch_size, verbose=2, shuffle=False) model.reset_states() testPredict = model.predict(X_test, batch_size=batch_size) ``` 注意，在这种情况下每次epoch结束后都需要手动调用`reset_states()`重置状态以便于下一轮循环正常运行[^2]。