Python实现EMD-LSTM时间序列预测（完整源码和数据)

#!/usr/bin/env python # coding: utf-8 # In[2]: # 导入必要的库 from keras.models import Sequential # 从Keras模型中导入Sequential类 from keras.layers import Activation, Dropout, Dense, LSTM # 从Keras中导入特定的层类 import tensorflow as tf # 导入TensorFlow库 from sklearn.preprocessing import MinMaxScaler # 从scikit-learn中导入MinMaxScaler类 from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score # 导入常见的评估指标类 from PyEMD import EMD # 导入PyEMD库，用于经验模态分解 import pandas as pd # 导入pandas库，用于处理和分析数据 import numpy as np # 导入NumPy库，用于数值计算 import matplotlib.pyplot as plt # 导入matplotlib库，用于绘图 import warnings # 导入warnings模块，用于警告处理 warnings.filterwarnings('ignore') # 忽略警告信息 plt.rcParams['font.sans-serif'] = ['Times New Roman'] # 设置图表中的字体为 Times New Roman plt.rcParams['axes.unicode_minus'] = False # 解决图表中显示负号时的乱码问题 #更多模型搜索机器学习之心，支持模型定制 LOOK_BACK = 10 # 定义时间序列窗口的长度 SPLIT = 0.8 # 定义训练集和测试集的划分比例 EPOCHS = 10 # 定义训练的迭代次数 BATCH_SIZE = 10 # 定义每个训练批次中的样本数量 #更多模型搜索机器学习之心，支持模型定制 data = pd.read_csv('焦作.csv', usecols=[1]) # 从 '焦作.csv' 文件中读取数据，并仅使用第二列 data.columns = ["AQI"] # 将数据框的列名更改为 "AQI" def get_EMD_residue(data: pd.DataFrame):#更多模型搜索机器学习之心，支持模型定制 data_np = data.to_numpy() # 将数据框转换为 NumPy 数组 print(data_np.shape) emd = EMD() # 创建 EMD 对象 emd.emd(data_np) # 对数据进行经验模态分解 IMFs, residue = emd.get_imfs_and_residue() # 获取分解后的各个内禀模态函数 (IMFs) 和残差 nIMFs = IMFs.shape[0] # 获取内禀模态函数的数量 plt.figure(figsize=(18,18), dpi=100) # 创建一个新的图表对象，指定图表的大小和分辨率 # 绘制原始数据 plt.subplot(nIMFs + 2, 1, 1) # 创建一个子图，并指定其在图表中的位置 plt.plot(data, 'r') # 绘制数据曲线，颜色为红色 plt.ylabel("AQI") # 设置y轴标签为 "AQI" # 绘制残差 plt.subplot(nIMFs + 2, 1, nIMFs + 2) # 创建一个子图，并指定其在图表中的位置 plt.plot(data.index.to_numpy(), residue) # 绘制残差曲线 plt.ylabel("Residue") # 设置y轴标签为 "Residue" # 绘制各个内禀模态函数 for n in range(nIMFs):#更多模型搜索机器学习之心，支持模型定制 plt.subplot(nIMFs + 2, 1, n + 2) # 创建一个子图，并指定其在图表中的位置 plt.plot(data.index.to_numpy(), IMFs[n], 'k') # 绘制内禀模态函数曲线，颜色为黑色 plt.ylabel("eIMF %i" % (n + 1)) # 设置y轴标签，显示当前内禀模态函数的索引 plt.locator_params(axis='y', nbins=4) # 设置y轴刻度数量为4 plt.tight_layout() # 调整子图的布局，使其紧凑显示 plt.show() # 显示图表 return IMFs, residue, nIMFs # 返回内禀模态函数、残差和内禀模态函数的数量 def create_dataset(dataset): dataX, dataY = [], [] # 创建空列表用于存储输入特征和目标变量 for i in range(len(dataset) - LOOK_BACK - 1): # 遍历数据集，从第一个样本开始 look_back_data = dataset[i:(i + LOOK_BACK), 0] # 获取长度为LOOK_BACK的输入特征数据 dataX.append(look_back_data) # 将输入特征添加到dataX列表中 dataY.append(dataset[i + LOOK_BACK, 0]) # 将目标变量添加到dataY列表中 return np.array(dataX), np.array(dataY) # 将dataX和dataY转换为NumPy数组并返回 def LSTM_CNN_CBAM(dataset, layer=128): # 将数据集转换为'float64'数据类型 dataset = dataset.astype('float64') # 重新调整数据集的形状为(-1, 1)，即一个特征列 dataset = np.reshape(dataset, (-1, 1)) # 创建MinMaxScaler对象 scaler = MinMaxScaler()#更多模型搜索机器学习之心，支持模型定制 # 对数据集进行归一化处理 dataset = scaler.fit_transform(dataset) # 根据划分比例计算训练集和测试集的大小#更多模型搜索机器学习之心，支持模型定制 train_size = int(len(dataset) * SPLIT) test_size = len(dataset) - train_size # 将数据集按照划分大小进行分割，得到训练集和测试集 train, test = dataset[:train_size, :], dataset[train_size:, :] # 使用create_dataset函数创建训练集和测试集的输入和输出数据 trainX, trainY = create_dataset(train) testX, testY = create_dataset(test)#更多模型搜索机器学习之心，支持模型定制 # 将训练集和测试集的输入数据调整为LSTM模型所需的形状 trainX = np.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1])) testX = np.reshape(testX, (testX.shape[0], 1, testX.shape[1])) # 创建Sequential模型 model = Sequential() # 添加第一层LSTM层，指定输入形状、隐藏单元数和激活函数 model.add(LSTM(layer, input_shape=(1, LOOK_BACK), activation='tanh', return_sequences=True)) # 添加Dropout层，防止过拟合 model.add(Dropout(0.2)) # 添加第二层LSTM层，指定隐藏单元数和激活函数 model.add(LSTM(32, activation='tanh', return_sequences=False)) # 添加Dropout层 model.add(Dropout(0.2)) # 添加全连接层，输出预测结果 model.add(Dense(1, activation='tanh')) # 编译模型，指定损失函数、优化器和评估指标 model.compile(loss='mse', optimizer='adam', metrics=['accuracy']) # 训练模型 model.fit(trainX, trainY, epochs=EPOCHS, batch_size=BATCH_SIZE, verbose=1, validation_split=0.1) # 使用训练好的模型对训练集和测试集进行预测 trainPredict = model.predict(trainX) testPredict = model.predict(testX)#更多模型搜索机器学习之心，支持模型定制 # 对预测结果进行逆归一化 trainPredict = scaler.inverse_transform(trainPredict) trainY = scaler.inverse_transform([trainY]) testPredict = scaler.inverse_transform(testPredict) testY = scaler.inverse_transform([testY])#更多模型搜索机器学习之心，支持模型定制 # 计算测试集的均方根误差（RMSE） testing_error = np.sqrt(mean_squared_error(testY[0], testPredict[:, 0])) # 返回预测结果、真实结果和测试误差 return testPredict, testY, testing_error def run_model(IMFs): # 创建空列表用于存储每个IMF的预测结果和测试误差 IMF_predict_list = [] error_list = [] # 遍历IMFs列表中的每个IMF for i, IMF in enumerate(IMFs):#更多模型搜索机器学习之心，支持模型定制 # 打印当前IMF的编号 print(f"IMF number {i+1}") # 调用LSTM_CNN_CBAM函数对当前IMF进行预测 IMF_predict, IMF_test, testing_error = LSTM_CNN_CBAM(IMF, layer=128) # 将当前IMF的预测结果和测试误差添加到对应的列表中 error_list.append(testing_error) IMF_predict_list.append(IMF_predict) # 返回所有IMF的预测结果和测试误差 return IMF_predict_list, error_list#更多模型搜索机器学习之心，支持模型定制 def visualize_results(IMF_predict_list, error_list): # 对每个IMF的预测结果进行处理，只保留第一列 for i, v in enumerate(IMF_predict_list): IMF_predict_list[i] = v[:, 0] # 创建空列表用于存储最终的预测结果 final_prediction = [] # 遍历IMF_predict_list中的每个时间步 for i in range(len(IMF_predict_list[0])): element = 0 # 对每个IMF的预测结果进行累加 for j in range(l