LSTM计算详解

写在前面

本文记录笔者在学习LSTM时的记录，相信读者已经在网上看过许多的LSTM博客与视频，与其他博客不同的是，本文会从数学公式的角度，剖析LSTM模型中各个部分的模型输入输出等维度信息，帮助初学者在公式层面理解LSTM模型，并且给出了相关计算的例子代入股票预测场景，并给出参考代码。

模型结构

LSTM的模型结构如下图所示。它由若干个重复的LSTM单元组成，每个单元内部包含遗忘门、输入门和输出门，以及当前时刻的单元状态和输出状态。

模型输入

LSTM模型，通常是处理一个序列（比如文本序列或时间序列） ，每个时间步的输入可以表示为，我们使用滑动窗口将序列分为若干个窗口大小为的窗口，步长为，当数据划分到最后，若不足为不能构成窗口时，缺少的数据使用pad填充，通常为0填充或使用最近数据填充。例如，假设我们有个时间步骤的输入，即，且假设窗口大小为，步长也为我们将数据分成三个窗口，即分为



由于的值不存在，我们将其值设为或者的值，即或者。

当步长为时，通常不会出现上面的情况，这也是我们使用的最多的一种滑动窗口划分方案。
例如，对于一个时序序列 ，窗口大小 ，滑动步长 ，滑动窗口划分结果为：

LSTM 单元的输入包含当前时刻的输入、上一时刻的输出状态以及上一时刻的单元状态。在进行运算第一层LSTM单元时，我们会手动初始化、，而在后面的LSTM的单元中和，都可以由上一次的LSTM单元获得。、、分别代表当前时刻的输入信息、上一时刻的输出信息以及上一时刻的记忆信息。其中，，是输入序列处理后的窗口大小（长度），上一时刻的输出状态，形状为，是LSTM单元的隐藏状态大小，是上一时刻的单元状态，形状为，与具有相同的形状。

我们通常会把和拼在一起形成更长的向量，我们通常竖着拼，即 ，如公式下所示，然后会传入各个门。当采用多批次时，。

遗忘门

遗忘门的输入为我们在模型输入中处理得到的。我们将与遗忘门中的权重矩阵相乘再加上置偏值，得到结果。然后对取Sigmoid，得到遗忘门的输出，其形状与单元状态相同，即 ，表示遗忘的程度。具体的计算公式如下所示。


其中，，，，。

在LSTM的许多门中，都使用Sigmoid函数，Sigmoid函数的绝大部分的值的取值范围为，这可以很有效的表示在Sigmoid函数的输入中哪些数据需要记忆，哪些数据需要遗忘的过程。当Sigmoid函数只越接近时表示遗忘，当接近时表示需要记忆。

输入门

输入门的输入为我们在模型输入中处理得到的，且。我们将与输入门中的权重矩阵相乘再加上置偏值，得到结果，然后对取Sigmoid，得到输入门的输出，表示输入的重要程度。具体的计算公式如下所示。


其中，，，，。

输出门

输出门的输入为我们在模型输入中处理得到的，且。我们将与输出门中的权重矩阵相乘再加上置偏值，得到结果，然后对取Sigmoid，得到输出门的输出，具体的计算公式如下所示。

其中，，，，。
​

当前输入单元状态

在计算之前，我们需要引入当前输入单元状态，并计算的值。是当前输入的单元状态，表示当前输入要保留多少内容到记忆中。我们将与当前时刻状态单元的权重矩阵相乘再加上置偏值，得到结果，然后对取tanh，得到的输出。的计算如公式下所示。


其中，，，，。

当前输入单元状态中，使用了tanh函数，tanh函数的取值范围为，当函数的值接近时代表着当前输入信息要被修正，当但函数值接近时，代码当前输入信息要被加强。

当前时刻单元状态

接下来我们进行当前时刻单元状态的计算。我们使用遗忘门和输入门得到的结果 、和上一时刻单元状态来计算当前时刻单元状态。我们分别将、按元素相乘，和按元素相乘，然后再将两者相加得到我们的当前时刻单元状态。具体计算如公式下所示。

其中，时遗忘门输出，是输入门输出，是当前输入状态单元， 是上一时刻状态单元，表示 __按元素乘__。

模型输出

