Розробка бота з ML/AI стратегією торгівлі

Розробка бота з ML/AI стратегією торговлі

ML в трейдингу — це не волшебна кнопка "profit". Це статистичний інструмент для нахождення паттернів у даних, які мають передбачувальну силу. Більшість спроб застосувати ML до торговлі провалюються через overfitting, lookahead bias або ігнорування transaction costs. Розповім як робити це правильно.

Чому ML в трейдингу складніше, ніж здається

Фундаментальні проблеми

Non-stationarity: ринки змінюються. Паттерн, який працював у 2020 році, може не працювати у 2024. Модель навчається на минулому, застосовується до майбутнього — яке по розподілу відрізняється від минулого.

Low signal-to-noise ratio: у фінансових даних співвідношення сигнал/шум вкрай низьке. Більшість паттернів, знайдених моделлю — шум, який був "значимим" у тренировочній вибірці случайно.

Lookahead bias: якщо при формуванні фічей случайно використовувалися дані з майбутнього — модель вивчає інформацію, якої в реальності немає. Backtest буде фантастичним, live trading — збитковим.

Overfitting: модель з 100 параметрами та 500 сделками в історії практично напевно переучена.

Правильний підхід

1. Чітка гіпотеза що саме предсказує модель та чому це працює
2. Коректне розділення даних train/validation/test без lookahead
3. Прості моделі як baseline перед складними
4. Transaction costs включені в backtest
5. Walk-forward validation

Feature Engineering

Типи фічей для криптотрейдингу

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

Критично важливо: всі індикатори, які "дивляться вперед" по часу, повинні бути зміщені на 1 крок назад:

# Неправильно: використовуємо close поточної свічки для генерації сигналу цієї ж свічки
signal = rsi > 70

# Правильно: сигнал поточної свічки використовує дані попередної
signal = rsi.shift(1) > 70

Вибір моделі

Gradient Boosting (XGBoost / LightGBM)

Найкращий baseline для структурованих даних. Швидко навчається, добре інтерпретується через feature importance, стійкий до выбросів.

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 для Sequence Modeling

Якщо гіпотеза в тому, що важлива послідовність подій (не просто значення індикатора, а його рух за N періодів), LSTM може бути корисним:

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:
        # x: (batch, sequence_len, features)
        lstm_out, _ = self.lstm(x)
        last_output = lstm_out[:, -1, :]  # берємо останній timestep
        return self.classifier(last_output)

На практиці LSTM рідко перевищує LightGBM на денних та часових даних. На тиковых даних або при роботі з послідовностями ордерів — може бути ефективніше.

Walk-forward Validation

Стандартний train/test split неприйнятний для часових рядів: модель навчається на даних, які йдуть після тестових — це lookahead bias.

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)

Walk-forward validation дає реалістичну оцінку продуктивності: модель ніколи не бачить тестових даних до моменту "real" застосування.

Інтеграція в торговий бот

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

Важливо: threshold 0.65 означає "вхожу тільки якщо модель з 65%+ впевненістю предсказує рост". Це зменшує кількість сделок, але покращує їх якість. Оптимальний threshold визначається на validation даних.

Ключові помилки

ML бот — це не запустити та забути. Ринки дрейфують, моделі деградують. Потребує моніторингу метрик моделі в live та переодичного ретрейну.