Разработка бота с 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 и переодический ретрейн.