自己造轮子——时间序列预测（股票数据）_基于LSTM

久闻LSTM的大名，觊觎其进行时间序列分析的能力，以本人非专业的能力理解起来相当之困难，先开篇学习一个项目，再尝试精进

关于LSTM

近年来在知网总是看到LSTM等神经网络方法用于时间序列分析的论文着实很眼馋，先来做一个简单的项目，后面尝试复现一些论文

LSTM（Long Short-Term Memory）是一种特殊类型的循环神经网络（RNN），它能够学习长期依赖信息，

LSTM的主要特点：

门控机制：LSTM引入了三个门控结构——输入门（Input Gate）、遗忘门（Forget Gate）和输出门（Output Gate），这使得LSTM能够有选择性地保留或遗忘信息。

记忆单元：LSTM的核心是记忆单元（Memory Cell），它负责存储长期信息。记忆单元的状态可以通过遗忘门和输入门的控制进行更新。

梯度消失和梯度爆炸的解决：由于门控机制的存在，LSTM能够有效地解决传统RNN在处理长序列数据时遇到的梯度消失和梯度爆炸问题。

参数共享：在处理序列数据时，LSTM的参数在时间步上是共享的，这减少了模型的参数数量，提高了模型的泛化能力。

LSTM的基本结构：

遗忘门：决定从上一个状态中丢弃哪些信息。它通过sigmoid激活函数输出一个0到1之间的值，0表示完全丢弃，1表示完全保留。

输入门：决定新的状态信息中包含哪些内容。输入门由两部分构成：一部分是sigmoid激活函数决定哪些信息可以进入，另一部分是tanh激活函数生成新的候选值。

记忆单元状态更新：记忆单元的状态通过遗忘门和输入门的信息进行更新。

输出门：决定输出哪些信息。输出门通过sigmoid激活函数决定哪些信息可以输出，并通过tanh激活函数对记忆单元的状态进行变换后输出。

（感谢Timi老师的解答，虽然学生看的是一知半解哈哈）

数据集

这里就体现出自己造轮子的方便了，直接用我之前学的爬取股票数据的脚本来建造数据集

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import baostock as bs
import pandas as pd

#### 登陆系统 ####
lg = bs.login()
# 显示登陆返回信息
print('login respond error_code:'+lg.error_code)
print('login respond  error_msg:'+lg.error_msg)

#### 等待用户输入参数 ####
user_code=input(['请输入股票代码：'])
user_startdate=input(['请输入开始日期YYYY-MM-DD'])
user_enddate=input(['请输入结束日期YYYY-MM-DD'])
user_frequency=input(['请输入 d=日k线、w=周、m=月、5=5分钟、15=15分钟、30=30分钟、60=60分钟k线数据'])
user_adjustflag=input(['请输入 复权类型 1后复权 2前复权 3不复权'])

#### 获取历史K线数据 ####
rs = bs.query_history_k_data_plus(user_code,
    "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST",
    start_date=user_startdate, end_date=user_enddate, 
    frequency=user_frequency, adjustflag=user_adjustflag) 
print('query_history_k_data_plus respond error_code:'+rs.error_code)
print('query_history_k_data_plus respond  error_msg:'+rs.error_msg)

#### 打印结果集 ####
data_list = []
while (rs.error_code == '0') & rs.next():
    # 获取一条记录，将记录合并在一起
    data_list.append(rs.get_row_data())
result = pd.DataFrame(data_list, columns=rs.fields)

#### 结果集输出到csv文件 ####
outlist=str('D:/'+user_code+'k线数据.csv')
result.to_csv(outlist, encoding="gbk", index=False)
print(result)

bs.logout()

库的安装和导入

LSTM神经网络项目需要使用Keras库，Keras库又需要在TensorFlow之上才能运行，所以要先安装TensorFlow。安装库直接pip即可，这么简单为何要在此多费笔墨呢，因为！背叛老黄的报应来了，TensorFlow的GPU加速需要cuda，本人只能CPU加速了。

这个库相当大，用清华镜像快多了

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple/ --upgrade tensorflow

1
2
3
4
5
6
7
8
9
10
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import math
import time
from keras.models import Sequential
from keras.layers import Dense, Activation, Dropout, LSTM
from keras import optimizers
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error

