前言

系列專欄:【深度學(xué)習：算法項目實戰(zhàn)】??
涉及醫(yī)療健康、財經(jīng)金融、商業(yè)零售、食品飲料、運動健身、交通運輸、環(huán)境科學(xué)、社交媒體以及文本和圖像處理等諸多領(lǐng)域，討論了各種復(fù)雜的深度神經(jīng)網(wǎng)絡(luò)思想，如卷積神經(jīng)網(wǎng)絡(luò)、循環(huán)神經(jīng)網(wǎng)絡(luò)、生成對抗網(wǎng)絡(luò)、門控循環(huán)單元、長短期記憶、自然語言處理、深度強化學(xué)習、大型語言模型和遷移學(xué)習。

在金融市場的分析中，股票價格預(yù)測一直是一個充滿挑戰(zhàn)且備受關(guān)注的領(lǐng)域。傳統(tǒng)的時序預(yù)測方法，如ARIMA、LSTM等，雖然在一定程度上能夠捕捉到時間序列數(shù)據(jù)的動態(tài)特性，但在處理復(fù)雜的非線性關(guān)系和長期依賴時往往力不從心。近年來，隨著深度學(xué)習技術(shù)的快速發(fā)展，尤其是Transformer模型的出現(xiàn)，為時序預(yù)測問題提供了新的解決思路。

Transformer模型通過其獨特的自注意力機制，能夠有效地捕捉到時間序列數(shù)據(jù)中的長期依賴關(guān)系，這在股票價格預(yù)測等金融時序預(yù)測任務(wù)中顯得尤為重要。然而，Transformer模型在處理局部依賴和時序信息方面可能不如LSTM等循環(huán)神經(jīng)網(wǎng)絡(luò)模型。因此，結(jié)合LSTM和Transformer的混合模型應(yīng)運而生，旨在充分利用LSTM在處理時序信息和短期依賴方面的優(yōu)勢，以及Transformer在捕捉長期依賴關(guān)系方面的能力。

本文將介紹一種基于LSTM-Transformer混合模型的股票價格多變量時序預(yù)測方法，該方法結(jié)合了LSTM和Transformer的優(yōu)點，旨在提高股票價格預(yù)測的準確性。我們將使用PyTorch框架來實現(xiàn)該模型，并通過實驗驗證其在股票價格預(yù)測任務(wù)中的有效性。希望通過本文的探討，能夠為金融市場的時序預(yù)測問題提供一些新的思路和解決方案。

1. 數(shù)據(jù)集介紹

AAPL股票數(shù)據(jù)集，是蘋果公司（Apple Inc.）在股票市場上的交易數(shù)據(jù)集合。這些數(shù)據(jù)集包含了蘋果公司的股票價格、交易量、市值等關(guān)鍵財務(wù)指標，是金融分析、量化交易、時間序列預(yù)測等領(lǐng)域的重要數(shù)據(jù)源。投資者可以通過分析AAPL股票數(shù)據(jù)集來評估蘋果公司的基本面和市場表現(xiàn)，從而做出更為明智的投資決策。

---

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as snsfrom sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, \mean_absolute_percentage_error, \mean_squared_error, root_mean_squared_error, \r2_scoreimport torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset, Dataset
from torchinfo import summarynp.random.seed(0)
torch.manual_seed(0)

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

使用 pandas.to_datetime 函數(shù)將標量、數(shù)組、Series 或 DataFrame/dict-like 轉(zhuǎn)換為時間數(shù)據(jù)類型。

data = pd.read_csv('AAPL.csv')
print(type(data['Close'].iloc[0]),type(data['Date'].iloc[0]))# Let's convert the data type of timestamp column to datatime format
data['Date'] = pd.to_datetime(data['Date'])
print(type(data['Close'].iloc[0]),type(data['Date'].iloc[0]))# Selecting subset
cond_1 = data['Date'] >= '2021-04-23 00:00:00'
cond_2 = data['Date'] <= '2024-04-23 00:00:00'
data = data[cond_1 & cond_2].set_index('Date')
print(data.shape)

