基于Keras的LSTM从Y时序数据预测X值的复现问题求助
嘿,我帮你把用Keras搭建LSTM从Y预测X的完整流程整理好了,结合你的CSV数据结构和需求,包含预处理、建模、预测、绘图全步骤,直接就能复用:
LSTM实现从Y预测X的完整流程(Keras框架)
1. 先把依赖库都导入
这一步是基础,把需要的工具包都加载进来:
import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn.preprocessing import MinMaxScaler from sklearn.metrics import mean_squared_error, r2_score from keras.models import Sequential from keras.layers import LSTM, Dense
2. 数据预处理(核心步骤)
LSTM对输入格式要求很严格,得把你的CSV数据转换成它能接受的形状,还要做归一化(不然模型很难收敛):
加载并提取数据
# 替换成你的CSV文件路径 df = pd.read_csv('your_data.csv') # 提取特征Y(用来预测的变量)和目标X(要预测的变量) data = df[['Y', 'X']].values
数据归一化
LSTM对数据尺度敏感,先把值缩到[0,1]区间:
scaler = MinMaxScaler(feature_range=(0, 1)) scaled_data = scaler.fit_transform(data)
构造时间序列数据集
LSTM需要用过去的Y值序列来预测当前的X,我们写个函数来生成这种输入输出对:
def create_dataset(dataset, time_steps=1): X, y = [], [] # 遍历数据,取前time_steps个Y值作为输入,对应下一个时间点的X作为目标 for i in range(len(dataset) - time_steps): X.append(dataset[i:(i+time_steps), 0]) # 输入:前time_steps个Y y.append(dataset[i + time_steps, 1]) # 目标:当前时间点的X return np.array(X), np.array(y) # 这里设置时间步长,比如用过去5个Y值预测X,你可以根据自己的数据调整 time_steps = 5 X, y = create_dataset(scaled_data, time_steps)
调整输入形状
LSTM要求输入是(样本数, 时间步长, 特征数)的三维数组,我们这里特征数是1(只有Y):
X = np.reshape(X, (X.shape[0], X.shape[1], 1))
拆分训练测试集
注意!时间序列不能随机拆分,必须按时间顺序来,比如按8:2的比例分:
train_size = int(len(X) * 0.8) test_size = len(X) - train_size X_train, X_test = X[0:train_size], X[train_size:len(X)] y_train, y_test = y[0:train_size], y[train_size:len(y)]
3. 搭建并训练LSTM模型
我们先搭一个基础的LSTM结构,后续你可以根据效果调参:
model = Sequential() # 添加LSTM层,units是神经元数量,input_shape对应时间步长和特征数 model.add(LSTM(units=50, return_sequences=False, input_shape=(time_steps, 1))) # 全连接层输出预测的X值 model.add(Dense(units=1)) # 编译模型,回归任务用adam优化器+均方误差损失 model.compile(optimizer='adam', loss='mean_squared_error') # 训练模型,batch_size和epochs可以根据数据量调整,validation_data用来监控验证集损失 history = model.fit(X_train, y_train, batch_size=32, epochs=50, validation_data=(X_test, y_test), verbose=1)
4. 预测与反归一化
因为之前做了归一化,预测出来的值是缩放过的,得转回到原始尺度才能和真实值对比:
# 生成训练集和测试集的预测值 train_predict = model.predict(X_train) test_predict = model.predict(X_test) # 反归一化:因为我们是对[Y,X]整体归一化的,所以要构造对应形状的数据来反变换 # 处理训练集预测值 train_predict_full = np.zeros((len(train_predict), 2)) train_predict_full[:, 1] = train_predict[:, 0] train_predict = scaler.inverse_transform(train_predict_full)[:, 1] # 处理训练集真实值 y_train_full = np.zeros((len(y_train), 2)) y_train_full[:, 1] = y_train[:] y_train = scaler.inverse_transform(y_train_full)[:, 1] # 处理测试集预测值 test_predict_full = np.zeros((len(test_predict), 2)) test_predict_full[:, 1] = test_predict[:, 0] test_predict = scaler.inverse_transform(test_predict_full)[:, 1] # 处理测试集真实值 y_test_full = np.zeros((len(y_test), 2)) y_test_full[:, 1] = y_test[:] y_test = scaler.inverse_transform(y_test_full)[:, 1]
5. 模型评估与绘图
现在可以和你的传统ML模型对比效果,还能画出可视化图:
评估指标
计算MSE和R²,和你之前的模型结果对比:
test_mse = mean_squared_error(y_test, test_predict) test_r2 = r2_score(y_test, test_predict) print(f"测试集均方误差(MSE): {test_mse:.4f}") print(f"测试集R²系数: {test_r2:.4f}")
绘图展示
训练损失曲线
看模型有没有过拟合或者欠拟合:
plt.figure(figsize=(12, 4)) plt.plot(history.history['loss'], label='训练损失') plt.plot(history.history['val_loss'], label='验证损失') plt.title('模型损失变化') plt.xlabel('迭代次数') plt.ylabel('均方误差损失') plt.legend() plt.show()
测试集真实值vs预测值
直观对比预测效果:
plt.figure(figsize=(12, 4)) plt.plot(y_test, label='真实X值') plt.plot(test_predict, label='预测X值') plt.title('测试集真实值vs预测值') plt.xlabel('时间步') plt.ylabel('X值') plt.legend() plt.show()
全数据集真实值vs预测值
展示整个数据集的拟合效果:
full_predict = np.concatenate([train_predict, test_predict]) full_y = np.concatenate([y_train, y_test]) plt.figure(figsize=(15, 5)) plt.plot(full_y, label='真实X值') plt.plot(full_predict, label='预测X值') plt.title('全数据集真实值vs预测值') plt.xlabel('时间步') plt.ylabel('X值') plt.legend() plt.show()
6. 几个关键调参建议
- 时间步长:可以试试3、5、10等不同值,找和你传统ML模型效果最接近的
- 模型结构:可以增加LSTM层(比如把
return_sequences=True,再加一层LSTM),或者调整units数量 - 训练参数:batch_size和epochs可以根据数据量调整,比如数据大就调大batch_size,epochs可以多跑点看损失是否收敛
- 归一化:如果MinMaxScaler效果不好,可以试试StandardScaler
内容的提问来源于stack exchange,提问作者Desta Haileselassie Hagos
相关产品推荐
相关产品推荐

