Bot Integration with OKX API

OKX API Bot Integration

OKX (formerly OKEx) — the third-largest centralized exchange by volume. Provides REST and WebSocket V5 API for spot, futures, options, and margin. Key feature: unified account (Unified Account) allows trading all products from a single balance.

Authentication

OKX requires three headers for private requests: API key, timestamp, signature, and passphrase:

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

class OKXClient:
    BASE_URL = "https://www.okx.com"

    def __init__(self, api_key: str, secret_key: str, passphrase: str, sandbox: bool = False):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
        if sandbox:
            self.BASE_URL = "https://www.okx.com"  # sandbox via header flag

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

    def _headers(self, method: str, path: str, body: str = "", sandbox: bool = False) -> dict:
        timestamp = time.strftime('%Y-%m-%dT%H:%M:%S.000Z', time.gmtime())
        headers = {
            "OK-ACCESS-KEY": self.api_key,
            "OK-ACCESS-SIGN": self._sign(timestamp, method, path, body),
            "OK-ACCESS-TIMESTAMP": timestamp,
            "OK-ACCESS-PASSPHRASE": self.passphrase,
            "Content-Type": "application/json"
        }
        if sandbox:
            headers["x-simulated-trading"] = "1"
        return headers

Trading Operations

async def place_order(
    self,
    inst_id: str,      # 'BTC-USDT' for spot, 'BTC-USDT-SWAP' for perpetual
    td_mode: str,      # 'cash' (spot), 'cross' or 'isolated' (futures)
    side: str,         # 'buy' or 'sell'
    ord_type: str,     # 'market', 'limit', 'post_only', 'fok', 'ioc'
    sz: str,           # size
    px: str = None     # price (for limit)
) -> dict:
    path = "/api/v5/trade/order"
    payload = {
        "instId": inst_id,
        "tdMode": td_mode,
        "side": side,
        "ordType": ord_type,
        "sz": sz
    }
    if px:
        payload["px"] = px

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

    if data["code"] != "0":
        raise OKXError(f"Order failed: {data['msg']}")
    return data["data"][0]

async def get_positions(self, inst_type: str = "SWAP") -> list:
    path = f"/api/v5/account/positions?instType={inst_type}"
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}{path}",
            headers=self._headers("GET", path)
        )
    return response.json().get("data", [])

WebSocket

class OKXWebSocket:
    WS_PUBLIC = "wss://ws.okx.com:8443/ws/v5/public"
    WS_PRIVATE = "wss://ws.okx.com:8443/ws/v5/private"

    async def subscribe_trades(self, inst_id: str):
        async with websockets.connect(self.WS_PUBLIC) as ws:
            await ws.send(json.dumps({
                "op": "subscribe",
                "args": [{"channel": "trades", "instId": inst_id}]
            }))

            async for msg in ws:
                data = json.loads(msg)
                if data.get("arg", {}).get("channel") == "trades":
                    for trade in data.get("data", []):
                        await self.on_trade(trade)

    async def login_private(self, ws):
        """Authentication in private WebSocket"""
        timestamp = str(int(time.time()))
        sign = base64.b64encode(
            hmac.new(
                self.secret_key.encode(),
                f"{timestamp}GET/users/self/verify".encode(),
                hashlib.sha256
            ).digest()
        ).decode()

        await ws.send(json.dumps({
            "op": "login",
            "args": [{
                "apiKey": self.api_key,
                "passphrase": self.passphrase,
                "timestamp": timestamp,
                "sign": sign
            }]
        }))

OKX Instrument Formats

OKX Sandbox is available via x-simulated-trading: 1 header — does not require a separate URL. Official Python SDK: pip install python-okx. Documentation: okx.com/docs-v5.