LSTM时间序列预测的原理以及代码实现_lstm代码

前言

时间序列是指将同一统计指标的数值按其发生的时间先后顺序排列而成的数列。时间序列分析的主要目的是根据已有的历史数据对未来进行预测。比如根据过去的股票数据预测未来股票趋势，结合过去天气数据和发电数据预测未来光伏发电量等等。时间序列预测的算法有很多，本文主要以LSTM为例，详细讲述时间序列预测的原理和代码详解。

LSTM基本原理

RNN介绍

循环神经网络（Recurrent Neural Network，RNN）是一类用于处理序列数据的神经网络，在自然语言处理、时间序列预测、语音识别等领域中得到了广泛应用。与传统的前馈神经网络不同，RNN具有“记忆”能力。RNN通过在网络中引入循环连接，使得网络可以利用先前的输入信息来影响当前的输出。这使得RNN非常适合处理具有时间依赖性或顺序性的任务。

下面是RNN的简单结构图：

• 输入层：接收输入数据，并将其传递给隐藏层。输入不仅仅是静态的，还包含着序列中的历史信息。

• 隐藏层：捕捉时序依赖性。隐藏层的输出不仅取决于当前的输入，还取决于前一时刻的隐藏状态。

• 输出层：根据隐藏层的输出生成最终的预测结果。

虽然RNN有一定的记忆能力，但是RNN还存在着梯度消失和爆炸以及对长期信息不敏感的问题，即长期依赖问题。RNN在处理长序列数据时，很难记住很久之前的输入信息。比如下图所示，虽然一开始的What（深绿色部分）输入通过隐藏层传递到了最后，但是在最后What（深绿色部分）的占比非常小。如果有更长的输入序列，那么一开始的信息传递到最后时，占比将非常小，就产生了遗忘。我们可以把神经元想象成有一定空间限制的大脑，每次隐藏层的传递就是一次学习。那么这个过程就可以理解为，学了后面的知识，忘记前面的知识。因为RNN没有巩固前面知识的操作，对于长时间序列，很容易产生遗忘。

而LSTM 就是为了解决这个问题而提出的。

LSTM介绍

LSTM引入了门还加入了细胞状态的新概念，以更有效地捕捉和处理序列数据中的长期依赖关系。LSTM是RNN的一种变体，更高级的RNN，它们的本质还是一样的，整体的结构还是分为三层：输入层、隐藏层、输出层。LSTM只是在隐藏层的处理上，和RNN有所区别，RNN什么信息它都存下来，因为它没有挑选的能力，而LSTM不一样，它会选择性的存储信息，因为它有门控装置，可以尽情的选择。LSTM在隐藏层传递的东西有两个：隐藏状态和细胞状态。LSTM的大概结构如下图所示：

LSTM引入了三个门控单元：输入门、遗忘门和输出门。这些门控单元可以控制信息的流动和遗忘，从而有效地解决了梯度消失的问题。所谓“门”结构就是用来去除或者增加信息到细胞状态的能力。这里的细胞状态是核心，它属于隐藏层，类似于传送带，在整个链上运行。LSTM的细节介绍如下（当前细胞称为时间步t）：

• 遗忘门：决定哪些信息需要被遗忘。输出是一个0到1之间的值，表示应该遗忘多少过去的信息。当接近1时，过去的信息会被完全保留；当接近0时，过去的信息会被完全遗忘。

  

• 输入门：决定哪些信息应该被保留并更新细胞状态。输出是一个0到1之间的值，表示哪些新的输入应该被保留。当接近1时，所有新的输入都会被完全保留；当接近0时，所有新的输入都会被完全忽略。接下来会计算候选细胞状态，它表示当前时间步的新输入可以对细胞状态产生多少影响。输入门的作用是控制新的输入在当前时间步t的权重。

• 输出门：控制哪些信息应该被输出。输出是一个0到1之间的值，表示哪些信息应该被输出。当接近1时，所有的信息都会被完全保留；当接近0时，所有的信息都会被完全屏蔽。LSTM会将细胞状态通过一个tanh函数进行处理，得到当前时间步的隐藏状态

  

• 细胞状态：细胞状态可以被看作是整个LSTM网络的核心，它可以存储和传递信息，同时也能够控制信息的流动和更新。LSTM的细胞状态会被更新和传递到下一个时间步t+1。

  

