基於LSTM的比特幣價格預測模型（系列1）

摘要：plt.plot(df.index[train_size+2:], yhat_inv, '-', color=red, label=label+ \。plt.plot(np.arange(train_size, train_size+test_size), sdf[label].iloc[train_size:], \。

標星★ 置頂 公衆號 愛你們 ♥

作者：Pawel 編譯：1+1=6

讓預測不斷逼近！

前言

設計並訓練由輸入/訓練數據（比特幣價格時間序列/60min）驅動的LSTM，預測一小時內的比特幣價格，從而在整個測試數據樣本中實現真實價格和預測價格之間的最小均方根誤差(RMSE)。

數據準備

基於Ethereum（ETH/USD）、Tezos（XTZ/USD）、Litecoin（LTC/USD）三種加密貨幣的收盤價序列，我們希望對BTC/USD的收盤價進行預測，有可能包含BTC/USD數據。

數據如下：

from ccrypto import getMultipleCrypoSeries
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, display_html, HTML
 
ccpairs = ['BTCUSD', 'ETHUSD', 'XTZUSD', 'LTCUSD']
df = getMultipleCrypoSeries(ccpairs, freq='h', exch='Coinbase',
                            start_date='2019-08-09 13:00', end_date='2020-03-13 23:00')
 
display(df)

可視化：

from sklearn.preprocessing import MinMaxScaler
 
scaler = MinMaxScaler(feature_range=(0, 1))
sdf_np = scaler.fit_transform(df)
sdf = pd.DataFrame(sdf_np, columns=df.columns, index=df.index)
 
plt.figure(figsize=(12,6))
plt.grid()
plt.plot(sdf)
plt.legend(sdf.columns)

之所以選擇 Litecoin 和 Ethereum，是因爲它們與比特幣的相關性非常高，無需考慮時間的長短。我們可以通過測量滾動窗口大小函數中的平均線性相關來判斷：

blue, orange, red = '#1f77b4', '#ff7f0e', '#d62728'  
plt.figure(figsize=(12,4))
plt.grid()
 
avg_corr1, avg_corr2, avg_corr3 = list(), list(), list()
 
for win in range(3, 72): # hours
    avg_corr1.append(df.ETHUSD.rolling(win).corr(df.BTCUSD) \
                     .replace([np.inf, -np.inf], np.nan).dropna().mean())
    avg_corr2.append(df.XTZUSD.rolling(win).corr(df.BTCUSD) \
                     .replace([np.inf, -np.inf], np.nan).dropna().mean())
    avg_corr3.append(df.LTCUSD.rolling(win).corr(df.BTCUSD) \
                     .replace([np.inf, -np.inf], np.nan).dropna().mean())
    
plt.plot(range(3, 72), avg_corr1, '.-', color=blue, label='ETH vs BTC')
plt.plot(range(3, 72), avg_corr2, '.-', color=orange, label='XTZ vs BTC')
plt.plot(range(3, 72), avg_corr3, '.-', color=red, label='LTC vs BTC')
 
plt.legend()
plt.xlabel('Rolling Window Length [hours]', fontsize=12)
plt.ylabel('Average Correlation', fontsize=12)

特徵工程

將特徵和標籤進行區分。通過標籤，我們將瞭解想要預測的值。比如，比特幣1小時、2小時、3小時的價格（標籤），或者只是1小時的價格（標籤）。

在訓練樣本中，標籤用於訓練。爲此，我們提供了一系列的特徵，並向計算機顯示相關的標籤。例如，如果我們設置的特徵是三個加密貨幣一小時前（T-1）和兩個小時前（T-2）的價格，而標籤是一小時後的比特幣價格（t），我們希望計算機學習其他加密貨幣價格的expected值與“預期”比特幣價格之間的關係。爲什麼呢？原因很簡單，基於過去的學習經驗，當新的、從未見過的加密貨幣價格序列或測試樣本出現相似性時，比特幣的價格預測應該更準確。這就是學習的目標。如果你赤手空拳地去碰一口煎鍋，你將學會不再以同樣的方式去碰它。

假設在上面的推理中我們是正確的。如果是這樣，我們可以看看特徵和標籤的準備過程。首先，讓我們準備一個函數，它允許在一行中並排顯示兩個DataFrames：

def displayS(dfs:list, captions:list, space=5):

    output = ""
    combined = dict(zip(captions, dfs))
    for caption, df in combined.items():
        output += df.style.set_table_attributes("style='display:inline'") \
                  .set_caption(caption)._repr_html_()
        output += space * "\xa0\xa0\xa0"
    display(HTML(output))

接下來，我們將使用 ccrypto library中的另一個函數，它只接受原始時間序列，允許我們指定哪個函數應該作爲標籤（在我們的例子中是BTC/USD），以及我們希望使用多少小時滯後來生成一組特徵：

label = 'BTCUSD'
 
from ccrypto import get_features_and_labels
 
out_X, _, out_y, _ = get_features_and_labels(df, label,
                                             timesteps=2,
                                             exclude_lagged_label=False,
                                             train_fraction=0  
                                            )

display(df.head())
displayS([out_X.head(), out_y.head()], ['Features:', 'Label:'])

返回結果：

訓練測試樣本分離

同樣的函數也可以用來將輸入數據拆分爲訓練樣本和測試樣本。我們所要做的就是決定使用時間序列的哪一部分來訓練 LSTM 網絡。我們取95%：

train_fract = 0.95
 
train_X, train_y, test_X, test_y = get_features_and_labels(df, label,
                                                           timesteps=2,
                                                           exclude_lagged_label=False,
                                                           train_fraction=train_fract
                                                          )
 
print(train_X.shape, train_y.shape, test_X.shape, test_y.shape)

（4972, 8) (4972, 1) (260, 8) (260, 1)

因此，我們將使用4972個觀測值乘以8個特徵進行訓練，並對260個新的比特幣收盤價進行樣本外測試。如果我們能夠非常準確地預測這260個值，那說明這次的研究成功了！

data_size = df.shape[0]
train_size = int(data_size * train_fract)
test_size = data_size - train_size
 
plt.figure(figsize=(12,6))
plt.grid()
 
for cn in sdf.columns:
    plt.plot(np.arange(train_size+test_size), sdf[cn], color=grey, \
            label='True Signal ' + cn)
 
plt.plot(np.arange(train_size, train_size+test_size), sdf[label].iloc[train_size:], \
            color=blue, label='Test Signal ' + label)
 
plt.axvspan(train_size, test_size+train_size, facecolor=blue, alpha=0.1)
plt.annotate("Train Sample", xy=(2075, 0.875), fontsize=15, color=grey)
plt.annotate("Test Sample", xy=(4350, 0.075), fontsize=15, color=blue)
plt.xlabel("Data Points")
plt.ylabel("Normalized Time-Series")
plt.legend(loc=3)

5

LSTM與比特幣

我們將設計一個簡單的網絡架構。公衆號將在今天推文的後續部分中解釋RNN、LSTM引擎的工作原理。現在，只要知道我們的LSTM將由8540個單元和一個Dropout層組成就足夠了。輸出是一個稠密層（一個單元）來返回預測的比特幣價格。

我們將使用TensorFlow 2.1.x（TF）構建LSTM網絡。

TF使用基於keras的wrapper，該wrapper要求輸入數據採用特定的格式。首先，我們需要重用get_features_and_label函數，將訓練和測試樣本作爲NumPy數組返回，而不是Pandas的DataFrame，然後使用重塑函數爲TF準備數據：

train_fract = 0.95
timesteps = 2
train_X, train_y, test_X, test_y = get_features_and_labels(sdf, label,
                                                           timesteps=timesteps,
                                                           train_fraction=train_fract,
                                                           as_np=True,
                                                           exclude_lagged_label=False)
 
train_X = train_X.reshape((train_X.shape[0], 1, train_X.shape[1]))
test_X = test_X.reshape((test_X.shape[0], 1, test_X.shape[1]))
 
print(train_X.shape, train_y.shape, test_X.shape, test_y.shape)

(4972, 1, 8) (4972, 1) (260, 1, 8) (260, 1)

準備好輸入數據後，我們開始設計網結構：

import tensorflow as tf
 
tf.random.set_seed(7)
 
dropout_fraction = 0.1
units = 8*5
ishape = train_X.shape[1], train_X.shape[2]
 
model = tf.keras.models.Sequential()
model.add(tf.keras.layers.LSTM(units, return_sequences=True, input_shape=ishape))
model.add(tf.keras.layers.Dropout(dropout_fraction))
model.add(tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(1)))
 