<class 'numpy.float64'> <class 'str'>
<class 'numpy.float64'> <class 'pandas._libs.tslibs.timestamps.Timestamp'>
(755, 6)

3. 數(shù)據(jù)可視化

通過 matplotlib 繪制收盤價格，收盤價是股票在正常交易日的最后交易價格，是投資者跟蹤其長期表現(xiàn)的標準基準。通過圖形可以快速識別股價的趨勢走向。

# plt.style.available
plt.style.use('_mpl-gallery')
plt.figure(figsize=(18,6))
plt.title('Close Price History')
plt.plot(AAPL['Close'],label='AAPL')
plt.ylabel('Close Price USD ($)', fontsize=18)
plt.legend()
plt.show()

4. 特征工程

4.1 特征縮放（歸一化）

MinMaxScaler() 函數(shù)是 scikit-learn 庫中預(yù)處理模塊的一個非常實用的工具，用于特征縮放，特別是將特征值縮放到一個指定的范圍內(nèi)，通常是[0, 1]。這種縮放方法對于許多機器學(xué)習算法來說是非常有用的，因為它可以幫助改善算法的收斂速度和性能，特別是在特征尺度差異較大的情況下。

# 使用選定的特征來訓(xùn)練模型
features = data.drop(['Adj Close', 'Volume'], axis=1)
target = data['Adj Close'].values.reshape(-1, 1)

# 創(chuàng)建 MinMaxScaler實例，對特征進行擬合和變換，生成NumPy數(shù)組
scaler = MinMaxScaler()
features_scaled = scaler.fit_transform(features)
target_scaled = scaler.fit_transform(target)
print(features_scaled.shape, target_scaled.shape)

4.2 構(gòu)建時間序列數(shù)據(jù)

我們創(chuàng)建一個時間序列數(shù)據(jù)，時間步 time_steps 假設(shè)設(shè)置為30

time_steps = 30
X_list = []
y_list = []for i in range(len(features_scaled) - time_steps):X_list.append(features_scaled[i:i+time_steps])y_list.append(target_scaled[i+time_steps])X = np.array(X_list) # [samples, time_steps, num_features]
y = np.array(y_list) # [target]

上述代碼的目的是進行時間序列數(shù)據(jù)的預(yù)處理，將原始的時間序列數(shù)據(jù)轉(zhuǎn)換為適合機器學(xué)習模型輸入的格式。具體來說，它通過滑動窗口的方式將時間序列數(shù)據(jù)分割成多個樣本，每個樣本包含一定數(shù)量的時間步 time_steps 的特征數(shù)據(jù)以及對應(yīng)的一個目標值。time_steps：表示每個樣本中包含的時間步數(shù)。它決定了模型在預(yù)測時考慮的歷史數(shù)據(jù)長度。X_list：用于存儲分割后的特征數(shù)據(jù)樣本的列表。y_list：用于存儲每個特征數(shù)據(jù)樣本對應(yīng)的目標值的列表。

X_list.append(features_scaled[i:i + time_steps])：將從當前位置 i 開始，長度為 time_steps 的特征數(shù)據(jù)切片添加到 X_list 中。這樣就得到了一系列連續(xù)的時間步的特征數(shù)據(jù)樣本。
y_list.append(target_scaled[i + time_steps])：將當前位置 i + time_steps 的目標值添加到 y_list 中。這個目標值對應(yīng)于當前特征數(shù)據(jù)樣本之后的一個時間步的目標值。

samples, time_steps, num_features = X.shape  # 賦值

4.3 數(shù)據(jù)集劃分

train_test_split 函數(shù)將數(shù)組或矩陣隨機分成訓(xùn)練子集和測試子集。