使用Sequential可以快速构建一个深度神经网络，Dense和LSTM层用于构建网络的架构，Dropout层用于正则化，optimizers用于定义如何更新网络权重，MinMaxScaler用于数据预处理，而mean_squared_error用于评估模型性能。

from keras.models import Sequential : 导入Keras的Sequential模型，这是一个线性堆叠的网络层结构，用于快速构建模型。

from keras.layers import Dense, Activation, Dropout, LSTM : 从Keras库中导入几种网络层：

Dense: 用于创建全连接的神经网络层。
Activation: 用于应用激活函数到网络层的输出。
Dropout: 用于在训练过程中随机丢弃一部分神经元，以防止过拟合。
LSTM: 用于创建长短期记忆网络层。

from keras import optimizers : 导入Keras的优化器模块，用于定义模型的训练算法。

from sklearn.preprocessing import MinMaxScaler : 导入sklearn库中的MinMaxScaler类，用于将数据缩放到指定的最小和最大值（默认是0到1）。

from sklearn.metrics import mean_squared_error : 导入sklearn库中的mean_squared_error函数，用于计算模型预测值和实际值之间的均方误差。

数据的导入和处理

1
2
dataframe = pd.read_csv('C:/Users/cyx94a/Desktop/Python/data/sz.001979k线数据.csv',header=0, parse_dates=[0],index_col=0, usecols=[0, 5])
dataset = dataframe.values

dataframe = pd.read_csv() : 使用Pandas的read_csv函数读取CSV文件。这里指定了以下参数：

header=0: 表示CSV文件的第一行包含列名。
parse_dates=[0]: 指示Pandas解析第一列（索引为0）作为日期。
index_col=0: 使用第一列作为DataFrame的索引。
usecols=[0, 5]: 只读取第一列和第六列。

dataset = dataframe.values: 将DataFrame中的数据转换为NumPy数组。

1
2
3
4
5
plt.figure(figsize=(20, 20))
dataframe.plot()
plt.ylabel('price')
plt.yticks(np.arange(0, 30, 0.01))
plt.show()

使用matplotlib库来创建一个图表，展示dataframe中的数据

1
2
scaler = MinMaxScaler(feature_range=(0, 1))
dataset = scaler.fit_transform(dataset.reshape(-1, 1))

使用sklearn.preprocessing模块中的MinMaxScaler类来对数据进行归一化处理。归一化是一种数据预处理技术，它将数据的数值范围调整到一个特定的区间，通常是[0, 1]。这样做有助于标准化不同特征的尺度，使得它们对模型训练的影响更加均匀。

scaler = MinMaxScaler(feature_range=(0, 1)) : 创建一个MinMaxScaler对象，feature_range=(0, 1)参数指定了归一化后的范围是0到1。

dataset = scaler.fit_transform(dataset.reshape(-1, 1)) : 使用fit_transform方法对数据进行归一化处理。首先，dataset被重塑为(-1, 1)的二维数组形式，其中-1表示自动计算行数以匹配原始数据的长度，1表示每行只有一个元素。然后，fit方法计算数据的最小值和最大值，接着transform方法使用这些最小值和最大值将数据缩放到[0, 1]范围内。

归一化处理后的dataset数组中的每个值都位于0到1之间，这有助于某些机器学习算法的性能，特别是那些对输入数据的尺度敏感的算法，如支持向量机（SVM）和一些梯度下降优化算法。

1
2
3
train_size = int(len(dataset)*0.8)
test_size = len(dataset)-train_size
train, test = dataset[0: train_size], dataset[train_size: len(dataset)]

分割训练集和测试集

1
2
3
4
5
6
7
def creat_dataset(dataset, look_back=1):
    dataX, dataY = [], []
    for i in range(len(dataset)-look_back-1):
        a = dataset[i: (i+look_back)]
        dataX.append(a)
        dataY.append(dataset[i+look_back])
    return np.array(dataX), np.array(dataY)

定义函数create_dataset用于从时间序列数据中创建一个数据集，其中每个数据点都是一个包含多个先前时间点的序列（即“look back”窗口）。注意：这种格式通常用于训练循环神经网络（RNN），如长短期记忆网络（LSTM）。

1
2
3
look_back = 1
trainX, trainY = creat_dataset(train, look_back)
testX, testY = creat_dataset(test, look_back)

