In diesem wegweisenden und bahnbrechenden Artikel schlugen acht Wissenschaftler erstmals eine neuronale Netzwerk-Architektur namens Transformer vor, die sich vollständig auf die Aufmerksamkeitsmechanismen stützt und die herkömmlichen Kreislauf- und Volumenoperationen ablehnt. Durch die Selbst-Aufmerksamkeitsmechanismen (self-attention) kann Transformer die Abhängigkeiten in der Länge der Eingangsreihen effektiv erfassen, wodurch das Modell bei der Verarbeitung von langem Text effizienter und genauer ist. Die Multi-Head-Aufmerksamkeitsmechanismen (multi-head attention) verbessern die Expressivität des Modells weiter und ermöglichen es, sich gleichzeitig auf verschiedene, kompliziertere semantische Beziehungen in den Eingangsreihen zu konzentrieren.

Transformer für die Anwendung von Finanzquantifizierung

Die Kerninnovation des Transformers liegt in der Selbst-Attention-Methode, die die Abhängigkeiten zwischen zwei beliebigen Elementen in einer Sequenz erfasst, unabhängig davon, wie weit sie voneinander entfernt sind. Diese Eigenschaft macht es besonders geeignet, Finanz-Zeitreihendaten mit komplexen Zeitabhängigkeiten zu verarbeiten.

Zeitreihenvorhersage

Herkömmliche Finanztechniken (wie ARIMA, LSTM) haben Schwierigkeiten, langfristige Abhängigkeiten zu erfassen, und Transformer ist in dieser Hinsicht hervorragend.

Mehrere Faktoren

Transformers können Hunderte von Faktoren gleichzeitig verarbeiten und lernen automatisch komplexe nichtlineare Beziehungen und zeitliche Eigenschaften zwischen Faktoren

• Factor-Importance-Analyse: Identifizierung von Schlüsselfaktoren durch Aufmerksamkeitsgewichtung

• Dynamische Faktor-Exposition: Zeitechnische Merkmale, um die Effektivität von Faktoren zu erfassen

Risikomanagement

Modellierung von Marktrisiken:

• VaR (Value-of-Risk) -Vorhersage: Nichtlineare Beziehungen unter extremen Marktbedingungen behandelt

• Volatilitätsprognose: Erfassung der Effekte von Volatilitätsaggregation und Leverage

• Relevanzprognose: Modellierung der Relevanz dynamischer Assets

Optimierung des Portfolios

Transformer ist in der Lage:

• Mehrwertsteuer- und Mehrfrequenzdaten

• Identifizierung und Vorhersage von Marktzuständen

Vermögenswahl basierend auf Aufmerksamkeitsgewicht
Optimierung der Transaktionsdurchführung:

• Modellierung von Marktschocks: Vorhersage der Auswirkungen von Großhandelsgeschäften auf die Märkte

• Optimale Ausführungsstrategien: Minimierung der Transaktionskosten

Hochfrequenzhandel und die Mikrostruktur des Marktes

Analyse der limitierten Bestellbücher
Transformer ist in der Lage, mit hochfrequenten Auftragsbuchdaten zu arbeiten:

• Auftragsflussprognose: Ungleichgewichte bei Kauf- und Verkaufsaufträgen

• Preis-Schock-Analyse: Bewertung der Auswirkungen von Großbestellungen auf die Marktpreise

• Liquiditätsprognosen: Veränderungen der Markttiefe

Marktmanipulation aufgedeckt

Durch die Analyse von Transaktionssequenzmustern identifiziert:

• Das ist eine Art "Spoofing".

• Pump and Dump-Methode

• Weitere Marktmanipulationen

Der Kampf

Transformer verarbeitet 5-dimensionale OHLCV-Zeitfolge-Daten (Eröffnungspreis, Höchstpreis, Tiefpreis, Abschlusspreis, Transaktionsvolumen):

• Eingabeform: ((batch_size, seq_len=30, feature_dim=5)

• Reichere Zeitenmerkmale durch das Lernen von zeitlichen Abhängigkeiten in der Sequenz

• Komplexe Marktmuster besser erfasst als einfache technische Indikatoren

Transformer-Modelldefinition (TimeSeriesTransformer-Klasse)

            
            
            
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)

Datenvorbereitungsfunktion

Transformator-Eingaben erhalten

            
            
            
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)

Transformer-Eingaben erhalten (Fest-Scaler)

            
            
            
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)

Transformer-Codes im Training

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

Transformer-Code in Echtzeit-Vorhersagen

            
            
            
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)

Modell gespeichert und geladen

Das Transformer-Modell wurde gespeichert:

            
            
            
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)

Das Transformer-Modell wird geladen:

            
            
            
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)

Konfigurationsparameter

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

Verwaltung von Feature Names

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

Transformer-Arbeitsabläufe

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

Zusammenfassen

Das Transformer-Modell revolutioniert die Finanzwelt. Seine leistungsstarke Serienmodellierung eignet sich besonders für die Analyse von Finanz-Zeitreihen. Transformer ist dabei, alle Bereiche der Fintech zu verändern, von der Quantifizierung von Transaktionen bis zum Risikomanagement, von der Alternativdatenanalyse bis zur Portfoliooptimierung.

Die erfolgreiche Anwendung von Transformer erfordert jedoch ein tiefes Verständnis der Besonderheiten der Finanzmärkte, einschließlich der Instabilität der Daten, der Lärmbelästigung der Signale und der Komplexität der Regulierung. In der Zukunft wird Transformer mit der Verbesserung der Modellinterpretationsfähigkeit, der Optimierung der Rechenleistung und der Entwicklung spezifischer Architekturen für den Finanzsektor einen größeren Wert in der Finanzwelt haben.