X_train, X_valid,\y_train, y_valid = train_test_split(X, y, test_size=0.2, random_state=45,shuffle=False)
print(X_train.shape, X_valid.shape, y_train.shape, y_valid.shape)

以上代碼中 random_state=45 設(shè)置了隨機種子，以確保每次運行代碼時分割結(jié)果的一致性。shuffle=False 表示在分割數(shù)據(jù)時不進行隨機打亂。如果設(shè)置為True（默認值），則會在分割之前對數(shù)據(jù)進行隨機打亂，這樣可以增加數(shù)據(jù)的隨機性，但時間序列數(shù)據(jù)具有連續(xù)性，所以設(shè)置為False。

4.4 數(shù)據(jù)加載器

# 將 NumPy數(shù)組轉(zhuǎn)換為 tensor張量
X_train_tensor = torch.from_numpy(X_train).type(torch.Tensor)
X_valid_tensor = torch.from_numpy(X_valid).type(torch.Tensor)
y_train_tensor = torch.from_numpy(y_train).type(torch.Tensor).view(-1, 1)
y_valid_tensor = torch.from_numpy(y_valid).type(torch.Tensor).view(-1, 1)print(X_train_tensor.shape, X_valid_tensor.shape, y_train_tensor.shape, y_valid_tensor.shape)

以上代碼通過 train_test_split 劃分得到的訓(xùn)練集和驗證集中的特征數(shù)據(jù) X_train、X_valid 以及標簽數(shù)據(jù) y_train、y_valid 從 numpy 數(shù)組轉(zhuǎn)換為 PyTorch 的張量（tensor）類型。.type(torch.Tensor) 確保張量的數(shù)據(jù)類型為標準的 torch.Tensor 類型，.view(-1, 1) 對張量進行維度調(diào)整

class DataHandler(Dataset):def __init__(self, X_train_tensor, y_train_tensor, X_valid_tensor, y_valid_tensor):self.X_train_tensor = X_train_tensorself.y_train_tensor = y_train_tensorself.X_valid_tensor = X_valid_tensorself.y_valid_tensor = y_valid_tensordef __len__(self):return len(self.X_train_tensor)def __getitem__(self, idx):sample = self.X_train_tensor[idx]labels = self.y_train_tensor[idx]return sample, labelsdef train_loader(self):train_dataset = TensorDataset(self.X_train_tensor, self.y_train_tensor)return DataLoader(train_dataset, batch_size=32, shuffle=True)def valid_loader(self):valid_dataset = TensorDataset(self.X_valid_tensor, self.y_valid_tensor)return DataLoader(valid_dataset, batch_size=32, shuffle=False)

在上述代碼中，定義了一個名為 DataHandler 的類，它繼承自 torch.utils.data.Dataset
__init__ 方法用于接收數(shù)據(jù)和標簽。__len__ 方法返回數(shù)據(jù)集的長度。__getitem__ 方法根據(jù)給定的索引 idx 返回相應(yīng)的數(shù)據(jù)樣本和標簽。

data_handler = DataHandler(X_train_tensor, y_train_tensor, X_valid_tensor, y_valid_tensor)
train_loader = data_handler.train_loader()
valid_loader = data_handler.valid_loader()

在上述代碼中，創(chuàng)建了一個數(shù)據(jù)處理對象 data_handler，并通過該對象生成訓(xùn)練集數(shù)據(jù)加載器 train_loader 和驗證集數(shù)據(jù)加載器valid_loader。通過這種方式，可以方便地管理和加載訓(xùn)練集和驗證集數(shù)據(jù)，為深度學(xué)習模型的訓(xùn)練和評估提供了數(shù)據(jù)支持。

5. 構(gòu)建時序模型（TSF）

5.1 構(gòu)建LSTM-Transformer模型

