LeNet5股票預(yù)測(cè)：預(yù)測(cè)股票走勢(shì)是上漲還是下跌

• • • 1.LeNet5股票預(yù)測(cè)
    • 2.數(shù)據(jù)預(yù)處理
    • • (1)歸一化
      • (2)滑動(dòng)平均
      • (3)加窗取股票樣本
      • (4)分割數(shù)據(jù)集
      • (5)OneHot編碼
    • 3.設(shè)計(jì)LeNet5網(wǎng)絡(luò)結(jié)構(gòu)
    • • (1)神經(jīng)網(wǎng)絡(luò)結(jié)構(gòu)設(shè)計(jì)
      • (2)模型訓(xùn)練
      • (3)模型評(píng)價(jià)
      • (4)可能遇到問題
    • 4.完整代碼
    • 5.數(shù)據(jù)集下載鏈接

1.LeNet5股票預(yù)測(cè)

實(shí)現(xiàn)股票長期預(yù)測(cè)：
數(shù)據(jù)集包括：開盤價(jià)，最高價(jià)，最低價(jià)，收盤價(jià)，交易量
基本思想：
從時(shí)間序列角度用過去的數(shù)據(jù)預(yù)測(cè)未來的走勢(shì)，用每只股票過去的開盤價(jià)，收盤價(jià)和最高價(jià)進(jìn)行預(yù)測(cè)。

2.數(shù)據(jù)預(yù)處理

(1)歸一化

因?yàn)閿?shù)據(jù)的量綱不一樣，所以要對(duì)訓(xùn)練數(shù)據(jù)進(jìn)行歸一化。這里采用最大/最小值歸一是對(duì)原始數(shù)據(jù)的線性變換，使結(jié)果映射到范圍[0,1]內(nèi)。代碼中調(diào)用sklearn機(jī)器學(xué)習(xí)庫的最小最大值歸一化

# 數(shù)據(jù)歸一化消除量綱 df['open'] = minmax_scale(df['open'])#開盤價(jià) df['high'] = minmax_scale(df['high'])#最高價(jià) df['low'] = minmax_scale(df['low'])#最低價(jià)

(2)滑動(dòng)平均

股票的擾動(dòng)很多，所以要做一個(gè)滑動(dòng)平均處理，即將兩天或三天的平均值作為一個(gè)新的樣本點(diǎn)。若是實(shí)時(shí)的股價(jià)數(shù)據(jù)，我們一般選擇一個(gè)小窗口，采樣若干個(gè)點(diǎn)求其平均值，這就類似于時(shí)間序列算法中的滑動(dòng)平均，我們不是直接采用原始數(shù)據(jù)，而是取相鄰兩個(gè)數(shù)的平均值作為訓(xùn)練數(shù)據(jù)集以此來弱化噪聲影響從而使模型更加穩(wěn)定.

(3)加窗取股票樣本

使用了加窗采樣的技術(shù)，每一個(gè)窗口代表一個(gè)樣本，統(tǒng)計(jì)窗口內(nèi)的漲跌次數(shù)作為此樣本的標(biāo)簽，窗內(nèi)漲多跌少，標(biāo)記為1，反之標(biāo)記為0，因此將股票走勢(shì)問題轉(zhuǎn)化為分類問題。具體方法如下：
在某個(gè)小窗口采樣300次，比較相鄰兩次采樣點(diǎn)并觀察走勢(shì)，如80%的采樣點(diǎn)中后邊都比前邊大，我們就認(rèn)為這個(gè)窗口是上升的。

代碼實(shí)現(xiàn)：
采用了open、high和low這3個(gè)屬性作為3個(gè)通道，這里窗口的大小設(shè)置為90，即每90條數(shù)據(jù)進(jìn)行一次加窗。

# 定義窗口函數(shù)：取樣時(shí)加窗的操作 def windows(data, size):start = 0while start < data.count():yield int(start), int(start + size)start += (size / 2)# 返回格式數(shù)據(jù) 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代表開盤價(jià)y = data["high"][start:end]#y代表開盤價(jià)z = data["low"][start:end]#z代表開盤價(jià)#將窗口格式進(jìn)行整理，即將窗口的數(shù)據(jù)和標(biāo)簽分別存起來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)分割數(shù)據(jù)集

#調(diào)用train_test_split庫函數(shù)，將數(shù)據(jù)集按照8:2的比例劃分為訓(xùn)練集和測(cè)試集 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ùn)練樣本集的條件入整理成一維向量形式 X_test = np.array(X_test).reshape(len(X_test), 90, 3)#將測(cè)試樣本集的條件整理成一維向量形式 y_train = np.array(y_train).reshape(-1, 1)#訓(xùn)練集的標(biāo)簽整理成向量形式 y_test = np.array(y_test).reshape(-1, 1)#將測(cè)試集的標(biāo)簽整理成向量形式

(5)OneHot編碼

引入sklearn中的OneHotEncoder()函數(shù)進(jìn)行獨(dú)熱編碼
使訓(xùn)練過程中不受分類值表示的問題對(duì)模型產(chǎn)生的負(fù)面影響。

