En 2017, un article intitulé “Attention Is All You Need” a été publié et a dominé le domaine de l’IA générative au cours des années suivantes. Dans ce document historique et révolutionnaire, huit chercheurs ont proposé pour la première fois l’architecture de réseau neuronal Transformer, qui est unique en ce qu’elle est entièrement basée sur le mécanisme d’attention, abandonnant les opérations traditionnelles de cycles et de volumes.

Transformer est une application de la quantification financière.

L’innovation centrale de Transformer réside dans le mécanisme d’auto-attention, qui permet de capturer la dépendance entre deux éléments de séquence, quelle que soit la distance entre eux. Cette caractéristique le rend particulièrement adapté pour traiter des données de séquences financières avec des dépendances temporelles complexes.

Prévisions de séquence temporelle

Les méthodes traditionnelles de prévision de séquences chronologiques financières (ARIMA, LSTM) ont du mal à capturer les dépendances à long terme, et Transformer est excellent dans ce domaine.

Modèle à facteurs multiples

Le Transformer peut traiter des centaines de facteurs simultanément, apprenant automatiquement les relations non linéaires complexes entre les facteurs et les propriétés de changement de temps

• Analyse de l’importance des facteurs: identifier les facteurs moteurs clés par le poids de l’attention

• Exposition à des facteurs dynamiques: capture des caractéristiques changeantes du temps de l’efficacité des facteurs

Gestion des risques

Modélisation du risque de marché:

• VaR (valeur de risque) prévision: traiter des relations non linéaires dans des conditions de marché extrêmes

• Prévision de la volatilité: capture des effets de concentration et de levier de la volatilité

• Prévision de la corrélation: modélisation de la corrélation des actifs dynamiques

Optimisation du portefeuille

Transformer est capable de:

• Données multi-actifs et multi-fréquences

• Identification et prévision de l’état du marché

Sélection d’actifs basée sur le poids de l’attention Optimisation de l’exécution des transactions:

• Modélisation des chocs de marché: prédire l’impact sur le marché des transactions à grande échelle

• Stratégie d’exécution optimale: minimiser les coûts de transaction

Les transactions à haute fréquence et la microstructure du marché

Analyse du carnet de commandes à prix fixe Transformer est capable de gérer efficacement les données de livret de commande à haute fréquence:

• Prévisions de flux de commandes: prévisions de déséquilibre des commandes d’achat et de vente

• Analyse des chocs de prix: évaluation de l’impact des commandes importantes sur les prix du marché

• Prévisions de liquidité: prévoir des variations de la profondeur du marché

Détection de la manipulation du marché

En analysant les modèles de séquences de transactions, nous avons identifié:

• Le spoofing est une forme de fraude.

• Pump and Dump (P&D) est une méthode de transfert de marchandises vers le marché.

• Autres pratiques de manipulation du marché

Bataille

Transformer est spécialisé dans le traitement de données en ordre chronologique OHLCV en 5 dimensions (prix d’ouverture, prix le plus élevé, prix le plus bas, prix de clôture, volume):

• La forme de l’entrée est: batch_size, seq_len = 30, feature_dim = 5

• Extraire des caractéristiques de chronologie plus riches en apprenant les dépendances temporelles dans les séquences

• Les indicateurs techniques sont plus efficaces pour capturer les modèles de marché complexes.

Définition du modèle de transformateur (classe TimeSeriesTransformer)

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_dim=5, d_model=32, nhead=4, num_encoder_layers=2, dim_feedforward=64, dropout=0.1, num_classes=3):
        super(TimeSeriesTransformer, self).__init__()
        self.input_proj = nn.Linear(input_dim, d_model)
        encoder_layer = nn.TransformerEncoderLayer(d_model, nhead, dim_feedforward, dropout, batch_first=True)
        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_encoder_layers)
        self.feature_layer = nn.Linear(d_model, d_model) # 用于提取特征
        self.classifier = nn.Linear(d_model, num_classes) # 用于Transformer自身的预测（如果需要）

    def forward(self, src, return_features=False):
        src = self.input_proj(src)
        memory = self.transformer_encoder(src)
        # 取序列的最后一个时间步的输出作为特征
        features = self.feature_layer(memory[:, -1, :])
        
        if return_features:
            return features
        else:
            return self.classifier(features)

Fonction de prétraitement de données

Obtenir les entrées du transformateur

