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

#!/usr/bin/env python # coding: utf-8 # In[1]: import time # 导入time模块，用于时间相关的操作 import datetime # 导入datetime模块，用于处理日期和时间 import pandas as pd # 导入pandas库，用于数据处理和分析 import numpy as np # 导入numpy库，用于数值计算 import matplotlib.pyplot as plt # 导入matplotlib库，用于绘图 from sampen import sampen2 # 从sampen模块导入sampen2函数，用于计算样本熵 from vmdpy import VMD # 从vmdpy模块导入VMD类，用于进行VMD分解 import tensorflow as tf # 导入tensorflow库，用于构建和训练深度学习模型 from sklearn.cluster import KMeans # 从sklearn.cluster模块导入KMeans类，用于进行K-Means聚类分析 from sklearn.metrics import r2_score, mean_squared_error, mean_absolute_error, mean_absolute_percentage_error # 从sklearn.metrics模块导入回归模型的评估指标 from sklearn.preprocessing import MinMaxScaler # 从sklearn.preprocessing模块导入MinMaxScaler类，用于特征缩放 from tensorflow.keras.models import Sequential # 从tensorflow.keras.models模块导入Sequential类，用于构建序贯模型 from tensorflow.keras.layers import Dense, Activation, Dropout, LSTM, GRU # 从tensorflow.keras.layers模块导入用于构建深度学习模型的各种层 from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping # 从tensorflow.keras.callbacks模块导入回调函数，用于在训练过程中进行学习率调整和早停 import warnings # 导入warnings模块，用于控制警告信息的显示 warnings.filterwarnings('ignore') # 忽略警告信息 # In[2]: plt.rcParams['font.sans-serif'] = ['Times New Roman'] # 设置字体为 Times New Roman，用于绘图时显示中文字符 plt.rcParams['axes.unicode_minus'] = False # 解决绘图时负号显示为方块的问题 # In[3]: #更多模型搜索机器学习之心，支持模型定制 # In[4]: # 从 '焦作.csv' 文件中读取数据，并只使用第一列和第二列的数据 df_raw_data = pd.read_csv('焦作.csv', usecols=[0, 1]) # 将 'AQI' 列的值转换为 Series，并将 'time' 列作为索引 series_close = pd.Series(df_raw_data['AQI'].values, index=df_raw_data['time']) # 获取 df_raw_data['AQI'] 列值的后 80% 的数据作为测试数据 test = df_raw_data['AQI'].values[int(len(df_raw_data['AQI'].values) * 0.8):] # In[5]: def vmd_decompose(series=None, alpha=2000, tau=0, K=7, DC=0, init=1, tol=1e-7, draw=True): # 调用 VMD 函数进行分解，得到 VMD 分解后的各个分量、分解后的信号和频率 imfs_vmd, imfs_hat, omega = VMD(series, alpha, tau, K, DC, init, tol) # 将 VMD 分解得到的各个分量转换为 DataFrame df_vmd = pd.DataFrame(imfs_vmd.T) # 为 DataFrame 的列命名为 'imf0', 'imf1', 'imf2', ... df_vmd.columns = ['imf'+str(i) for i in range(K)] # 返回 VMD 分解得到的各个分量的 DataFrame return df_vmd # In[6]: def LSTM_model(trainset_shape): # 创建 Sequential 模型 model = Sequential() # 添加第一层 LSTM，输入形状为 (trainset_shape[1], trainset_shape[2])，激活函数为 'tanh'，返回序列 model.add(GRU(128, input_shape=(trainset_shape[1], trainset_shape[2]), activation='tanh', return_sequences=True)) # 添加 Dropout 层，丢弃比例为 0.2 model.add(Dropout(0.2)) # 添加第二层 LSTM，激活函数为 'tanh'，返回序列 model.add(GRU(64, activation='tanh', return_sequences=True)) # 添加 Dropout 层，丢弃比例为 0.2 model.add(Dropout(0.2)) # 添加第三层 LSTM，激活函数为 'tanh'，返回单个输出值（不返回序列） model.add(GRU(32, activation='tanh', return_sequences=False)) # 添加 Dropout 层，丢弃比例为 0.2 model.add(Dropout(0.2)) # 添加全连接层，输出维度为 1，激活函数为 'tanh' model.add(Dense(1, activation='tanh')) # 编译模型，损失函数为 'mse'，优化器为 'adam' model.compile(loss='mse', optimizer='adam') # 返回构建好的模型 return model # In[7]: def evaluation_model(y_test, y_pred): # 将 y_test 和 y_pred 转换为一维数组 y_test, y_pred = np.array(y_test).ravel(), np.array(y_pred).ravel() # 计算 R^2 分数 r2 = r2_score(y_test, y_pred) # 计算均方根误差（RMSE） rmse = mean_squared_error(y_test, y_pred, squared=False) # 计算平均绝对误差（MAE） mae = mean_absolute_error(y_test, y_pred) # 计算平均绝对百分比误差（MAPE） mape = mean_absolute_percentage_error(y_test, y_pred) # 创建包含评估指标的 DataFrame df_evaluation = pd.DataFrame({'r2': r2, 'rmse': rmse, 'mae': mae, 'mape': mape}, index=range(1)) # 返回评估结果的 DataFrame return df_evaluation#更多模型搜索机器学习之心，支持模型定制 # In[8]: def create_train_test_set(data=None, timestep=30, co_imf_predict_for_fitting=None): # 检查 data 是否为 DataFrame，如果是，则将数据分为输入特征和目标变量 if isinstance(data, pd.DataFrame): dataY = data['sum'].values.reshape(-1, 1) # 提取目标变量并转换为二维数组 dataX = data.drop('sum', axis=1, inplace=False) # 提取输入特征 else: dataY = data.values.reshape(-1, 1) # 将目标变量转换为二维数组 dataX = dataY # 将输入特征设置为目标变量 # 对输入特征进行归一化处理 scalarX = MinMaxScaler(feature_range=(0, 1)) # 创建归一化器对象 dataX = scalarX.fit_transform(dataX) # 对输入特征进行归一化处理 # 如果有 co_imf_predict_for_fitting 参数，对其进行归一化处理 if co_imf_predict_for_fitting is not None: co_imf_predict_for_fitting = scalarX.transform(co_imf_predict_for_fitting) # 对目标变量进行归一化处理 scalarY = MinMaxScaler(feature_range=(0, 1)) # 创建归一化器对象 dataY = scalarY.fit_transform(dataY) # 对目标变量进行归一化处理 trainX, trainY = [], []#更多模型搜索机器学习之心，支持模型定制 for i in range(len(dataY) - timestep):#更多模型搜索机器学习之心，支持模型定制 # 构建训练集，每个样本包含 timestep 个时间步的输入特征 trainX.append(np.array(dataX[i:(i + timestep)])) # 对应的目标变量为 timestep+1 的值 trainY.append(np.array(dataY[i + timestep])) # 如果有 co_imf_predict_for_fitting 参数，将其插入到训练集的末尾 if co_imf_predict_for_fitting is not None: if i < (len(dataY) - timestep - len(co_imf_predict_for_fitting)):#更多模型搜索机器学习之心，支持模型定制 trainX[i] = np.insert(trainX[i], timestep, dataX[i + timestep], 0)#更多模型搜索机器学习之心，支持模型定制 else: trainX[i] = np.insert(trainX[i], timestep, co_imf_predict_for_fitting[i - (len(dataY) - timestep - len(co_imf_predict_for_fitting))], 0) # 返回训练集的输入特征、目标变量以及目标变量的归一化器 return np.array(trainX), np.array(trainY), scalarY # In[9]: def LSTM_predict(data=None, predict_duration=len(test), fitting=None): # 创建训练集和测试集 trainX, trainY, scalarY = create_train_test_set(data, co_imf_predict_for_fitting=fitting) x_train, x_test = trainX[:-predict_duration], trainX[-predict_duration:] y_train, y_test = trainY[:-predict_duration], trainY[-predict_duration:] # 将训练集和测试集进行形状调整以符合 LSTM 模型的输入要求 train_X = x_train.reshape((x_train.shape[0], x_train.shape[1], x_train.shape[2])) test_X = x_test.reshape((x_test.shape[0], x_test.shape[1], x_test.shape[2])) # 创建 LSTM 模型 model = LSTM_model(train_X.shape) # 使用训练集训练模型 history = model.fit(train_X,