class LSTM_Transformer(nn.Module):def __init__(self, input_dim, hidden_dim, lstm_layers, transformer_heads, transformer_layers, output_dim, dropout=0.5):super(LSTM_Transformer, self).__init__()# LSTM 層self.lstm = nn.LSTM(input_dim, hidden_dim, lstm_layers, batch_first=True)# Transformer 編碼器層transformer_encoder_layer = nn.TransformerEncoderLayer(d_model=hidden_dim, nhead=transformer_heads, dim_feedforward=hidden_dim * 2, dropout=dropout, batch_first=True)self.transformer_encoder = nn.TransformerEncoder(transformer_encoder_layer, num_layers=transformer_layers)# 輸出層self.fc = nn.Linear(hidden_dim, output_dim)def forward(self, x):# LSTM 輸出lstm_out, _ = self.lstm(x)# Transformer 輸入transformer_input = lstm_out# Transformer 輸出transformer_out = self.transformer_encoder(transformer_input)# 預(yù)測輸出output = self.fc(transformer_out[:, -1, :])return output

5.2 實例化模型、損失函數(shù)和優(yōu)化器

input_dim = num_features  # 輸入特征維度
hidden_dim = 64  # LSTM 和 Transformer 的隱藏維度
lstm_layers = 1  # LSTM 層數(shù)
transformer_heads = 8  # Transformer 頭數(shù)
transformer_layers = 1  # Transformer 層數(shù)
output_dim = 1  # 輸出維度

model = LSTM_Transformer(input_dim, hidden_dim, lstm_layers, transformer_heads, transformer_layers, output_dim)
criterion_mse = nn.MSELoss()  # 定義均方誤差損失函數(shù)
criterion_mae = nn.L1Loss()  # 定義平均絕對誤差損失
optimizer = torch.optim.Adam(model.parameters(), lr=0.0001) # 定義優(yōu)化器

1. criterion_mse = nn.MSELoss()：
   使用 PyTorch 中的 nn.MSELoss 定義了均方誤差（Mean Squared Error）損失函數(shù)。均方誤差是回歸問題中常用的損失函數(shù)，它計算預(yù)測值與真實值之間的平方差的平均值。在時序預(yù)測中，MSE 損失可以衡量模型預(yù)測值與實際值之間的差異程度。
2. criterion_mae = nn.L1Loss()：
   定義了平均絕對誤差（Mean Absolute Error）損失函數(shù)。平均絕對誤差計算預(yù)測值與真實值之間的絕對差值的平均值。與 MSE 損失相比，MAE 損失對異常值不那么敏感。
3. optimizer = torch.optim.Adam(model.parameters(), lr=0.0001)：
   • 使用 Adam 優(yōu)化器來優(yōu)化模型的參數(shù)。Adam 是一種自適應(yīng)學(xué)習率的優(yōu)化算法，結(jié)合了動量和 RMSProp 的優(yōu)點，能夠在訓(xùn)練過程中自動調(diào)整學(xué)習率，加快收斂速度。
   • model.parameters() 表示要優(yōu)化的模型參數(shù)，lr=0.0001是學(xué)習率，控制每次參數(shù)更新的步長。學(xué)習率的選擇對于模型的訓(xùn)練效果很重要，過小的學(xué)習率可能導(dǎo)致收斂速度過慢，而過大的學(xué)習率可能導(dǎo)致模型無法收斂或振蕩。

5.3 模型概要

# batch_size, seq_len(time_steps), input_size(in_channels)
summary(model, (32, time_steps, num_features)) 

===============================================================================================
Layer (type:depth-idx)                        Output Shape              Param #
===============================================================================================
LSTM_Transformer                              [32, 1]                   --
├─LSTM: 1-1                                   [32, 30, 64]              17,920
├─TransformerEncoder: 1-2                     [32, 30, 64]              --
│    └─ModuleList: 2-1                        --                        --
│    │    └─TransformerEncoderLayer: 3-1      [32, 30, 64]              33,472
├─Linear: 1-3                                 [32, 1]                   65
===============================================================================================
Total params: 51,457
Trainable params: 51,457
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 17.21
===============================================================================================
Input size (MB): 0.02
Forward/backward pass size (MB): 0.49
Params size (MB): 0.07
Estimated Total Size (MB): 0.58
===============================================================================================

