Разработка бота с кастомной торговой стратегией

Разработка бота с кастомной торговой стратегией

Кастомный торговый бот — это реализация конкретной торговой идеи в виде автоматизированного алгоритма. В отличие от готовых DCA или grid ботов, кастомная стратегия предполагает уникальную логику входа/выхода, специфические условия и параметры. Разберём как это строится от идеи до production.

Анатомия торговой стратегии

Компоненты стратегии

Любая торговая стратегия состоит из:

1. Signal generation: когда входить в позицию (условия открытия)
2. Position sizing: сколько вкладывать
3. Exit logic: когда и как закрывать позицию
4. Risk management: stoploss, максимальные потери, drawdown limits

class TradingStrategy:
    """Базовый интерфейс кастомной стратегии"""

    def generate_signal(self, market_data: MarketData) -> Signal:
        """Возвращает BUY, SELL или HOLD"""
        raise NotImplementedError

    def calculate_position_size(
        self,
        signal: Signal,
        account_balance: Decimal,
        current_price: Decimal
    ) -> Decimal:
        """Размер позиции в базовой валюте"""
        raise NotImplementedError

    def should_exit(
        self,
        position: Position,
        market_data: MarketData
    ) -> ExitReason | None:
        """Возвращает причину выхода или None"""
        raise NotImplementedError

Пример: пробой уровня с подтверждением

Стратегия покупки при пробое локального максимума с подтверждением объёмом:

class BreakoutStrategy(TradingStrategy):
    def __init__(self, lookback: int = 20, volume_multiplier: float = 1.5):
        self.lookback = lookback
        self.volume_multiplier = volume_multiplier

    def generate_signal(self, data: MarketData) -> Signal:
        closes = data.close[-self.lookback:]
        volumes = data.volume[-self.lookback:]

        resistance = max(closes[:-1])  # максимум без последней свечи
        current_close = closes[-1]
        current_volume = volumes[-1]
        avg_volume = sum(volumes[:-1]) / len(volumes[:-1])

        # Условие пробоя: цена пробила resistance + объём выше среднего
        if (current_close > resistance and
                current_volume > avg_volume * self.volume_multiplier):
            return Signal.BUY

        # Пробой вниз — шорт
        support = min(closes[:-1])
        if (current_close < support and
                current_volume > avg_volume * self.volume_multiplier):
            return Signal.SELL

        return Signal.HOLD

Position sizing: Kelly Criterion и Fixed Fraction

Размер позиции — один из важнейших элементов стратегии. Слишком большой → один убыток уничтожит счёт. Слишком маленький → доходность мала.

Fixed Fraction (% от капитала): самый простой подход:

def fixed_fraction_sizing(balance: Decimal, risk_percent: float = 2.0) -> Decimal:
    return balance * Decimal(str(risk_percent / 100))

Kelly Criterion: оптимизация роста капитала:

Kelly % = (Win Rate × Avg Win / Avg Loss - (1 - Win Rate)) / (Avg Win / Avg Loss)

def kelly_sizing(win_rate: float, avg_win: float, avg_loss: float) -> float:
    profit_ratio = avg_win / avg_loss
    kelly = (win_rate * profit_ratio - (1 - win_rate)) / profit_ratio
    # Используем половину Kelly для снижения риска (Half-Kelly)
    return max(0, kelly * 0.5)

На практике используют Half-Kelly или Quarter-Kelly: полная Kelly формула агрессивна и требует точных исторических данных.

Event-driven архитектура бота

Основной цикл

class TradingBot:
    def __init__(self, strategy: TradingStrategy, exchange_client, risk_manager):
        self.strategy = strategy
        self.exchange = exchange_client
        self.risk = risk_manager
        self.position: Position | None = None

    async def run(self):
        async for candle in self.exchange.subscribe_candles(self.symbol, '1h'):
            await self.on_candle(candle)

    async def on_candle(self, candle: Candle):
        # Обновляем рыночные данные
        self.market_data.append(candle)

        # Проверяем текущую позицию
        if self.position:
            exit_reason = self.strategy.should_exit(self.position, self.market_data)
            if exit_reason:
                await self.close_position(exit_reason)
                return

        # Проверяем новые сигналы
        if not self.position:
            signal = self.strategy.generate_signal(self.market_data)
            if signal != Signal.HOLD:
                await self.open_position(signal)

    async def open_position(self, signal: Signal):
        # Risk checks
        if not self.risk.can_open_position():
            logger.info("Risk manager blocked new position")
            return

        balance = await self.exchange.get_balance()
        price = self.market_data.last_close
        size = self.strategy.calculate_position_size(signal, balance, price)

        order = await self.exchange.place_order(
            side='buy' if signal == Signal.BUY else 'sell',
            size=size,
            order_type='market'
        )

        self.position = Position(
            entry_price=order.fill_price,
            size=size,
            side=signal,
            opened_at=datetime.utcnow()
        )
        logger.info(f"Opened {signal} position: {size} @ {order.fill_price}")

Risk management модуль

class RiskManager:
    def __init__(self, config: RiskConfig):
        self.max_daily_loss = config.max_daily_loss_percent  # % от баланса
        self.max_drawdown = config.max_drawdown_percent
        self.max_consecutive_losses = config.max_consecutive_losses

        self.daily_pnl = Decimal(0)
        self.peak_balance = None
        self.consecutive_losses = 0

    def can_open_position(self) -> bool:
        # Дневной лимит потерь
        if self.daily_pnl < -self.max_daily_loss:
            logger.warning("Daily loss limit reached")
            return False

        # Максимальная просадка
        if self.peak_balance:
            drawdown = (self.peak_balance - self.current_balance) / self.peak_balance
            if drawdown > self.max_drawdown:
                logger.warning(f"Max drawdown {drawdown:.1%} exceeded")
                return False

        # Серия убытков
        if self.consecutive_losses >= self.max_consecutive_losses:
            logger.warning(f"{self.consecutive_losses} consecutive losses")
            return False

        return True

Параметризация и оптимизация

Хорошая кастомная стратегия не должна быть hardcoded — все ключевые параметры выносятся в конфигурацию и оптимизируются через backtesting:

# strategy_config.yaml
strategy:
  type: breakout
  lookback_period: 20
  volume_multiplier: 1.5
  entry_delay_candles: 1  # ждём закрытия свечи пробоя

risk:
  position_size_percent: 2.0
  stop_loss_percent: 2.5
  take_profit_percent: 5.0
  max_daily_loss_percent: 6.0
  max_drawdown_percent: 15.0
  max_consecutive_losses: 5

execution:
  exchange: binance
  symbol: BTCUSDT
  timeframe: 1h
  order_type: limit  # или market
  max_slippage_percent: 0.1

Grid search по параметрам на исторических данных даёт понимание робастности стратегии. Стратегия, которая работает только при lookback=20 и ни при каких других значениях — скорее всего overfit на историю.

Логирование и мониторинг

Production бот обязан иметь:

• Структурированное логирование каждого действия с timestamp
• Telegram/email алерты на критические события (стоп по drawdown, ошибки API)
• Dashboard с текущим P&L, открытыми позициями, количеством сделок
• Автоматический рестарт при сбоях (supervisord или systemd)
• Heartbeat мониторинг (Uptime Robot или Grafana alerting)

Бот без мониторинга — это черный ящик, который может незаметно потерять деньги неделями.