形成训练集和测试集的x和y

构建神经网络

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
model = Sequential()

model.add(LSTM(units=50, return_sequences=True,input_shape=(look_back, 1)))

model.add(LSTM(units=100, return_sequences=True))

model.add(LSTM(units=200, return_sequences=True))

model.add(LSTM(units=300, return_sequences=False))
model.add(Dropout(0.2))

model.add(Dense(100))
model.add(Dense(units=1))  # 修改前一个Dense层的units为100

model.add(Activation('relu'))
start = time.time()
model.compile(loss='mean_squared_error', optimizer='Adam')
print(model.summary())

model = Sequential() : 创建一个Keras的Sequential模型，这是一个线性堆叠的网络层结构。

model.add(LSTM(units=50, return_sequences=True,input_shape=(look_back, 1))) :

向 Keras 模型中添加了一个 LSTM 层

units=50: 指 LSTM 层中的单元（或神经元）数量为 50。这些单元可以捕捉和记忆输入序列的不同特征。

return_sequences=True: 这个参数指出该 LSTM 层的输出也将是一个序列。这很重要，当在堆叠多个 LSTM 层时（即在一个 LSTM 层后面紧跟另一个 LSTM 层），第二个 LSTM 层期望其输入是一个序列。

input_shape=(look_back, 1): 这个参数定义了输入数据的形状。look_back 是在创建数据集时定义的时间步长，它指定了每个输入样本中有多少个时间点。在这里，look_back 表示每个输入样本将包含相同数量的时间步长，而 1 表示每个时间步长只有一个特征值（股票价格）。

input_shape 参数指定了输入数据的空间形状，这对于 Keras 来说是必须的，特别是当创建网络的第一个层时。对于 LSTM 层，input_shape 需要是三元组，其中：

第一个元素是时间步长（timesteps），即数据集中每个样本的长度。
第二个元素是特征数量（features），即每个时间步长中的数据点数。
第三个元素通常是用于三维输入的，但在大多数时间序列预测的情况下，只处理一维数据（例如，每个时间步长只有一个数据点），因此这个值是 1。

如果使用 create_dataset 函数创建了具有 look_back 个时间步长的数据集，那么应该在第一个 LSTM 层中指定 input_shape=(look_back, 1)。对于后续的 LSTM 层，不需要指定 input_shape Keras 将自动推断它。

继续添加更多的LSTM层，每一层的输出单元数递增，并且每个LSTM层的输出都是一个序列，以便可以作为下一个LSTM层的输入。

model.add(LSTM(300, return_sequences=False)) : 添加最后一个LSTM层，具有300个输出单元。return_sequences=False意味着这个LSTM层的输出将不是一个序列，它将为后续的全连接层提供输入。

model.add(Dropout(0.2)) : 在最后一个LSTM层之后添加Dropout层。

model.add(Dense(100)) : 添加一个具有100个单元的全连接层。

model.add(Dense(units=1)) : 添加最后一个全连接层，它将输出单个值，这在回归任务中是常见的。

model.add(Activation('relu')) : 添加一个ReLU激活函数，为模型的输出提供非线性。

start = time.time() : 记录模型编译开始的时间。

model.compile(loss='mean_squared_error', optimizer='Adam') : 编译模型，指定使用均方误差作为损失函数，以及Adam优化器。

model.summary() : 打印模型的摘要信息，包括每一层的输出形状和参数数量。

到此为止生成了一个943,001层的神经网络

训练LSTM

1
2
history = model.fit(trainX, trainY, batch_size=64, epochs=50, validation_split=0.1, verbose=2)
print('compilatiom time:', time.time()-start)

model.fit() : 训练 LSTM 模型。需要以下参数：

trainX: 训练数据集，应该是一个经过预处理和重塑的 NumPy 数组。
trainY: 与训练数据集相对应的目标值（标签）。
batch_size=64: 每个批次中的样本数量。较小的批量大小可以提供更好的泛化能力，但训练速度可能较慢。
epochs=50: 训练模型的轮数。每个 epoch 都会遍历整个数据集一次。
validation_split=0.1: 从训练数据中分离出的比例作为验证集，用于在训练过程中评估模型性能。
verbose=2: 日志的详细程度。1 表示输出进度条，2 表示每 epoch 结束后输出一行信息。

