如何用TensorFlow LSTM训练好的模型预测单变量时间序列未来n周期？

问题背景 #

已完成单变量时间序列的LSTM模型训练与验证，模型效果良好，需实现n个周期（如30/50天）的超前预测。

现有模型回顾 #

现有模型基于窗口化数据训练，每次输入window_size长度的序列，预测下一个时间步的值（单步预测）。要实现多步预测，可采用递归预测法（无需修改模型结构）或直接多步预测法（需调整模型输出），以下分别介绍。

方法一：递归多步预测（无需修改现有模型） #

核心逻辑：用模型预测出的第一个值，添加到输入序列末尾，再用新序列预测下一个值，循环往复直到生成n个预测值。

1. 准备初始输入序列 #

取归一化后时间序列的最后window_size个时间步作为初始输入窗口：

# X为归一化后的完整时间序列数据
initial_sequence = X[-window_size:]

2. 编写递归预测函数 #

def recursive_forecast(model, initial_seq, n_steps):
    """
    参数:
        model: 训练完成的LSTM模型
        initial_seq: 初始输入序列，形状为(window_size, num_features)
        n_steps: 需要预测的总步数
    返回:
        预测结果数组，形状为(n_steps,)
    """
    # 转换为模型要求的输入格式：(batch_size, time_steps, features)
    current_seq = initial_seq.reshape(1, window_size, len(features))
    predictions = []
    
    for _ in range(n_steps):
        # 预测当前序列的下一个值
        pred = model.predict(current_seq, verbose=0)
        # 提取预测结果（因模型return_sequences=True，取最后一个时间步的输出）
        pred_value = pred[0, -1, 0]
        predictions.append(pred_value)
        
        # 更新输入序列：移除最旧值，添加新预测值
        current_seq = np.concatenate([current_seq[:, 1:, :], [[pred_value]]], axis=1)
    
    return np.array(predictions)

3. 执行预测并反归一化 #

训练时数据做了归一化（除以列最大值），需将预测结果还原为真实值：

# df为原始数据集，features为目标列名
max_value = df[features].max().values[0]

# 计算30天对应的步数（假设数据是15分钟间隔，一天96步：24*4）
n_days = 30
n_steps = n_days * 24 * 4

# 生成预测
predicted_values = recursive_forecast(model, initial_sequence, n_steps)

# 反归一化
predicted_values = predicted_values * max_value

4. 可视化预测结果 #

# 生成预测时间轴（基于原始数据的最后时间戳）
last_timestamp = dataframe.index[-1]
pred_timestamps = pd.date_range(start=last_timestamp, periods=n_steps+1, freq='15T')[1:]

# 绘制历史数据与预测结果
plt.figure(figsize=(12,6), dpi=120)
sns.lineplot(data=dataframe[-7*24*4:], x=dataframe.index[-7*24*4:], y=features[0], label='历史数据（最近7天）')
sns.lineplot(x=pred_timestamps, y=predicted_values, label=f'预测数据（{n_days}天）', color='orange')
plt.title('时间序列多步超前预测')
plt.xlabel('时间')
plt.ylabel('目标值')
plt.xticks(rotation=45)
plt.show()

方法二：直接多步预测（需修改模型结构） #

让模型直接输出未来n步的预测值，需调整数据生成逻辑与模型输出层。

1. 修改窗口化数据生成函数 #

将目标从单步改为n步：

def multi_step_window_dataset(series, window_size, n_steps, batch_size, shuffle_buffer):
    data = tf.data.Dataset.from_tensor_slices(series)
    # 窗口大小为输入长度+预测步数，输入取前window_size，目标取后n_steps
    data = data.window(window_size + n_steps, shift=1, drop_remainder=True)
    data = data.flat_map(lambda w: w.batch(window_size + n_steps))
    data = data.shuffle(shuffle_buffer)
    data = data.map(lambda w: (w[:-n_steps], w[-n_steps:]))
    return data.batch(batch_size).prefetch(1)

# 定义预测步数
n_steps = 30*24*4
train_set_multi = multi_step_window_dataset(x_train, window_size, n_steps, batch_size, buffer_size)

2. 调整模型输出层 #

将输出改为对应n步预测的维度：

# 复用原有模型的特征提取层，修改输出部分
inputs = layers.Input(shape=(None, len(features)))
x = conv1(inputs)
x = lstm1(x)
x = lstm2(x)
x = conv2(x)
x = layers.Dense(60, activation='relu')(x)
x = layers.Dropout(.1)(x)
# 输出n_steps个值，对应未来n步的预测结果
outputs = layers.Dense(n_steps, activation='tanh')(x)
outputs = layers.Lambda(lambda x: 25*abs(x))(x)

# 重新编译并训练模型
model_multi = keras.Model(inputs=inputs, outputs=outputs)
model_multi.compile(optimizer=optimizer, loss=loss, metrics=["mae"])
history_multi = model_multi.fit(train_set_multi, epochs=epochs, verbose=1)

3. 执行直接多步预测 #

def multi_step_forecast(model, X, window_size):
    data = tf.data.Dataset.from_tensor_slices(X)
    data = data.window(window_size, shift=1, drop_remainder=True)
    data = data.flat_map(lambda w: w.batch(window_size))
    data = data.batch(32).prefetch(1)
    forecast = model.predict(data)
    return forecast

# 取最后一个窗口预测未来n_steps
final_window = X[-window_size:].reshape(1, window_size, len(features))
predicted_values_multi = model_multi.predict(final_window, verbose=0)[0]

# 反归一化
predicted_values_multi = predicted_values_multi * max_value

注意事项 #

• 递归预测的误差会随步数增加累积，适合中短期预测；若需长期预测，建议结合周期性特征优化模型。
• 直接多步预测避免了误差累积，但需要重新训练模型，且需更多数据支撑模型学习输入到多步输出的映射关系。

内容的提问来源于stack exchange，提问作者Hemil Shah