基于LSTM的时间序列预测算法Python实现

import torch import pandas as pd import numpy as np import torch.nn as nn from sklearn.preprocessing import MinMaxScaler from handle import handleData from LSTM import LSTM import matplotlib.pyplot as plt from pic_settings import pic_settings from sklearn.metrics import mean_squared_error plt.rcParams["font.sans-serif"] = ["SimHei"] # 设置字体 plt.rcParams["axes.unicode_minus"] = False # 该语句解决图像中的“-”负号的乱码问题 # 读取数据 dataset = pd.read_excel('data/陕西省近20年货运量.xls') data_copy = dataset.copy() train_data = data_copy.iloc[:-2, -1:] test_data = data_copy.iloc[-2:, -1:] # 可视化数据 year = dataset.loc[:,'年份'] rail_trans = dataset.loc[:,'铁路货运量(万吨)'] plt.plot(year,rail_trans,label='铁路货运量(万吨)',color='orange') plt.legend() plt.title('年份与铁路货运量(万吨)') plt.xticks(list(range(2000,2021)),rotation=45) plt.xlabel(pic_settings()['analyze_data']['xlabel']) plt.ylabel(pic_settings()['analyze_data']['ylabel']) plt.savefig('images/analyze_data/ori_data.jpg') plt.clf() # 数据预处理 scaler = MinMaxScaler(feature_range=(-1, 1)) # 将数据缩放到(-1，1) ，防止梯度爆炸 X = scaler.fit_transform(data_copy.iloc[:, -1:]) data = np.array(X)[:-2, -1:] testData = np.array(X)[-2:, -1:] year = dataset.loc[:,'年份'] plt.plot(year,X,label='铁路货运量(万吨)',color='orange') plt.legend() plt.title('年份与铁路货运量预处理后') plt.xticks(list(range(2000,2021)),rotation=45) plt.xlabel(pic_settings()['analyze_data']['xlabel']) plt.ylabel(pic_settings()['analyze_data']['ylabel1']) plt.savefig('images/analyze_data/pre_data.jpg') plt.clf() # trainData = data[:-2,-1:] # print(trainData.shape) def train(model, loss_fn, optimizer, num_epochs, x_train, y_train): """ 训练lstm模型 :param model: lstm模型 :param loss_fn: 损失函数 :param optimizer: 优化器 :param num_epochs: 训练次数 :return: 训练结果，损失值列表 """ import time hist = np.zeros(num_epochs) start_time = time.time() lstm = [] the_best_model = None the_min_loss = 1 for t in range(num_epochs): y_train_pred = model(x_train) # print(y_train_pred) loss = loss_fn(y_train_pred, y_train) # print("Epoch ", t, "MSE: ", loss.item()) hist[t] = loss.item() if the_min_loss > hist[t]: the_min_loss = hist[t] the_best_model = model optimizer.zero_grad() loss.backward() optimizer.step() return (y_train_pred, hist, the_best_model, the_min_loss) def predict(time_span, train_data, model, test_data, loss_fn): # goal:201 loss_list = [] predict_list = [] x0 = np.array(train_data[-(time_span - 1):, -1:]).reshape(1, time_span - 1, 1) x = torch.from_numpy(x0).type(torch.Tensor) ori_pred = model(x) # 2021 loss1 = loss_fn(ori_pred, torch.from_numpy(test_data[-2:-1, -1:]).type(torch.Tensor)).item() new = pd.DataFrame(scaler.inverse_transform(ori_pred.detach().numpy())) predict_list.append(new.iloc[0, 0]) for i in range(1): # print(f'正在预测第{i}次') x1 = x0[0, 1:] x2 = ori_pred.detach().numpy() x0 = np.concatenate((x1, x2), axis=0).reshape(1, time_span - 1, 1) x = torch.from_numpy(x0).type(torch.Tensor) ori_pred = model(x) loss2 = loss_fn(ori_pred, torch.from_numpy(test_data[-2:-1, -1:]).type(torch.Tensor)).item() new = pd.DataFrame(scaler.inverse_transform(ori_pred.detach().numpy()).astype("int64")) predict_list.append(new.iloc[0, 0]) # print(f'预测值为：{new.iloc[0, 0]}') test_loss = (loss1 + loss2) / 2 return predict_list, test_loss def plot_train_pred(y_train_pred, time_span, image_path='images/y_train_pred_time_span/'): # 绘制训练结果图 year = dataset.iloc[time_span - 1:-2, 0] # year first_year = 1999 + time_span end_year = 2018 y = dataset.iloc[time_span - 1:-2, 1] # y y_train_pred = np.array(pd.DataFrame(scaler.inverse_transform(y_train_pred.detach().numpy()))) # 训练预测值 plt.plot(year, y, label="y", color='r') label = f"y_train_pred-time_span={time_span}" title = f'{first_year}-{end_year} LSTM 模型预测货运量与真实货运量对比图' # print(y_train_pred) plt.plot(year, y_train_pred.reshape(-1), linestyle='--', label=f"y_train_pred-time_span={time_span}", color='g') plt.legend() plt.grid() plt.xlabel('年份') plt.ylabel('铁路货运量（万吨）') plt.xticks(year, rotation=45) plt.title(title) image_name = image_path + f"{label}.jpg" plt.savefig(image_name) # plt.show() plt.clf() def plot_epoch_loss(mLL, time_span_list, image_path='images\epoch_loss/', epoch=100): x = list(range(1, epoch + 1)) title = f'不同time_span epochs={epoch} 损失变化趋势图 ' for i in range(len(time_span_list)): time_span=time_span_list[i] mse_list = mLL[i] label = f'time_span={time_span} epochs={epoch}' plt.plot(x, mse_list, label=label) plt.legend() plt.grid(True) plt.xticks(list(range(0, epoch + 1, 10)), rotation=45) plt.xlabel('迭代次数') plt.ylabel('MSE损失') plt.title(title) # plt.show() image_name = image_path + f"不同time_span epochs={epoch}.jpg" plt.savefig(image_name) plt.clf() def plot_time_span_mse(min_mse_list, time_span_list, image_path='images/time_span_loss/'): x = time_span_list label = f'Minimum-Mse' title = f'不同time_span下的Minimum-MSE变化图' plt.plot(x, min_mse_list, label=label) plt.legend() plt.grid(True) # plt.xticks(list(range(0, epoch + 1, 10)), rotation=45) plt.title(title) # plt.show() plt.xlabel('time_span') plt.ylabel('最小MSE损失') image_name = image_path + f"{label}.jpg" plt.savefig(image_name) plt.clf() def plot_time_span_test_mse(test_mse_list, time_span_list, image_path='images/test_loss/'): x = time_span_list label = f'Test-Mse' title = f'不同time_span下的Test-Mse变化图' plt.plot(x, test_mse_list, label=label) plt.legend() plt.grid(True) # plt.xticks(list(range(0, epoch + 1, 10)), rotation=45) plt.title(title) plt.xlabel('time_span') plt.ylabel('测试MSE损失') # plt.show() image_name = image_path + f"{label}.jpg" plt.savefig(image_name) plt.clf() def plot_hyper_time_span_test_mse(hyper, hyper_list, tLL, time_span_list, image_path='images/different_hyper-parameter_test_loss/'): x = time_span_list title = f'不同{hyper}对应time_span下的Test-Mse变化图' for i in range(len(hyper_list)): label = f'{hyper}={hyper_list[i]}-Test-Mse' test_mse_list = tLL[i] plt.plot(x, test_mse_list, label=label) plt.legend() plt.grid(True) # plt.xticks(list(range(0, epoch + 1, 10)), rotation=45) plt.title(title) plt.xlabel('time_span') plt.ylabel('测试MSE损失') # plt.show() image_name = image_path + f"{title}.jpg" plt.savefig(image_name) def main(hidden_num=32, lr=0.01, num_layers=2, num_epochs=100): input_dim = 1 hidden_num = hidden_num output_dim = 1 num_layers = num_layers num_epochs = num_epochs time_span_list = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11] model_list = [] info_list = [] y_train_pred_list = [] min_mse_list = [] mLL = [] for time_span in time_span_list: x_train, y_train = handleData(data, time_span=time_span) model = LSTM(input_dim=input_dim, hidden_dim=hidden_num, output_dim=output_dim, num_layers=num_layers) loss_