Разработка торгового бота для спотовой торговли

Разработка торгового бота для спотовой торговли

Спотовый торговый бот — алгоритм, торгующий реальными активами: покупает BTC за USDT и продаёт обратно. В отличие от фьючерсных ботов, нет кредитного плеча и маржин-коллов, что делает это наиболее безопасным стартом для алготрейдинга.

Архитектура спотового бота

import asyncio
from decimal import Decimal

class SpotTradingBot:
    def __init__(
        self,
        exchange,
        strategy: Strategy,
        symbol: str,
        base_asset: str,  # BTC
        quote_asset: str, # USDT
        risk_manager: RiskManager,
    ):
        self.exchange = exchange
        self.strategy = strategy
        self.symbol = symbol
        self.base = base_asset
        self.quote = quote_asset
        self.risk_manager = risk_manager
        self.is_running = False

    async def start(self):
        self.is_running = True
        
        # Запускаем параллельно: получение данных и мониторинг ордеров
        await asyncio.gather(
            self.data_loop(),
            self.order_monitor_loop(),
            self.heartbeat_loop(),
        )

    async def data_loop(self):
        """Получаем рыночные данные и генерируем сигналы"""
        async for candle in self.exchange.watch_candles(self.symbol, '1h'):
            if not self.is_running:
                break

            signal = self.strategy.on_candle(candle)

            if signal == Signal.BUY:
                await self.open_long()
            elif signal == Signal.SELL:
                await self.close_long()

    async def open_long(self):
        balance = await self.exchange.fetch_balance()
        available_quote = Decimal(str(balance[self.quote]['free']))

        if available_quote < Decimal('10'):  # минимум $10
            return

        # Размер позиции от risk manager
        position_size_usd = self.risk_manager.get_position_size(
            available_capital=available_quote,
            risk_pct=0.02,  # 2% риска на сделку
        )

        current_price = await self.exchange.get_price(self.symbol)
        quantity = (position_size_usd / current_price).quantize(Decimal('0.00001'))

        order = await self.exchange.create_order(
            symbol=self.symbol,
            type='market',
            side='buy',
            amount=float(quantity),
        )
        
        logger.info(f"Opened long: {quantity} {self.base} at ${current_price:.2f}")

    async def close_long(self):
        balance = await self.exchange.fetch_balance()
        base_available = Decimal(str(balance[self.base]['free']))

        if base_available <= Decimal('0.00001'):
            return  # нет позиции

        order = await self.exchange.create_order(
            symbol=self.symbol,
            type='market',
            side='sell',
            amount=float(base_available),
        )

        logger.info(f"Closed long: {base_available} {self.base}")

Стратегия: DCA (Dollar Cost Averaging) бот

class DCAStrategy:
    """
    Простая DCA стратегия:
    - Покупаем фиксированную сумму каждый период
    - Продаём при достижении target profit
    """
    def __init__(self, dca_amount_usd: float = 100, take_profit_pct: float = 0.05):
        self.dca_amount = Decimal(str(dca_amount_usd))
        self.take_profit = take_profit_pct
        self.avg_entry: Decimal = Decimal(0)
        self.total_invested: Decimal = Decimal(0)
        self.total_quantity: Decimal = Decimal(0)

    def on_scheduled_trigger(self, current_price: Decimal) -> Signal:
        """Вызывается по расписанию (раз в день/неделю)"""
        return Signal.BUY

    def on_price_update(self, current_price: Decimal) -> Signal:
        """Проверяем take profit"""
        if self.avg_entry > 0:
            unrealized_pnl_pct = (current_price - self.avg_entry) / self.avg_entry
            if unrealized_pnl_pct >= Decimal(str(self.take_profit)):
                return Signal.SELL
        return Signal.HOLD

Order Management и Fill Tracking

class OrderManager:
    async def place_with_retry(
        self,
        symbol: str,
        side: str,
        quantity: Decimal,
        max_retries: int = 3,
    ) -> Order:
        for attempt in range(max_retries):
            try:
                # Для больших объёмов — limit order немного агрессивнее mid
                price = await self.get_aggressive_limit_price(symbol, side)

                order = await self.exchange.create_order(
                    symbol=symbol,
                    type='limit',
                    side=side,
                    amount=float(quantity),
                    price=float(price),
                    params={'postOnly': False, 'timeInForce': 'IOC'},  # Immediate or Cancel
                )

                if order['status'] == 'closed':
                    return order

                # Если не исполнился — пробуем market order
                if attempt == max_retries - 1:
                    return await self.exchange.create_order(
                        symbol=symbol, type='market', side=side, amount=float(quantity)
                    )

            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

Мониторинг и алерты

Спотовый бот должен иметь мониторинг: текущий баланс (base и quote), открытая позиция, P&L за день/неделю/всё время, последний сигнал и действие. Telegram уведомления на каждый ордер и ежедневный summary.

Спотовая торговля — безопасный старт: максимальный убыток ограничен размером инвестиции, нет риска ликвидации как на фьючерсах. Для новых стратегий начинайте именно со спота.