模型的输出是和当前时刻的单元状态，而由当前时刻的单元状态​和输出门的输出确定。我们将当前时刻的单元状态取 tanh得到，然后将 与 按元素相乘得到最后的，计算公式如下所示。通常，会进一步传递给模型的上层或者作为最终的预测结果。


其中 为当前层隐藏状态，为输出门的输出，为当前时刻状态单元。

日期	开盘价	收盘价	最高价	最低价
4月23日	3038.6118	3021.9775	3044.9438	3016.5168
4月24日	3029.4028	3044.8223	3045.6399	3019.1238
4月25日	3037.9272	3052.8999	3060.2634	3034.6499
4月26日	3054.9793	3088.6357	3092.4300	3054.9793

Table: SH000001

简单的LSTM例子

接下来我们根据上面的模型结构中的计算方法来简单计算一个LSTM的例子。

我们以取中国A股上证指数（SH000001）2024年4月23日-25日共3个交易日的数据为例，取开盘价、收盘价、最高价、最低价作为特征，具体数据如表格所示。使用LSTM模型计算预测2024年4月26日的开盘价、收盘价、最高价、最低价，损失函数使用MSE。我们取隐藏层状态的大小为，然后进行计算，预测下一天的数据。

我们把表格数据处理成的形式，也就是把每天的个特征，转换成即的向量，然后我们得到以的结果。

由于隐藏层大小为 ，所以 、的维度都是 ，我们将 和进行初始化为向量，即

随后我们初始化 、、、（维度为，即 以及、、、，的元素值 ，W是随机矩阵，如下所示。

全部初始化为单位列向量即

然后我们将与拼在一起作为 ，即

我们依次计算遗忘门，输入门，输出门，即

随后我们进行计算当前输入单元状态，即

接着我们计算当前时刻单元状态，即

最后我们计算当前层隐藏层输出 ，即

这样我们就完成了一次LSTM单元的正向传播计算，我们得到了 和，我们将其传入下一层。

同理我们可以进行接下来 第个交易日 的计算。
我们将与拼在一起作为 ，即

我们依次计算遗忘门，输入门，输出门，即

随后我们进行计算当前输入单元状态，即

接着我们计算当前时刻单元状态，即

最后我们计算当前层隐藏层输出 ，即

同理我们可以进行接下来 第个交易日 的计算。
我们将与拼在一起作为 ，即

我们依次计算遗忘门，输入门，输出门。

随后我们进行计算当前输入单元状态，即

接着我们计算当前时刻单元状态，即

最后我们计算当前层隐藏层输出 ，即

得到了 之后，我们可以简单将的结果作为预测的结果，然后使用MSE进行计算损失，MSE的计算公式如下所示。

然后我们就得到我们的损失为 。

以上就完成了一次将LSTM用于预测的计算。可以看到误差很大，实际应用中会先将数据输入到LSTM前，会进行一次归一化，在LSTM的输出后，会将隐藏层的结果进行一层线性映射，然后使用逆归一化，这样得到结果会比较接近我们的指数。

小结

LSTM模型的具体训练步骤如下：

1.LSTM 单元的输入包含当前时刻的输入、上一时刻的输出状态以及上一时刻的单元状态。在进行运算第一层LSTM单元时，我们会手动初始化、，而在后面的LSTM的单元中和，都可以由上一次的LSTM单元获得。其中，，是输入特征的维度，上一时刻的输出状态，形状为，是LSTM单元的隐藏状态大小，是上一时刻的单元状态，形状为。

我们通常会把和拼在一起形成更长的向量，我们通常竖着拼，即 ，然后会传入各个门。

2.随后是计算各个门的输出，各个门的输入是。我们将与门中的权重矩阵相乘再加上置偏值，得到中间结果。然后对取Sigmoid，得到门的输出，其形状与单元状态相同，即 。

其中，，，，。

​

3.计算当前输入单元状态的值，表示当前输入要保留多少内容到记忆中。我们将与当前时刻状态单元的权重矩阵相乘再加上置偏值，得到中间结果，然后对取tanh，得到输出。

其中，，，，。

4.接下来我们进行当前时刻单元状态的计算。我们使用遗忘门和输入门得到的结果 、和上一时刻单元状态来计算当前时刻单元状态。我们分别将、按元素相乘，和按元素相乘，然后再将两者相加得到我们的但钱时刻单元状态。

