Розробка бота для міжбіржового арбітражу

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

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

Економіка маркет-мейкингу

МM заробляє на захопленні спреду: якщо bid = $42,000 і ask = $42,100, а обидва виконані — прибуток = $100 (мінус комісії). Але якщо ціна впала до $41,800 протягом цього часу, МM купив по $42,000 і тепер має збиткову позицію.

Успішний МM повинен:

• Заробити достатньо на захопленні спреду, щоб покрити збитки від adverse selection
• Керувати ризиком інвентарю (не накопичувати занадто багато позиції в одному напрямку)
• Адаптувати спред до поточної волатильності

Базова стратегія котирування

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 цикл помилка: {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
        
        # Коригування волатильності: при високій волатильності — розширюємо спред
        vol_multiplier = Decimal('1') + volatility * Decimal('10')  # 1x при vol=0, 2x при vol=10%
        adjusted_spread = half_spread * vol_multiplier
        
        # Дисбаланс інвентарю: зміщуємо котирування для зменшення дисбалансу
        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:
        """
        Зміщуємо котирування на основі поточного інвентарю.
        При накопленні long позиції — знижуємо bid, піднімаємо ask (продаємо активніше).
        """
        target = Decimal('0')  # цільовий інвентар = 0 (нейтральний)
        max_inventory = self.max_position_usd
        
        inventory_ratio = inventory_usd / max_inventory
        inventory_ratio = max(Decimal('-1'), min(Decimal('1'), inventory_ratio))
        
        # Skew пропорційний інвентарю: при 100% позиції — зміщуємо на 1 повний спред
        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:
    """
    Реалізована волатильність за останні 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)))

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

Уникнення ненав'язних скасувань/заміни

Кожне скасування та новий ордер — комісія та 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)
    
    # Пакетне скасування + розміщення для мінімізації 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
        ])

Аварійне скасування

При різкому русі ринку або технічній помилці — негайне скасування всіх ордерів:

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_change:.2%}. Скасовуємо всі ордери.")
        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("Ціна стабілізована, відновлюємо маркет-мейкинг")
            self.reference_price = new_price

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

Adverse Selection

Крупний учасник знає, що ціна виросте, і агресивно купує у МM за bid. МM опиняється з long позицією перед падінням.

Виявлення: якщо останні N fillsів усі на одній стороні — стоп і переоцінка. Ознака інформованої торгівлі.

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

Ризик голої позиції

Жорсткий ліміт на максимальну позицію — circuit breaker:

async def check_position_limits(self):
    position = await self.get_net_position()  # $ в базовому активі
    
    if abs(position) > self.max_position_usd:
        logger.critical(f"Ліміт позиції перевищений: {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')  # у базовій валюті
        self.avg_cost = Decimal('0')
    
    def on_fill(self, fill: Fill):
        self.fill_history.append(fill)
        
        if fill.side == 'buy':
            # Оновлюємо середню вартість
            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)

Сроки розробки

• Базовий МM бот з однорівневим котируванням: 3–4 тижні
• Багаторівневий з керуванням інвентарем: 5–7 тижнів
• Production-ready з адаптацією волатильності та виявленням adverse selection: 8–12 тижнів
• Фреймворк backtesting для стратегії: +2–3 тижні