Bot Integration with Gate.io API

Gate.io API Bot Integration

Gate.io provides REST and WebSocket API v4 for spot, margin, futures (perpetual), and delivery. Key feature: Gate has a wide listing of altcoins — often lists new tokens first, making their API popular for listing bots and arbitrage strategies on new assets.

Authentication

Gate.io v4 uses HMAC-SHA512 with request body hashing:

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()))

        # Hash request body
        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 Trading

async def place_spot_order(
    self,
    currency_pair: str,  # 'BTC_USDT'
    side: str,           # 'buy' or 'sell'
    amount: str,         # quantity
    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:
    """Public endpoint — no signature required"""
    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"]  # depth, update interval
                }))

            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):
        """Private channel for own trades"""
        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 returns errors via HTTP status code (4xx/5xx) and JSON with label and message. Gate Futures uses the same BASE_URL but /futures/usdt/ path for USDT-M and /futures/btc/ for inverse. Official library: pip install gate-api.