Создание кастомного адаптера CCXT.Pro для парсинга ликвидаций Bitget

Суть: Создание кастомного класса-наследника ccxt.pro.bitget для реализации отсутствующего в стандартной библиотеке метода watch_liquidations(), позволяющего в реальном времени получать данные о принудительных закрытиях позиций на фьючерсном рынке Bitget через WebSocket API v2.

Содержание

Исходный код

Для реализации стриминга ликвидаций мы расширим базовый класс ccxt.pro.bitget. Мы переопределим метод handle_message для перехвата событий из канала liquidation и напишем метод watch_liquidations, который будет управлять подпиской и возвращать стандартизированные данные. Это позволяет бесшовно интегрировать кастомный фид в существующую кодовую базу на CCXT.

import asyncio
import ccxt.pro as ccxtpro

class CustomBitget(ccxtpro.bitget):
    def describe(self):
        return self.deep_extend(super().describe(), {
            'has': {
                'watchLiquidations': True,
            },
            'urls': {
                'api': {
                    'ws': {
                        'public': 'wss://ws.bitget.com/v2/mix/public',
                    },
                },
            },
        })

    async def watch_liquidations(self, symbol, since=None, limit=None, params={}):
        await self.load_markets()
        market = self.market(symbol)
        messageHash = 'liquidation:' + market['symbol']
        url = self.urls['api']['ws']['public']
        instType = 'USDT-FUTURES' if market['linear'] else 'COIN-FUTURES'
        
        request = {
            'op': 'subscribe',
            'args': [
                {
                    'instType': instType,
                    'channel': 'liquidation',
                    'instId': market['id']
                }
            ]
        }
        return await self.watch(url, messageHash, request, messageHash)

    def handle_message(self, client, message):
        arg = self.safe_value(message, 'arg', {})
        channel = self.safe_string(arg, 'channel')
        if channel == 'liquidation':
            return self.handle_liquidation(client, message)
        return super().handle_message(client, message)

    def handle_liquidation(self, client, message):
        data = self.safe_value(message, 'data', [])
        arg = self.safe_value(message, 'arg', {})
        instId = self.safe_string(arg, 'instId')
        market = self.safe_market(instId)
        symbol = market['symbol']
        messageHash = 'liquidation:' + symbol
        
        result = []
        for i in range(len(data)):
            item = data[i]
            parsed = self.parse_liquidation(item, market)
            result.append(parsed)
        
        client.resolve(result, messageHash)
        return result

    def parse_liquidation(self, liquidation, market=None):
        timestamp = self.safe_integer(liquidation, 'ts')
        return {
            'symbol': market['symbol'] if market else None,
            'timestamp': timestamp,
            'datetime': self.iso8601(timestamp),
            'price': self.safe_float(liquidation, 'price'),
            'amount': self.safe_float(liquidation, 'size'),
            'side': self.safe_string(liquidation, 'side'),
            'info': liquidation
        }

async def main():
    exchange = CustomBitget()
    symbol = 'BTC/USDT:USDT'
    print('Подключение к Bitget WebSocket...')
    try:
        while True:
            liquidations = await exchange.watch_liquidations(symbol)
            for liq in liquidations:
                print(f"[{liq['datetime']}] Ликвидация {liq['symbol']}: {liq['side'].upper()} {liq['amount']} по цене {liq['price']}")
    except Exception as e:
        print(f'Ошибка: {e}')
    finally:
        await exchange.close()

if __name__ == '__main__':
    asyncio.run(main())

Разбор параметров

watch_liquidations(symbol): Основной асинхронный метод для подписки на поток ликвидаций конкретной торговой пары. Принимает унифицированный символ CCXT (например, BTC/USDT:USDT).
handle_message(client, message): Внутренний обработчик входящих WebSocket-сообщений. Идентифицирует пакеты от канала ликвидаций и перенаправляет их в специализированный парсер.
parse_liquidation(liquidation, market): Метод нормализации сырых данных Bitget в единый формат CCXT, содержащий временную метку, цену, объем и сторону сделки.

Как запустить

Для запуска скрипта вам понадобится установленная библиотека ccxt с поддержкой Pro-версии и установленный asyncio. Скопируйте код адаптера и запустите асинхронный цикл событий.

Для построения полноценной торговой инфраструктуры и снижения задержек, рекомендуем изучить статью Интеграция Redis для кэширования стакана ордеров L2 на Python.

Полученные данные о крупных ликвидациях можно отправлять в мессенджеры, используя подходы из статьи Интеграция Python скрипта с Telegram Bot API для отправки PnL отчета CCXT.

Если вам интересен низкоуровневый разбор вебсокетов без CCXT, обратите внимание на материал Парсинг стакана ордеров Binance через WebSockets на Python asyncio.