model.compile(loss='mse', optimizer='adam')
 
print(model.summary())

LSTM網絡的訓練可還以通過以下擬合函數來執行：

n_epochs = 3000
batch_size = train_X.shape[0]//5
 
history = model.fit(train_X, train_y, epochs=n_epochs, batch_size=batch_size,
                    validation_data=(test_X, test_y), verbose=0, shuffle=False)

epochs的數量僅僅意味着我們需要對網絡進行多長時間的訓練，其中批量大小是訓練中使用的數據點的大小。如果verbose設置爲0，model.fit函數將不會顯示擬合的進度。如何verbose=1，我們可以在進程運行時預覽它:

爲了可視化損失函數的演變，使用以下代碼：

plt.figure(figsize=(12,2))
plt.grid()
plt.plot(history.history['loss'], label='train')
plt.plot(history.history['val_loss'], label='test')
plt.ylabel('Loss')
_ = plt.legend()
 
plt.figure(figsize=(12,2))
plt.grid()
plt.ylabel('Loss')
plt.semilogy(history.history['loss'], label='train')
plt.semilogy(history.history['val_loss'], label='test')
_ = plt.legend()

測試樣本中比特幣收盤價的預測

TF允許使用一行代碼對測試樣本的模型進行預測。爲了檢驗擬合優度（預測的比特幣價格與實際價值），我們將使用均方誤差迴歸損失（MSE）來度量：