print('compilation time:', time.time()-start) : 这行代码计算并打印出模型编译所需的时间。这里 start 是一个之前定义的时间戳变量，用于记录模型编译开始的时间。

预测与评价

1
2
3
4
5
6
7
8
9
10
11
12
trainPredict = model.predict(trainX)
testPredict = model.predict(testX) # 预测

trainPredict = scaler.inverse_transform(trainPredict)
trainY = scaler.inverse_transform(trainY)
testPredict = scaler.inverse_transform(testPredict)
testY = scaler.inverse_transform(testY) # 反归一化

trainScore = math.sqrt(mean_squared_error(trainY, trainPredict[:, 0]))
print('Train Score %.2f RMSE' %(trainScore))
testScore = math.sqrt(mean_squared_error(testY, testPredict[:, 0]))
print('Test Score %.2f RMSE' %(testScore)) # 评价

使用model.predict() 方法对训练集 trainX 和测试集 testX 进行预测。

使用scaler.inverse_transform() 方法将预测值和实际值从归一化的范围反归一化回原始的范围。

使用mean_squared_error 函数计算均方误差（MSE），然后使用math.sqrt() 函数计算均方根误差（RMSE）。

可视化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
trainPredictPlot = np.empty_like(dataset)
trainPredictPlot[:] = np.nan
trainPredictPlot = np.reshape(trainPredictPlot, (dataset.shape[0], 1))
trainPredictPlot[look_back: len(trainPredict)+look_back, :] = trainPredict

testPredictPlot = np.empty_like(dataset)
testPredictPlot[:] = np.nan
testPredictPlot = np.reshape(testPredictPlot, (dataset.shape[0], 1))
testPredictPlot[len(trainPredict)+(look_back*2)+1: len(dataset)-1, :] = testPredict

fig1 = plt.figure(figsize=(12, 8))
plt.plot(history.history['loss'])
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epochs')
plt.show()

fig2 = plt.figure(figsize=(20, 15))
plt.plot(scaler.inverse_transform(dataset))
plt.plot(trainPredictPlot)
plt.plot(testPredictPlot)
plt.ylabel('price')
plt.xlabel('date')
plt.show()

蓝色是原始数据，黄色是训练数据训练完再进行预测的。绿色是测试数据。

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import baostock as bs
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import math
import time
from keras.models import Sequential
from keras.layers import Dense, Activation, Dropout, LSTM
from keras import optimizers
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error

#### 登陆系统 ####
lg = bs.login()
# 显示登陆返回信息
print('login respond error_code:'+lg.error_code)
print('login respond  error_msg:'+lg.error_msg)

#### 等待用户输入参数 ####
user_code=input(['请输入股票代码：'])
user_startdate=input(['请输入开始日期YYYY-MM-DD'])
user_enddate=input(['请输入结束日期YYYY-MM-DD'])
user_frequency=input(['请输入 d=日k线、w=周、m=月、5=5分钟、15=15分钟、30=30分钟、60=60分钟k线数据'])
user_adjustflag=input(['请输入 复权类型 1后复权 2前复权 3不复权'])

#### 获取历史K线数据 ####
rs = bs.query_history_k_data_plus(user_code,
    "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST",
    start_date=user_startdate, end_date=user_enddate, 
    frequency=user_frequency, adjustflag=user_adjustflag) 
print('query_history_k_data_plus respond error_code:'+rs.error_code)
print('query_history_k_data_plus respond  error_msg:'+rs.error_msg)

#### 打印结果集 ####
data_list = []
while (rs.error_code == '0') & rs.next():
    # 获取一条记录，将记录合并在一起
    data_list.append(rs.get_row_data())
result = pd.DataFrame(data_list, columns=rs.fields)

#### 结果集输出到csv文件 ####
outlist=str('C:/Users/cyx94a/Desktop/Python/data/'+user_code+'k线数据.csv')
result.to_csv(outlist, encoding="gbk", index=False)
print(result)

bs.logout()

dataframe = pd.read_csv('C:/Users/cyx94a/Desktop/Python/data/'+user_code+'k线数据.csv',header=0, parse_dates=[0],index_col=0)
dataset = dataframe.values[:, 5] # 将DataFrame中的数据转换为NumPy数组