6. 模型訓(xùn)練與可視化

6.1 定義訓(xùn)練與評估函數(shù)

在模型訓(xùn)練之前，我們需先定義 train 函數(shù)來執(zhí)行模型訓(xùn)練過程

def train(model, iterator, optimizer):epoch_loss_mse = 0epoch_loss_mae = 0model.train()  # 確保模型處于訓(xùn)練模式for batch in iterator:optimizer.zero_grad()  # 清空梯度inputs, targets = batch  # 獲取輸入和目標值outputs = model(inputs)  # 前向傳播loss_mse = criterion_mse(outputs, targets)  # 計算損失loss_mae = criterion_mae(outputs, targets)combined_loss = loss_mse + loss_mae  # 可以根據(jù)需要調(diào)整兩者的權(quán)重combined_loss.backward()optimizer.step()epoch_loss_mse += loss_mse.item()  # 累計損失epoch_loss_mae += loss_mae.item()average_loss_mse = epoch_loss_mse / len(iterator)  # 計算平均損失average_loss_mae = epoch_loss_mae / len(iterator)return average_loss_mse, average_loss_mae

上述代碼定義了一個名為 train 的函數(shù)，用于訓(xùn)練給定的模型。它接收模型、數(shù)據(jù)迭代器、優(yōu)化器作為參數(shù)，并返回訓(xùn)練過程中的平均損失。

def evaluate(model, iterator):epoch_loss_mse = 0epoch_loss_mae = 0model.eval()  # 將模型設(shè)置為評估模式，例如關(guān)閉 Dropout 等with torch.no_grad():  # 不需要計算梯度for batch in iterator:inputs, targets = batchoutputs = model(inputs)  # 前向傳播loss_mse = criterion_mse(outputs, targets)  # 計算損失loss_mae = criterion_mae(outputs, targets)epoch_loss_mse += loss_mse.item()  # 累計損失epoch_loss_mae += loss_mae.item()return epoch_loss_mse / len(iterator), epoch_loss_mae / len(iterator)

上述代碼定義了一個名為 evaluate 的函數(shù)，用于評估給定模型在給定數(shù)據(jù)迭代器上的性能。它接收模型、數(shù)據(jù)迭代器作為參數(shù)，并返回評估過程中的平均損失。這個函數(shù)通常在模型訓(xùn)練的過程中定期被調(diào)用，以監(jiān)控模型在驗證集或測試集上的性能。通過評估模型的性能，可以了解模型的泛化能力和訓(xùn)練的進展情況。

epoch = 1000
train_mselosses = []
valid_mselosses = []
train_maelosses = []
valid_maelosses = []for epoch in range(epoch):train_loss_mse, train_loss_mae = train(model, train_loader, optimizer)valid_loss_mse, valid_loss_mae = evaluate(model, valid_loader)train_mselosses.append(train_loss_mse)valid_mselosses.append(valid_loss_mse)train_maelosses.append(train_loss_mae)valid_maelosses.append(valid_loss_mae)print(f'Epoch: {epoch+1:02}, Train MSELoss: {train_loss_mse:.5f}, Train MAELoss: {train_loss_mae:.3f}, Val. MSELoss: {valid_loss_mse:.5f}, Val. MAELoss: {valid_loss_mae:.3f}')

上述代碼主要進行了模型的訓(xùn)練和評估過程，并記錄了每個 epoch 的訓(xùn)練和驗證集上的均方誤差損失(MSE Loss)和平均絕對誤差損失(MAE Loss)。

Epoch: 1000, Train MSELoss: 0.00149, Train MAELoss: 0.029, Val. MSELoss: 0.00093, Val. MAELoss: 0.023

