使用LSTM模型预测多特征变量的时间序列

Python编程爱好者

发布于 2024-07-12 13:36:59

2160

发布于 2024-07-12 13:36:59

文章被收录于专栏：Python编程爱好者

Hi，我是Johngo~

今儿和大家聊聊关于「使用LSTM模型预测多特征变量的时间序列」的一个简单项目。

使用LSTM模型预测多特征变量的时间序列，能够帮助我们在各种实际应用中进行更准确的预测。这些应用包括金融市场预测、气象预报、能源消耗预测等。

本项目使用Python和TensorFlow/Keras框架来实现一个LSTM模型，对多特征变量的时间序列数据进行预测。

实现流程

数据准备
- 收集和准备时间序列数据集。
- 处理缺失值和异常值。
- 归一化数据。
数据预处理
- 创建输入特征和目标变量。
- 将数据分为训练集和测试集。
- 将数据重塑为适合LSTM模型的格式。
构建和训练LSTM模型
- 使用Keras构建LSTM模型。
- 编译模型并设置优化器和损失函数。
- 训练模型并进行验证。
模型评估和预测
- 评估模型的性能。
- 使用模型进行未来时间点的预测。
- 可视化预测结果和实际值。

代码实现

在这个示例中，创建一个模拟的多特征时间序列数据集，并保存为CSV文件以供使用。你可以使用以下代码生成一个模拟的数据集，然后保存为multi_feature_time_series.csv文件。

生成模拟数据集并保存为CSV文件

import numpy as np
import pandas as pd

# 设置随机种子以确保可重复性
np.random.seed(42)

# 生成模拟时间序列数据
time_steps = 1000
data = {
    'temperature': np.random.normal(20, 5, time_steps),   # 模拟温度数据
    'humidity': np.random.normal(50, 10, time_steps),     # 模拟湿度数据
    'wind_speed': np.random.normal(10, 2, time_steps),    # 模拟风速数据
    'power_consumption': np.random.normal(200, 50, time_steps)  # 模拟电力消耗数据
}

# 创建DataFrame
df = pd.DataFrame(data)

# 保存为CSV文件
df.to_csv('multi_feature_time_series.csv', index=False)

print("模拟数据集已保存为 multi_feature_time_series.csv")

运行上述代码生成模拟数据集并保存为CSV文件。

然后，大家可以使用生成的CSV文件进行后续的LSTM时间序列预测模型的构建和训练。

完整代码实现

下面是完整的代码实现，包括生成数据集、数据预处理、LSTM模型构建和训练，以及模型评估和预测。

1. 生成模拟数据集并保存为CSV文件

import numpy as np
import pandas as pd

# 设置随机种子以确保可重复性
np.random.seed(42)

# 生成模拟时间序列数据
time_steps = 10000
data = {
    'temperature': np.random.normal(20, 5, time_steps),   # 模拟温度数据
    'humidity': np.random.normal(50, 10, time_steps),     # 模拟湿度数据
    'wind_speed': np.random.normal(10, 2, time_steps),    # 模拟风速数据
    'power_consumption': np.random.normal(200, 50, time_steps)  # 模拟电力消耗数据
}

# 创建DataFrame
df = pd.DataFrame(data)

# 保存为CSV文件
df.to_csv('multi_feature_time_series.csv', index=False)

print("模拟数据集已保存为 multi_feature_time_series.csv")

2. 数据准备和预处理

import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split

# 加载数据
data = pd.read_csv('multi_feature_time_series.csv')

# 检查数据
print(data.head())

# 处理缺失值（如果有）
data = data.dropna()

# 归一化数据
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data)

# 创建输入特征和目标变量
def create_dataset(dataset, time_step=1):
    dataX, dataY = [], []
    for i in range(len(dataset) - time_step - 1):
        a = dataset[i:(i + time_step), :]
        dataX.append(a)
        dataY.append(dataset[i + time_step, -1])  # 假设目标变量是最后一列
    return np.array(dataX), np.array(dataY)

time_step = 10
X, y = create_dataset(scaled_data, time_step)

# 将数据分为训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 检查形状
print(X_train.shape, y_train.shape, X_test.shape, y_test.shape)

3. 构建和训练LSTM模型

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense

# 构建LSTM模型
model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(time_step, X_train.shape[2])))
model.add(LSTM(50, return_sequences=False))
model.add(Dense(25))
model.add(Dense(1))

# 编译模型
model.compile(optimizer='adam', loss='mean_squared_error')

# 训练模型
history = model.fit(X_train, y_train, epochs=100, batch_size=64, validation_data=(X_test, y_test), verbose=1)

# 可视化训练过程
plt.plot(history.history['loss'], label='Training Loss')
plt.plot(history.history['val_loss'], label='Validation Loss')
plt.title('Model Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.legend()
plt.show()

4. 模型评估和预测

# 评估模型
train_predict = model.predict(X_train)
test_predict = model.predict(X_test)

# 反归一化预测值
train_predict = scaler.inverse_transform(np.concatenate((np.zeros((train_predict.shape[0], scaled_data.shape[1]-1)), train_predict), axis=1))[:, -1]
test_predict = scaler.inverse_transform(np.concatenate((np.zeros((test_predict.shape[0], scaled_data.shape[1]-1)), test_predict), axis=1))[:, -1]

# 反归一化实际值
y_train_actual = scaler.inverse_transform(np.concatenate((np.zeros((y_train.shape[0], scaled_data.shape[1]-1)), y_train.reshape(-1, 1)), axis=1))[:, -1]
y_test_actual = scaler.inverse_transform(np.concatenate((np.zeros((y_test.shape[0], scaled_data.shape[1]-1)), y_test.reshape(-1, 1)), axis=1))[:, -1]

# 可视化预测结果
plt.figure(figsize=(14, 5))
plt.plot(y_test_actual, color='blue', label='Actual Value')
plt.plot(test_predict, color='red', label='Predicted Value')
plt.title('Actual vs Predicted')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.show()