# -*- coding: utf-8 -*- """ OKX API 封装 """ import hmac import hashlib import base64 import json import requests from datetime import datetime, timezone from config import OKX_API_KEY, OKX_SECRET_KEY, OKX_PASSPHRASE, OKX_BASE_URL from exchange.http import http_get, http_post class OKX: def __init__(self): self.base = OKX_BASE_URL self.key = OKX_API_KEY self.secret = OKX_SECRET_KEY self.passphrase = OKX_PASSPHRASE # session 仅保留鉴权头，代理由 http_get/http_post 统一管理 self.session = requests.Session() def _sign(self, ts, method, path, body=""): msg = ts + method + path + body mac = hmac.new(self.secret.encode(), msg.encode(), hashlib.sha256) return base64.b64encode(mac.digest()).decode() def _headers(self, method, path, body=""): ts = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.') + \ f"{datetime.now(timezone.utc).microsecond // 1000:03d}Z" return { "OK-ACCESS-KEY": self.key, "OK-ACCESS-SIGN": self._sign(ts, method, path, body), "OK-ACCESS-TIMESTAMP": ts, "OK-ACCESS-PASSPHRASE": self.passphrase, "Content-Type": "application/json" } def get(self, path, params=None): url = self.base + path path_qs = path if params: path_qs = path + "?" + "&".join(f"{k}={v}" for k, v in params.items()) headers = self._headers("GET", path_qs) r = http_get(url, params=params, headers=headers, timeout=10, session=self.session) return r.json() def post(self, path, data): body = json.dumps(data) headers = self._headers("POST", path, body) r = http_post(self.base + path, data=body, headers=headers, timeout=10, session=self.session) return r.json() def ticker(self, inst_id): r = self.get("/api/v5/market/ticker", {"instId": inst_id}) return r.get("data", [{}])[0] def candles(self, inst_id, bar="5m", limit=50): r = self.get("/api/v5/market/candles", {"instId": inst_id, "bar": bar, "limit": str(limit)}) return r.get("data", []) def orderbook(self, inst_id, depth=20): r = self.get("/api/v5/market/books", {"instId": inst_id, "sz": str(depth)}) return r.get("data", [{}])[0] def trades(self, inst_id, limit=200): r = self.get("/api/v5/market/trades", {"instId": inst_id, "limit": str(limit)}) return r.get("data", []) def open_interest(self, inst_id): r = self.get("/api/v5/public/open-interest", {"instId": inst_id}) return r.get("data", [{}])[0] def oi_history(self, ccy, period="5m"): """OI + 成交量历史""" r = self.get("/api/v5/rubik/stat/contracts/open-interest-volume", {"ccy": ccy, "period": period}) return r.get("data", []) def taker_volume(self, ccy, period="5m"): """主动买卖量历史 [ts, buy_vol, sell_vol]""" r = self.get("/api/v5/rubik/stat/taker-volume", {"ccy": ccy, "instType": "CONTRACTS", "period": period}) return r.get("data", []) def liquidations(self, inst_id): """最近清算订单""" r = self.get("/api/v5/public/liquidation-orders", {"instType": "SWAP", "instId": inst_id, "state": "filled", "uly": inst_id.replace("-SWAP", "")}) return r.get("data", []) def long_short_ratio(self, ccy="ETH", period="5m"): """多空持仓人数比""" r = self.get("/api/v5/rubik/stat/contracts-long-short-ratio", {"ccy": ccy, "period": period}) return r.get("data", []) def open_interest_hist(self, inst_id, period="5m"): """历史OI""" r = self.get("/api/v5/rubik/stat/contracts-open-interest-volume", {"ccy": inst_id.split("-")[0], "period": period}) return r.get("data", []) def taker_volume_ratio(self, ccy="ETH", inst_type="CONTRACTS", period="5m", limit=12): """Taker买卖量比 — 返回最近N个周期的buy/sell数据""" r = self.get("/api/v5/rubik/stat/taker-volume", {"ccy": ccy, "instType": inst_type, "period": period}) data = r.get("data", [])[:limit] return data # [[ts, sellVol, buyVol], ...] def funding_rate(self, inst_id): r = self.get("/api/v5/public/funding-rate", {"instId": inst_id}) return r.get("data", [{}])[0] def balance(self): r = self.get("/api/v5/account/balance", {"ccy": "USDT"}) details = r.get("data", [{}])[0].get("details", []) for d in details: if d.get("ccy") == "USDT": return float(d.get("availBal", 0)), float(d.get("eqUsd", 0)) return 0.0, 0.0 def positions(self, inst_id=None): params = {"instId": inst_id} if inst_id else {} r = self.get("/api/v5/account/positions", params) return r.get("data", []) def set_leverage(self, inst_id, lever, side): return self.post("/api/v5/account/set-leverage", { "instId": inst_id, "lever": str(lever), "mgnMode": "isolated", "posSide": side }) def place_order(self, inst_id, side, pos_side, sz, tp=None, sl=None): data = { "instId": inst_id, "tdMode": "isolated", "side": side, "posSide": pos_side, "ordType": "market", "sz": str(sz) } if tp or sl: algo = {} if tp: algo["tpTriggerPx"] = str(round(tp, 2)) algo["tpOrdPx"] = "-1" algo["tpTriggerPxType"] = "last" if sl: algo["slTriggerPx"] = str(round(sl, 2)) algo["slOrdPx"] = "-1" algo["slTriggerPxType"] = "last" data["attachAlgoOrds"] = [algo] return self.post("/api/v5/trade/order", data) def close_position(self, inst_id, pos_side): return self.post("/api/v5/trade/close-position", { "instId": inst_id, "mgnMode": "isolated", "posSide": pos_side }) def get_algo_orders(self, inst_id, ord_type="conditional"): """获取当前algo订单""" r = self.get("/api/v5/trade/orders-algo-pending", { "instId": inst_id, "ordType": ord_type }) return r.get("data", []) if isinstance(r, dict) else [] def cancel_algo(self, algo_id, inst_id): """取消algo订单""" return self.post("/api/v5/trade/cancel-algos", [{ "algoId": algo_id, "instId": inst_id }]) def place_algo_tpsl(self, inst_id, pos_side, sz, tp=None, sl=None): """下止盈止损algo单""" side = "sell" if pos_side == "long" else "buy" data = { "instId": inst_id, "tdMode": "isolated", "side": side, "posSide": pos_side, "ordType": "conditional", "sz": str(sz), } if tp: data["tpTriggerPx"] = str(round(tp, 2)) data["tpOrdPx"] = "-1" data["tpTriggerPxType"] = "last" if sl: data["slTriggerPx"] = str(round(sl, 2)) data["slOrdPx"] = "-1" data["slTriggerPxType"] = "last" return self.post("/api/v5/trade/order-algo", data) def amend_algo(self, algo_id, inst_id, new_tp=None, new_sl=None): """修改algo订单的TP/SL价格""" data = {"algoId": algo_id, "instId": inst_id} if new_tp: data["newTpTriggerPx"] = str(round(new_tp, 2)) if new_sl: data["newSlTriggerPx"] = str(round(new_sl, 2)) return self.post("/api/v5/trade/amend-algos", data)