בוט ML למסחר בקריפטו: פיתוח אסטרטגיות עם LightGBM ו-LSTM

import pandas as pd
import numpy as np
from ta import trend, momentum, volatility

class FeatureEngineer:
    def generate_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """df содержит: open, high, low, close, volume"""
        features = pd.DataFrame(index=df.index)

        # === Технические индикаторы ===
        # Trend
        features['ema_9'] = trend.EMAIndicator(df.close, 9).ema_indicator()
        features['ema_21'] = trend.EMAIndicator(df.close, 21).ema_indicator()
        features['macd'] = trend.MACD(df.close).macd()
        features['macd_signal'] = trend.MACD(df.close).macd_signal()
        features['adx'] = trend.ADXIndicator(df.high, df.low, df.close).adx()

        # Momentum
        features['rsi_14'] = momentum.RSIIndicator(df.close, 14).rsi()
        features['stoch_k'] = momentum.StochasticOscillator(df.high, df.low, df.close).stoch()
        features['cci'] = momentum.CCIIndicator(df.high, df.low, df.close).cci()

        # Volatility
        features['atr'] = volatility.AverageTrueRange(df.high, df.low, df.close).average_true_range()
        features['bb_width'] = (
            volatility.BollingerBands(df.close).bollinger_hband() - volatility.BollingerBands(df.close).bollinger_lband()
        ) / df.close

        # === Price-derived features ===
        # Returns на разных горизонтах
        for period in [1, 3, 6, 12, 24]:
            features[f'return_{period}h'] = df.close.pct_change(period)

        # Расстояние от скользящих средних (нормализованное)
        for period in [20, 50, 200]:
            ma = df.close.rolling(period).mean()
            features[f'dist_ma_{period}'] = (df.close - ma) / ma

        # === Volume features ===
        features['volume_ratio'] = df.volume / df.volume.rolling(20).mean()
        features['obv'] = (np.sign(df.close.diff()) * df.volume).cumsum()
        features['obv_ratio'] = features['obv'] / features['obv'].rolling(20).mean()

        # === Market microstructure ===
        features['high_low_range'] = (df.high - df.low) / df.close
        features['close_position'] = (df.close - df.low) / (df.high - df.low + 1e-10)

        return features.dropna()

ניואנס חשוב: כל האינדיקטורים ש"מסתכלים קדימה" בזמן חייבים להיות מוזזים צעד אחד אחורה. האות של הנר הנוכחי משתמש בנתוני הנר הקודם—זה מבטל מבט קדימה.

איזה מודל לבחור: LightGBM או LSTM?

עבור נתונים מובנים, הבסיס הטוב ביותר הוא Gradient Boosting (LightGBM). הוא מתאמן מהר, ניתן לפרשנות טובה באמצעות חשיבות תכונות, ועמיד בפני חריגים. תיעוד LightGBM ממליץ עליו לסדרות עתיות עם תכונות רבות.

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit

class DirectionPredictor:
    def __init__(self, horizon: int = 4):
        self.horizon = horizon  # предсказываем направление через N свечей
        self.model = None
        self.feature_cols = None

    def prepare_target(self, df: pd.DataFrame) -> pd.Series:
        """Target: 1 если цена вырастет на X% за horizon периодов, иначе 0"""
        future_return = df.close.shift(-self.horizon) / df.close - 1
        threshold = 0.005  # 0.5%
        return (future_return > threshold).astype(int)

    def train(self, features: pd.DataFrame, prices: pd.DataFrame):
        y = self.prepare_target(prices)
        # Выравниваем индексы
        common_idx = features.index.intersection(y.dropna().index)
        X = features.loc[common_idx]
        y = y.loc[common_idx]
        # Walk-forward validation: обучаем на первых 70%, тестируем на последних 30%
        split = int(len(X) * 0.7)
        X_train, X_test = X.iloc[:split], X.iloc[split:]
        y_train, y_test = y.iloc[:split], y.iloc[split:]
        params = {
            'objective': 'binary',
            'metric': 'auc',
            'learning_rate': 0.05,
            'num_leaves': 31,
            'min_data_in_leaf': 50,
            'feature_fraction': 0.8,
            'bagging_fraction': 0.8,
            'bagging_freq': 5,
            'verbose': -1
        }
        train_data = lgb.Dataset(X_train, label=y_train)
        val_data = lgb.Dataset(X_test, label=y_test)
        self.model = lgb.train(
            params,
            train_data,
            valid_sets=[val_data],
            num_boost_round=500,
            callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)]
        )
        self.feature_cols = X.columns.tolist()

    def predict_proba(self, features: pd.DataFrame) -> float:
        X = features[self.feature_cols].iloc[-1:]
        return float(self.model.predict(X)[0])

