python量化之搭建Transformer模型用于股票價格預測
前言
下面的這篇文章主要教大家如何搭建一個基于Transformer的簡單預測模型,并將其用于股票價格預測當中。原代碼在文末進行獲取。
1、Transformer模型
Transformer 是 Google 的團隊在 2017 年提出的一種 NLP 經(jīng)典模型,現(xiàn)在比較火熱的 Bert 也是基于 Transformer。Transformer 模型使用了 Self-Attention 機制,不采用 RNN 的順序結構,使得模型可以并行化訓練,而且能夠擁有全局信息。這篇文章的目的主要是將帶大家通過Pytorch框架搭建一個基于Transformer的簡單股票價格預測模型。
Transformer的基本架構:
具體地,我們用到了上證指數(shù)的收盤價數(shù)據(jù)為例,進行預測t+1時刻的收盤價。需要注意的是,本文只是通過這樣一個簡單的基本模型,帶大家梳理一下數(shù)據(jù)預處理,模型構建以及模型評估的流程。模型還有很多可以改進的地方,例如選擇更有意義的特征,如何進行有效的多步預測等。
2、環(huán)境準備
本地環(huán)境:
Python 3.7
IDE:Pycharm
庫版本:
numpy 1.18.1
pandas 1.0.3
sklearn 0.22.2
matplotlib 3.2.1
torch 1.10.1
3、代碼實現(xiàn)
3.1. 導入庫以及定義超參
首先,需要導入用到庫,以及模型的一些超參數(shù)的設置。其中,input_window和output_window分別用于設置輸入數(shù)據(jù)的長度以及輸出數(shù)據(jù)的長度。當然,這些參數(shù)大家也可以根據(jù)實際應用場景進行修改。
import torch import torch.nn as nn import numpy as np import time import math import matplotlib.pyplot as plt from sklearn.preprocessing import MinMaxScaler import pandas as pd torch.manual_seed(0) np.random.seed(0) input_window = 20 output_window = 1 batch_size = 64 device = torch. device("cuda" if torch.cuda.is_available() else "cpu") print(device)
3.2. 模型構建
Transformer中很重要的一個組件是提出了一種新的位置編碼的方式。我們知道,循環(huán)神經(jīng)網(wǎng)絡本身就是一種順序結構,天生就包含了詞在序列中的位置信息。當拋棄循環(huán)神經(jīng)網(wǎng)絡結構,完全采用Attention取而代之,這些詞序信息就會丟失,模型就沒有辦法知道每個詞在句子中的相對和絕對的位置信息。因此,有必要把詞序信號加到詞向量上幫助模型學習這些信息,位置編碼(PositionalEncoding)就是用來解決這種問題的方法。它的原理是將生成的不同頻率的正弦和余弦數(shù)據(jù)作為位置編碼添加到輸入序列中,從而使得模型可以捕捉輸入變量的相對位置關系。
class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super(PositionalEncoding, self).__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0).transpose(0, 1) self.register_buffer('pe', pe) def forward(self, x): return x + self.pe[:x.size(0), :]
之后,搭建Transformer
的基本結構,在Pytorch中有已經(jīng)實現(xiàn)的封裝好的Transformer組件,可以很方便地進行調用和修改。其中需要注意的是,文中并沒有采用原論文中的Encoder-Decoder的架構,而是將Decoder用了一個全連接層進行代替,用于輸出預測值。另外,其中的create_mask將輸入進行mask,從而避免引入未來信息。
class TransAm(nn.Module): def __init__(self, feature_size=250, num_layers=1, dropout=0.1): super(TransAm, self).__init__() self.model_type = 'Transformer' self.src_mask = None self.pos_encoder = PositionalEncoding(feature_size) self.encoder_layer = nn.TransformerEncoderLayer(d_model=feature_size, nhead=10, dropout=dropout) self.transformer_encoder = nn.TransformerEncoder(self.encoder_layer, num_layers=num_layers) self.decoder = nn.Linear(feature_size, 1) self.init_weights() def init_weights(self): initrange = 0.1 self.decoder.bias.data.zero_() self.decoder.weight.data.uniform_(-initrange, initrange) def forward(self, src): if self.src_mask is None or self.src_mask.size(0) != len(src): device = src.device mask = self._generate_square_subsequent_mask(len(src)).to(device) self.src_mask = mask src = self.pos_encoder(src) output = self.transformer_encoder(src, self.src_mask) output = self.decoder(output) return output def _generate_square_subsequent_mask(self, sz): mask = (torch.triu(torch.ones(sz, sz)) == 1).transpose(0, 1) mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0)) return mask
3.3. 數(shù)據(jù)預處理
接下來需要對數(shù)據(jù)進行預處理,首先定義一個窗口劃分的函數(shù)。它的作用是將輸入按照延遲output_windw的方式來劃分數(shù)據(jù)以及其標簽,文中是進行單步預測,所以假設輸入是1到20,則其標簽就是2到21,以適應Transformer
的seq2seq
的形式的輸出。
def create_inout_sequences(input_data, tw): inout_seq = [] L = len(input_data) for i in range(L - tw): train_seq = input_data[i:i + tw] train_label = input_data[i + output_window:i + tw + output_window] inout_seq.append((train_seq, train_label)) return torch.FloatTensor(inout_seq)
之后劃分訓練集和測試集,其中前70%條數(shù)據(jù)用于模型訓練,后面的數(shù)據(jù)用于模型測試。具體地,我們用到了前input_window個收盤價來預測下一時刻的收盤價數(shù)據(jù)。
def get_data(): series = pd.read_csv('./000001_Daily.csv', usecols=['Close']) # series = pd.read_csv('./daily-min-temperatures.csv', usecols=['Temp']) scaler = MinMaxScaler(feature_range=(-1, 1)) series = scaler.fit_transform(series.values.reshape(-1, 1)).reshape(-1) train_samples = int(0.7 * len(series)) train_data = series[:train_samples] test_data = series[train_samples:] train_sequence = create_inout_sequences(train_data, input_window) train_sequence = train_sequence[:-output_window] test_data = create_inout_sequences(test_data, input_window) test_data = test_data[:-output_window] return train_sequence.to(device), test_data.to(device)
接下來實現(xiàn)一個databatch generator
,便于從數(shù)據(jù)中按照batch的形式進行讀取數(shù)據(jù)。
def get_batch(source, i, batch_size): seq_len = min(batch_size, len(source) - 1 - i) data = source[i:i + seq_len] input = torch.stack(torch.stack([item[0] for item in data]).chunk(input_window, 1)) target = torch.stack(torch.stack([item[1] for item in data]).chunk(input_window, 1)) return input, target
3.4. 模型訓練以及評估
下面是模型訓練的代碼。具體地,就是通過遍歷訓練集,通過既定的loss,對參數(shù)進行反向傳播,其中用到了梯度裁剪的技巧用于防止梯度爆炸,然后每間隔幾個間隔打印一下loss。
def train(train_data): model.train() for batch_index, i in enumerate(range(0, len(train_data) - 1, batch_size)): start_time = time.time() total_loss = 0 data, targets = get_batch(train_data, i, batch_size) optimizer.zero_grad() output = model(data) loss = criterion(output, targets) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 0.7) optimizer.step() total_loss += loss.item() log_interval = int(len(train_data) / batch_size / 5) if batch_index % log_interval == 0 and batch_index > 0: cur_loss = total_loss / log_interval elapsed = time.time() - start_time print('| epoch {:3d} | {:5d}/{:5d} batches | lr {:02.6f} | {:5.2f} ms | loss {:5.5f} | ppl {:8.2f}' .format(epoch, batch_index, len(train_data) // batch_size, scheduler.get_lr()[0], elapsed * 1000 / log_interval, cur_loss, math.exp(cur_loss)))
接下來是對模型進行評估的代碼。
def evaluate(eval_model, data_source): eval_model.eval() total_loss = 0 eval_batch_size = 1000 with torch.no_grad(): for i in range(0, len(data_source) - 1, eval_batch_size): data, targets = get_batch(data_source, i, eval_batch_size) output = eval_model(data) total_loss += len(data[0]) * criterion(output, targets).cpu().item() return total_loss / len(data_source)
最后,是模型運行過程的可視化:
def plot_and_loss(eval_model, data_source, epoch): eval_model.eval() total_loss = 0. test_result = torch.Tensor(0) truth = torch.Tensor(0) with torch.no_grad(): for i in range(0, len(data_source) - 1): data, target = get_batch(data_source, i, 1) output = eval_model(data) total_loss += criterion(output, target).item() test_result = torch.cat((test_result, output[-1].view(-1).cpu()), 0) truth = torch.cat((truth, target[-1].view(-1).cpu()), 0) plt.plot(test_result, color="red") plt.plot(truth, color="blue") plt.grid(True, which='both') plt.axhline(y=0, color='k') plt.savefig('graph/transformer-epoch%d.png' % epoch) plt.close() return total_loss / i
3.5. 模型運行
最后,對模型進行運行。其中用到了mse作為loss,adam作為優(yōu)化器,以及設定學習率的調度器,最后運行200個epoch,每隔10個epoch在測試集上評估一下模型。
train_data, val_data = get_data() model = TransAm().to(device) criterion = nn.MSELoss() lr = 0.005 optimizer = torch.optim.AdamW(model.parameters(), lr=lr) scheduler = torch.optim.lr_scheduler.StepLR(optimizer, 1, gamma=0.95) epochs = 200 for epoch in range(1, epochs + 1): epoch_start_time = time.time() train(train_data) if (epoch % 10 is 0): val_loss = plot_and_loss(model, val_data, epoch) else: val_loss = evaluate(model, val_data) print('-' * 89) print('| end of epoch {:3d} | time: {:5.2f}s | valid loss {:5.5f} | valid ppl {:8.2f}'.format(epoch, ( time.time() - epoch_start_time), val_loss, math.exp(val_loss))) print('-' * 89) scheduler.step()
下面是運行的結果,可以看到loss明顯降低了:
cuda| epoch 1 | 2/ 10 batches | lr 0.005000 | 7.83 ms | loss 39.99368 | ppl 233902099994043520.00| epoch 1 |
4/ 10 batches | lr 0.005000 | 7.81 ms | loss 7.20889 | ppl 1351.39| epoch 1 | 6/ 10 batches | lr 0.005000 | 11.10 ms | loss 1.68758 | ppl 5.41| epoch 1 |
8/ 10 batches | lr 0.005000 | 9.35 ms | loss 0.00833 | ppl 1.01| epoch 1 | 10/ 10 batches | lr 0.005000 | 7.81 ms | loss 1.18041 | ppl 3.26-----------------------------------------------------------------------------------------| end of epoch 1 | time: 1.96s | valid loss 2.58557 | valid ppl 13.27
...
| end of epoch 198 | time: 0.30s | valid loss 0.00032 | valid ppl 1.00-----------------------------------------------------------------------------------------| epoch 199 |
2/ 10 batches | lr 0.000000 | 15.62 ms | loss 0.00057 | ppl 1.00| epoch 199 | 4/ 10 batches | lr 0.000000 | 15.62 ms | loss 0.00184 | ppl 1.00| epoch 199 |
6/ 10 batches | lr 0.000000 | 15.62 ms | loss 0.00212 | ppl 1.00| epoch 199 | 8/ 10 batches | lr 0.000000 | 7.81 ms | loss 0.00073 | ppl 1.00| epoch 199 | 10/ 10 batches | lr 0.000000 |
7.81 ms | loss 0.00057 | ppl 1.00-----------------------------------------------------------------------------------------| end of epoch 199 | time: 0.30s | valid loss 0.00032 | valid ppl 1.00-----------------------------------------------------------------------------------------| epoch 200 | 2/ 10 batches | lr 0.000000 | 15.62 ms | loss 0.00053 | ppl 1.00| epoch 200 |
4/ 10 batches | lr 0.000000 | 7.81 ms | loss 0.00177 | ppl
1.00| epoch 200 | 6/ 10 batches | lr 0.000000 | 7.81 ms | loss 0.00224 | ppl 1.00| epoch 200 | 8/ 10 batches | lr 0.000000 | 15.62 ms | loss 0.00069 | ppl 1.00| epoch 200 | 10/ 10 batches | lr 0.000000 | 7.81 ms | loss 0.00049 | ppl 1.00-----------------------------------------------------------------------------------------| end of epoch 200 | time: 0.62s | valid loss 0.00032 | valid ppl
1.00-----------------------------------------------------------------------------------------
最后是模型的擬合效果,從實驗結果中可以看出我們搭建的簡單的Transformer模型可以實現(xiàn)相對不錯的數(shù)據(jù)擬合效果。
4、總結
在這篇文章中,我們介紹了如何基于Pytorch框架搭建一個基于Transformer的股票預測模型,并通過真實股票數(shù)據(jù)對模型進行了實驗,可以看出Transformer模型對股價預測具有一定的效果。另外,文中只是做了一個簡單的demo,其中仍然有很多可以改進的地方,如采用更多有意義的輸入數(shù)據(jù),優(yōu)化其中的一些組件等。除此之外,目前基于Transformer的模型層出不窮,其中也有很多值得我們去學習,大家也可以采用更先進的Transformer模型進行試驗。
到此這篇關于python量化之搭建Transformer模型用于股票價格預測的文章就介紹到這了,更多相關python搭建Transformer模型內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
python 根據(jù)列表批量下載網(wǎng)易云音樂的免費音樂
這篇文章主要介紹了python 根據(jù)列表下載網(wǎng)易云音樂的免費音樂,幫助大家更好的理解和學習python,感興趣的朋友可以了解下2020-12-12Python模擬登錄requests.Session應用詳解
這篇文章主要介紹了Python模擬登錄requests.Session應用詳解,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2020-11-11python數(shù)字圖像處理skimage讀取顯示與保存圖片
這篇文章主要為大家介紹了python數(shù)字圖像處理使用skimage讀取顯示與保存圖片示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2022-06-06通過Python實現(xiàn)對SQL Server 數(shù)據(jù)文件大小的監(jiān)控告警功能
這篇文章主要介紹了通過Python實現(xiàn)對SQL Server 數(shù)據(jù)文件大小的監(jiān)控告警,本文給大家分享問題報錯信息及解決方案,需要的朋友可以參考下2021-04-04