Em 2017, um artigo intitulado “Attention Is All You Need” (Attenção é tudo o que você precisa) foi publicado e, nos anos seguintes, dominou todo o campo da IA geradora. Neste artigo de marco e de grande importância, oito pesquisadores propuseram pela primeira vez a arquitetura da rede neural Transformer, que é única em se basear inteiramente no mecanismo de atenção, abandonando as operações tradicionais de ciclo e volume.

Transformer para aplicações financeiras de quantificação

A inovação central do Transformer é o mecanismo de auto-atenção, que é capaz de capturar a dependência entre dois elementos arbitrários na sequência, independentemente da distância entre eles. Esta característica torna-o especialmente adequado para processar dados de sequências financeiras com complexas dependências temporais.

Previsão de sequência temporal

Os métodos tradicionais de previsão de sequências de tempo financeiras (como ARIMA, LSTM) têm dificuldade em capturar dependências de longo prazo, e o Transformer é excelente nesse sentido.

Modelo multifator

O Transformer pode processar centenas de fatores ao mesmo tempo, aprendendo automaticamente as complexas relações não-lineares entre os fatores e as propriedades de transição

• Análise da importância dos fatores: Identificação dos fatores-chave através da ponderação da atenção

• Exposição a fatores dinâmicos: características de mudança de tempo para capturar a eficácia dos fatores

Gestão de Riscos

Modelagem de risco de mercado:

• Previsão VaR (valor de risco): trata de relações não lineares em condições de mercado extremas

• Previsão de volatilidade: captura de efeitos de agregação e de alavancagem de volatilidade

• Previsão de correlação: Modelagem de correlação de ativos dinâmicos

Otimização de portfólio

O Transformer é capaz de:

• Dados de múltiplos ativos e frequências

• Identificação e previsão do estado do mercado

Seleção de ativos com base no peso da atenção Optimização de Execução de Transações:

• Modelagem de choques de mercado: previsão do impacto de grandes transações no mercado

• Estratégias de execução ótimas: minimizar os custos de transação

Transações de alta frequência e a microestrutura do mercado

Análise de livros de pedidos limitados O Transformer é capaz de lidar com dados de livros de pedidos de alta frequência:

• Previsão de fluxo de pedidos: Previsão de desequilíbrio de pedidos de compra e venda

• Análise de choques de preços: avaliação do impacto de pedidos em grandes volumes nos preços de mercado

• Previsão de liquidez: previsão de mudanças na profundidade do mercado

Detecção de manipulação de mercado

Analisando os padrões de sequência de transações, identificamos:

• O que é o Spoofing?

• Pump and Dump (Método Pump e Dump)

• Outros tipos de manipulação de mercado

Batalhas reais

O Transformer é especializado em processar dados de sequência OHLCV de 5 dimensões (preço de abertura, preço máximo, preço mínimo, preço de fechamento, volume de transação):

• A forma de entrada: batch_size, seq_len=30, feature_dim=5

• Extrair características mais ricas da ordem temporal, aprendendo as dependências temporais na sequência

• Os indicadores técnicos são mais capazes de capturar padrões de mercado complexos.

Definição do modelo Transformer (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)

Funções de pré-processamento de dados

Obter a entrada do transformador (scaler de fita)

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)

Obter a entrada do transformador (scaler fixo)

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)

Código Transformer em treinamento

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()

Código do Transformer em previsão em tempo real

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)

Salvar e carregar o modelo

Guarde o modelo do 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)

Carregar o modelo 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)

Parâmetros de configuração

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"]  # 学习率

Gerenciamento de nomes de características

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个特征

Transformador de fluxo de trabalho

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

Resumir

O modelo Transformer está revolucionando o setor financeiro, e sua poderosa capacidade de modelagem de sequências é especialmente adequada para a análise de sequências de tempo financeiras. Desde a quantificação de transações até o gerenciamento de riscos e a análise de dados alternativos até a otimização de carteiras, o Transformer está reinventando todas as áreas da Fintech.

No entanto, a aplicação bem-sucedida do Transformer requer uma compreensão profunda das especificidades dos mercados financeiros, incluindo a instabilidade dos dados, o ruído dos sinais e a complexidade da regulamentação. No futuro, o Transformer deverá ter maior valor no setor financeiro com o aumento da interpretabilidade do modelo, a otimização da eficiência da computação e o desenvolvimento de arquiteturas específicas para o setor financeiro.