# 采用獨(dú)熱編碼（One-Hot） enc = OneHotEncoder() enc.fit(y_train) y_train = enc.transform(y_train).toarray() y_test = enc.transform(y_test).toarray()

3.設(shè)計(jì)LeNet5網(wǎng)絡(luò)結(jié)構(gòu)

(1)神經(jīng)網(wǎng)絡(luò)結(jié)構(gòu)設(shè)計(jì)

• 一維卷積
  二維卷積圖像編碼是一個(gè)矩形，如果是彩色圖像它的編碼是三個(gè)矩形也就是一個(gè)張量，而這里股票的走勢(shì)我們用的是一維的卷積，卷積核肯定是一個(gè)更小向量，比如用一個(gè)1×10的一維向量對(duì)300個(gè)股票的樣本進(jìn)行卷積。卷積操作就是將樣本前10個(gè)與卷積核向量相乘相加，然后移動(dòng)一個(gè)采樣點(diǎn)再對(duì)新的10個(gè)位置相乘相加，步長為1沒有Padding的話，根據(jù)輸入輸出與步長很容易求出輸出向量的尺寸。

• 使用開盤價(jià)（open）、最高價(jià)（high）和最低價(jià)（low）作為輸入數(shù)據(jù)，對(duì)股票趨勢(shì)進(jìn)行建模預(yù)測(cè)分析，因此將open、high和low作為CNN的3個(gè)通道。

• 構(gòu)建3層的CNN進(jìn)行訓(xùn)練，首先定義迭代次數(shù)、輸入通道、隱藏層神經(jīng)元數(shù)目等結(jié)構(gòu)參數(shù)。

#通道數(shù)量為3個(gè) in_channels = 3#訓(xùn)練迭代次數(shù) epoch = 10000 #定義批大小 batch_size = 5#即每一批的樣本個(gè)數(shù)為5 batch = X_train.shape[0] / batch_size#批數(shù)# 創(chuàng)建占位符：存取讀入的樣本，X是輸入，Y是標(biāo)簽 X = tf.placeholder(tf.float32, shape=(None, 90, in_channels)) Y = tf.placeholder(tf.float32, shape=(None, 2))#LeNet5網(wǎng)絡(luò)結(jié)構(gòu)：卷積池化卷積池化卷積池化 # 第一層 h1 = tf.layers.conv1d(X, 256, 4, 2, 'SAME', name='h1', use_bias=True, activation=tf.nn.relu)#conv1d一維卷積，參數(shù)：256是第一個(gè)卷積層核函數(shù)的數(shù)量，卷積核大小為4，步長為2；'SAME'表示Padding，激活函數(shù)用的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)模型訓(xùn)練

#定義損失函數(shù) loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(logits=res, labels=Y))# 定義正確率評(píng)價(jià)指標(biāo) ac = tf.cast(tf.equal(tf.argmax(res, 1), tf.argmax(Y, 1)), tf.float32) acc = tf.reduce_mean(ac)# 創(chuàng)建優(yōu)化器：學(xué)習(xí)步長的優(yōu)化 optim = tf.train.AdamOptimizer(0.0001).minimize(loss)#tf.Session執(zhí)行訓(xùn)練 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)#應(yīng)用測(cè)試集測(cè)試ccc,bbb = sess.run([tf.argmax(res, 1),tf.argmax(Y,1)], feed_dict={X: X_test, Y: y_test})#輸出測(cè)試結(jié)果for i in range(0,len(ccc)):f.write(str(ccc[i])+" "+str(bbb[i])+"\n")f.close()

運(yùn)行結(jié)果：
結(jié)果中會(huì)輸出定義的網(wǎng)絡(luò)的結(jié)構(gòu)以及每訓(xùn)練100次的損失和準(zhǔn)確度，可以看到準(zhǔn)確度在一直上升。

測(cè)試樣本的輸出會(huì)報(bào)錯(cuò)在result.txt文件中，輸出1表示上漲📈，輸出0表示下跌📉

(3)模型評(píng)價(jià)

