২০১৭ সালে, “Attention Is All You Need” নামে একটি গবেষণাপত্র প্রকাশিত হয়েছিল, এবং পরবর্তী কয়েক বছর ধরে এখন পর্যন্ত পুরো জেনারেটর এআই ক্ষেত্রকে নিয়ন্ত্রণে রেখেছে। এই মাইলফলক এবং যুগান্তকারী গবেষণাপত্রে, আটজন গবেষক প্রথমবারের মতো ট্রান্সফর্মারের স্নায়ুসংক্রান্ত নেটওয়ার্ক আর্কিটেকচারটি উপস্থাপন করেছিলেন, যা সম্পূর্ণরূপে মনোযোগের উপর ভিত্তি করে, প্রচলিত চক্র এবং ক্লিপ অপারেশনগুলিকে বাদ দিয়েছিল। স্ব-মনোযোগের মাধ্যমে, ট্রান্সফর্মার ইনপুট ক্রমের মধ্যে দীর্ঘ দূরত্বের নির্ভরতা কার্যকরভাবে ক্যাপচার করতে সক্ষম হয়েছিল, যা মডেলটিকে দীর্ঘ পাঠ্য প্রক্রিয়া করার সময় আরও দক্ষ এবং নির্ভুল করে তোলে।

ট্রান্সফরমার ফাইন্যান্সিয়াল কোয়ান্টামাইজেশনের জন্য

ট্রান্সফরমারের মূল উদ্ভাবন হল স্ব-মনোযোগ ব্যবস্থা (Self-Attention), যা একটি ক্রমের মধ্যে যে কোনও দুটি উপাদানের মধ্যে নির্ভরতা ধরতে সক্ষম, তাদের মধ্যে দূরত্ব যাই হোক না কেন। এই বৈশিষ্ট্যটি এটিকে জটিল সময় নির্ভরতার সাথে আর্থিক টাইম সিকোয়েন্স ডেটা পরিচালনার জন্য বিশেষভাবে উপযুক্ত করে তোলে।

টাইম সিকোয়েন্স পূর্বাভাস

ঐতিহ্যগত আর্থিক টাইম সিকোয়েন্স ভবিষ্যদ্বাণী পদ্ধতি (যেমন ARIMA, LSTM) দীর্ঘমেয়াদী নির্ভরতা ক্যাপচার করতে অসুবিধা হয়, এবং ট্রান্সফর্মার এই ক্ষেত্রে ভাল কাজ করে

মাল্টিফ্যাক্টর মডেল

ট্রান্সফরমার একসাথে শত শত ফ্যাক্টরকে ম্যানেজ করতে পারে, ফ্যাক্টরগুলির মধ্যে জটিল অ-রৈখিক সম্পর্ক এবং সময় পরিবর্তন বৈশিষ্ট্যগুলি স্বয়ংক্রিয়ভাবে শিখতে পারে

• ফ্যাক্টর গুরুত্ব বিশ্লেষণঃ মনোযোগের ওজন দ্বারা মূল চালিত ফ্যাক্টরগুলি সনাক্ত করা

• ডায়নামিক ফ্যাক্টর এক্সপোজারঃ ফ্যাক্টরের কার্যকারিতা ক্যাপচার করার সময় পরিবর্তনশীল বৈশিষ্ট্য

ঝুঁকি ব্যবস্থাপনা

মার্কেট রিস্ক মডেলিংঃ

• VaR (ঝুঁকি-মূল্য) পূর্বাভাসঃ চরম বাজার অবস্থার অধীনে নন-লাইনার সম্পর্কগুলি মোকাবেলা করা

• ওঠানামা পূর্বাভাসঃ ওঠানামা সংগ্রহের প্রভাব এবং লিভারেজ প্রভাব ক্যাপচার

• প্রাসঙ্গিকতা পূর্বাভাসঃ গতিশীল সম্পদ প্রাসঙ্গিকতা মডেলিং

পোর্টফোলিও অপ্টিমাইজেশন

ট্রান্সফরমারঃ

• মাল্টি-অ্যাসেট, মাল্টি-ফ্রিকোয়েন্সি ডেটা

• বাজারের অবস্থা সনাক্তকরণ এবং পূর্বাভাস

