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

#!/usr/bin/env python # coding: utf-8 # In[2]: import itertools # 导入itertools库，用于迭代工具 import sys # 导入sys库，用于系统相关的功能 import math # 导入math库，用于数学运算 import numpy as np # 导入numpy库，并将其别名为np，用于数值计算 import pandas as pd # 导入pandas库，并将其别名为pd，用于数据处理和分析 from numpy import concatenate # 从numpy库中导入concatenate函数，用于数组拼接 from pandas import concat, DataFrame # 从pandas库中导入concat和DataFrame函数，用于数据合并和构造数据框 import matplotlib.pyplot as plt # 导入matplotlib.pyplot库，并将其别名为plt，用于绘图 # from sklearn.preprocessing import MinMaxScaler # 导入MinMaxScaler类，用于特征缩放 from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score # 导入评估指标类，用于模型评估 # from keras.layers import Dense, Flatten, RepeatVector # 导入Dense、Flatten和RepeatVector类，用于神经网络层的构建 # from keras.models import Sequential # 导入Sequential类，用于构建序列模型 from keras.layers import LSTM # 导入LSTM类，用于构建LSTM模型 from keras.layers import GRU # 导入GRU类，用于构建GRU模型 from keras.layers.convolutional import Conv1D # 导入Conv1D类，用于构建卷积神经网络模型 from keras.layers.convolutional import MaxPooling1D # 导入MaxPooling1D类，用于构建卷积神经网络模型中的池化层 from keras.layers import Dropout # 导入Dropout类，用于构建模型中的Dropout层 from statsmodels.tsa.arima_model import ARIMA # 导入ARIMA类，用于构建ARIMA模型 import tensorflow as tf # 导入tensorflow库，并将其别名为tf，用于深度学习 import statsmodels.api as sm # 导入statsmodels.api库，并将其别名为sm，用于统计分析 from keras.layers import TimeDistributed # 导入TimeDistributed类，用于构建时间分布层 import matplotlib # 导入matplotlib库，用于绘图 import warnings # 导入warnings库，用于忽略警告信息 import statsmodels # 导入statsmodels库，用于统计分析 from scipy import stats # 导入stats模块，用于统计分析 # In[3]: # 调用GPU加速 gpus = tf.config.experimental.list_physical_devices(device_type='GPU') for gpu in gpus: tf.config.experimental.set_memory_growth(gpu, True) # In[4]: # 用来正常显示中文标签 plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示负号 plt.rcParams['axes.unicode_minus'] = False # In[5]: # 读入数据 data_raw = pd.read_csv('焦作.csv', usecols=[0, 1]) # # 选择特征列 features = ['AQI'] data_raw = data_raw[features] # In[6]: plt.figure(figsize=(10, 3), dpi=150) # 创建一个图形窗口，设置图形的大小为宽度10英寸，高度3英寸，像素密度为150 plt.title('AQI') # 设置图形的标题为'AQI' plt.xlabel('time') # 设置x轴标签为'time' plt.ylabel('AQI') # 设置y轴标签为'AQI' plt.plot(data_raw, 'r', label='AQI') # 绘制数据折线图，使用红色('r')的线条表示，标签为'AQI' plt.legend() # 显示图例，将图例放置在合适的位置 plt.show() # 显示绘制的图形 # In[7]: p_min = 0 # 定义p的最小值为0，用于ARIMA模型中的自回归阶数 p_max = 5 # 定义p的最大值为5，用于ARIMA模型中的自回归阶数 d_min = 0 # 定义d的最小值为0，用于ARIMA模型中的差分阶数 d_max = 2 # 定义d的最大值为2，用于ARIMA模型中的差分阶数 q_min = 0 # 定义q的最小值为0，用于ARIMA模型中的移动平均阶数 q_max = 5 # 定义q的最大值为5，用于ARIMA模型中的移动平均阶数 # # In[8]: # 使用信息准则（AIC和BIC）选择最佳ARMA模型的阶数 # data_raw: 用于模型阶数选择的时间序列数据 # ic=['aic', 'bic']: 用于评估的信息准则，即AIC和BIC # trend='c': 模型的趋势参数（常数项） # max_ar: AR阶数（p）的最大值 # max_ma: MA阶数（q）的最大值 train_results = sm.tsa.arma_order_select_ic(data_raw, ic=['aic', 'bic'], trend='c', max_ar=p_max, max_ma=q_max) # In[9]: # 打印AIC的最小阶数 print('AIC', train_results.aic_min_order) # 打印BIC的最小阶数 print('BIC', train_results.bic_min_order) # # In[10]: # 创建ARIMA模型对象，指定阶数为(3, 2, 1) model = sm.tsa.ARIMA(data_raw, order=(3, 2, 1)) # 使用给定的时间序列数据拟合ARIMA模型 fit = model.fit() # # In[11]: # 使用拟合好的模型进行预测，预测从1到len(data_raw)，预测类型为'levels' preds = fit.predict(1, len(data_raw), typ='levels') # # 将预测结果转换为DataFrame格式 preds_pd = preds.to_frame() # # 将DataFrame的索引减去1，以对齐原始数据的索引 preds_pd.index -= 1 # In[12]: # 创建一个空的DataFrame，列名为'AQI' arima_result = pd.DataFrame(columns=['AQI']) # # 将原始数据赋值给'AQI'列 arima_result['AQI'] = data_raw # # 将预测结果赋值给'predicted'列 arima_result['predicted'] = preds_pd # # 计算残差，并将结果赋值给'residuals'列 arima_result['residuals'] = arima_result['AQI'] - arima_result['predicted'] # In[13]: # 将arima_result赋值给新的变量new_data new_data = arima_result # # 从new_data中提取'residuals'列的值，并转换为浮点型数组 lstm_data = new_data['residuals'][:].values.astype(float) # In[14]: def series_to_supervised(data, n_in, n_out, dropnan=True): # 确定输入数据的变量数目 n_vars = 1 if type(data) is list else data.shape[1] # 创建一个DataFrame对象来存储转换后的数据 df = DataFrame(data) # 初始化列和列名的列表 cols, names = list(), list() # 创建输入序列的列 for i in range(n_in, 0, -1): cols.append(df.shift(i)) names += [('var%d(t-%d)' % (j + 1, i)) for j in range(n_vars)] # # 创建输出序列的列 for i in range(0, n_out): cols.append(df.shift(-i)) if i == 0: names += [('var%d(t)' % (j + 1)) for j in range(n_vars)] else: names += [('var%d(t+%d)' % (j + 1, i)) for j in range(n_vars)] # 合并列为一个DataFrame agg = concat(cols, axis=1) agg.columns = names # 删除包含NaN值的行 if dropnan: agg.dropna(inplace=True) return agg # # In[15]: # 生成LSTM所需要的3维数据格式：(样本数，时间步长，特征数) def dataprepare(values, timestep): # 使用series_to_supervised函数将数据转换为监督学习问题的数据集 reframed = series_to_supervised(values, timestep, 1) # X,y values = reframed.values # # 划分训练集和测试集 train = values[1:train_len, :] test = values[train_len:, :] # # 获取输入X和标签（即y） train_X, train_y = train[:, :-1], train[:, -1] test_X, test_y = test[:, :-1], test[:, -1] # 将输入重塑为3D格式 [样例，时间步，特征] train_X = train_X.reshape((train_X.shape[0], 1, train_X.shape[1])) test_X = test_X.reshape((test_X.shape[0], 1, test_X.shape[1])) print("train_X.shape:%s train_y.shape:%s test_X.shape:%s test_y.shape:%s" % ( train_X.shape, train_y.shape, test_X.shape, test_y.shape)) return train_X, train_y, test_X, test_y # In[16]: # 创建一个MinMaxScaler对象，指定特征范围为(-1, 1) scaler = MinMaxScaler(feature_range=(-1, 1)) # # 使用MinMaxScaler对象对数据进行缩放和转换，将数据转换为指定的特征范围(-1, 1) scaler_data = scaler.fit_transform(lstm_data.reshape(-1, 1)) # # In[17]: # 计算训练集长度 train_len = int(len(data_raw) * 0.80) # 计算测试集长度 test_len = len(data_raw) - train_len # 打印训练集长度 print(train_len) # In[18]: timestep = 13 # 滑动窗口 # 调用dataprepare函数准备训练集和测试集的数据 x_train,