【图像识别|卷积神经网络实战之LeNet5股票预测代码实现及遇到各种问题的解决方案】
LeNet5股票预测：预测股票走势是上涨还是下跌
-
-
- 1.LeNet5股票预测
- 2.数据预处理
-
- (1)归一化
- (2)滑动平均
- (3)加窗取股票样本
- (4)分割数据集
- (5)OneHot编码
- 3.设计LeNet5网络结构
-
- (1)神经网络结构设计
- (2)模型训练
- (3)模型评价
- (4)可能遇到问题
- 4.完整代码
- 5.数据集下载链接
-
1.LeNet5股票预测
实现股票长期预测：
数据集包括：开盘价，最高价，最低价，收盘价，交易量
基本思想：
从时间序列角度用过去的数据预测未来的走势，用每只股票过去的开盘价，收盘价和最高价进行预测。
文章图片
2.数据预处理
(1)归一化 因为数据的量纲不一样，所以要对训练数据进行归一化。这里采用最大/最小值归一是对原始数据的线性变换，使结果映射到范围[0,1]内。代码中调用sklearn机器学习库的最小最大值归一化
# 数据归一化消除量纲
df['open'] = minmax_scale(df['open'])#开盘价
df['high'] = minmax_scale(df['high'])#最高价
df['low'] = minmax_scale(df['low'])#最低价
(2)滑动平均 股票的扰动很多，所以要做一个滑动平均处理，即将两天或三天的平均值作为一个新的样本点。若是实时的股价数据，我们一般选择一个小窗口，采样若干个点求其平均值，这就类似于时间序列算法中的滑动平均，我们不是直接采用原始数据，而是取相邻两个数的平均值作为训练数据集以此来弱化噪声影响从而使模型更加稳定.
(3)加窗取股票样本 使用了加窗采样的技术，每一个窗口代表一个样本，统计窗口内的涨跌次数作为此样本的标签，窗内涨多跌少，标记为1，反之标记为0，因此将股票走势问题转化为分类问题。具体方法如下：
在某个小窗口采样300次，比较相邻两次采样点并观察走势，如80%的采样点中后边都比前边大，我们就认为这个窗口是上升的。
文章图片
代码实现：
采用了open、high和low这3个属性作为3个通道，这里窗口的大小设置为90，即每90条数据进行一次加窗。
# 定义窗口函数：取样时加窗的操作
def windows(data, size):
start = 0
while start < data.count():
yield int(start), int(start + size)
start += (size / 2)# 返回格式数据
def segment_signal(data, window_size=90):
segments = np.empty((0, window_size, 3))
labels = np.empty((0))
for (start, end) in windows(data["timestamp"], window_size):
x = data["open"][start:end]#x代表开盘价
y = data["high"][start:end]#y代表开盘价
z = data["low"][start:end]#z代表开盘价
#将窗口格式进行整理，即将窗口的数据和标签分别存起来
if (len(df["timestamp"][start:end]) == window_size):
segments = np.vstack([segments, np.dstack([x, y, z])])
labels = np.append(labels, stats.mode(data["label"][start:end])[0][0])
return segments, labels(4)分割数据集
#调用train_test_split库函数，将数据集按照8:2的比例划分为训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(data, label, test_size=0.2)
X_train = np.array(X_train).reshape(len(X_train), 90, 3)#将训练样本集的条件入整理成一维向量形式
X_test = np.array(X_test).reshape(len(X_test), 90, 3)#将测试样本集的条件整理成一维向量形式
y_train = np.array(y_train).reshape(-1, 1)#训练集的标签整理成向量形式
y_test = np.array(y_test).reshape(-1, 1)#将测试集的标签整理成向量形式
(5)OneHot编码 引入sklearn中的OneHotEncoder()函数进行独热编码
使训练过程中不受分类值表示的问题对模型产生的负面影响。
# 采用独热编码（One-Hot）
enc = OneHotEncoder()
enc.fit(y_train)
y_train = enc.transform(y_train).toarray()
y_test = enc.transform(y_test).toarray()
3.设计LeNet5网络结构
(1)神经网络结构设计
- 一维卷积
二维卷积图像编码是一个矩形，如果是彩色图像它的编码是三个矩形也就是一个张量，而这里股票的走势我们用的是一维的卷积，卷积核肯定是一个更小向量，比如用一个1×10的一维向量对300个股票的样本进行卷积。卷积操作就是将样本前10个与卷积核向量相乘相加，然后移动一个采样点再对新的10个位置相乘相加，步长为1没有Padding的话，根据输入输出与步长很容易求出输出向量的尺寸。
- 使用开盘价（open）、最高价（high）和最低价（low）作为输入数据，对股票趋势进行建模预测分析，因此将open、high和low作为CNN的3个通道。
- 构建3层的CNN进行训练，首先定义迭代次数、输入通道、隐藏层神经元数目等结构参数。
#通道数量为3个
in_channels = 3#训练迭代次数
epoch = 10000
#定义批大小
batch_size = 5#即每一批的样本个数为5
batch = X_train.shape[0] / batch_size#批数# 创建占位符：存取读入的样本，X是输入，Y是标签
X = tf.placeholder(tf.float32, shape=(None, 90, in_channels))
Y = tf.placeholder(tf.float32, shape=(None, 2))#LeNet5网络结构：卷积池化卷积池化卷积池化
# 第一层
h1 = tf.layers.conv1d(X, 256, 4, 2, 'SAME', name='h1', use_bias=True, activation=tf.nn.relu)#conv1d一维卷积，参数：256是第一个卷积层核函数的数量，卷积核大小为4，步长为2；'SAME'表示Padding，激活函数用的relu
p1 = tf.layers.max_pooling1d(h1, 2, 2, padding='VALID')#池化核为2，步长为2，无Padding
print(h1)
print(p1)# 第二层
h2 = tf.layers.conv1d(p1, 256, 4, 2, 'SAME', use_bias=True, activation=tf.nn.relu)
p2 = tf.layers.max_pooling1d(h2, 2, 2, padding='VALID')
print(h2)
print(p2)# 第三层
h3 = tf.layers.conv1d(p1, 2, 4, 2, 'SAME', use_bias=True, activation=tf.nn.relu)
p3 = tf.layers.max_pooling1d(h3, 11, 1, padding='VALID')
res = tf.reshape(p3, shape=(-1, 2))print(h3)
print(p3)
print(res)(2)模型训练
#定义损失函数
loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(logits=res, labels=Y))# 定义正确率评价指标
ac = tf.cast(tf.equal(tf.argmax(res, 1), tf.argmax(Y, 1)), tf.float32)
acc = tf.reduce_mean(ac)# 创建优化器：学习步长的优化
optim = tf.train.AdamOptimizer(0.0001).minimize(loss)#tf.Session执行训练
f=open('result/result.txt','w')
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for i in range(epoch):
sess.run(optim, feed_dict={X: X_train, Y: y_train})
if i % 100 == 0:
los, accuracy = sess.run([loss, acc], feed_dict={X: X_train, Y: y_train})
print(los, accuracy)
#应用测试集测试
ccc,bbb = sess.run([tf.argmax(res, 1),tf.argmax(Y,1)], feed_dict={X: X_test, Y: y_test})
#输出测试结果
for i in range(0,len(ccc)):
f.write(str(ccc[i])+" "+str(bbb[i])+"\n")
f.close()
运行结果：
结果中会输出定义的网络的结构以及每训练100次的损失和准确度，可以看到准确度在一直上升。
文章图片
测试样本的输出会报错在result.txt文件中，输出1表示上涨，输出0表示下跌
文章图片
(3)模型评价
#读取训练后输出的结果txt文件
f=open('result/result.txt','r')
pre=[]
t=[]
for row in f.readlines():
row=row.strip() #去掉每行头尾空白
row=row.split(" ")
pre.append((row[0]))
t.append((row[1]))
#混淆矩阵绘制
def plot_confusion_matrix(cm, labels_name, title):
cm = cm.astype(np.float64)
if(cm.sum(axis=0)[0]!=0):
cm[:,0] = cm[:,0] / cm.sum(axis=0)[0]# 归一化
if(cm.sum(axis=0)[1]!=0):
cm[:,1] = cm[:,1] / cm.sum(axis=0)[1]# 归一化plt.imshow(cm, interpolation='nearest')# 在特定的窗口上显示图像plt.title(title)# 图像标题
plt.colorbar()
num_local = np.array(range(len(labels_name)))
plt.xticks(num_local, labels_name)# 将标签印在x轴坐标上
plt.yticks(num_local, labels_name)# 将标签印在y轴坐标上
plt.ylabel('True label')
plt.xlabel('Predicted label')
cm=confusion_matrix(t,pre)
y_true = np.array(list(map(int,t)))
y_scores = np.array(list(map(int,pre)))roc=str(roc_auc_score(y_true, y_scores))
precision, recall, _thresholds = precision_recall_curve(y_true, y_scores)
pr =str(auc(recall, precision))
title="ROC AUC:"+roc+"\n"+"PR AUC:"+pr
labels_name=["0.0","1.0"]
plot_confusion_matrix(cm, labels_name, title)
for x in range(len(cm)):
for y in range(len(cm[0])):
plt.text(y,x,cm[x][y],color='white',fontsize=10, va='center')
plt.show()
(4)可能遇到问题 1)OSError: [Errno 22] Invalid argument: ‘.\dataset\tt.csv’
解决方案：
df = pd.read_csv(r".\dataset\tt.csv")
2)AttributeError: module ‘tensorflow’ has no attribute ‘placeholder’
解决方案：
import tensorflow.compat.v1 as tf
tf.disable_v2_behavior()
3)No mappable was found to use for colorbar creation.
唯一有点遗憾的是暂未解决这个问题，尝试了将近一个小时，找了各种方法，下午继续，再接再厉，加油小赵！
解决方案：待定
注：一定要仔细检查python中各种模块的代入，多数问题都是由于未导入相关模块引起的
4.完整代码
import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
import tensorflow.compat.v1 as tf
tf.disable_v2_behavior()
import pandas as pd
from sklearn.preprocessing import minmax_scale
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import OneHotEncoder
import osfrom sklearn.metrics import confusion_matrix
from sklearn.metrics import roc_auc_score
from sklearn.metrics import precision_recall_curve,aucos.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'# 读取数据集
df = pd.read_csv(r".\dataset\tt.csv")
# 数据归一化消除量纲：调用sklearn机器学习库的最小最大值归一化
df['open'] = minmax_scale(df['open'])#开盘价
df['high'] = minmax_scale(df['high'])#最高价
df['low'] = minmax_scale(df['low'])#最低价# 构建向量矩阵
# 总数据矩阵/标签
data = https://www.it610.com/article/[]
label = []# 定义窗口函数：取样时加窗的操作
def windows(data, size):
start = 0
while start < data.count():
yield int(start), int(start + size)
start += (size / 2)# 返回格式数据
def segment_signal(data, window_size=90):
segments = np.empty((0, window_size, 3))
labels = np.empty((0))
for (start, end) in windows(data["timestamp"], window_size):
x = data["open"][start:end]#x代表开盘价
y = data["high"][start:end]#y代表开盘价
z = data["low"][start:end]#z代表开盘价
#将窗口格式进行整理，即将窗口的数据和标签分别存起来
if (len(df["timestamp"][start:end]) == window_size):
segments = np.vstack([segments, np.dstack([x, y, z])])
labels = np.append(labels, stats.mode(data["label"][start:end])[0][0])
return segments, labels#调用segment_signal函数，将窗口返回的数据作为网络的输入
data, label = segment_signal(df)#label表示窗口内股票的走势，上升编码为1下降编码为0，data和label分别作为网络的输入和输出
# 对标签数据进行处理
for i in range(0, len(label)):#将label进行归一化，使用OneHot编码
if label[i] == -1:#将所有原始数据中label为-1的编码为0
label[i] = 0#调用train_test_split库函数，将数据集按照8:2的比例划分为训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(data, label, test_size=0.2)
X_train = np.array(X_train).reshape(len(X_train), 90, 3)#将训练样本集的条件入整理成一维向量形式
X_test = np.array(X_test).reshape(len(X_test), 90, 3)#将测试样本集的条件整理成一维向量形式
y_train = np.array(y_train).reshape(-1, 1)#训练集的标签整理成向量形式
y_test = np.array(y_test).reshape(-1, 1)#将测试集的标签整理成向量形式# 涨跌标签采用独热编码（One-Hot）
enc = OneHotEncoder()
enc.fit(y_train)
y_train = enc.transform(y_train).toarray()
y_test = enc.transform(y_test).toarray()#通道数量为3个
in_channels = 3#训练迭代次数
epoch = 10000
#定义批大小
batch_size = 5#即每一批的样本个数为5
batch = X_train.shape[0] / batch_size#批数# 创建占位符：存取读入的样本，X是输入，Y是标签
X = tf.placeholder(tf.float32, shape=(None, 90, in_channels))
Y = tf.placeholder(tf.float32, shape=(None, 2))#LeNet5网络结构：卷积池化卷积池化卷积池化
# 第一层
h1 = tf.layers.conv1d(X, 256, 4, 2, 'SAME', name='h1', use_bias=True, activation=tf.nn.relu)#conv1d一维卷积，参数：256是第一个卷积层核函数的数量，卷积核大小为4，步长为2；'SAME'表示Padding，激活函数用的relu
p1 = tf.layers.max_pooling1d(h1, 2, 2, padding='VALID')#池化核为2，步长为2，无Padding
print(h1)
print(p1)# 第二层
h2 = tf.layers.conv1d(p1, 256, 4, 2, 'SAME', use_bias=True, activation=tf.nn.relu)
p2 = tf.layers.max_pooling1d(h2, 2, 2, padding='VALID')
print(h2)
print(p2)# 第三层
h3 = tf.layers.conv1d(p1, 2, 4, 2, 'SAME', use_bias=True, activation=tf.nn.relu)
p3 = tf.layers.max_pooling1d(h3, 11, 1, padding='VALID')
res = tf.reshape(p3, shape=(-1, 2))print(h3)
print(p3)
print(res)#定义损失函数
loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(logits=res, labels=Y))# 定义正确率评价指标
ac = tf.cast(tf.equal(tf.argmax(res, 1), tf.argmax(Y, 1)), tf.float32)
acc = tf.reduce_mean(ac)# 创建优化器：学习步长的优化
optim = tf.train.AdamOptimizer(0.0001).minimize(loss)#tf.Session执行训练
f=open('result/result.txt','w')
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for i in range(epoch):
sess.run(optim, feed_dict={X: X_train, Y: y_train})
if i % 100 == 0:
los, accuracy = sess.run([loss, acc], feed_dict={X: X_train, Y: y_train})
print(los, accuracy)
#应用测试集测试
ccc,bbb = sess.run([tf.argmax(res, 1),tf.argmax(Y,1)], feed_dict={X: X_test, Y: y_test})
#输出测试结果
for i in range(0,len(ccc)):
f.write(str(ccc[i])+" "+str(bbb[i])+"\n")
f.close()#读取训练后输出的结果txt文件
f=open('result/result.txt','r')
pre=[]
t=[]
for row in f.readlines():
row=row.strip() #去掉每行头尾空白
row=row.split(" ")
pre.append((row[0]))
t.append((row[1]))
#混淆矩阵绘制
def plot_confusion_matrix(cm, labels_name, title):
cm = cm.astype(np.float64)
if(cm.sum(axis=0)[0]!=0):
cm[:,0] = cm[:,0] / cm.sum(axis=0)[0]# 归一化
if(cm.sum(axis=0)[1]!=0):
cm[:,1] = cm[:,1] / cm.sum(axis=0)[1]# 归一化plt.imshow(cm, interpolation='nearest')# 在特定的窗口上显示图像plt.title(title)# 图像标题
plt.colorbar()
plt.imshow()
plt.show()
num_local = np.array(range(len(labels_name)))
plt.xticks(num_local, labels_name)# 将标签印在x轴坐标上
plt.yticks(num_local, labels_name)# 将标签印在y轴坐标上
plt.ylabel('True label')
plt.xlabel('Predicted label')
cm=confusion_matrix(t,pre)
y_true = np.array(list(map(int,t)))
y_scores = np.array(list(map(int,pre)))roc=str(roc_auc_score(y_true, y_scores))
precision, recall, _thresholds = precision_recall_curve(y_true, y_scores)
pr =str(auc(recall, precision))
title="ROC AUC:"+roc+"\n"+"PR AUC:"+pr
labels_name=["0.0","1.0"]
plot_confusion_matrix(cm, labels_name, title)
for x in range(len(cm)):
for y in range(len(cm[0])):
plt.text(y,x,cm[x][y],color='white',fontsize=10, va='center')
plt.show()5.数据集下载链接
https://download.csdn.net/download/fencecat/85104287
推荐阅读
- Numpy数据类型对象详细指南
- Python中的OrderedDict介绍和用法指南
- Python MongoDB – create_index查询用法详细示例
- Python focus_set()和focus_get()方法用法介绍
- Python OpenCV – cv2.polylines()方法使用示例
- python做一个微型美颜图片处理器，十行代码即可完成
- Python Kivy的相对布局用法介绍和代码示例
- python中的numpy.sum()用法详细介绍
- python3 time.sleep()函数解释 – Python3教程