মনোযোগের উপর ভিত্তি করে সম্পদ নির্বাচন ট্রেড এক্সিকিউশন অপ্টিমাইজেশানঃ

• মার্কেট শক মডেলিংঃ বড় লেনদেনের প্রভাবের পূর্বাভাস

• সর্বোত্তম কার্যকরকরণ কৌশলঃ লেনদেনের খরচ কমানো

হাই-ফ্রিকোয়েন্সি ট্রেডিং এবং মার্কেট মাইক্রোস্ট্রাকচার

লিমিটেড অর্ডার বুক বিশ্লেষণ ট্রান্সফরমার উচ্চ-ফ্রিকোয়েন্সি অর্ডার বুকের ডেটা কার্যকরভাবে পরিচালনা করতে পারেঃ

• অর্ডার প্রবাহের পূর্বাভাসঃ ক্রয় ও বিক্রয় আদেশের ভারসাম্যহীনতার পূর্বাভাস

• মূল্য শক বিশ্লেষণঃ বাজারের দামের উপর বড় আকারের আদেশের প্রভাব মূল্যায়ন করা

• তরলতা পূর্বাভাসঃ বাজার গভীরতার পরিবর্তন পূর্বাভাস

বাজার ম্যানিপুলেশন পরীক্ষা

ট্রেডিং সিকোয়েন্স প্যাটার্ন বিশ্লেষণ করে, নিম্নলিখিতগুলি চিহ্নিত করুনঃ

• স্পুফিং

• পাম্প এবং ডাম্প মোড

• অন্যান্য বাজার ম্যানিপুলেশন

যুদ্ধ

ট্রান্সফরমার বিশেষভাবে 5 ডি ওএইচএলসিভি টাইমলাইন ডেটা প্রক্রিয়াকরণ করে (খোলার মূল্য, সর্বোচ্চ মূল্য, সর্বনিম্ন মূল্য, বন্ধ মূল্য, লেনদেনের পরিমাণ):

• ইনপুট আকৃতিঃ ((batch_size, seq_len=30, feature_dim=5)

• ধারাবাহিকতার মধ্যে সময় নির্ভরতা শেখার মাধ্যমে আরও সমৃদ্ধ সময়সূচী বৈশিষ্ট্যগুলি বের করা

• সহজ প্রযুক্তিগত সূচকগুলির চেয়ে জটিল বাজার মডেলগুলিকে আরও ভালভাবে ধরা যায়

ট্রান্সফরমার মডেল সংজ্ঞা (টাইমসেরিজ ট্রান্সফরমার ক্লাস)

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. 输出：上涨/下跌/盘整的概率分布

সারসংক্ষেপ

ট্রান্সফর্মার মডেলটি আর্থিক ক্ষেত্রে বিপ্লব এনেছে, এর শক্তিশালী ক্রমিক মডেলিং ক্ষমতা বিশেষত আর্থিক সময় ক্রম বিশ্লেষণের জন্য উপযুক্ত। ট্রেডিংয়ের পরিমাণ থেকে ঝুঁকি পরিচালনা, বিকল্প ডেটা বিশ্লেষণ থেকে পোর্টফোলিও অপ্টিমাইজেশান পর্যন্ত ট্রান্সফর্মার ফিনান্সিয়াল প্রযুক্তির বিভিন্ন ক্ষেত্রকে নতুন করে তৈরি করছে।

যাইহোক, সফলভাবে ট্রান্সফরমার প্রয়োগের জন্য আর্থিক বাজারের বিশেষত্বগুলি গভীরভাবে বোঝার প্রয়োজন, যার মধ্যে রয়েছে ডেটার অস্থিরতা, সংকেতের গোলমাল এবং নিয়ন্ত্রনের জটিলতা। ভবিষ্যতে, মডেলের ব্যাখ্যাযোগ্যতার উন্নতি, গণনার দক্ষতার অপ্টিমাইজেশন এবং আর্থিক ক্ষেত্রে বিশেষ কাঠামোর বিকাশের সাথে সাথে, ট্রান্সফরমার আর্থিক ক্ষেত্রে আরও বেশি মূল্যবান হওয়ার সম্ভাবনা রয়েছে।