Розробка бота з кастомною торговою стратегією

Розробка бота з кастомною торговою стратегією

Кастомний торговий бот — це реалізація конкретної торговельної ідеї в виді автоматизованого алгоритму. На відміну від готових 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):
        # Перевірки ризику
        if not self.risk.can_open_position():
            logger.info("Менеджер ризику заблокував нову позицію")
            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"Відкрили {signal} позицію: {size} @ {order.fill_price}")

Модуль управління ризиком

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("Дневний ліміт втрат досягнутий")
            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%} перевищений")
                return False

        # Серія збитків
        if self.consecutive_losses >= self.max_consecutive_losses:
            logger.warning(f"{self.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 і ні при яких інших значеннях — скоріше за все переучена на історію.

Логування та моніторинг

Production бот обов'язково має мати:

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

Бот без моніторингу — це чорна скринька, яка може непомітно втратити гроші тижнями.