2017년, “Attention Is All You Need”라는 논문이 발표되었고, 그 후 몇 년 동안 현재까지 생성형 AI 분야를 지배하고 있다. 이 획기적이고 획기적인 논문에서 8명의 연구자들은 Transformer의 신경망 구조를 처음으로 제안했다. Transformer의 특징은 완전히 주의 메커니즘에 기반하고, 전통적인 회전 및 축적 작업을 포기한다는 것이다.

트랜스포머의 금융량화에 대한 응용

트랜스포머의 핵심 혁신은 자기주의 메커니즘 (Self-Attention) 으로, 시퀀스의 임의의 두 요소 사이의 의존 관계를 포착할 수 있다. 이 특징은 복잡한 시간 의존 관계를 가진 금융 시퀀스 데이터를 처리하는 데 특히 적합하다.

시간계 예측

전통적인 금융 시간계열 예측 방법들 (ARIMA, LSTM 등) 은 장기적인 의존성을 파악하기 어렵고, 트랜스포머는 이 부분에서 우수합니다.

다인자 모델

트랜스포머는 수백 개의 인자를 동시에 처리할 수 있으며, 인자 간의 복잡한 비선형 관계와 시간 변동 특성을 자동으로 학습한다.

• 요인 중요성 분석: 주의 중량으로 핵심 동기를 식별

• 동적 인자 노출: 인자의 유효성을 포착하는 시간의 변화 특징

위험 관리

시장의 위험 모형화:

• VaR (위험의 가치) 예측: 극한 시장 조건의 비선형 관계를 다루는

• 변동률 예측: 변동률 집적 효과와 레버리지 효과를 포착

• 연관성 예측: 동적 자산 연관성 모델링

포트폴리오 최적화

트랜스포머는

• 다자산, 다주파 데이터

• 시장 상태를 인식하고 예측

집중력을 바탕으로 한 자산 선택 거래 실행 최적화:

• 시장 충격 모델링: 대규모 거래가 시장에 미치는 영향을 예측하는 방법

• 최적의 실행 전략: 거래 비용 최소화

고주파 거래와 시장 미시 구조

제한 가격 주문 서적 분석 트랜스포머는 높은 주파수의 주문서치 데이터를 효율적으로 처리할 수 있습니다.

• 주문 흐름 예측: 구매/판매 주문의 불균형 예측

• 가격 충격 분석: 시장 가격에 대량 주문의 영향을 평가

• 유동성 예측: 시장 깊이의 변화를 예측하는 것

시장 조작 탐지

트랜잭션 시퀀스 패턴을 분석하여 다음을 식별합니다.

• 스푸핑 (Spoofing)

• 펌프와 덤프 방식

• 다른 시장 조작 행위

실전

트랜스포머는 5차원 OHLCV 시간 순서 데이터를 처리합니다. (개시 가격, 최고 가격, 최저 가격, 폐시 가격, 거래량):

• 입력 형태: ((batch_size, seq_len=30, feature_dim=5)

• 시퀀스의 시간 의존성을 학습하여 더 풍부한 시간 순서 특성을 추출합니다.

• 단순한 기술적인 지표보다 복잡한 시장 패턴을 포착할 수 있습니다.

트랜스포머 모델 정의 (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)

데이터 전처리 함수

트랜스포머 입력을 얻습니다.

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)

트랜스포머 입력을 얻습니다.

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)

트랜스포머 관련 코드가 훈련 중

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

실시간 예측의 트랜스포머 코드

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)

모델 저장 및 로드

트랜스포머 모델 저장:

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)

트랜스포머 모델 로드:

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)

구성 변수

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

특징 이름 관리

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

트랜스포머 작업 프로세스

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

요약하다

트랜스포머 모델은 금융 분야에 혁명적인 변화를 가져오고 있으며, 강력한 시퀀스 모델링 능력은 금융 시간 순서 분석에 특히 적합합니다. 트랜스포머는 거래의 수량화에서 위험 관리, 대안 데이터 분석에서 포트폴리오 최적화까지 다양한 분야를 재구성하고 있습니다.

그러나, Transformer를 성공적으로 적용하려면 데이터의 불안정성, 신호의 소음성, 규제의 복잡성 등 금융 시장의 특수성을 깊이 이해해야합니다. 미래에는 모델의 해석성이 향상되고, 계산의 효율성이 최적화되고, 금융 분야에 특화된 구조가 개발됨에 따라 Transformer는 금융 분야에서 더 큰 가치를 발휘할 것으로 예상됩니다.