Pada tahun 2017, sebuah makalah berjudul “Attention Is All You Need” muncul secara horizontal dan selama beberapa tahun berikutnya hingga saat ini telah menguasai seluruh bidang AI generatif. Dalam makalah yang bersejarah dan terobosan ini, delapan peneliti akademisi pertama kali mengusulkan arsitektur jaringan saraf Transformer, yang unik karena sepenuhnya didasarkan pada mekanisme perhatian, meninggalkan operasi siklus dan volume tradisional.

Aplikasi Transformer untuk Kuantitasitas Keuangan

Inovasi inti dari Transformer adalah mekanisme Self-Attention, yang mampu menangkap ketergantungan antara dua elemen dalam urutan arbitrer, tidak peduli seberapa jauh mereka berada di antara satu sama lain. Fitur ini membuatnya sangat cocok untuk menangani data urutan waktu keuangan dengan ketergantungan waktu yang kompleks.

Prediksi waktu

Metode prediksi urutan waktu keuangan tradisional (seperti ARIMA, LSTM) sulit untuk menangkap ketergantungan jangka panjang, dan Transformer sangat baik dalam hal ini.

Model multi faktor

Transformer dapat memproses ratusan faktor sekaligus, secara otomatis mempelajari hubungan non-linear yang kompleks antara faktor-faktor dan sifat perubahan waktu

• Analisis pentingnya faktor: mengidentifikasi faktor pendorong utama melalui bobot perhatian

• Paparan faktor dinamis: karakteristik perubahan waktu untuk menangkap efektivitas faktor

Manajemen Risiko

Pemodelan risiko pasar:

• VaR (value of risk) forecast: menangani hubungan non-linear dalam kondisi pasar ekstrim

• Prediksi volatilitas: menangkap efek agregat dan efek leverage

• Prediksi Relevansi: Pemodelan Relevansi Aset Dinamis

Optimalisasi portofolio

Transformer dapat menangani:

• Data multi-aset, multi-frekuensi

• Identifikasi dan Prediksi Kondisi Pasar

Pilihan Aset Berdasarkan Berat Perhatian Optimalisasi Eksekusi Transaksi:

• Pemodelan kejutan pasar: memprediksi dampak transaksi besar pada pasar

• Strategi Eksekusi Optimal: Meminimalkan Biaya Transaksi

Transaksi frekuensi tinggi dan struktur mikro pasar

Analisis buku pesanan terbatas Transformer mampu menangani data buku pesanan frekuensi tinggi dengan efisien:

• Perkiraan arus pesanan: Perkiraan ketidakseimbangan pesanan jual beli

• Analisis dampak harga: menilai dampak pesanan besar pada harga pasar

• Prediksi likuiditas: memprediksi perubahan kedalaman pasar

Deteksi Manipulasi Pasar

Dengan menganalisis pola urutan transaksi, dapat diidentifikasi:

• Spoofing (Pelapisan)

• Model Pump and Dump

• Manipulasi Pasar Lainnya

Pertempuran

Transformer khusus untuk menangani data berurutan OHLCV 5 dimensi (harga buka, harga tertinggi, harga terendah, harga tutup, volume transaksi):

• Input shape: ((batch_size, seq_len=30, feature_dim=5)

• Mengekstrak karakteristik waktu yang lebih kaya dengan mempelajari ketergantungan waktu dalam urutan

• Lebih baik menangkap pola pasar yang kompleks daripada indikator teknis sederhana

Definisi model Transformer (Kelas Transformer TimeSeries)

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)

Fungsi pemrosesan data

Mendapatkan Transformer input (scaler pita)

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)

Dapatkan Transformer input (scaler tetap)

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)

Kode Transformer dalam pelatihan

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

Kode Transformer dalam Prediksi Waktu Nyata

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)

Model disimpan dan dimuat

Simpan model 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)

Transformer model yang dimuat:

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)

Parameter Konfigurasi

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

Pengelolaan Nama Fitur

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 Workflow

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

Meringkaskan

Model Transformer telah membawa perubahan revolusioner di bidang keuangan, dengan kemampuan pemodelan sekuens yang kuat yang sangat cocok untuk analisis time series keuangan. Transformer sedang mengubah berbagai bidang Fintech mulai dari kuantitatif transaksi hingga manajemen risiko, dari analisis data alternatif hingga optimasi portofolio.

Namun, aplikasi Transformer yang sukses membutuhkan pemahaman yang mendalam tentang kekhususan pasar keuangan, termasuk ketidakstabilan data, kebisingan sinyal, dan kompleksitas peraturan. Di masa depan, Transformer diharapkan dapat memberikan nilai yang lebih besar di bidang keuangan dengan peningkatan interpretasi model, pengoptimalan efisiensi komputasi, dan pengembangan arsitektur khusus di bidang keuangan.