dekun
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""OKX REST 只读行情。不调用任何交易类接口。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .instruments import safe_float
|
|
from .types import BookLevel
|
|
|
|
|
|
class OkxRestClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str = "https://www.okx.com",
|
|
timeout: float = 15.0,
|
|
proxy: str | None = None,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.proxy = (proxy or "").strip() or None
|
|
self._client = httpx.Client(
|
|
base_url=self.base_url,
|
|
timeout=timeout,
|
|
proxy=self.proxy,
|
|
headers={"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.1"},
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def __enter__(self) -> OkxRestClient:
|
|
return self
|
|
|
|
def __exit__(self, *args: object) -> None:
|
|
self.close()
|
|
|
|
def _get(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
|
|
r = self._client.get(path, params=params or {})
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
if str(body.get("code")) != "0":
|
|
raise RuntimeError(f"OKX REST error code={body.get('code')} msg={body.get('msg')}")
|
|
data = body.get("data") or []
|
|
return [x for x in data if isinstance(x, dict)]
|
|
|
|
def fetch_instruments(self, *, inst_type: str, inst_family: str | None = None) -> list[dict[str, Any]]:
|
|
params: dict[str, Any] = {"instType": inst_type}
|
|
if inst_family:
|
|
params["instFamily"] = inst_family
|
|
return self._get("/api/v5/public/instruments", params)
|
|
|
|
def fetch_option_instruments(self, inst_family: str) -> list[dict[str, Any]]:
|
|
rows = self.fetch_instruments(inst_type="OPTION", inst_family=inst_family)
|
|
return [r for r in rows if str(r.get("state") or "").lower() == "live"]
|
|
|
|
def fetch_index_ticker(self, inst_id: str) -> float | None:
|
|
rows = self._get("/api/v5/market/index-tickers", {"instId": inst_id})
|
|
if not rows:
|
|
return None
|
|
return safe_float(rows[0].get("idxPx"))
|
|
|
|
def fetch_mark_price(self, inst_id: str) -> float | None:
|
|
rows = self._get("/api/v5/public/mark-price", {"instId": inst_id})
|
|
if not rows:
|
|
t = self._get("/api/v5/market/ticker", {"instId": inst_id})
|
|
if not t:
|
|
return None
|
|
return safe_float(t[0].get("markPx")) or safe_float(t[0].get("last"))
|
|
return safe_float(rows[0].get("markPx"))
|
|
|
|
def fetch_books(self, inst_id: str, sz: int = 5) -> tuple[list[BookLevel], list[BookLevel], int | None]:
|
|
rows = self._get(
|
|
"/api/v5/market/books",
|
|
{"instId": inst_id, "sz": str(max(1, min(int(sz), 400)))},
|
|
)
|
|
if not rows:
|
|
return [], [], None
|
|
row = rows[0]
|
|
ts = safe_float(row.get("ts"))
|
|
ts_ms = int(ts) if ts is not None else None
|
|
return (
|
|
_levels(row.get("bids") or []),
|
|
_levels(row.get("asks") or []),
|
|
ts_ms,
|
|
)
|
|
|
|
|
|
def _levels(raw: list[Any]) -> list[BookLevel]:
|
|
out: list[BookLevel] = []
|
|
for item in raw:
|
|
if not isinstance(item, (list, tuple)) or len(item) < 2:
|
|
continue
|
|
px = safe_float(item[0])
|
|
sz = safe_float(item[1])
|
|
if px is None or sz is None or px <= 0 or sz <= 0:
|
|
continue
|
|
out.append(BookLevel(px=px, sz=sz))
|
|
return out
|