Pada tahun 2017, sebuah makalah berjudul “Attention Is All You Need” muncul secara mendadak, dan dalam beberapa tahun yang akan datang hingga kini telah menguasai seluruh bidang AI generatif. Dalam makalah yang bersejarah dan terobosan ini, 8 penyelidik akademik pertama kali mengemukakan seni bina rangkaian saraf Transformer, yang unik kerana sepenuhnya berdasarkan mekanisme perhatian, meninggalkan operasi kitaran dan pengumpulan tradisional.

Aplikasi Transformer untuk Kuantitifikasi Kewangan

Inovasi utama Transformer adalah mekanisme perhatian diri (Self-Attention), yang dapat menangkap hubungan pergantungan antara dua elemen dalam urutan, tidak kira berapa jauh jarak antara mereka. Ciri ini menjadikannya sangat sesuai untuk menangani data urutan masa kewangan yang mempunyai hubungan pergantungan masa yang kompleks.

Ramalan siri masa

Kaedah ramalan siri masa kewangan tradisional (seperti ARIMA, LSTM) sukar untuk menangkap ketergantungan jangka panjang, dan Transformer sangat baik dalam hal ini

Model multi-faktor

Transformer boleh memproses beratus-ratus faktor pada masa yang sama, secara automatik mempelajari hubungan bukan linear yang rumit antara faktor dan sifat perubahan masa

• Analisis kepentingan faktor: mengenal pasti faktor penggerak utama dengan berat perhatian

• Pendedahan faktor dinamik: ciri perubahan masa untuk menangkap keberkesanan faktor

Pengurusan Risiko

Model risiko pasaran:

• Ramalan VaR: menangani hubungan bukan linear dalam keadaan pasaran yang melampau

• Ramalan turun naik: menangkap kesan pengumpulan turun naik dan kesan pengaruh

• Ramalan keserasian: Pemodelan keserasian aset dinamik

Optimumkan portfolio

Transformer boleh mengendalikan:

• Data pelbagai aset, pelbagai frekuensi

• Pengiktirafan dan ramalan keadaan pasaran

Pilihan Aset Berasaskan Berat Perhatian Pengoptimuman pelaksanaan transaksi:

• Pemodelan kejutan pasaran: meramalkan kesan perdagangan besar ke atas pasaran

• Strategi pelaksanaan optimum: mengurangkan kos transaksi

Perdagangan frekuensi tinggi dan struktur mikro pasaran

Analisis buku pesanan terhad Transformer mampu menangani data buku pesanan frekuensi tinggi dengan cekap:

• Ramalan aliran pesanan: ramalan ketidakseimbangan pesanan beli dan jual

• Analisis kejutan harga: menilai kesan pesanan besar terhadap harga pasaran

• Ramalan kecairan: meramalkan perubahan kedalaman pasaran

Pengesanan Manipulasi Pasaran

Dengan menganalisis corak urutan transaksi, ia mengenal pasti:

• Perbuatan Spoofing

• Pump and Dump

• Tindakan lain untuk memanipulasi pasaran

Pertempuran

Transformer mengkhususkan diri untuk memproses data masa OHLCV 5 dimensi ((harga pembukaan, harga tertinggi, harga terendah, harga penutupan, jumlah transaksi):

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

• Mengekstrak ciri masa yang lebih kaya dengan mempelajari ketergantungan masa dalam urutan

• Lebih mudah untuk menangkap corak pasaran yang rumit daripada petunjuk teknikal yang mudah

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 pemprosesan data

Dapatkan input Transformer (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)

Mendapatkan input Transformer (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)

Kod berkaitan Transformer dalam latihan

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

Kod Transformer dalam ramalan masa 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)

Simpan dan muat naik model

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)

Muat naik model 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)

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

Pengurusan nama ciri

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

Proses kerja Transformer

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

ringkaskan

Model Transformer membawa perubahan revolusioner dalam bidang kewangan, dengan keupayaan pemodelan sekuens yang kuat yang sangat sesuai untuk analisis urutan masa kewangan. Dari kuantiti perdagangan hingga pengurusan risiko, dari analisis data alternatif hingga pengoptimuman portfolio, Transformer mengubah pelbagai bidang Fintech.

Walau bagaimanapun, penerapan Transformer yang berjaya memerlukan pemahaman yang mendalam mengenai kekhususan pasaran kewangan, termasuk ketidakstabilan data, kebisingan isyarat, dan kerumitan peraturan. Di masa depan, Transformer dijangka dapat memberikan nilai yang lebih besar dalam bidang kewangan dengan peningkatan keterangan model, pengoptimuman kecekapan pengiraan, dan pembangunan struktur khusus dalam bidang kewangan.