yhat = model.predict(test_X)
yorg = sdf[label].iloc[train_X.shape[0]+timesteps:]
 
yhat_f = yhat.flatten()
yorg_f = yorg.values.flatten()
 
from sklearn.metrics import mean_squared_error
rmse = np.sqrt(mean_squared_error(yhat_f, yorg_f))
print('Test MSE: %.5f' % rmse)

Test MSE: 0.01957

這是什麼意思？這是我們在玩了幾個小時後得到的最低的誤差測量，最終的fit測試樣本如下:

z = np.zeros((test_size-timesteps, 4))
z[:,0] = yhat_f
yhat_inv = scaler.inverse_transform(z)[:,0]
yorg_inv = scaler.inverse_transform(sdf)[:,0]
 
plt.figure(figsize=(12,6))
plt.grid()
plt.plot(df.index[0:train_size], yorg_inv[:train_size], '-', color=grey, \
         label=label+' True Signal (train)')
plt.plot(df.index[train_size+2:], yorg_inv[train_size+2:], '-', color=blue, \
         label=label+' True Signal (test)')
plt.plot(df.index[train_size+2:], yhat_inv, '-', color=red, label=label+ \
         ' Close Price Prediction')
 
plt.title('Test MSE: %.5f' % rmse)
plt.ylabel('BTC/USD', fontsize=12)
plt.xlim([pd.to_datetime('2020-03-01'), df.index[-1]])
plt.ylim([4000, 9500])
_ = plt.legend(loc=3)

看看預測值和實際比特幣收盤價之間的差異：

btc = yorg_inv[train_size+2:]
btc_hat = yhat_inv
 
plt.figure(figsize=(12,4))
plt.grid()
res = btc_hat - btc
plt.plot(df.index[train_size+2:], res)
plt.title('Residuals')
plt.xlim([pd.to_datetime('2020-03-01'), df.index[-1]])
plt.savefig("/Users/pawel/Desktop/resid.png", bbox_inches='tight')