LSTM解决时间序列预测问题

代码地址：cdy-123/time-series: Time series forecasting (github.com)

单变量时间序列预测

单变量时间序列预测指的是，除了时间属性数据，只有单一属性的一组数据，即只通过历史数据，预测未来数据。

数据集准备：数据集格式为.csv格式，存放在data文件夹里。数据集一共两列，一列时间，一列需要预测的数据。

首先需要根据自己的数据集更改配置文件

{
	"data": {
		"filename": "sp500.csv",
		"columns": [
			"Close"
		],
		"sequence_length": 50,
		"train_test_split": 0.85,
		"normalise": true
	},
	"training": {
		"epochs": 1,
		"batch_size": 32
	},
	"model": {
		"loss": "mse",
		"optimizer": "adam",
		"save_dir": "saved_models",
		"layers": [
			{
				"type": "lstm",
				"neurons": 100,
				"input_timesteps": 49,
				"input_dim": 1,
				"return_seq": true
			},
			{
				"type": "dropout",
				"rate": 0.2
			},
			{
				"type": "lstm",
				"neurons": 100,
				"return_seq": true
			},
			{
				"type": "lstm",
				"neurons": 100,
				"return_seq": false
			},
			{
				"type": "dropout",
				"rate": 0.2
			},
			{
				"type": "dense",
				"neurons": 1,
				"activation": "linear"
			}
		]
	}
}

• data.filename是数据集的文件名
• data.columns是需要预测变量的列名
• data.sequence_length是单位预测序列的长度
• data.train_test_split是训练集和测试集的比例
• data.normalise是否对数据进行标准化，一般都选择true
• model里面是模型的参数，其中model.layers.input_timesteps的值等于data.sequence_length-1

运行train.py代码即可进行预测。主要代码如下：

def main():
    #读取所需参数
    configs = json.load(open('config_1.json', 'r')) #读取配置文件
    if not os.path.exists(configs['model']['save_dir']): os.makedirs(configs['model']['save_dir'])
    #读取数据
    data = DataLoader(
        os.path.join('data', configs['data']['filename']),
        configs['data']['train_test_split'],
        configs['data']['columns']
    )
    #标准化数据
    scaler_train = MinMaxScaler(feature_range=(-1, 1))
    scaler_test = MinMaxScaler(feature_range=(-1, 1))
    data.data_train = scaler_train.fit_transform(data.data_train.reshape(-1, 1))
    data.data_test = scaler_test.fit_transform(data.data_test.reshape(-1, 1))
 
    #创建LSTM模型
    model = Model()
    mymodel = model.build_model(configs)
    
    #准备训练数据
    x, y = data.get_train_data(
        seq_len=configs['data']['sequence_length'],
        normalise=configs['data']['normalise']
    )
    print (x.shape)
    print (y.shape)
    
	# 训练模型
    model.train(
		x,
		y,
		epochs = configs['training']['epochs'],
		batch_size = configs['training']['batch_size'],
		save_dir = configs['model']['save_dir']
	)
	
   #准备测试数据
    x_test, y_test = data.get_test_data(
        seq_len=configs['data']['sequence_length'],
        normalise=configs['data']['normalise']
    )
    y_test = scaler_test.inverse_transform(y_test)#数据标准化
    
    #展示测试效果
    predictions_multiseq = model.predict_sequences_multiple(x_test, configs['data']['sequence_length'], configs['data']['pred_length'])
    predictions_multiseq = scaler_test.inverse_transform(predictions_multiseq)
    plot_results_multiple(predictions_multiseq, y_test, configs['data']['sequence_length'])

多变量时间序列预测

平常我们构建的大多属于单变量时间序列，在构建预测模型的时候往往只有时间和变量本身可以参考，而这样的变量往往会显得很单薄。多变量时间序列是指除了时间依赖外，还有其他因素共同影响某一结果的发生。比如预测某地的温度，云层覆盖率，露点、湿度、风速及风向都有可能被认为是影响温度的因素。这时候就可以建立多变量时间序列模型来预测。

LSTM预测

首先使用LSTM模型来解决多变量时间序列预测问题。

