Обучение модели на основе Transformer для прогноза цены

Обучение модели на основе Transformer для прогноза цены

Transformer архитектура, разработанная для NLP в 2017 году, показала отличные результаты и в задачах временных рядов. Механизм self-attention позволяет модели напрямую обращаться к любому историческому моменту без рекуррентного прохода, что устраняет ключевой недостаток LSTM — затухание градиентов на длинных последовательностях.

Temporal Fusion Transformer (TFT)

TFT — специализированная Transformer архитектура для временных рядов. Разработана Google Brain в 2021 году. Умеет работать с:

• Наблюдаемыми временными features (цена, объём)
• Статичными мета-features (символ, биржа)
• Known future inputs (час дня, день недели)

from pytorch_forecasting import TemporalFusionTransformer, TimeSeriesDataSet
from pytorch_forecasting.metrics import QuantileLoss

# Создаём датасет в формате TFT
training = TimeSeriesDataSet(
    data=train_df,
    time_idx='time_idx',
    target='close_return',
    group_ids=['symbol'],  # можно учить на нескольких парах одновременно
    max_encoder_length=120,   # история
    max_prediction_length=24, # горизонт прогноза
    
    # Time-varying known inputs
    time_varying_known_reals=['hour_of_day', 'day_of_week'],
    
    # Time-varying unknown (наблюдаемые только в прошлом)
    time_varying_unknown_reals=[
        'close_return', 'volume_ratio', 'rsi', 'macd', 
        'funding_rate', 'open_interest_change'
    ],
    
    # Целевое преобразование
    target_normalizer=None  # returns уже в нормализованном диапазоне
)

tft = TemporalFusionTransformer.from_dataset(
    training,
    hidden_size=64,
    attention_head_size=4,
    dropout=0.1,
    hidden_continuous_size=16,
    loss=QuantileLoss(quantiles=[0.1, 0.25, 0.5, 0.75, 0.9]),
    optimizer='ranger'  # AdamW + Lookahead
)

Quantile Loss позволяет модели предсказывать не точку, а распределение: «50% вероятность что return между -1% и +2%». Это значительно полезнее для торговли, чем точечный прогноз.

Vanilla Transformer для Price Forecasting

import torch
import torch.nn as nn
import math

class FinancialTransformer(nn.Module):
    def __init__(self, input_size, d_model=64, nhead=8, num_encoder_layers=3,
                 dim_feedforward=256, dropout=0.1, max_seq_len=512):
        super().__init__()
        
        # Positional Encoding
        self.pos_encoding = PositionalEncoding(d_model, dropout, max_seq_len)
        
        # Input projection
        self.input_projection = nn.Linear(input_size, d_model)
        
        # Transformer Encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=dim_feedforward,
            dropout=dropout,
            batch_first=True,
            norm_first=True  # Pre-norm для стабильности обучения
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_encoder_layers)
        
        # Output
        self.output_norm = nn.LayerNorm(d_model)
        self.fc = nn.Linear(d_model, 1)
    
    def forward(self, x):
        # x: (batch, seq_len, input_size)
        x = self.input_projection(x)
        x = self.pos_encoding(x)
        
        # Causal mask: модель видит только прошлое
        seq_len = x.size(1)
        causal_mask = nn.Transformer.generate_square_subsequent_mask(seq_len)
        
        encoded = self.transformer(x, mask=causal_mask)
        out = self.output_norm(encoded[:, -1, :])
        return self.fc(out)

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, dropout=0.1, max_len=5000):
        super().__init__()
        self.dropout = nn.Dropout(p=dropout)
        
        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)
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        x = x + self.pe[:, :x.size(1), :]
        return self.dropout(x)

PatchTST — эффективный Transformer для временных рядов

PatchTST (2023) применяет patching аналогично Vision Transformer: разбивает временной ряд на патчи (подпоследовательности) и применяет self-attention к патчам, а не к отдельным timestep'ам. Это значительно эффективнее по вычислениям и лучше захватывает локальные паттерны.

class PatchTST(nn.Module):
    def __init__(self, seq_len, patch_len=16, stride=8, d_model=128, 
                 nhead=8, n_layers=3, dropout=0.1):
        super().__init__()
        self.patch_len = patch_len
        self.stride = stride
        n_patches = (seq_len - patch_len) // stride + 1
        
        self.patch_embedding = nn.Linear(patch_len, d_model)
        self.pos_embedding = nn.Parameter(torch.randn(1, n_patches, d_model))
        
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model, nhead=nhead, 
            dim_feedforward=d_model*4, dropout=dropout, batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, n_layers)
        self.head = nn.Linear(d_model * n_patches, 1)
    
    def forward(self, x):
        # x: (batch, seq_len, n_features) — берём один feature
        # Patching
        patches = x.unfold(1, self.patch_len, self.stride)
        # patches: (batch, n_patches, patch_len)
        
        embedded = self.patch_embedding(patches) + self.pos_embedding
        encoded = self.transformer(embedded)
        out = self.head(encoded.flatten(1))
        return out

Мультиактивное обучение

Transformer хорошо масштабируется на несколько активов одновременно. Обучение на 50+ парах:

• Более разнообразный обучающий сигнал
• Модель учится общим паттернам рынка
• Symbol embedding для различения пар

class MultiAssetTransformer(nn.Module):
    def __init__(self, n_symbols, input_size, d_model=128, **kwargs):
        super().__init__()
        # Embedding для каждого символа
        self.symbol_embedding = nn.Embedding(n_symbols, 16)
        # Проекция: input_size + 16 (symbol emb) -> d_model
        self.input_projection = nn.Linear(input_size + 16, d_model)
        # Остальная архитектура аналогична

Обучение и регуляризация

Learning rate schedule: Warmup + Cosine Annealing. Transformer чувствителен к начальному learning rate.

from transformers import get_cosine_schedule_with_warmup

scheduler = get_cosine_schedule_with_warmup(
    optimizer,
    num_warmup_steps=100,
    num_training_steps=total_steps
)

Gradient clipping: обязателен для Transformer: clip_grad_norm_(model.parameters(), 1.0)

Mixup augmentation: случайное смешивание примеров для регуляризации — уменьшает переобучение.

Сравнение LSTM vs Transformer

На крупных датасетах (2+ лет 1h данных, 50+ пар) Transformer обычно лучше LSTM. На малых — LSTM или LightGBM.

Разрабатываем и обучаем Transformer модели (TFT для probabilistic forecasting, PatchTST для эффективности) с walk-forward validation, мультиактивным обучением и production deployment через FastAPI.