Інтеграція бота з API Gate.io

Інтеграція бота з API Gate.io

Gate.io надає REST та WebSocket API v4 для spot, margin, futures (perpetual) та delivery. Ключова особливість: Gate має широкий листинг альткоїнів — часто першими листять нові токени, що робить їхній API популярним для listing ботів та арбітражних стратегій на нових активах.

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

Gate.io v4 використовує HMAC-SHA512 з хешуванням тела запиту:

import hmac
import hashlib
import time
import json
import httpx

class GateClient:
    BASE_URL = "https://api.gateio.ws/api/v4"

    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret

    def _sign(
        self,
        method: str,
        url: str,
        query_string: str = "",
        body: str = ""
    ) -> dict:
        timestamp = str(int(time.time()))

        # Хешуємо тело запиту
        body_hash = hashlib.sha512(body.encode('utf-8')).hexdigest()

        sign_string = f"{method}\n{url}\n{query_string}\n{body_hash}\n{timestamp}"

        signature = hmac.new(
            self.api_secret.encode('utf-8'),
            sign_string.encode('utf-8'),
            hashlib.sha512
        ).hexdigest()

        return {
            "KEY": self.api_key,
            "Timestamp": timestamp,
            "SIGN": signature,
            "Content-Type": "application/json"
        }

Spot торгівля

async def place_spot_order(
    self,
    currency_pair: str,  # 'BTC_USDT'
    side: str,           # 'buy' або 'sell'
    amount: str,         # кількість
    order_type: str = 'limit',
    price: str = None,
    time_in_force: str = 'gtc'  # 'gtc', 'ioc', 'poc' (post-only)
) -> dict:
    endpoint = "/spot/orders"
    payload = {
        "currency_pair": currency_pair,
        "side": side,
        "amount": amount,
        "type": order_type,
        "time_in_force": time_in_force
    }
    if price:
        payload["price"] = price

    body = json.dumps(payload)
    headers = self._sign("POST", endpoint, body=body)

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

    if response.status_code not in (200, 201):
        raise GateAPIError(f"HTTP {response.status_code}: {response.text}")
    return response.json()

async def get_spot_balances(self) -> list:
    endpoint = "/spot/accounts"
    headers = self._sign("GET", endpoint)

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

async def get_order_book(self, currency_pair: str, limit: int = 20) -> dict:
    """Публічний endpoint — без підпису"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}/spot/order_book",
            params={"currency_pair": currency_pair, "limit": limit}
        )
    return response.json()

WebSocket

class GateWebSocket:
    WS_URL = "wss://api.gateio.ws/ws/v4/"

    async def subscribe_order_book(self, pairs: list[str]):
        async with websockets.connect(self.WS_URL) as ws:
            for pair in pairs:
                await ws.send(json.dumps({
                    "time": int(time.time()),
                    "channel": "spot.order_book",
                    "event": "subscribe",
                    "payload": [pair, "20", "100ms"]  # глибина, інтервал оновлення
                }))

            async for message in ws:
                data = json.loads(message)
                if data.get("channel") == "spot.order_book":
                    await self.on_orderbook(data["result"])

    async def subscribe_user_trades(self):
        """Приватний канал для власних угод"""
        channel = "spot.usertrades"
        timestamp = int(time.time())
        sign_msg = f"channel={channel}&event=subscribe&time={timestamp}"
        signature = hmac.new(
            self.api_secret.encode(),
            sign_msg.encode(),
            hashlib.sha512
        ).hexdigest()

        async with websockets.connect(self.WS_URL) as ws:
            await ws.send(json.dumps({
                "time": timestamp,
                "channel": channel,
                "event": "subscribe",
                "auth": {
                    "method": "api_key",
                    "KEY": self.api_key,
                    "SIGN": signature
                }
            }))

            async for message in ws:
                data = json.loads(message)
                if data.get("channel") == channel and data.get("event") == "update":
                    for trade in data.get("result", []):
                        await self.on_user_trade(trade)

Gate.io API повертає помилки через HTTP статус код (4xx/5xx) та JSON з label та message. Gate Futures використовує той же BASE_URL але шлях /futures/usdt/ для USDT-M та /futures/btc/ для inverse. Офіційна бібліотека: pip install gate-api.