plt.figure(figsize=(20, 20))
dataframe.plot()
plt.ylabel('price')
plt.yticks(np.arange(0, 30, 0.01))
plt.show()

scaler = MinMaxScaler(feature_range=(0, 1))
dataset = scaler.fit_transform(dataset.reshape(-1, 1)) # 归一化处理

train_size = int(len(dataset)*0.8)
test_size = len(dataset)-train_size
train, test = dataset[0: train_size], dataset[train_size: len(dataset)] # 分割训练集和测试集

def create_dataset(dataset, look_back=1):
    dataX, dataY = [], []
    for i in range(len(dataset) - look_back - 1):
        a = dataset[i:(i + look_back), np.newaxis]
        dataX.append(a)
        dataY.append(dataset[i + look_back])
    return np.array(dataX), np.array(dataY) # 训练循环神经网络的常用数据处理函数

look_back = 1
trainX, trainY = create_dataset(train, look_back)
testX, testY = create_dataset(test, look_back)

trainX = np.reshape(trainX, (trainX.shape[0], trainX.shape[1], 1)) 
testX = np.reshape(testX, (testX.shape[0], testX.shape[1], 1)) # 重塑数据以匹配LSTM

model = Sequential()

model.add(LSTM(units=50, return_sequences=True,input_shape=(look_back, 1)))

model.add(LSTM(units=100, return_sequences=True))

model.add(LSTM(units=200, return_sequences=True))

model.add(LSTM(units=300, return_sequences=False))
model.add(Dropout(0.2))

model.add(Dense(100))
model.add(Dense(units=1))  # 修改前一个Dense层的units为100

model.add(Activation('relu'))
start = time.time()
model.compile(loss='mean_squared_error', optimizer='Adam')
print(model.summary())

history = model.fit(trainX, trainY, batch_size=64, epochs=50, validation_split=0.1, verbose=2) # 训练LSTM
print('compilatiom time:', time.time()-start) # 计算并打印编译时间

trainPredict = model.predict(trainX)
testPredict = model.predict(testX) # 预测

trainPredict = scaler.inverse_transform(trainPredict)
trainY = scaler.inverse_transform(trainY)
testPredict = scaler.inverse_transform(testPredict)
testY = scaler.inverse_transform(testY) # 反归一化

trainScore = math.sqrt(mean_squared_error(trainY, trainPredict[:, 0]))
print('Train Score %.2f RMSE' %(trainScore))
testScore = math.sqrt(mean_squared_error(testY, testPredict[:, 0]))
print('Train Score %.2f RMSE' %(testScore)) # 评价

trainPredictPlot = np.empty_like(dataset)
trainPredictPlot[:] = np.nan
trainPredictPlot = np.reshape(trainPredictPlot, (dataset.shape[0], 1))
trainPredictPlot[look_back: len(trainPredict)+look_back, :] = trainPredict

testPredictPlot = np.empty_like(dataset)
testPredictPlot[:] = np.nan
testPredictPlot = np.reshape(testPredictPlot, (dataset.shape[0], 1))
testPredictPlot[len(trainPredict)+(look_back*2)+1: len(dataset)-1, :] = testPredict

fig1 = plt.figure(figsize=(12, 8))
plt.plot(history.history['loss'])
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epochs')
plt.show()

fig2 = plt.figure(figsize=(20, 15))
plt.plot(scaler.inverse_transform(dataset))
plt.plot(trainPredictPlot)
plt.plot(testPredictPlot)
plt.ylabel('price')
plt.xlabel('date')
plt.show()

模型性能挺好的，RMSE比较低，但是也领略到了LSTM的滞后性，整个预测图象就像是原图平移的结果。

真是一场酣畅淋漓的学习，模仿前人的代码，频频出bug，再苦苦debug，头脑晕晕，好在有一定收获，至少对LSTM有一个概括性的浅薄认识了，也大概了解了类似的时间序列分析项目的思路。

Python
#神经网络 #LSTM
自己造轮子——时间序列预测（股票数据）_基于LSTM
http://example.com/2024/04/23/自己造轮子——时间序列预测（股票数据）_基于LSTM/
作者
cyx94a
发布于
2024年4月23日
许可协议