Интеграция бота с API Bitget

Интеграция бота с API Bitget

Bitget предоставляет REST и WebSocket API для spot, futures и copy-trading. Особенность Bitget: активный фокус на copy trading — у них есть специальный API для trader profiles и follower управления.

Аутентификация Bitget

Bitget использует HMAC-SHA256 с timestamp + method + path + body:

import hmac
import hashlib
import base64
import time
import json
import httpx

class BitgetClient:
    BASE_URL = "https://api.bitget.com"

    def __init__(self, api_key: str, secret_key: str, passphrase: str):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase  # Bitget требует passphrase в отличие от Bybit

    def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
        prehash = timestamp + method.upper() + path + body
        signature = hmac.new(
            self.secret_key.encode('utf-8'),
            prehash.encode('utf-8'),
            hashlib.sha256
        ).digest()
        return base64.b64encode(signature).decode()

    def _get_headers(self, method: str, path: str, body: str = "") -> dict:
        timestamp = str(int(time.time() * 1000))
        return {
            "ACCESS-KEY": self.api_key,
            "ACCESS-SIGN": self._sign(timestamp, method, path, body),
            "ACCESS-TIMESTAMP": timestamp,
            "ACCESS-PASSPHRASE": self.passphrase,
            "Content-Type": "application/json",
            "locale": "en-US"
        }

Базовые торговые операции

async def place_spot_order(
    self,
    symbol: str,      # 'BTCUSDT_SPBL'
    side: str,        # 'buy' или 'sell'
    order_type: str,  # 'limit' или 'market'
    size: str,        # количество
    price: str = None
) -> dict:
    path = "/api/spot/v1/trade/orders"
    payload = {
        "symbol": symbol,
        "side": side,
        "orderType": order_type,
        "force": "normal",  # GTC
        "size": size
    }
    if price:
        payload["price"] = price

    body = json.dumps(payload)
    headers = self._get_headers("POST", path, body)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.BASE_URL}{path}",
            content=body,
            headers=headers
        )
    return response.json()

async def get_account_info(self) -> dict:
    path = "/api/spot/v1/account/assets"
    headers = self._get_headers("GET", path)

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}{path}",
            headers=headers
        )
    return response.json()

Futures API (Mix)

Bitget называет futures API "Mix". Для USDT-M perpetuals:

async def place_futures_order(
    self,
    symbol: str,      # 'BTCUSDT_UMCBL'
    side: str,        # 'open_long', 'open_short', 'close_long', 'close_short'
    order_type: str,  # 'limit' или 'market'
    size: str,        # количество контрактов
    price: str = None,
    margin_mode: str = 'crossed'  # 'crossed' или 'fixed'
) -> dict:
    path = "/api/mix/v1/order/placeOrder"
    payload = {
        "symbol": symbol,
        "marginCoin": "USDT",
        "size": size,
        "side": side,
        "orderType": order_type,
        "marginMode": margin_mode
    }
    if price:
        payload["price"] = price

    body = json.dumps(payload)
    headers = self._get_headers("POST", path, body)

    async with httpx.AsyncClient() as client:
        response = await client.post(f"{self.BASE_URL}{path}", content=body, headers=headers)
    return response.json()

WebSocket подписки

class BitgetWebSocket:
    WS_URL = "wss://ws.bitget.com/spot/v1/stream"

    async def subscribe_ticker(self, symbols: list[str]):
        async with websockets.connect(self.WS_URL) as ws:
            sub_args = [{"instType": "sp", "channel": "ticker", "instId": s} for s in symbols]
            await ws.send(json.dumps({"op": "subscribe", "args": sub_args}))

            # Keepalive ping каждые 30 секунд
            async def ping():
                while True:
                    await asyncio.sleep(30)
                    await ws.send("ping")

            asyncio.create_task(ping())

            async for message in ws:
                if message == "pong":
                    continue
                data = json.loads(message)
                if "data" in data:
                    await self.on_ticker(data)

Симовлы и форматы

Bitget использует нестандартные форматы символов:

Эта особенность — первое с чем сталкиваются при интеграции. Bitget sandbox доступен по https://api-sandbox.bitget.com — идентичный production API, но с тестовыми средствами.

Документация: bitget.com/api-doc. SDK официальный на Python: pip install pybitget. Для production рекомендуем официальный SDK как wrapper — он уже обрабатывает подпись и базовые retry.