基于pytorch的LSTM预测实现（入门级别）

# User: 廖宇 # Data Development:2023/5/14 17:07 import torch from datetime import datetime from torchvision import transforms, datasets from torch import nn, optim from torch.autograd import Variable from torch.utils.data import DataLoader import torch.utils.data as Data import numpy as np import pandas as pd import torch.nn.functional as F import matplotlib.pyplot as plt from sklearn.preprocessing import MinMaxScaler,StandardScaler from sklearn.metrics import mean_squared_error from sklearn.metrics import mean_absolute_error import os os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE' import warnings warnings.filterwarnings("ignore") plt.style.use('ggplot') seq_length =32#time step input_size =7#input dim hidden_size = 5 num_layers = 1 num_classes = 1#输出维度 learning_rate = 0.0001 batch_size = 64 n_iters = 40000 split_ratio = 0.8 # data_path = 'WTH.csv' data_path = './data/ETTm1.csv' torch.manual_seed(10) def read_data(data_path): ''' read original data, show and plot ''' # data = pd.read_excel(data_path) # data = pd.read_csv(data_path,parse_dates=['date'],index_col="date") data = pd.read_csv(data_path,parse_dates=['date'],index_col=['date']) # data = data[:0.3*len(data)] # data = data[data.index.year.isin([2016])].copy() #将'2016-07-01 00:00:00转换成dataframe日期格式 # data["date"]= pd.to_numeric(data["date"],errors="ignore") # data.drop(['date'],axis=1,inplace=True) # data = data.values # 将pd的系列格式转换为np的数组格式 data['HUFL']= data['HUFL'].astype(float) data['HULL']= data['HULL'].astype(float) data['MUFL']= data['MUFL'].astype(float) data['MULL']= data['MULL'].astype(float) data['LUFL']= data['LUFL'].astype(float) data['LULL']= data['LULL'].astype(float) data['HUFL']= data['HUFL'].astype(float) # Remove $ from prices and cast as float # df['Close'] = df['Close'].str.replace('$', '').astype(float) # df['Open'] = df['Open'].str.replace('$', '').astype(float) # df['High'] = df['High'].str.replace('$', '').astype(float) # df['Low'] = df['Low'].str.replace('$', '').astype(float) # Convert Date column to numeric data type # df["Date"] = pd.to_numeric(df["Date"], errors='ignore') # data = data.iloc[:int(0.1*len(data)),:] # plt.ylabel('output_temperature') # plt.autoscale(axis='x', tight=True) # plt.plot(data.iloc[:, 0]) # plt.legend(['data_temperature']) # plt.show() # data = data.dropna() # data = data.drop(['date'], axis=1,inplace=True) # data=data[:,:7].values#不包括最后一行 loc包括 # label=data[:,-1].values # data = data.iloc[:, :6] label = data.iloc[:,-1] # print(data.head()) return data,label def normalization(data,label): ''' normalization ''' # mm_x = MinMaxScaler() # mm_y = MinMaxScaler() mm_x = StandardScaler() mm_y = StandardScaler() # data = data.values # 将pd的系列格式转换为np的数组格式 # data["HUFL"]= mm_x.fit_transform(data[['HUFL']]) # data["HULL"]= mm_x.fit_transform(data[['HULL']]) # data["MUFL"]= mm_x.fit_transform(data[['MUFD']]) # data["MULL"]= mm_x.fit_transform(data[['MULL']]) # # data["LUFL"]= mm_x.fit_transform(data[['LUFL']]) # data["LULL"]= mm_x.fit_transform(data[['LULL']]) # data["HUFL"]= mm_x.fit_transform(data[['HUFD']]) data = mm_x.fit_transform(data) # data = pd.DataFrame(data) label = mm_y.fit_transform(label.values.reshape(-1, 1)) # label = pd.DataFrame(label) return data, label, mm_y def sliding_windows(data, seq_length): ''' Output:The data form we can feed GRU ''' x = [] y = [] for i in range(len(data)-seq_length-1):#子序列的个数 长度为seq_length _x = data[i:(i+seq_length),:]#子序列 seq_length行 _y = data[i+seq_length,-1]#子序列最后一列的下一列最后一个元素 作为标签 # _y = data[i + seq_length-1, -1] x.append(_x) y.append(_y) x, y = np.array(x),np.array(y)#这段代码的作用是为了将x和y转换为NumPy数组 print('x.shape,y.shape:\n',x.shape,y.shape) return x, y def data_split(x, y, split_ratio): ''' convert to torch format and split train&test ''' train_size = int(len(y) * split_ratio) test_size = int(len(y) *(1 - split_ratio -0.1) ) x_data = Variable(torch.FloatTensor(np.array(x))) y_data = Variable(torch.FloatTensor(np.array(y))) # y_data = Variable(torch.Tensor(np.array(y))) x_train = Variable(torch.FloatTensor(np.array(x[0:train_size]))) y_train = Variable(torch.FloatTensor(np.array(y[0:train_size]))) x_test = Variable(torch.FloatTensor(np.array(x[-test_size:]))) y_test = Variable(torch.FloatTensor(np.array(y[-test_size:]))) print('x_train.shape,y_train.shape,x_test.shape,y_test.shape:\n',x_train.shape,y_train.shape,x_test.shape,y_test.shape) return x_data, y_data, x_train, y_train, x_test, y_test def data_generator(x_train, y_train, x_test, y_test, n_iters, batch_size): ''' generate DataLoader ''' num_epochs = n_iters / (len(x_train) / batch_size) num_epochs = int(num_epochs) train_dataset = Data.TensorDataset(x_train, y_train) test_dataset = Data.TensorDataset(x_test, y_test) train_loader = torch.utils.data.DataLoader(dataset=train_dataset, batch_size=batch_size, shuffle=False) test_loader = torch.utils.data.DataLoader(dataset=test_dataset, batch_size=batch_size, shuffle=False) # val_loader = torch.utils.data.DataLoader() return train_loader, test_loader, num_epochs data,label = read_data(data_path) data, label, mm_y = normalization(data, label) x, y = sliding_windows(data, seq_length) x_data, y_data, x_train, y_train, x_test, y_test = data_split(x, y, split_ratio) # for index,date in enumerate(x_train): # # print(index,date) # for index, date in enumerate(y_train): # print(index, date) train_loader, test_loader, num_epochs = data_generator(x_train, y_train, x_test, y_test, n_iters, batch_size) # for i, (features,targets) in enumerate(train_loader): # print("Size of the features",features.shape) # print("Printing features:\n", features) # print("Size of the features", targets.shape) # print("Printing targets:\n", targets) # break # for i, (features,targets) in enumerate(y_train): # print("Size of the features",features.shape) # print("Printing features:\n", features) # print("Printing targets:\n", targets) # break class BP(nn.Module): def __init__(self, num_classes, input_size, hidden_size, num_layers): super(BP, self).__init__() self.num_classes = num_classes self.num_layers = num_layers self.input_size = input_size self.hidden_size = hidden_size self.seq_length = seq_length self.fc1 = nn.Linear(seq_length * input_size, 20) self.fc2 = nn.Linear(20, 20) self.fc3 = nn.Linear(20, num_classes) self.dropout = nn.Dropout(0.2) def forward(self, x): out = x.view(-1, self.seq_length * self.input_size) out = F.relu(self.fc1(out)) out = F.relu(self.fc2(out)) # F.relu() out = self.fc3(out) return out class GRU(nn.Module): def __init__(self, num_classes, input_size, hidden_size, num_layers): super(GRU, self).__init__() self.num_classes = num_classes self.num_layers = num_layers self.input_size = input_size self.hidden_size = hidden_size self.seq_length = seq_length self.gru = nn.GRU(input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True) self.fc1 = nn.Linear(hidden_size, 20) self.