Разработка бота для межбиржевого арбитража

Разработка маркет-мейкер бота

Market-making — это одновременное выставление ордеров на покупку и продажу с целью заработка на спреде. Маркет-мейкер (ММ) обеспечивает ликвидность рынку и зарабатывает когда его bid и ask исполняются. Главный враг ММ — adverse selection: торговля против информированного трейдера, который знает что цена скоро сдвинется.

Экономика маркет-мейкинга

ММ зарабатывает spread capture: если bid = $42,000 и ask = $42,100, и оба исполнились — прибыль $100 (минус комиссии). Но если за это время цена упала до $41,800, ММ купил по $42,000 и теперь держит убыточную позицию.

Успешный ММ должен:

• Зарабатывать достаточно spread capture чтобы покрыть adverse selection losses
• Управлять inventory risk (нельзя накопить слишком много позиции в одну сторону)
• Адаптировать спред к текущей волатильности

Базовая стратегия цитирования

from decimal import Decimal
import asyncio

class MarketMaker:
    def __init__(self, config: MMConfig):
        self.pair = config.pair
        self.base_spread_bps = Decimal(str(config.spread_bps))  # базовые пункты
        self.order_levels = config.levels  # количество уровней в стакане
        self.level_size_usd = Decimal(str(config.level_size_usd))
        self.max_position_usd = Decimal(str(config.max_position_usd))
        
    async def run(self):
        while True:
            try:
                reference_price = await self.get_reference_price()
                inventory = await self.get_inventory()
                volatility = await self.estimate_volatility()
                
                quotes = self.calculate_quotes(reference_price, inventory, volatility)
                await self.update_orders(quotes)
                
                await asyncio.sleep(1)  # обновляем каждую секунду
            except Exception as e:
                logger.error(f"MM cycle error: {e}")
                await self.cancel_all()
                await asyncio.sleep(5)  # пауза перед повторной попыткой
    
    def calculate_quotes(self, mid: Decimal, inventory: Decimal, volatility: Decimal) -> Quotes:
        # Базовый спред
        half_spread = mid * self.base_spread_bps / Decimal('10000') / 2
        
        # Волатильный adjustment: при высокой волатильности — расширяем спред
        vol_multiplier = Decimal('1') + volatility * Decimal('10')  # 1x при vol=0, 2x при vol=10%
        adjusted_spread = half_spread * vol_multiplier
        
        # Inventory skew: смещаем котировки чтобы уменьшить дисбаланс
        skew = self.calculate_inventory_skew(inventory, mid)
        
        best_bid = mid - adjusted_spread + skew
        best_ask = mid + adjusted_spread + skew
        
        # Генерируем несколько уровней
        bids = []
        asks = []
        for i in range(self.order_levels):
            level_adjustment = adjusted_spread * Decimal(str(i)) * Decimal('0.5')
            size = self.level_size_usd / (best_bid - level_adjustment)
            
            bids.append(Quote(price=best_bid - level_adjustment, size=size))
            asks.append(Quote(price=best_ask + level_adjustment, size=size))
        
        return Quotes(bids=bids, asks=asks)
    
    def calculate_inventory_skew(self, inventory_usd: Decimal, mid: Decimal) -> Decimal:
        """
        Смещаем цитирование на основе текущего inventory.
        При накоплении long позиции — снижаем bid, повышаем ask (продаём активнее).
        """
        target = Decimal('0')  # целевой inventory = 0 (нейтральный)
        max_inventory = self.max_position_usd
        
        inventory_ratio = inventory_usd / max_inventory
        inventory_ratio = max(Decimal('-1'), min(Decimal('1'), inventory_ratio))
        
        # Skew пропорционален inventory: при 100% позиции — сдвигаем на 1 full spread
        skew_pct = Decimal('-0.0001') * inventory_ratio  # -0.01% * ratio
        return mid * skew_pct

Оценка волатильности

def estimate_volatility(self, prices: list[Decimal], window: int = 20) -> Decimal:
    """
    Realized volatility за последние N периодов.
    Используем для адаптивного расширения спреда.
    """
    if len(prices) < window + 1:
        return Decimal('0.01')  # default если недостаточно данных
    
    returns = []
    for i in range(1, window + 1):
        ret = float(prices[-i] / prices[-(i+1)] - 1)
        returns.append(ret ** 2)
    
    variance = sum(returns) / len(returns)
    vol = Decimal(str(variance ** 0.5))
    
    # Аннуализированная волатильность: для 1-минутных данных
    # multiply by sqrt(525600 minutes per year)
    return vol * Decimal(str((525600 ** 0.5)))

Управление ордерами

Избегаем лишних cancel/replace

