Deep Implementation of Transformer-based Deep Learning Models for Financial Time Series Forecasting: Detecting Stock Market Anomalies and Optimizing Returns
Existing forecasting models have faced limitations due to the nonlinearity, non-stationarity, and complex long-term dependencies of financial markets. This article introduces the Transformer architecture, which revolutionized the field of natural language processing, to financial time series forecasting. It presents a concrete implementation plan for precisely detecting anomalies in the stock market and ultimately optimizing investment returns. This approach will be a game-changer, learning dynamic market changes, identifying hidden patterns, and elevating prediction accuracy to the next level.
1. The Challenge / Context
Financial time series data is inherently complex and difficult to predict. High noise, low signal-to-noise ratio (SNR), frequent changes in market structure (non-stationarity), and sensitive reactions to various external factors such as macroeconomic indicators, corporate earnings, and news make it difficult for traditional time series analysis models (e.g., ARIMA, GARCH) to capture effective long-term patterns.
While recurrent neural networks (RNNs) and Long Short-Term Memory (LSTM) networks have partially overcome these limitations, they faced vanishing or exploding gradient problems with extremely long sequences, or suffered from long training times due to the difficulty of parallelization caused by their sequential computation method. Especially in the stock market, it is crucial to synthesize numerous trading points and diverse information to discern subtle market 'sentiment' and 'signs of change,' an area difficult to explain with only linear relationships or local patterns. The ability to detect seemingly unpredictable 'anomalies' early is directly linked to risk management and alpha (excess return) generation.
2. Deep Dive: Transformer Architecture and its Application to Financial Time Series
The Transformer is a deep learning model introduced in the "Attention Is All You Need" paper published by the Google Brain team in 2017, proposed to overcome the limitations of existing recurrent neural network-based sequence models. Its core lies in the Self-Attention Mechanism.
- Self-Attention Mechanism: Calculates how related all elements within a sequence are to all other elements. This allows for direct modeling of dependencies between all time points, regardless of distance, solving the difficulty RNNs had in capturing long-term dependencies. In financial time series, a price fluctuation at a specific time point might have a strong correlation with a past event (e.g., earnings surprise, interest rate hike announcement), and self-attention effectively captures this.
- Positional Encoding: Self-attention is inherently indifferent to the order of elements within a sequence. Therefore, to inject sequence order information into the model, the Transformer encodes the positional information of each token (each time point in the case of time series data) and adds it to the input. This is essential for distinguishing the importance of 'recent information' and 'old information' in financial time series.
- Multi-Head Attention: Executes multiple attention mechanisms in parallel to understand and synthesize information from the input sequence from different perspectives. For example, one head might focus on the relationship between trading volume and price fluctuations, while another focuses on the relationship between specific technical indicators, allowing the model to learn various patterns simultaneously.
- Encoder-Decoder Structure (or Encoder-Only): While most sequence-to-sequence tasks use both an encoder and a decoder, for financial time series forecasting, which predicts future values, an 'encoder-only' structure is often adopted to effectively extract complex features of time series data.
Thanks to these characteristics, the Transformer has excellent potential for modeling the non-stationarity, nonlinearity, and complex long-term dependencies of financial data. In particular, its parallel computation capability also enhances learning efficiency for large amounts of financial data and long sequence lengths.
3. Step-by-Step Guide / Implementation
Here, we explain the specific steps to implement a Transformer-based financial time series forecasting model using Python and PyTorch. The main goal is to predict future price movements and detect anomalies using stock market data.
Step 1: Data Collection and Preprocessing
The most important aspects of financial time series forecasting are high-quality data and appropriate preprocessing. Stock prices (OHLCV), trading volume, technical indicators (RSI, MACD, Bollinger Bands, etc.), and, in some cases, news sentiment data or macroeconomic indicators can be utilized. Here, we use price, trading volume, and a few technical indicators.
import pandas as pd
import numpy as np
import torch
import torch.nn as nn
from sklearn.preprocessing import MinMaxScaler
import yfinance as yf # 주가 데이터 수집 라이브러리
# 1. 데이터 수집 (예시: 삼성전자 주가 데이터)
ticker = "005930.KS" # 삼성전자 티커 (한국 코스피)
start_date = "2010-01-01"
end_date = "2023-12-31"
df = yf.download(ticker, start=start_date, end=end_date)
# 2. 기술적 지표 추가 (예시: 이동평균선, RSI)
def add_technical_indicators(df):
df['SMA_5'] = df['Close'].rolling(window=5).mean()
df['SMA_20'] = df['Close'].rolling(window=20).mean()
# RSI 계산
delta = df['Close'].diff()
gain = (delta.where(delta > 0, 0)).fillna(0)
loss = (-delta.where(delta < 0, 0)).fillna(0)
avg_gain = gain.rolling(window=14).mean()
avg_loss = loss.rolling(window=14).mean()
rs = avg_gain / avg_loss
df['RSI_14'] = 100 - (100 / (1 + rs))
df.fillna(method='bfill', inplace=True) # 초기 NaN 값 처리
df.fillna(method='ffill', inplace=True) # 남은 NaN 값 처리
return df
df = add_technical_indicators(df)
# 사용할 특성 선택
features = ['Open', 'High', 'Low', 'Close', 'Volume', 'SMA_5', 'SMA_20', 'RSI_14']
data = df[features].values
# 3. 데이터 스케일링
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data)
# 4. 시퀀스 데이터 생성
# lookback_window: 과거 몇 일의 데이터를 보고 예측할 것인가
# forecast_horizon: 미래 몇 일 뒤의 값을 예측할 것인가 (여기서는 1일 뒤 Close 가격)
def create_sequences(data, lookback_window=60, forecast_horizon=1):
X, y = [], []
for i in range(len(data) - lookback_window - forecast_horizon + 1):
X.append(data[i:(i + lookback_window), :])
y.append(data[i + lookback_window + forecast_horizon - 1, features.index('Close')]) # 다음 날 Close 예측
return np.array(X), np.array(y)
lookback_window = 60 # 60일치 과거 데이터 사용
forecast_horizon = 1 # 1일 뒤 예측
X, y = create_sequences(scaled_data, lookback_window, forecast_horizon)
# 5. 학습/검증/테스트 데이터 분할 (시계열 특성상 순서 유지)
train_size = int(len(X) * 0.8)
val_size = int(len(X) * 0.1)
test_size = len(X) - train_size - val_size
X_train, y_train = X[:train_size], y[:train_size]
X_val, y_val = X[train_size:train_size+val_size], y[train_size:train_size+val_size]
X_test, y_test = X[train_size+val_size:], y[train_size+val_size:]
# PyTorch Tensor로 변환
X_train, y_train = torch.tensor(X_train, dtype=torch.float32), torch.tensor(y_train, dtype=torch.float32)
X_val, y_val = torch.tensor(X_val, dtype=torch.float32), torch.tensor(y_val, dtype=torch.float32)
X_test, y_test = torch.tensor(X_test, dtype=torch.float32), torch.tensor(y_test, dtype=torch.float32)
print(f"X_train shape: {X_train.shape}, y_train shape: {y_train.shape}")
print(f"X_val shape: {X_val.shape}, y_val shape: {y_val.shape}")
print(f"X_test shape: {X_test.shape}, y_test shape: {y_test.shape}")This code uses `yfinance` to fetch stock price data and adds technical indicators such as Simple Moving Average (SMA) and RSI. Then, `MinMaxScaler` normalizes the data to the [0, 1] range, and the `create_sequences` function structures the data into a sequence format suitable for the Transformer model's input. Finally, considering the characteristics of time series data, the training, validation, and test sets are split in chronological order.
Step 2: Transformer Model Design
The model is constructed using PyTorch's `nn.TransformerEncoderLayer` and `nn.TransformerEncoder`. Positional encoding can be implemented directly or by utilizing a library. Here, Sinusoidal Positional Encoding is implemented.
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() * (-np.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) # (max_len, 1, d_model)
self.register_buffer('pe', pe)
def forward(self, x):
# x shape: (seq_len, batch_size, d_model)
return x + self.pe[:x.size(0), :]
class TransformerModel(nn.Module):
def __init__(self, input_dim, d_model, nhead, num_encoder_layers, dim_feedforward, dropout=0.1):
super(TransformerModel, self).__init__()
self.encoder_input_layer = nn.Linear(input_dim, d_model) # 입력 특성을 d_model 차원으로 임베딩
self.pos_encoder = PositionalEncoding(d_model)
encoder_layers = nn.TransformerEncoderLayer(d_model, nhead, dim_feedforward, dropout, batch_first=False)
self.transformer_encoder = nn.TransformerEncoder(encoder_layers, num_encoder_layers)
self.decoder_output_layer = nn.Linear(d_model, 1) # 최종 예측은 1차원 (Close 가격)
def forward(self, src):
# src shape: (seq_len, batch_size, input_dim)
src = self.encoder_input_layer(src) * np.sqrt(self.transformer_encoder.layers[0].d_model) # 스케일링
src = self.pos_encoder(src)
output = self.transformer_encoder(src) # (seq_len, batch_size, d_model)
output = self.decoder_output_layer(output[-1, :, :]) # 마지막 시점의 출력을 사용하여 예측
return output
# 모델 파라미터 설정
input_dim = len(features) # 입력 특성 수
d_model = 64 # 임베딩 차원
nhead = 4 # 멀티-헤드 어텐션 헤드 수
num_encoder_layers = 2 # 인코더 레이어 수
dim_feedforward = 128 # Feedforward 네트워크 차원
dropout = 0.1
model = TransformerModel(input_dim, d_model, nhead, num_encoder_layers, dim_feedforward)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
print(model)The `PositionalEncoding` class provides time series order information to the model. The `TransformerModel` embeds input features into `d_model` dimensions, adds positional encoding, and then performs final prediction through multiple `TransformerEncoderLayer`s. `batch_first=False` means that the input tensor has the shape (sequence length, batch size, feature dimension).
Step 3: Model Training and Validation
Model training is similar to the general deep learning model training process. MSE (Mean Squared Error) is used as the loss function, and Adam is used as the optimizer. Early Stopping and learning rate scheduling can be considered to prevent overfitting.
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
# 데이터셋 및 데이터 로더 생성
train_dataset = TensorDataset(X_train.permute(1, 0, 2), y_train) # (batch_size, seq_len, input_dim)으로 변환
val_
