"""OKX V5 私有交易 REST(下单)。""" from __future__ import annotations import base64 import hashlib import hmac import json import logging import time from dataclasses import dataclass from typing import Any import httpx from ..config import Settings, get_settings from ..exchange.okx.parse import safe_float from .rate_limit import RateLimitError, get_throttle, parse_retry_after_header logger = logging.getLogger(__name__) @dataclass(slots=True) class LiveFill: inst_id: str side: str avg_px: float sz: float # 张或币,取决于合约 fee: float ord_id: str raw: dict[str, Any] class OkxTradeClient: def __init__(self, settings: Settings | None = None) -> None: self.settings = settings or get_settings() proxy = (self.settings.okx_http_proxy or "").strip() or None self._client = httpx.Client( base_url=self.settings.okx_rest_base.rstrip("/"), timeout=20.0, proxy=proxy, headers={"Accept": "application/json", "User-Agent": "eth-hedge-live/0.1"}, ) self._ct_val_cache: dict[str, float] = {} self._throttle = get_throttle("okx_trade", min_interval_sec=1.0) def close(self) -> None: self._client.close() def _ts(self) -> str: # OKX: ISO8601 with milliseconds return ( time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + f".{int(time.time() * 1000) % 1000:03d}Z" ) def _sign(self, ts: str, method: str, path: str, body: str) -> str: secret = (self.settings.okx_api_secret or "").encode("utf-8") msg = f"{ts}{method.upper()}{path}{body}".encode("utf-8") dig = hmac.new(secret, msg, hashlib.sha256).digest() return base64.b64encode(dig).decode("utf-8") def _headers(self, ts: str, sign: str) -> dict[str, str]: return { "OK-ACCESS-KEY": self.settings.okx_api_key or "", "OK-ACCESS-SIGN": sign, "OK-ACCESS-TIMESTAMP": ts, "OK-ACCESS-PASSPHRASE": self.settings.okx_api_passphrase or "", "Content-Type": "application/json", } def _request( self, method: str, path: str, body: dict[str, Any] | None = None ) -> list[dict[str, Any]]: self._throttle.before_request() payload = "" if body is None else json.dumps(body, separators=(",", ":")) ts = self._ts() sign = self._sign(ts, method, path, payload) headers = self._headers(ts, sign) if method.upper() == "GET": r = self._client.get(path, headers=headers) else: r = self._client.request(method.upper(), path, content=payload, headers=headers) if r.status_code in (418, 429): ra = parse_retry_after_header(r.headers) self._throttle.mark_http(r.status_code, ra) raise RateLimitError( f"OKX HTTP {r.status_code}: {r.text[:200]}", retry_after=self._throttle.remaining_cooldown(), ) try: r.raise_for_status() except httpx.HTTPStatusError as e: raise RuntimeError(f"OKX HTTP {r.status_code}: {r.text[:300]}") from e data = r.json() code = str(data.get("code") or "") msg = str(data.get("msg") or "") # OKX 业务层频率类错误 if code != "0": low = f"{code} {msg}".lower() if code in ("50011", "50061") or "too many" in low or "频率" in msg: self._throttle.mark_seconds(20.0) raise RateLimitError( f"OKX trade rate-limited code={code} msg={msg}", retry_after=self._throttle.remaining_cooldown(), ) raise RuntimeError( f"OKX trade error code={code} msg={msg} data={data.get('data')}" ) rows = data.get("data") or [] return [x for x in rows if isinstance(x, dict)] def get_ct_val(self, inst_id: str, *, inst_type: str) -> float: if inst_id in self._ct_val_cache: return self._ct_val_cache[inst_id] r = self._client.get( "/api/v5/public/instruments", params={"instType": inst_type, "instId": inst_id}, ) r.raise_for_status() body = r.json() rows = body.get("data") or [] for row in rows: if str(row.get("instId")) == inst_id: v = safe_float(row.get("ctVal")) or safe_float(row.get("ctMult")) if v and v > 0: self._ct_val_cache[inst_id] = float(v) return float(v) default = 0.01 self._ct_val_cache[inst_id] = default return default def place_market( self, *, inst_id: str, side: str, # buy|sell sz: str, td_mode: str, pos_side: str | None = None, reduce_only: bool = False, ) -> LiveFill: body: dict[str, Any] = { "instId": inst_id, "tdMode": td_mode, "side": side, "ordType": "market", "sz": str(sz), } if pos_side: body["posSide"] = pos_side if reduce_only: body["reduceOnly"] = True rows = self._request("POST", "/api/v5/trade/order", body) if not rows: raise RuntimeError("OKX 下单无返回") ord_id = str(rows[0].get("ordId") or "") # 查单取均价 fill = self._wait_fill(inst_id, ord_id) return fill def _wait_fill(self, inst_id: str, ord_id: str, *, tries: int = 8) -> LiveFill: path = f"/api/v5/trade/order?instId={inst_id}&ordId={ord_id}" last: dict[str, Any] = {} for _ in range(tries): rows = self._request("GET", path) if rows: last = rows[0] state = str(last.get("state") or "") avg = safe_float(last.get("avgPx")) if state in ("filled", "partially_filled") and avg and avg > 0: fee = abs(safe_float(last.get("fee")) or 0.0) sz = safe_float(last.get("accFillSz")) or safe_float(last.get("sz")) or 0.0 return LiveFill( inst_id=inst_id, side=str(last.get("side") or ""), avg_px=float(avg), sz=float(sz), fee=float(fee), ord_id=ord_id, raw=last, ) if state in ("canceled", "failed"): raise RuntimeError(f"OKX 订单失败 state={state} {last}") time.sleep(0.25) raise RuntimeError(f"OKX 订单未成交 ordId={ord_id} last={last}")