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 = 0while 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 = [] label = []# 定義窗口函數：取樣時加窗的操作 def windows(data, size):start = 0while 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的編碼為0label[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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。