#讀取訓(xùn)練后輸出的結(jié)果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])) #混淆矩陣?yán)L制 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) # 圖像標(biāo)題plt.colorbar()num_local = np.array(range(len(labels_name)))plt.xticks(num_local, labels_name) # 將標(biāo)簽印在x軸坐標(biāo)上plt.yticks(num_local, labels_name) # 將標(biāo)簽印在y軸坐標(biāo)上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.
唯一有點(diǎn)遺憾的是暫未解決這個(gè)問題，嘗試了將近一個(gè)小時(shí)，找了各種方法，下午繼續(xù)，再接再厲，加油小趙！
解決方案：待定
注：一定要仔細(xì)檢查python中各種模塊的代入，多數(shù)問題都是由于未導(dǎo)入相關(guān)模塊引起的

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'# 讀取數(shù)據(jù)集 df = pd.read_csv(r".\dataset\tt.csv") # 數(shù)據(jù)歸一化消除量綱：調(diào)用sklearn機(jī)器學(xué)習(xí)庫的最小最大值歸一化 df['open'] = minmax_scale(df['open'])#開盤價(jià) df['high'] = minmax_scale(df['high'])#最高價(jià) df['low'] = minmax_scale(df['low'])#最低價(jià)# 構(gòu)建向量矩陣 # 總數(shù)據(jù)矩陣/標(biāo)簽 data = [] label = []# 定義窗口函數(shù)：取樣時(shí)加窗的操作 def windows(data, size):start = 0while start < data.count():yield int(start), int(start + size)start += (size / 2)# 返回格式數(shù)據(jù) 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代表開盤價(jià)y = data["high"][start:end]#y代表開盤價(jià)z = data["low"][start:end]#z代表開盤價(jià)#將窗口格式進(jìn)行整理，即將窗口的數(shù)據(jù)和標(biāo)簽分別存起來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#調(diào)用segment_signal函數(shù)，將窗口返回的數(shù)據(jù)作為網(wǎng)絡(luò)的輸入 data, label = segment_signal(df)#label表示窗口內(nèi)股票的走勢(shì)，上升編碼為1下降編碼為0，data和label分別作為網(wǎng)絡(luò)的輸入和輸出 # 對(duì)標(biāo)簽數(shù)據(jù)進(jìn)行處理 for i in range(0, len(label)):#將label進(jìn)行歸一化，使用OneHot編碼if label[i] == -1:#將所有原始數(shù)據(jù)中l(wèi)abel為-1的編碼為0label[i] = 0#調(diào)用train_test_split庫函數(shù)，將數(shù)據(jù)集按照8:2的比例劃分為訓(xùn)練集和測(cè)試集 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ùn)練樣本集的條件入整理成一維向量形式 X_test = np.array(X_test).reshape(len(X_test), 90, 3)#將測(cè)試樣本集的條件整理成一維向量形式 y_train = np.array(y_train).reshape(-1, 1)#訓(xùn)練集的標(biāo)簽整理成向量形式 y_test = np.array(y_test).reshape(-1, 1)#將測(cè)試集的標(biāo)簽整理成向量形式# 漲跌標(biāo)簽采用獨(dú)熱編碼（One-Hot） enc = OneHotEncoder() enc.fit(y_train) y_train = enc.transform(y_train).toarray() y_test = enc.transform(y_test).toarray()#通道數(shù)量為3個(gè) in_channels = 3#訓(xùn)練迭代次數(shù) epoch = 10000 #定義批大小 batch_size = 5#即每一批的樣本個(gè)數(shù)為5 batch = X_train.shape[0] / batch_size#批數(shù)# 創(chuàng)建占位符：存取讀入的樣本，X是輸入，Y是標(biāo)簽 X = tf.placeholder(tf.float32, shape=(None, 90, in_channels)) Y = tf.placeholder(tf.float32, shape=(None, 2))#LeNet5網(wǎng)絡(luò)結(jié)構(gòu)：卷積池化卷積池化卷積池化 # 第一層 h1 = tf.layers.conv1d(X, 256, 4, 2, 'SAME', name='h1', use_bias=True, activation=tf.nn.relu)#conv1d一維卷積，參數(shù)：256是第一個(gè)卷積層核函數(shù)的數(shù)量，卷積核大小為4，步長為2；'SAME'表示Padding，激活函數(shù)用的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)#定義損失函數(shù) loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(logits=res, labels=Y))# 定義正確率評(píng)價(jià)指標(biāo) ac = tf.cast(tf.equal(tf.argmax(res, 1), tf.argmax(Y, 1)), tf.float32) acc = tf.reduce_mean(ac)# 創(chuàng)建優(yōu)化器：學(xué)習(xí)步長的優(yōu)化 optim = tf.train.AdamOptimizer(0.0001).minimize(loss)#tf.Session執(zhí)行訓(xùn)練 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)#應(yīng)用測(cè)試集測(cè)試ccc,bbb = sess.run([tf.argmax(res, 1),tf.argmax(Y,1)], feed_dict={X: X_test, Y: y_test})#輸出測(cè)試結(jié)果for i in range(0,len(ccc)):f.write(str(ccc[i])+" "+str(bbb[i])+"\n")f.close()#讀取訓(xùn)練后輸出的結(jié)果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])) #混淆矩陣?yán)L制 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) # 圖像標(biāo)題plt.colorbar()plt.imshow()plt.show()num_local = np.array(range(len(labels_name)))plt.xticks(num_local, labels_name) # 將標(biāo)簽印在x軸坐標(biāo)上plt.yticks(num_local, labels_name) # 將標(biāo)簽印在y軸坐標(biāo)上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.數(shù)據(jù)集下載鏈接

https://download.csdn.net/download/fencecat/85104287

總結(jié)

以上是生活随笔為你收集整理的卷积神经网络实战之LeNet5股票预测代码实现及遇到各种问题的解决方案的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。