def get_transformer_input(klines, index):
    if index < config.TRANSFORMER_SEQ_LEN - 1:
        return torch.zeros(1, config.TRANSFORMER_SEQ_LEN, 5)
    
    # 提取序列数据
    start_idx = index - config.TRANSFORMER_SEQ_LEN + 1
    end_idx = index + 1
    seq_data = klines[start_idx:end_idx]
    
    # 转换为OHLCV数组
    ohlcv = np.array([[k["open"], k["high"], k["low"], k["close"], k["volume"]] 
                     for k in seq_data])
    
    # 自动创建并拟合scaler（仅在训练时）
    if ModelRegistry.transformer_scaler is None:
        ModelRegistry.transformer_scaler = StandardScaler()
        if len(klines) > config.TRANSFORMER_SEQ_LEN:
            all_ohlcv = np.array([[k["open"], k["high"], k["low"], k["close"], k["volume"]] 
                                 for k in klines])
            ModelRegistry.transformer_scaler.fit(all_ohlcv)
    
    # 标准化数据
    normalized_ohlcv = ModelRegistry.transformer_scaler.transform(ohlcv)
    return torch.tensor(normalized_ohlcv, dtype=torch.float32).unsqueeze(0)

Obtenir une entrée de transformateur (scaler fixe)

def get_transformer_input_with_fixed_scaler(klines, index, fixed_scaler):
    """使用固定的scaler生成Transformer输入（用于实时预测）"""

    seq_data = klines[start_idx:end_idx]
    ohlcv = np.array([[k["open"], k["high"], k["low"], k["close"], k["volume"]] 
                     for k in seq_data])
    
    normalized_ohlcv = fixed_scaler.transform(ohlcv)
    return torch.tensor(normalized_ohlcv, dtype=torch.float32).unsqueeze(0)

Le code de Transformers en cours de formation

if config.TRANSFORMER_ENABLED:
    Log("开始联合训练Transformer和分类器...")
    
    # 创建Transformer模型
    ModelRegistry.transformer_model = TimeSeriesTransformer(
        input_dim=5, 
        d_model=config.TRANSFORMER_D_MODEL, 
        nhead=config.TRANSFORMER_NHEAD, 
        num_encoder_layers=config.TRANSFORMER_NUM_LAYERS,
        num_classes=3
    )
    
    # 优化器和损失函数
    optimizer_transformer = torch.optim.Adam(
        ModelRegistry.transformer_model.parameters(), 
        lr=config.TRANSFORMER_LEARNING_RATE
    )
    criterion_transformer = nn.CrossEntropyLoss()
    
    # 准备Transformer训练数据（避免数据泄露）
    X_transformer_sequences = []
    y_transformer_sequences = []
    
    for i in range(len(klines_for_transformer_training) - config.PREDICT_HORIZON - config.TRANSFORMER_SEQ_LEN + 1):
        # 提取序列
        seq_start = i
        seq_end = i + config.TRANSFORMER_SEQ_LEN
        seq_ohlcv = np.array([[k["open"], k["high"], k["low"], k["close"], k["volume"]] 
                            for k in klines_for_transformer_training[seq_start:seq_end]])
        
        # 标准化
        seq_normalized = ModelRegistry.transformer_scaler.transform(seq_ohlcv)
        X_transformer_sequences.append(seq_normalized)
        
        # 计算标签（价格变化）
        current_kline = klines_for_transformer_training[seq_end - 1]
        future_kline = klines_for_transformer_training[seq_end - 1 + config.PREDICT_HORIZON]
        # ... 标签计算逻辑 ...
    
    # 训练Transformer模型
    for epoch in range(config.TRANSFORMER_TRAIN_EPOCHS):
        ModelRegistry.transformer_model.train()
        optimizer_transformer.zero_grad()
        outputs = ModelRegistry.transformer_model(X_transformer_sequences)
        loss = criterion_transformer(outputs, y_transformer_sequences)
        loss.backward()
        optimizer_transformer.step()

Le code Transformer en temps réel

model_expects_transformer = any("transformer_feat" in name for name in ModelRegistry.feature_names)

if model_expects_transformer:
    if config.TRANSFORMER_ENABLED and ModelRegistry.transformer_model and ModelRegistry.transformer_scaler:
        # 检查数据是否足够
        if len(FeatureStore.klines_1min) >= config.TRANSFORMER_SEQ_LEN:
            # 使用固定scaler获取输入
            transformer_input = get_transformer_input_with_fixed_scaler(
                FeatureStore.klines_1min, 
                len(FeatureStore.klines_1min) - 1,
                ModelRegistry.transformer_scaler  # 使用训练好的scaler
            )
            
            # 提取Transformer特征
            with torch.no_grad():
                transformer_feature_np = ModelRegistry.transformer_model(
                    transformer_input, 
                    return_features=True
                ).squeeze(0).numpy()
        else:
            # 数据不足时使用零填充
            transformer_feature_np = np.zeros(config.TRANSFORMER_D_MODEL)
    else:
        # 配置禁用或模型丢失时使用零填充
        transformer_feature_np = np.zeros(config.TRANSFORMER_D_MODEL)