数据集准备：分为有预测值的训练数据和没有预测值的预测数据，比如下面的数据集：

有2001年1月25日到2021年9月29日的数据，我们需要预测的是open这一列，其他4列都是影响因素。训练数据需要包括时间在内的这6列，并且要把时间放第一列，需要预测的open列放第二列。

如果要预测2021年9月30日到2021年11月10日的open值，我们需要时间和4列影响因素作为数据集，构成预测数据（比训练数据少一列）。将预测数据输入模型，模型会预测对应的open列。

主要代码如下：

import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM
from tensorflow.keras.layers import Dense, Dropout
from scikeras.wrappers import KerasRegressor
from sklearn.model_selection import GridSearchCV
import pandas as pd
from matplotlib import pyplot as plt
from sklearn.preprocessing import MinMaxScaler
import joblib
 
## 读取数据集
df=pd.read_csv("train.csv",parse_dates=["Date"],index_col=[0])
print(df.shape)
print(df.head())
print(df.tail())
test_split=round(len(df)*0.20) # 测试集和训练集划分
df_for_training=df[:-test_split]
df_for_testing=df[-test_split:]
print(df_for_training.shape)
print(df_for_testing.shape)
 
## 标准化数据
scaler_train = MinMaxScaler(feature_range=(-1, 1))
df_for_training_scaled = scaler_train.fit_transform(df_for_training)
df_for_testing_scaled=scaler_train.transform(df_for_testing)
print(df_for_training_scaled)
print(df_for_testing_scaled)
 
def createXY(dataset,n_past):
    dataX = []
    dataY = []
    for i in range(n_past, len(dataset)):
        dataX.append(dataset[i - n_past:i, 0:dataset.shape[1]])
        dataY.append(dataset[i,0])
    return np.array(dataX),np.array(dataY)   
 
## 准备模型输入数据
trainX,trainY=createXY(df_for_training_scaled,30) ## 单位时间序列长度30，可根据需求更改
testX,testY=createXY(df_for_testing_scaled,30) ## 单位时间序列长度30，可根据需求更改
print("trainX Shape: ",trainX.shape)
print("trainY Shape: ",trainY.shape)
print("testX Shape: ",testX.shape)
print("testY Shape: ",testY.shape)
 
## 构建模型
def build_model():
    grid_model = Sequential()
    grid_model.add(LSTM(50,return_sequences=True,input_shape=(30,5))) ## input_shape根据单位时间序列长度更改
    grid_model.add(LSTM(50))
    grid_model.add(Dropout(0.2))
    grid_model.add(Dense(1))
 
    grid_model.compile(loss = 'mse')
    return grid_model
 
grid_model = KerasRegressor(build_fn=build_model,verbose=1)
parameters = {'batch_size' : [8,16],
              'epochs' : [8,10],
              'optimizer' : ['adam','Adadelta'] }
# 使用 GridSearchCV 训练过程中会自动调优超参数，训练时间可能会比较长
grid_search  = GridSearchCV(estimator = grid_model,
                            param_grid = parameters,
                            cv = 2)
 
## 训练
grid_search = grid_search.fit(trainX,trainY,validation_data=(testX,testY)) 
print(grid_search.best_params_)
my_model=grid_search.best_estimator_
joblib.dump(my_model, 'Model.h5')
print(my_model)
print('Model Saved!')
 
## 预测
my_model = joblib.load('Model.h5') # 加载模型
prediction=my_model.predict(testX)
print("prediction\n", prediction)
print("\nPrediction Shape-",prediction.shape)
prediction_copies_array = np.repeat(prediction,5, axis=-1)
pred=scaler_train.inverse_transform(np.reshape(prediction_copies_array,(len(prediction),5)))[:,0]
original_copies_array = np.repeat(testY,5, axis=-1)
original=scaler_train.inverse_transform(np.reshape(original_copies_array,(len(testY),5)))[:,0]
print("Pred Values-- " ,pred)
print("\nOriginal Values-- ",original)
 
## 绘制预测结果
plt.plot(original, color = 'red', label = 'Real  Stock Price')
plt.plot(pred, color = 'blue', label = 'Predicted  Stock Price')
plt.title(' Stock Price Prediction')
plt.xlabel('Time')
plt.ylabel(' Stock Price')
plt.legend()
plt.show()
 