LSTM למודל רצפים—אם ההשערה היא שרצף האירועים חשוב, LSTM יכול להיות יעיל יותר, אך על נתונים יומיים ושעתיים LightGBM לרוב מתפקד באופן דומה.

import torch
import torch.nn as nn

class PriceLSTM(nn.Module):
    def __init__(self, input_size: int, hidden_size: int = 64, num_layers: int = 2):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=0.2
        )
        self.classifier = nn.Sequential(
            nn.Linear(hidden_size, 32),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(32, 1),
            nn.Sigmoid()
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        lstm_out, _ = self.lstm(x)
        last_output = lstm_out[:, -1, :]
        return self.classifier(last_output)

אימות מתגלגל קדימה

חלוקת אימון/בדיקה רגילה אינה מקובלת עבור סדרות עתיות. אנו משתמשים בחלון נע:

def walk_forward_backtest(
    model_class, features: pd.DataFrame, prices: pd.DataFrame,
    train_window: int = 365, test_window: int = 30, step: int = 30
) -> pd.DataFrame:
    results = []
    n = len(features)
    for start in range(0, n - train_window - test_window, step):
        train_end = start + train_window
        test_end = train_end + test_window
        X_train = features.iloc[start:train_end]
        X_test = features.iloc[train_end:test_end]
        p_train = prices.iloc[start:train_end]
        p_test = prices.iloc[train_end:test_end]
        model = model_class()
        model.train(X_train, p_train)
        predictions = [model.predict_proba(X_test.iloc[:i+1]) for i in range(len(X_test))]
        period_results = simulate_trading(predictions, p_test)
        results.append(period_results)
    return pd.concat(results)

כיצד לשלב מודל ML בבוט מסחר

class MLTradingBot:
    def __init__(self, model: DirectionPredictor, threshold: float = 0.65):
        self.model = model
        self.threshold = threshold

    async def on_candle(self, candle: Candle):
        features = self.feature_eng.update(candle)
        prob_up = self.model.predict_proba(features)
        if prob_up > self.threshold and not self.has_position():
            await self.open_long()
        elif prob_up < (1 - self.threshold) and not self.has_position():
            await self.open_short()
        elif self.has_position():
            current_side = self.position.side
            if current_side == 'long' and prob_up < 0.5:
                await self.close_position("model_signal_weak")

סף של 0.65 אומר "היכנס רק כשהמודל בטוח ב-65% ומעלה." זה מפחית את מספר העסקאות אך משפר את איכותן. בנוסף, אנו מיישמים ניטור סחיפה של המודל ואימון מחדש אוטומטי כל 30–90 יום.

שלבי פיתוח בוט ML

מה כלול בפיתוח

• גיבוש השערת מסחר ואימות על נתונים היסטוריים
• פיתוח צינור תכונות (אינדיקטורים טכניים, מדדים על-רשת, חוסר איזון בספר ההזמנות)
• מודל בסיס ואימות מתגלגל קדימה כולל עלויות עסקה (0.1-0.2% לעסקה)
• בחירה ואימון של המודל הסופי (LightGBM, LSTM, transformers)
• שילוב בורסה באמצעות WebSocket/REST API (Binance, Bybit, OKX וכו')
• ניטור סחיפה של המודל ואימון מחדש אוטומטי כל 30-90 יום
• תיעוד ארכיטקטורה והכשרת צוות
• תמיכת אחריות ל-3 חודשים

טעויות נפוצות במסחר מבוסס ML

בוט ML אינו פתרון של "הגדר ושכח". השווקים נסחפים, המודלים מתדרדרים. זה דורש ניטור חי של מדדי המודל ואימון מחדש תקופתי. אך עם גישה נכונה, אסטרטגיית ML מספקת יתרון אמיתי.

רוצה ייעוץ על ארכיטקטורת בוט ML לאסטרטגיה שלך? צור קשר—נעריך את המשימה שלך ונציע את הפתרון האופטימלי. הזמן פיתוח של אסטרטגיה אדפטיבית עם ערבות ליישום טכני.

קישורים שימושיים

• אימות מתגלגל קדימה — שיטת אימות לסדרות עתיות
• תיעוד LightGBM — תיעוד רשמי של הספרייה