CoinOX/lib/exchange/okx_public_http_lib.py

"""OKX 公共 REST 轻量拉取(绕开 ccxt load_markets).

部分线路上 Python requests 偶发连 www.okx.com 超时,而 curl -4 仍可用;
对冲页标记价/指数优先走本模块,避免整页空白.
"""
from __future__ import annotations

import json
import subprocess
import urllib.parse
import urllib.request
from typing import Any, Optional


OKX_BASE = "https://www.okx.com"


def _curl_get_json(url: str, *, timeout_sec: float = 8.0) -> dict[str, Any]:
    proc = subprocess.run(
        [
            "curl",
            "-4",
            "-fsS",
            "--connect-timeout",
            str(max(1, int(timeout_sec))),
            "--max-time",
            str(max(2, int(timeout_sec + 2))),
            url,
        ],
        capture_output=True,
        text=True,
        timeout=max(3.0, timeout_sec + 3.0),
        check=False,
    )
    if proc.returncode != 0:
        raise RuntimeError((proc.stderr or proc.stdout or "curl failed").strip()[:240])
    return json.loads(proc.stdout or "{}")


def _urllib_get_json(url: str, *, timeout_sec: float = 8.0) -> dict[str, Any]:
    req = urllib.request.Request(url, headers={"User-Agent": "CoinOX/1.0"})
    with urllib.request.urlopen(req, timeout=timeout_sec) as resp:
        raw = resp.read().decode("utf-8", errors="replace")
    return json.loads(raw or "{}")


def okx_public_get(
    path: str,
    params: Optional[dict[str, Any]] = None,
    *,
    timeout_sec: float = 8.0,
) -> dict[str, Any]:
    q = urllib.parse.urlencode({k: v for k, v in (params or {}).items() if v is not None})
    url = f"{OKX_BASE}{path}"
    if q:
        url = f"{url}?{q}"
    last_err: Exception | None = None
    for getter in (_curl_get_json, _urllib_get_json):
        try:
            data = getter(url, timeout_sec=timeout_sec)
            if isinstance(data, dict):
                return data
        except Exception as e:
            last_err = e
            continue
    raise RuntimeError(str(last_err or "okx public get failed"))


def _sf(v: Any) -> Optional[float]:
    if v is None or v == "":
        return None
    try:
        x = float(v)
    except (TypeError, ValueError):
        return None
    return x if x > 0 else None


def fetch_swap_mark_px(base: str) -> Optional[float]:
    u = (base or "ETH").strip().upper() or "ETH"
    try:
        payload = okx_public_get(
            "/api/v5/public/mark-price",
            {"instId": f"{u}-USDT-SWAP"},
            timeout_sec=8.0,
        )
        rows = payload.get("data") or []
        if rows:
            return _sf(rows[0].get("markPx"))
    except Exception:
        pass
    return None


def fetch_swap_ticker(base: str) -> dict[str, Any]:
    """返回 ETH/BTC USDT 永续公共 ticker 行(补齐 markPx)."""
    u = (base or "ETH").strip().upper() or "ETH"
    payload = okx_public_get(
        "/api/v5/market/ticker",
        {"instId": f"{u}-USDT-SWAP"},
        timeout_sec=8.0,
    )
    rows = payload.get("data") or []
    if not rows:
        raise RuntimeError(payload.get("msg") or "ticker empty")
    row = dict(rows[0])
    if not _sf(row.get("markPx")):
        mp = fetch_swap_mark_px(u)
        if mp:
            row["markPx"] = str(mp)
        elif _sf(row.get("last")):
            row["markPx"] = row.get("last")
    return row


def fetch_index_px(uly: str) -> Optional[float]:
    u = (uly or "ETH").strip().upper() or "ETH"
    candidates = [u] if "-" in u else [f"{u}-USD", f"{u}-USDT"]
    for inst in candidates:
        try:
            payload = okx_public_get(
                "/api/v5/market/index-tickers",
                {"instId": inst},
                timeout_sec=8.0,
            )
            rows = payload.get("data") or []
            if not rows:
                continue
            v = _sf(rows[0].get("idxPx"))
            if v:
                return v
        except Exception:
            continue
    try:
        row = fetch_swap_ticker(u.split("-")[0])
        return _sf(row.get("markPx")) or _sf(row.get("last"))
    except Exception:
        return None


def fetch_option_instruments_public(inst_family: str) -> list[dict[str, Any]]:
    family = str(inst_family or "").strip()
    if not family:
        return []
    payload = okx_public_get(
        "/api/v5/public/instruments",
        {"instType": "OPTION", "instFamily": family},
        timeout_sec=12.0,
    )
    rows = payload.get("data") or []
    return [r for r in rows if isinstance(r, dict) and r.get("state") == "live"]


def fetch_option_tickers_public(inst_family: str) -> dict[str, dict[str, Any]]:
    family = str(inst_family or "").strip()
    if not family:
        return {}
    payload = okx_public_get(
        "/api/v5/market/tickers",
        {"instType": "OPTION", "instFamily": family},
        timeout_sec=12.0,
    )
    out: dict[str, dict[str, Any]] = {}
    for r in payload.get("data") or []:
        if isinstance(r, dict) and r.get("instId"):
            out[str(r["instId"])] = r
    return out