基于Keras/TensorFlow/LSTM的金融时间序列未来3个15分钟周期预测问询

我已经完成了基于Keras/TensorFlow的LSTM模型开发，实现了EURUSD 15分钟时间序列数据的训练、测试及样本内预测，以下是完整基础代码及未来3步实时预测的实现方案：

一、基础模型代码（训练+样本内预测） #

1. 导入依赖库 #

import math
import plotly.graph_objects as go
import yfinance as yf
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler 
import matplotlib.pyplot as plt
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
from datetime import datetime

2. 获取15分钟时间序列数据 #

# 获取EURUSD的5天15分钟K线数据
dataset_train = yf.download(tickers='EURUSD=x', period='5d', interval='15m')
# 查看数据
print(dataset_train)

3. 绘制历史收盘价曲线 #

plt.figure(figsize=(15, 8))
plt.title('EURUSD 历史收盘价')
plt.plot(dataset_train['Close'])
plt.xlabel('日期')
plt.ylabel('收盘价 ($)')
plt.show()

4. 预处理训练数据 #

close_prices = dataset_train['Close']
values = close_prices.values
# 取80%数据作为训练集
training_data_len = math.ceil(len(values)* 0.8)

# 归一化数据到0-1区间
scaler = MinMaxScaler(feature_range=(0,1))
scaled_data = scaler.fit_transform(values.reshape(-1,1))
train_data = scaled_data[0: training_data_len, :]

# 构造LSTM输入序列：用前60个时刻的数据预测下一个时刻的收盘价
x_train = []
y_train = []

for i in range(60, len(train_data)):
    x_train.append(train_data[i-60:i, 0])
    y_train.append(train_data[i, 0])
    
# 转换为模型可接受的数组格式
x_train, y_train = np.array(x_train), np.array(y_train)
# 调整为[样本数, 时间步长, 特征数]的LSTM输入格式
x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

5. 预处理测试数据 #

# 测试集需要包含训练集最后60个数据，用于构造初始输入序列
test_data = scaled_data[training_data_len-60: , : ]
x_test = []
y_test = values[training_data_len:]  # 真实收盘价

for i in range(60, len(test_data)):
  x_test.append(test_data[i-60:i, 0])

# 转换为模型可接受的数组格式
x_test = np.array(x_test)
x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

6. 构建LSTM模型 #

model = keras.Sequential()
# 第一层LSTM，输出序列用于下一层LSTM
model.add(layers.LSTM(100, return_sequences=True, input_shape=(x_train.shape[1], 1)))
# 第二层LSTM，不输出序列
model.add(layers.LSTM(100, return_sequences=False))
# 全连接层
model.add(layers.Dense(25))
# 输出层：预测单步收盘价
model.add(layers.Dense(1))
# 打印模型结构
model.summary()

7. 编译并训练模型 #

model.compile(optimizer='adam', loss='mean_squared_error')
# 批量大小1，训练3轮
model.fit(x_train, y_train, batch_size= 1, epochs=3)

8. 样本内预测与结果评估 #

predictions = model.predict(x_test)
# 反归一化，还原真实价格
predictions = scaler.inverse_transform(predictions)
# 计算RMSE评估模型误差
rmse = np.sqrt(np.mean((predictions - y_test)**2))
print(f"模型RMSE: {rmse}")

9. 绘制预测结果对比图 #

data = dataset_train.filter(['Close'])
train = data[:training_data_len]
validation = data[training_data_len:]
validation['Predictions'] = predictions

plt.figure(figsize=(16,8))
plt.title('模型预测结果对比')
plt.xlabel('日期')
plt.ylabel('收盘价 USD ($)')
plt.plot(train)
plt.plot(validation[['Close', 'Predictions']])
plt.legend(['训练集', '验证集真实值', '验证集预测值'], loc='lower right')
plt.show()

print("训练集数据：")
print(train)
print("\n验证集真实值与预测值：")
print(validation[['Close', 'Predictions']])

二、未来3个15分钟K线实时预测实现 #

多步未来预测的核心逻辑是：将模型预测出的结果作为下一次预测的输入数据，循环迭代3次即可得到未来3个时刻的收盘价。

实现代码 #

def predict_future_steps(model, scaler, last_60_data, steps=3):
    """
    预测未来N个时间步的收盘价
    :param model: 训练好的LSTM模型
    :param scaler: 数据归一化器
    :param last_60_data: 最近60个时刻的归一化收盘价数组
    :param steps: 预测步数，默认3步
    :return: 反归一化后的未来N步预测价格列表
    """
    future_predictions = []
    current_input = last_60_data.reshape(1, 60, 1)  # 调整为模型输入格式

    for _ in range(steps):
        # 预测当前步
        pred_scaled = model.predict(current_input, verbose=0)
        # 反归一化
        pred = scaler.inverse_transform(pred_scaled)[0][0]
        future_predictions.append(pred)
        
        # 更新输入序列：去掉最旧的一个数据，加入当前预测值（归一化后）
        pred_scaled = pred_scaled.reshape(1, 1, 1)
        current_input = np.concatenate([current_input[:, 1:, :], pred_scaled], axis=1)
    
    return future_predictions

# ------------------- 执行未来3步预测 -------------------
# 获取最近60个归一化的收盘价数据
last_60_scaled = scaled_data[-60:]
# 预测未来3个15分钟收盘价
future_prices = predict_future_steps(model, scaler, last_60_scaled, steps=3)

# 打印结果
print("\n未来3个15分钟K线收盘价预测值：")
for i, price in enumerate(future_prices, 1):
    print(f"第{i}个15分钟：{price:.5f}")

关键说明 #

1. 输入序列更新：每次预测出一个未来价格后，将其归一化，从当前输入序列中移除最早的一条数据，把新预测值加入序列末尾，作为下一次预测的输入。
2. 实时场景适配：实际实时预测时，需每次获取最新的15分钟K线收盘价，更新last_60_scaled数组后再调用预测函数。
3. 误差特性：多步预测的误差会随步数增加累积，可通过优化模型结构（如改用序列到序列模型）或加入更多特征（如开盘价、成交量）来改善。

内容的提问来源于stack exchange，提问作者A.M. T