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

#!/usr/bin/env python # coding: utf-8 # In[1]: import itertools # 导入 itertools 库用于迭代工具 import math # 导入 math 库进行数学运算 import numpy as np # 导入 numpy 库进行数组操作 import pandas as pd # 导入 pandas 库进行数据处理和分析 import matplotlib.pyplot as plt # 导入 matplotlib 库进行数据可视化 from numpy import concatenate # 导入 concatenate 函数从 numpy from pandas import concat, DataFrame # 导入特定函数和类从 pandas # from statsmodels.tsa.arima_model import ARIMA # 导入 ARIMA 模型从 statsmodels from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score # 导入性能评估指标从 sklearn from sklearn.preprocessing import MinMaxScaler # 导入 MinMaxScaler 数据预处理类从 sklearn from tensorflow.python.keras import Sequential # 导入 Sequential 类从 tensorflow.python.keras from tensorflow.python.keras.layers import LSTM, Dropout, Dense # 导入 LSTM、Dropout 和 Dense 层从 tensorflow.python.keras # import statsmodels.api as sm # 导入 statsmodels 库 import matplotlib # 导入 matplotlib 库 import statsmodels # 导入 statsmodels 库 import tensorflow as tf # 导入 TensorFlow 库 # from tensorflow.python.keras.callbacks import EarlyStopping # 导入 EarlyStopping 回调函数从 tensorflow.python.keras.callbacks from tensorflow.python.keras.layers.core import Activation # 导入 Activation 类从 tensorflow.python.keras.layers.core # In[2]: # # In[3]: # 用来正常显示中文标签 plt.rcParams['font.sans-serif'] = ['SimHei'] # 用来正常显示负号 plt.rcParams['axes.unicode_minus'] = False # In[4]: # 读入数据 data_raw = pd.read_csv('焦作.csv', usecols=[0, 1]) # 从数据中选择列索引为0和1的列作为数据 # features = ['AQI'] # 定义要保留的特征列 data_raw = data_raw[features] # 从数据中选择特定的特征列 # In[5]: 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模型中的移动平均阶数 # # 使用信息准则（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[6]: # 打印AIC的最小阶数 print('AIC', train_results.aic_min_order) # 打印BIC的最小阶数 print('BIC', train_results.bic_min_order) # # In[7]: # 创建ARIMA模型对象，指定阶数为(3, 2, 1) model = sm.tsa.ARIMA(data_raw, order=(3, 2, 1)) # 使用给定的时间序列数据拟合ARIMA模型 fit = model.fit() # # In[8]: # 使用拟合好的模型进行预测，预测从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[9]: # 创建一个空的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[10]: # 将arima_result赋值给新的变量new_data new_data = arima_result # # 从new_data中提取'residuals'列的值，并转换为浮点型数组 lstm_data = new_data['residuals'][:].values.astype(float) # In[11]: 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[12]: # 生成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[13]: # 创建一个MinMaxScaler对象，指定特征范围为(-1, 1) scaler = MinMaxScaler(feature_range=(-1, 1)) # 使用MinMaxScaler对象对数据进行缩放和转换，将数据转换为指定的特征范围(-1, 1) scaler_data = scaler.fit_transform(lstm_data.reshape(-1, 1)) # # In[14]: # 计算训练集长度 train_len = int(len(data_raw) * 0.80) # # 计算测试集长度 test_len = len(data_raw) - train_len # In[15]: def build_model(timestep, neurons1, neurons2, neurons3, dropout): timestep = timestep # 滑动窗口大小 x_train, y_train, x_test, y_test = dataprepare(scaler_data, timestep) # 数据准备函数的调用，返回训练和测试数据 # model = Sequential() # 创建一个顺序型模型 model.add(LSTM(units=neurons1, return_sequences=True, activation='tanh', input_shape=(x_train.shape[1], x_train.shape[2]))) # 添加一个 LSTM 层，包含 neurons1 个神经元，返回序列数据，激活函数为 tanh，输入形状为 (x_train.shape[1], x_train.shape[2]) model.add(LSTM(units=neurons2, return_sequences=True, activation='tanh', input_shape=(x_train.shape[1], x_train.shape[2]))) # 添加另一个 LSTM 层，包含 neurons2 个神经元，返回序列数据，激活函数为 tanh，输入形状为 (x_train.shape[1], x_train.shape[2]) model.add(LSTM(neurons3, return_sequences=False)) # 添加另一个 LSTM 层，包含 neurons3 个神经元，不返回序列数据 model.add(Dropout(dropout)) # 添加一个 Dropout 层，以指定比例的神经元失活 model.add(Dense(1)) # 添加一个全连接层，输出维度为 1 model.compile(loss='mean_squared_error', optimizer='Adam') # 编译模型，指定损失函数和优化器 return model, x_train, y_train, x_test, y_test # In[16]: def training(X): timestep = int(X[0]) # 获取滑动窗口大小 neurons1 = int(X[1]) # 获取第一