Python实现CEEMDAN-CNN-BILSTM-attention时间序列预测（完整源码和数据)

#!/usr/bin/env python # coding: utf-8 # In[1]: from keras.models import Sequential # 导入Sequential模型，用于构建深度学习模型 import tensorflow as tf # 导入tensorflow库 from sklearn.preprocessing import MinMaxScaler # 导入MinMaxScaler，用于数据标准化 from PyEMD import CEEMDAN # 导入CEEMDAN类，用于进行经验模态分解 import pandas as pd # 导入pandas库，用于数据处理 import numpy as np # 导入numpy库，用于数组操作 import matplotlib.pyplot as plt # 导入matplotlib库，用于数据可视化 import random # 导入random库，用于生成随机数 import warnings # 导入warnings库，用于警告处理 warnings.filterwarnings('ignore') # 忽略警告信息 from keras.layers import Input, Dense, LSTM, Conv1D, Dropout, Bidirectional, Multiply, Concatenate, BatchNormalization, MaxPooling1D, Flatten, Activation, Permute, Lambda # 导入各种神经网络层和函数 from keras.models import Model # 导入Model模块，用于构建模型 from sklearn.metrics import r2_score, mean_squared_error, mean_absolute_error, mean_absolute_percentage_error # 导入各种性能评估指标 from keras.utils import plot_model # 导入plot_model函数，用于绘制模型结构图 from keras import optimizers # 导入优化器模块，用于设置模型优化算法 import numpy # 导入numpy库 import math # 导入math库，用于数学计算 import datetime # 导入datetime库，用于日期和时间处理 from pandas import read_csv # 导入read_csv函数，用于读取CSV文件 from keras import backend as K # 导入K模块，用于底层操作 # In[2]: plt.rcParams['font.sans-serif'] = ['Times New Roman'] # 设置字体样式为Times New Roman，用于图形中的文本显示 plt.rcParams['axes.unicode_minus'] = False # 设置是否显示坐标轴上的负号（减号），将其设置为False表示显示负号 # In[3]: # # In[4]: LOOK_BACK = 10 # 定义观察窗口大小为10，表示使用过去10个时间步的数据来预测未来的值 SPLIT = 0.8 # 定义数据集拆分比例为0.8，表示将80%的数据用于训练，20%的数据用于测试 EPOCHS = 100 # 定义训练的轮数为100，表示将训练数据迭代100次 BATCH_SIZE = 100 # 定义批量大小为100，表示每次训练使用100个样本进行参数更新 # In[5]: data = pd.read_csv('焦作.csv', usecols=[1]) # 从名为'焦作.csv'的文件中读取数据，仅使用第二列数据（索引为1的列） data.columns = ["AQI"] # 将数据的列名设置为"AQI" # In[6]: def get_CEEMD_residue(data: pd.DataFrame): data_np = data.to_numpy() # 将数据转换为NumPy数组 ceemd = CEEMDAN() # 创建CEEMDAN对象 ceemd.extrema_detection = "parabol" # 设置极值点检测方法为"parabol" ceemd.ceemdan(data_np) # 对数据进行CEEMDAN分解 IMFs, residue = ceemd.get_imfs_and_residue() # 获取分解后的IMFs和残差 nIMFs = IMFs.shape[0] # 获取IMFs的数量 plt.figure(figsize=(18, 18), dpi=100) # 创建一个图形窗口，设置大小为18x18，分辨率为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, residue) # 绘制残差曲线 plt.ylabel("Residue") # 设置y轴标签为"Residue" for n in range(nIMFs): # 遍历每个IMF plt.subplot(nIMFs + 2, 1, n + 2) # 创建子图，显示每个IMF plt.plot(data.index, IMFs[n], 'k') # 绘制IMF曲线，颜色为黑色 plt.ylabel("eIMF %i" % (n + 1)) # 设置y轴标签为"eIMF %i"，其中%i为IMF的索引+1 plt.locator_params(axis='y', nbins=4) # 设置y轴刻度数量为4 # plt.tight_layout() # 调整子图布局 plt.show() # 显示图形 return IMFs, residue, nIMFs # 返回IMFs、残差和IMFs的数量 # In[7]: def attention_function(inputs, single_attention_vector=False): TimeSteps = K.int_shape(inputs)[1] # 获取输入的时间步数 input_dim = K.int_shape(inputs)[2] # 获取输入的特征维度 a = Permute((2, 1))(inputs) # 对输入进行维度置换，将特征维度放在前面 a = Dense(TimeSteps, activation='softmax')(a) # 在特征维度上应用全连接层，并使用softmax激活函数 if single_attention_vector: a = Lambda(lambda x: K.mean(x, axis=1))(a) # 如果指定单个注意力向量，将在时间步维度上取平均值 a = RepeatVector(input_dim)(a) # 将注意力向量复制到每个时间步上 a_probs = Permute((2, 1))(a) # 对注意力权重进行维度置换，恢复原始维度顺序 output_attention_mul = Multiply()([inputs, a_probs]) # 将输入与注意力权重相乘 return output_attention_mul # 返回加权后的输出 # In[9]: def create_dataset(dataset): dataX, dataY = [], [] # 创建空列表用于存储输入数据和目标数据 for i in range(len(dataset) - LOOK_BACK - 1): # 遍历数据集，从第一个样本到倒数第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数组并返回 # In[10]: def LSTM_CNN_CBAM(dataset, layer=128): dataset = dataset.astype('float64') # 将数据集转换为float64类型 dataset = np.reshape(dataset, (-1, 1)) # 调整数据集的形状为(-1, 1) scaler = MinMaxScaler() # 创建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:, :] # 将数据集分割为训练集和测试集 trainX, trainY = create_dataset(train) # 创建训练集的输入数据和目标数据 testX, testY = create_dataset(test) # 创建测试集的输入数据和目标数据 # trainX = np.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1])) # 调整训练集输入数据的形状 testX = np.reshape(testX, (testX.shape[0], 1, testX.shape[1])) # 调整测试集输入数据的形状 model = Sequential() # 创建Sequential模型 # inputs = Input(shape=(1, LOOK_BACK)) # 定义输入层形状 x = Conv1D(filters=128, kernel_size=1, activation='relu')(inputs) # 添加一维卷积层 BiLSTM_out = Bidirectional(LSTM(128, return_sequences=True, activation="relu"))(x) # 添加双向LSTM层 BiLSTM_out = Bidirectional(LSTM(64, return_sequences=True, activation="relu"))(BiLSTM_out) # 添加双向LSTM层 Batch_Normalization = BatchNormalization()(BiLSTM_out) # 添加批标准化层 Drop_out = Dropout(0.1)(Batch_Normalization) # 添加Dropout层# attention = attention_function(Drop_out) # 添加注意力层 Batch_Normalization = BatchNormalization()(attention) # 添加批标准化层 Drop_out = Dropout(0.1)(Batch_Normalization) # 添加Dropout层 Flatten_ = Flatten()(Drop_out) # 添加Flatten层 output = Dropout(0.1)(Flatten_) # 添加Dropout层 output = Dense(1, activation='sigmoid')(output) # 添加全连接层 model = Model(inputs=[inputs], outputs=output) # 创建模型 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(tr