基于LSTM、GRU、RNN的时间预测模型python源码+模型+数据集.zip

import numpy as np import torch import pandas as pd from torch import nn from sklearn.preprocessing import MinMaxScaler from sklearn.preprocessing import StandardScaler import warnings import torch.optim as optimizer import time from torch.utils.data import DataLoader, TensorDataset from sklearn.metrics import r2_score from sklearn import metrics import matplotlib.pyplot as plt import os warnings.filterwarnings("ignore") '读取数据' filename = '../00Data/月平均气温(land+ocean)1850-2022.10.xlsx' dataset = pd.read_excel(filename) dataset = dataset.values # 选择原始数据库中需要的数据,并将DataFrame格式转换为ndarray数据类型 '将数据集划分为训练集和测试集' def split_dataset(dataset,rate_split): len = dataset.shape[0] train_dataset = dataset[0:int(len * rate_split) , 3] test_dataset = dataset[int(len * rate_split):-1 ,3] return train_dataset,test_dataset rate_split = 0.7 #原始数据集切分比例 train_dataset,test_dataset = split_dataset(dataset = dataset,rate_split = rate_split) train_dataset = train_dataset.reshape(-1,1) test_dataset = test_dataset.reshape(-1,1) '对dataset中的数据进行归一化处理,并记录特征和标签的均值和方差' choice_Normalization = '最大最小值归一化' feature_X= np.append(train_dataset,test_dataset,axis=0) #对全部数据（包括训练集和测试集）的特征（X）进行读取 label_Y = feature_X #对全部数据（包括训练集和测试集）的标签（Y）进行读取 feature_X = feature_X.reshape(-1,1) label_Y = label_Y.reshape(-1,1) #进行维度改变 if choice_Normalization == '最大最小值归一化': Standardscaler_X = StandardScaler() Standardscaler_Y = StandardScaler() feature_X = Standardscaler_X.fit_transform(feature_X) label_Y = Standardscaler_Y.fit_transform(label_Y) X_mean = Standardscaler_X.mean_ X_var = Standardscaler_X.var_ X_scale = Standardscaler_X.scale_ Y_mean = Standardscaler_Y.mean_ Y_var = Standardscaler_Y.var_ Y_scale = Standardscaler_Y.scale_ elif choice_Normalization == '均值方差归一化' : Standardscaler_X = StandardScaler() Standardscaler_Y = StandardScaler() feature_X = Standardscaler_X.fit_transform(feature_X) label_Y = Standardscaler_Y.fit_transform(label_Y) X_mean = Standardscaler_X.mean_ X_var = Standardscaler_X.var_ X_scale = Standardscaler_X.scale_ Y_mean = Standardscaler_Y.mean_ Y_var = Standardscaler_Y.var_ Y_scale = Standardscaler_Y.scale_ '构建时间序列需要的数据集' def createDataset(dataset, look_back, ahead_step, feature_num): ''' :param dataset: 表示需要进行操作的数据库 :param look_back: 选择多少个历史数据进行分析 :param ahead_step: 预测步长为多少 :param feature_num: 选择多少个特征进行预测 :return: ''' dataX = [] dataY = [] for i in range(len(dataset) - look_back - ahead_step + 1): dataX.append(dataset[i : i + look_back]) dataY.append(dataset[i + (look_back - 1) + ahead_step]) dataX = torch.tensor(dataX) dataX = dataX.reshape(-1 , feature_num) dataY = torch.tensor(dataY) dataY = dataY[: , 0] dataY = dataY.reshape(-1 , 1) return dataX, dataY feature_num = 1 # 特征个数 ahead_step = 1 # 预测步数 look_back = 1 # 使用的历史数据的个数 #对训练集进行归一化 Standardscaler_trainX = StandardScaler() Standardscaler_trainX.mean_ = X_mean Standardscaler_trainX.var_ = X_var Standardscaler_trainX.scale_ = X_scale #对测试集进行归一化 Standardscaler_testX = StandardScaler() Standardscaler_testX.mean_ = X_mean Standardscaler_testX.var_ = X_var Standardscaler_testX.scale_ = X_scale train_dataset = Standardscaler_trainX.transform(train_dataset) #训练集归一化 test_dataset = Standardscaler_testX.transform(test_dataset) #测试集归一化 train_dataX, train_dataY = createDataset(dataset = train_dataset,look_back = look_back,ahead_step = ahead_step,feature_num = feature_num) test_dataX , test_dataY = createDataset(dataset = test_dataset ,look_back = look_back,ahead_step = ahead_step,feature_num = feature_num) '将dataX，dataY转换为LSTM模型的输入格式类型' train_dataX = train_dataX.reshape(-1,look_back,feature_num) train_dataY = train_dataY.reshape(-1,1,1) test_dataX = test_dataX.reshape (-1,look_back,feature_num) test_dataY = test_dataY.reshape (-1,1,1) '搭建模型LSTM、GRU、RNN' class LSTMModel(nn.Module): def __init__(self, input_Size, hidden_Size, num_layers , dropout , batch_first): super(LSTMModel, self).__init__() # LSTM层-> 两个LSTM单元叠加 self.lstm = nn.LSTM(input_size = input_Size,hidden_size = hidden_Size, num_layers = num_layers,dropout = dropout,batch_first = batch_first) self.linear = nn.Linear(hidden_Size,1) # 线性输出 def forward(self,x): lstm_out,hidden = self.lstm(x) linear_out = self.linear(lstm_out) return linear_out class RNNModel(nn.Module): def __init__(self, input_Size, hidden_Size,num_layers, dropout , batch_first): super(RNNModel, self).__init__() self.rnn = nn.RNN(input_size = input_Size,hidden_size = hidden_Size, num_layers = num_layers,dropout = dropout,batch_first = batch_first) self.linear = nn.Linear(hidden_Size,1) # 线性输出 def forward(self,x): lstm_out,hidden = self.rnn(x) linear_out = self.linear(lstm_out) return linear_out class GRUModel(nn.Module): def __init__(self, input_Size, hidden_Size,num_layers,dropout , batch_first): super(GRUModel, self).__init__() self.GRU = nn.GRU(input_size = input_Size,hidden_size = hidden_Size, num_layers = num_layers,dropout = dropout,batch_first = batch_first) self.linear = nn.Linear(hidden_Size,1) # 线性输出 def forward(self,x): lstm_out,hidden = self.GRU(x) linear_out = self.linear(lstm_out) return linear_out '网络参数' dropout = 0 batch_first = True hidden_Size = 10 num_layers = 1 Model_LSTM = RNNModel(input_Size = feature_num,hidden_Size = hidden_Size,num_layers = num_layers,dropout = dropout,batch_first = batch_first) Model_LSTM.double() #将所有的浮点类型的参数和缓冲转换为(双浮点)double数据类型. num_epoch = 500 batch_size = 100 optimizer = optimizer.SGD(Model_LSTM.parameters(), lr = 0.001) lossFn = nn.MSELoss() #损失函数 train_loader = DataLoader(TensorDataset(train_dataX, train_dataY), batch_size = batch_size) '模型训练过程' T_start = time.time() train_loss = [] # 用来保存训练集的对数均方根误差 for epoch in range(num_epoch): for i,data in enumerate(train_loader): train_X,train_Y = data optimizer.zero_grad() pred_y = Model_LSTM(train_X) loss = lossFn(pred_y,train_Y) loss.backward() optimizer.step() if epoch % 20 == 0: print(f"Epoch: {epoch}, Loss: {loss.item()}") train_time = time.time() - T_start print( "网络训练完毕,训练时长为:%.2f s" % train_time) Save_path = '../03Save_Model/Model_LSTM时间序列预测1.pt' torch.save(Model_LSTM.state_dict(), Save_path) '进行预测' # 创建一个一模一样的模型，加载预训练模型的参数 Model_LSTM = RNNModel(input_Size = feature_num,hidden_Size = hidden_Size,num_layers = num_layers,dropout = dropout,batch_first = batch_first) Model_LSTM.double() #将所有的浮点类型的参数和缓冲转换为(双浮点)double数据类型. Model_LSTM.load_state_dict(torch.load("../03Save_Model/Model_LSTM时间序列预测1.pt")) Model_LSTM.eval() # 不启用 BatchNormalization 和 Dropout，保证BN和dropout不发生变化 pred_testX = Model_LSTM(test_dataX) # 对测试集进行拟合 pred_train