Каждая отмена и новый ордер — комиссия и latency. Оптимизация: обновляем ордер только если цена сдвинулась достаточно.

async def update_orders(self, new_quotes: Quotes):
    current_orders = await self.get_open_orders()
    
    # Сравниваем текущие ордера с новыми котировками
    orders_to_cancel = []
    orders_to_place = []
    
    for new_bid in new_quotes.bids:
        matching = self.find_matching_order(current_orders.bids, new_bid.price)
        
        if matching:
            # Проверяем нужно ли обновить
            price_diff = abs(matching.price - new_bid.price) / matching.price
            if price_diff > Decimal('0.0005'):  # изменение > 0.05% — обновляем
                orders_to_cancel.append(matching.id)
                orders_to_place.append(new_bid)
            # Иначе оставляем существующий ордер
        else:
            orders_to_place.append(new_bid)
    
    # Batch cancel + place для минимизации latency
    if orders_to_cancel:
        await self.exchange.cancel_orders(orders_to_cancel)
    
    if orders_to_place:
        await asyncio.gather(*[
            self.exchange.place_order(quote) for quote in orders_to_place
        ])

Emergency cancel

При резком движении рынка или технической ошибке — немедленная отмена всех ордеров:

async def handle_price_shock(self, current_price: Decimal, reference_price: Decimal):
    price_change = abs(current_price - reference_price) / reference_price
    
    if price_change > Decimal('0.02'):  # > 2% за период обновления
        logger.warning(f"Price shock detected: {price_change:.2%}. Cancelling all orders.")
        await self.exchange.cancel_all_orders(self.pair)
        
        # Пауза и мониторинг
        await asyncio.sleep(30)
        
        # Проверяем стабилизировалась ли цена
        new_price = await self.get_reference_price()
        new_change = abs(new_price - reference_price) / reference_price
        
        if new_change < Decimal('0.005'):  # < 0.5% — возобновляем
            logger.info("Price stabilized, resuming market making")
            self.reference_price = new_price

Риски маркет-мейкинга

Adverse Selection

Крупный участник знает что цена вырастет и агрессивно покупает у ММ по bid. ММ оказывается с long позицией перед падением.

Детектирование: если последние N fills все по одной стороне — stop и переоценка. Признак информированной торговли.

def detect_adverse_selection(self, recent_fills: list[Fill]) -> bool:
    if len(recent_fills) < 5:
        return False
    
    # Все последние 5 fills по одной стороне — подозрительно
    sides = [f.side for f in recent_fills[-5:]]
    if len(set(sides)) == 1:
        return True
    
    # Объём fills резко вырос
    avg_fill_volume = sum(f.quantity for f in recent_fills[:-3]) / max(len(recent_fills) - 3, 1)
    recent_volume = sum(f.quantity for f in recent_fills[-3:]) / 3
    
    if recent_volume > avg_fill_volume * 3:
        return True
    
    return False

Naked Position Risk

Limit на максимальную позицию — жёсткий circuit breaker:

async def check_position_limits(self):
    position = await self.get_net_position()  # $ в base asset
    
    if abs(position) > self.max_position_usd:
        logger.critical(f"Position limit exceeded: {position} USD")
        await self.cancel_all_orders()
        
        # Принудительная ликвидация части позиции
        excess = abs(position) - self.max_position_usd
        if position > 0:  # long excess
            await self.market_sell(excess / self.last_price)
        else:  # short excess
            await self.market_buy(abs(excess) / self.last_price)

P&L расчёт

class MMPnLTracker:
    def __init__(self):
        self.realized_pnl = Decimal('0')
        self.fill_history: list[Fill] = []
        self.position = Decimal('0')  # в base currency
        self.avg_cost = Decimal('0')
    
    def on_fill(self, fill: Fill):
        self.fill_history.append(fill)
        
        if fill.side == 'buy':
            # Обновляем avg cost
            new_qty = self.position + fill.quantity
            self.avg_cost = (self.position * self.avg_cost + fill.quantity * fill.price) / new_qty
            self.position = new_qty
        else:
            # Реализованный P&L при продаже
            pnl = (fill.price - self.avg_cost) * fill.quantity
            self.realized_pnl += pnl
            self.position -= fill.quantity
    
    def get_unrealized_pnl(self, current_price: Decimal) -> Decimal:
        return (current_price - self.avg_cost) * self.position
    
    def get_total_pnl(self, current_price: Decimal) -> Decimal:
        return self.realized_pnl + self.get_unrealized_pnl(current_price)

Сроки разработки

• Базовый ММ бот с single-level цитированием: 3–4 недели
• Multi-level с inventory management: 5–7 недель
• Production-ready с volatility adaptation и adverser selection detection: 8–12 недель
• Backtesting framework для стратегии: +2–3 недели