其中，时遗忘门输出，是输入门输出，是当前输入状态单元， 是上一时刻状态单元，表示 __按元素乘__。

5.最后模型的输出是和当前时刻的单元状态，而由当前时刻的单元状态​和输出门的输出确定。我们将当前时刻的单元状态取 tanh得到，然后将 与 按元素相乘得到最后的。

其中 为当前层隐藏状态，为输出门的输出，为当前时刻状态单元。

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler


# 读取数据
df = pd.read_csv('sh_data.csv')
df = df.iloc[-30:, [2, 5, 3, 4]]
df1 = df[25:28].reset_index(drop=True)
df2 = df1.reset_index(drop=True)		

data = df[['open', 'close', 'high', 'low']].values.astype(float)

# 标准化数据
scaler = MinMaxScaler(feature_range=(0, 1))
data = scaler.fit_transform(data)

# 创建时间序列数据
def create_sequences(data, time_step=1):
	X, y = [], []
	for i in range(len(data) - time_step):
		X.append(data[i:(i + time_step)])
		y.append(data[i + time_step])
		return np.array(X), np.array(y)

time_step = 2  # 时间步长设置为2天
X, y = create_sequences(data, time_step)

# 转换为PyTorch张量
X = torch.FloatTensor(X)
y = torch.FloatTensor(y)

class LSTM(nn.Module):
	def __init__(self, input_size, hidden_layer_size, output_size):
		super(LSTM, self).__init__()
		self.hidden_layer_size = hidden_layer_size
		self.lstm = nn.LSTM(input_size, hidden_layer_size)
		self.linear = nn.Linear(hidden_layer_size, output_size)
		self.hidden_cell = (torch.zeros(1, 1, self.hidden_layer_size),
		torch.zeros(1, 1, self.hidden_layer_size))

	def forward(self, input_seq):
		lstm_out, self.hidden_cell = self.lstm(input_seq.view(len(input_seq), 1, -1), self.hidden_cell)
		predictions = self.linear(lstm_out.view(len(input_seq), -1))
		return predictions[-1]


input_size = 4  # 输入特征数量
hidden_layer_size = 4
output_size = 4  # 输出特征数量

model = LSTM(input_size=input_size, hidden_layer_size=hidden_layer_size, output_size=output_size)
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# epochs = 1
# for i in range(epochs):
#     for seq, labels in zip(X, y):
#         optimizer.zero_grad()
#         model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
#                              torch.zeros(1, 1, model.hidden_layer_size))
#         y_pred = model(seq)

#         single_loss = loss_function(y_pred, labels)
#         single_loss.backward()
#         optimizer.step()

#     if i % 10 == 0:
#         print(f'epoch: {i:3} loss: {single_loss.item():10.8f}')

# 只进行一次训练
seq, labels = X[0], y[0]
optimizer.zero_grad()
model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
torch.zeros(1, 1, model.hidden_layer_size))
y_pred = model(seq)
single_loss = loss_function(y_pred, labels)
single_loss.backward()
optimizer.step()

print(f'Single training loss: {single_loss.item():10.8f}')

model.eval()

# 预测下一天的四个特征
with torch.no_grad():
	seq = torch.FloatTensor(data[-time_step:])
	model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
	torch.zeros(1, 1, model.hidden_layer_size))
	next_day = model(seq).numpy()

# 将预测结果逆归一化
next_day = scaler.inverse_transform(next_day.reshape(-1, output_size))

print(f'Predicted features for the next day: open={next_day[0][0]}, close={next_day[0][1]}, high={next_day[0][2]}, low={next_day[0][3]}')


# 获取训练集的预测值
train_predict = []
for seq in X:
	with torch.no_grad():
	model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
	torch.zeros(1, 1, model.hidden_layer_size))
	train_predict.append(model(seq).numpy())

# 将预测结果逆归一化
train_predict = scaler.inverse_transform(np.array(train_predict).reshape(-1, output_size))
actual = scaler.inverse_transform(data)

# 绘制图形
plt.figure(figsize=(10, 6))

for i, col in enumerate(['open', 'close', 'high', 'low']):
	plt.subplot(2, 2, i+1)
	plt.plot(actual[:, i], label=f'Actual {col}')
	plt.plot(range(time_step, time_step + len(train_predict)), train_predict[:, i], label=f'Train Predict {col}')
	plt.legend()

plt.tight_layout()
plt.show()