# 特征拼接
if transformer_feature_np is not None:
    transformer_cols = [f"transformer_feat_{i}" for i in range(config.TRANSFORMER_D_MODEL)]
    transformer_df = pd.DataFrame(transformer_feature_np.reshape(1, -1), columns=transformer_cols)
    
    # 合并表格特征和Transformer特征
    final_features_for_model_df = pd.concat([scaled_features_df, transformer_df], axis=1)

Enregistrement et chargement du modèle

Pour sauvegarder le modèle Transformer:

if config.TRANSFORMER_ENABLED and ModelRegistry.transformer_model:
    transformer_path = os.path.join(config.MODEL_DIR, f"transformer_model_{version}.pth")
    torch.save(ModelRegistry.transformer_model.state_dict(), transformer_path)
    
    # 同时保存scaler
    transformer_scaler_path = os.path.join(config.MODEL_DIR, f"transformer_scaler_{version}.pkl")
    with open(transformer_scaler_path, 'wb') as f:
        pickle.dump(ModelRegistry.transformer_scaler, f)

Pour télécharger le modèle Transformer:

if config.TRANSFORMER_ENABLED:
    transformer_path = os.path.join(config.MODEL_DIR, f"transformer_model_{ModelRegistry.current_model_version}.pth")
    
    if os.path.exists(transformer_path):
        ModelRegistry.transformer_model = TimeSeriesTransformer(
            input_dim=5, 
            d_model=config.TRANSFORMER_D_MODEL, 
            nhead=config.TRANSFORMER_NHEAD,
            num_encoder_layers=config.TRANSFORMER_NUM_LAYERS, 
            num_classes=3
        )
        ModelRegistry.transformer_model.load_state_dict(torch.load(transformer_path))
        ModelRegistry.transformer_model.eval()
        
        # 加载scaler
        transformer_scaler_path = os.path.join(config.MODEL_DIR, f"transformer_scaler_{ModelRegistry.current_model_version}.pkl")
        with open(transformer_scaler_path, "rb") as f:
            ModelRegistry.transformer_scaler = pickle.load(f)

Paramètres de configuration

self.TRANSFORMER_ENABLED = self.config["transformer"]["enabled"]  # 是否启用
self.TRANSFORMER_SEQ_LEN = self.config["transformer"]["seq_len"]  # 序列长度（30）
self.TRANSFORMER_D_MODEL = self.config["transformer"]["d_model"]  # 特征维度（32）
self.TRANSFORMER_NHEAD = self.config["transformer"]["nhead"]      # 注意力头数（4）
self.TRANSFORMER_NUM_LAYERS = self.config["transformer"]["num_layers"]  # 层数（2）
self.TRANSFORMER_TRAIN_EPOCHS = self.config["transformer"]["train_epochs"]  # 训练轮数
self.TRANSFORMER_LEARNING_RATE = self.config["transformer"]["learning_rate"]  # 学习率

Gestion des noms de fonctionnalités

def update_feature_names_with_transformer():
    """更新特征名称列表以包含Transformer特征"""
    base_features = [...]  # 29个基础特征
    
    if config.TRANSFORMER_ENABLED:
        transformer_features = [f"transformer_feat_{i}" for i in range(config.TRANSFORMER_D_MODEL)]
        ModelRegistry.feature_names = base_features + transformer_features  
    else:
        ModelRegistry.feature_names = base_features  # 仅29个特征

Processus de travail du transformateur

实时预测时：
1. 获取最近30根K线的OHLCV数据 → (30, 5)
2. 使用训练时确定的scaler标准化 → 标准化后的(30, 5)
3. 输入Transformer模型 → 提取32维特征向量
4. 与29维表格特征拼接 → 61维特征向量
5. 输入LightGBM进行最终预测
6. 输出：上涨/下跌/盘整的概率分布

Résumer

Les modèles Transformer ont révolutionné le monde de la finance, et leurs puissantes capacités de modélisation de séquences sont particulièrement adaptées à l’analyse de séquences chronologiques financières. De la quantification des transactions à la gestion des risques, de l’analyse de données alternatives à l’optimisation de portefeuilles, Transformer est en train de réinventer les domaines de la Fintech.

Cependant, une application réussie de Transformer nécessite une compréhension approfondie des spécificités des marchés financiers, y compris l’instabilité des données, le bruit des signaux et la complexité de la réglementation. À l’avenir, Transformer devrait avoir une plus grande valeur dans le domaine financier avec l’amélioration de l’interprétabilité des modèles, l’optimisation de l’efficacité du calcul et le développement d’architectures spécifiques au domaine financier.