## 预测未来
df_30_days_past=df.iloc[-30:,:] # 过去30天的数据 
df_30_days_future=pd.read_csv("test.csv",parse_dates=["Date"],index_col=[0]) # 未来30天的其他数据
print(df_30_days_future.shape)
 
df_30_days_future["Open"]=0
df_30_days_future=df_30_days_future[["Open","High","Low","Close","Adj Close"]]
old_scaled_array=scaler_train.transform(df_30_days_past)
new_scaled_array=scaler_train.transform(df_30_days_future)
new_scaled_df=pd.DataFrame(new_scaled_array)
new_scaled_df.iloc[:,0]=np.nan
full_df=pd.concat([pd.DataFrame(old_scaled_array),new_scaled_df]).reset_index().drop(["index"],axis=1)
print(full_df.shape)
print(full_df)
full_df_scaled_array=full_df.values
all_data=[]
time_step=30
for i in range(time_step,len(full_df_scaled_array)):
    data_x=[]
    data_x.append(full_df_scaled_array[i-time_step:i,0:full_df_scaled_array.shape[1]])
    data_x=np.array(data_x)
    prediction=my_model.predict(data_x)
    all_data.append(prediction)
    full_df.iloc[i,0]=prediction
 
new_array=np.array(all_data)
new_array=new_array.reshape(-1,1)
prediction_copies_array = np.repeat(new_array,5, axis=-1)
y_pred_future_30_days = scaler_train.inverse_transform(np.reshape(prediction_copies_array,(len(new_array),5)))[:,0]
print(y_pred_future_30_days)

训练结果如下：

线性回归

多变量时间序列预测也可以利用回归模型来解决。线性回归是一个比较普遍的方法。线性回归算法是一种基于统计学的方法，能够捕捉时间序列的线性关系，从而进行准确的预测。

数据集准备：使用的数据集和LSTM预测的相同，有训练数据和预测数据。这里用到的是另外一个预测年度发电量的数据集。要预测的是发电量power列，也就是y，另外4列是影响因素，也就是x。也可以根据自己的数据集更改代码。

主要代码如下：

import pandas as pd
import csv
import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error,mean_squared_error,r2_score
 
# 读取数据
data = pd.read_csv("data-train.csv")#训练数据
print(data.head())
print("shape:",data.shape)
 
# 使用 pandas 构建 X（特征向量）和 y（标签列）
feature_cols = ["people","urban","gdp","industry"] #影响因素列名
X = data[feature_cols]
y = data["power"] #预测列名
 
# 构建线性回归模型并训练
model = LinearRegression().fit(X,y)
 
# 输出模型结果
print("截距：",model.intercept_)
coef = zip(feature_cols, model.coef_)
print("回归系数：",list(coef))
y_pred=model.predict(X)
print("平均绝对误差（MAE）：",mean_absolute_error(y,y_pred))
print("均方误差（MSE）：",mean_squared_error(y,y_pred))
print("R方值：",r2_score(y,y_pred))
 
# 展示拟合结果（根据自己数据集修改）
x=np.arange(72)
plt.plot(range(len(y_pred)),y_pred,"b",label="predict")
plt.plot(range(len(y)),y,"r",label="true")
plt.xticks(x[::10],range(1952,2023,10))
plt.legend(loc='upper left')
plt.xlabel("year")
plt.ylabel("power")
plt.show()
 
 
# 预测
data = pd.read_csv("data-test.csv") #预测数据
feature_cols = ["people","urban","gdp","industry"] #影响因素列名
X_test = data[feature_cols]
y_pred = model.predict(X_test)
print(y_pred)
 
# 保存预测的结果（根据自己的预测数据集更改下面代码）
with open('fa-pre.csv','w',newline='') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['year','power'])
    for i in range(0,27):
        writer.writerow([i+2024,y_pred[i]])
 
# 绘制预测曲线
x=np.arange(27)
plt.plot(range(len(y_pred)),y_pred,"b",label="predict")
plt.xticks(x[::5],range(2024,2050,5)) #根据自己的数据集更改
plt.legend(loc='upper left')
plt.xlabel("year")
plt.ylabel("power")
plt.show()

拟合曲线和预测曲线如下图所示：