En 2017, se publicó un artículo titulado “Attention Is All You Need” que dominó el campo de la IA generativa en los años siguientes hasta el momento. En este artículo de importancia histórica y revolucionaria, ocho académicos propusieron por primera vez la arquitectura de la red neuronal Transformer, que es única en que se basa exclusivamente en el mecanismo de atención, renunciando a las operaciones tradicionales de ciclo y acumulación.

Transformer para la aplicación de la cuantificación financiera

La innovación central de Transformer radica en el mecanismo de autoatención, que es capaz de capturar la dependencia entre dos elementos arbitrarios de la secuencia, independientemente de la distancia entre ellos. Esta característica lo hace especialmente adecuado para procesar datos de secuencias de tiempo financieras con dependencias de tiempo complejas.

Pronóstico de la secuencia temporal

Los métodos tradicionales de predicción de secuencias de tiempo financieras (como ARIMA, LSTM) tienen dificultades para capturar la dependencia a largo plazo, y Transformer es excelente en esto.

Modelo multifactorial

Transformer puede procesar cientos de factores al mismo tiempo, aprendiendo automáticamente las complejas relaciones no lineales y las propiedades de cambio de tiempo entre los factores

• Análisis de la importancia de los factores: identificación de los factores impulsores clave mediante el peso de la atención

• Exposición a factores dinámicos: características cambiantes del tiempo para capturar la eficacia de los factores

Gestión de riesgos

Modelado de riesgo de mercado:

• VaR (valor de riesgo) pronóstico: tratamiento de relaciones no lineales en condiciones extremas de mercado

• Pronóstico de la volatilidad: captura de los efectos de agregación y de la influencia de la volatilidad

• Pronóstico de correlación: Modelado de correlación de activos dinámicos

Optimización de la cartera

Transformer es capaz de:

• Datos de múltiples activos y frecuencias

• Identificación y predicción del estado del mercado

Selección de activos basados en el peso de la atención Optimización de la ejecución de transacciones:

• Modelado de impactos en el mercado: predicción del impacto de las grandes transacciones en el mercado

• Estrategias de ejecución óptima: minimizar los costos de las transacciones

Las operaciones de alta frecuencia y la microestructura del mercado

Análisis de la cartera de pedidos Transformer es capaz de procesar datos de libretas de pedidos de alta frecuencia de manera eficiente:

• Pronóstico de flujo de pedidos: pronóstico de desequilibrio de pedidos de compra y venta

• Análisis de impacto en los precios: evaluar el impacto de los pedidos de gran volumen en los precios del mercado

• Pronóstico de liquidez: predicción de cambios en la profundidad del mercado

Detección de manipulación de mercado

A través del análisis de patrones de secuencias de transacciones, se identifican:

• El contenido de la página web de Facebook de la empresa es un engaño.

• El modelo de Pump and Dump

• Otras prácticas de manipulación de mercado

La guerra en acción

Transformer se especializa en el procesamiento de datos de orden temporal en 5 dimensiones de OHLCV (precio de apertura, precio máximo, precio mínimo, precio de cierre, volumen de transacción):

• La forma de entrada es: ((batch_size, seq_len=30, feature_dim=5)

• Extraer características de la secuencia de tiempo más ricas mediante el aprendizaje de la dependencia de tiempo en la secuencia

• Un modelo de mercado más complejo que un simple indicador técnico

Definición del modelo de transformador (clase de transformadores de serie de tiempo)

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)

Funciones de procesamiento de datos

Obtener las entradas del transformador (scaler de banda)

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)

Obtener la entrada del transformador (scaler fijo)

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 de Transformers en el entrenamiento

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 Transformer en predicción en tiempo 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)

Salvando y cargando modelos

Para guardar el modelo de 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)

El modelo de Transformer se cargará en:

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 configuración

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

Administración de nombres 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个特征

El flujo de trabajo de Transformer

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

Resumir

El modelo Transformer ha revolucionado el mundo de las finanzas, y sus potentes capacidades de modelado de secuencias son especialmente adecuadas para el análisis de secuencias de tiempo financieras. Desde la cuantificación de transacciones hasta la gestión de riesgos, y desde el análisis de datos alternativos hasta la optimización de carteras, Transformer está reinventando todos los campos de la Fintech.

Sin embargo, la aplicación exitosa de Transformer requiere una comprensión profunda de las peculiaridades de los mercados financieros, incluida la inestabilidad de los datos, el ruido de las señales y la complejidad de la regulación. En el futuro, Transformer podría tener un mayor valor en el sector financiero con la mejora de la interpretabilidad de los modelos, la optimización de la eficiencia de la computación y el desarrollo de una arquitectura específica para el sector financiero.