6.2 繪制訓(xùn)練與驗證損失曲線

# 繪制 MSE損失圖
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(train_mselosses, label='Train MSELoss')
plt.plot(valid_mselosses, label='Validation MSELoss')
plt.xlabel('Epoch')
plt.ylabel('MSELoss')
plt.title('Train and Validation MSELoss')
plt.legend()
plt.grid(True)# 繪制 MAE損失圖
plt.subplot(1, 2, 2)
plt.plot(train_maelosses, label='Train MAELoss')
plt.plot(valid_maelosses, label='Validation MAELoss')
plt.xlabel('Epoch')
plt.ylabel('MAELoss')
plt.title('Train and Validation MAELoss')
plt.legend()
plt.grid(True)plt.show()

7. 模型評估與可視化

7.1 構(gòu)建預(yù)測函數(shù)

定義預(yù)測函數(shù) prediction 方便調(diào)用

# 定義 prediction函數(shù)
def prediction(model, iterator): all_targets = []all_predictions = []model.eval()with torch.no_grad():for batch in iterator:inputs, targets = batchpredictions = model(inputs)all_targets.extend(targets.numpy())all_predictions.extend(predictions.numpy())return all_targets, all_predictions

這段代碼定義了一個名為 prediction 的函數(shù)，其主要目的是使用給定的模型對輸入數(shù)據(jù)進行預(yù)測，并收集所有的目標值和預(yù)測值。

7.2 驗證集預(yù)測

# 模型預(yù)測
targets, predictions = prediction(model, valid_loader)

# 反歸一化
denormalized_targets = scaler.inverse_transform(targets)
denormalized_predictions = scaler.inverse_transform(predictions)

targets 是經(jīng)過歸一化處理后的目標值數(shù)組，predictions 是經(jīng)過歸一化處理后的預(yù)測值數(shù)組。scaler 是 MinMaxScaler() 歸一化類的實例，inverse_transform 方法會將歸一化后的數(shù)組還原為原始數(shù)據(jù)的尺度，即對預(yù)測值進行反歸一化操作。

# Visualize the data
plt.figure(figsize=(12,6))
plt.style.use('_mpl-gallery')
plt.title('Comparison of validation set prediction results')
plt.plot(denormalized_targets, color='blue',label='Actual Value')
plt.plot(denormalized_predictions, color='orange', label='Valid Value')
plt.legend()
plt.show()

7.3 回歸擬合圖

使用 regplot() 函數(shù)繪制數(shù)據(jù)圖并擬合線性回歸模型。

plt.figure(figsize=(5, 5), dpi=100)
sns.regplot(x=denormalized_targets, y=denormalized_predictions, scatter=True, marker="*", color='orange',line_kws={'color': 'red'})
plt.show()

7.4 評估指標

以下代碼使用了一些常見的評估指標：平均絕對誤差（MAE）、平均絕對百分比誤差（MAPE）、均方誤差（MSE）、均方根誤差（RMSE）和決定系數(shù)（R2）來衡量模型預(yù)測的性能。這里我們將通過調(diào)用 sklearn.metrics 模塊中的 mean_absolute_error mean_absolute_percentage_error mean_squared_error root_mean_squared_error r2_score 函數(shù)來對模型的預(yù)測效果進行評估。

mae = mean_absolute_error(targets, predictions)
print(f"MAE: {mae:.4f}")mape = mean_absolute_percentage_error(targets, predictions)
print(f"MAPE: {mape * 100:.4f}%")mse = mean_squared_error(targets, predictions)
print(f"MSE: {mse:.4f}")rmse = root_mean_squared_error(targets, predictions)
print(f"RMSE: {rmse:.4f}")r2 = r2_score(targets, predictions)
print(f"R2: {r2:.4f}")

MAE: 0.0229
MAPE: 3.0055%
MSE: 0.0009
RMSE: 0.0302
R2: 0.9346