"""币安只读 REST：USDT 永续 (fapi) + 欧洲期权 (eapi)。""" from __future__ import annotations from typing import Any import httpx from ..types import BookLevel from .parse import rows_to_option_contracts, safe_float class BinanceRestClient: def __init__( self, *, fapi_base: str = "https://fapi.binance.com", eapi_base: str = "https://eapi.binance.com", timeout: float = 15.0, proxy: str | None = None, ) -> None: self.fapi_base = fapi_base.rstrip("/") self.eapi_base = eapi_base.rstrip("/") self.proxy = (proxy or "").strip() or None headers = {"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.3"} self._fapi = httpx.Client( base_url=self.fapi_base, timeout=timeout, proxy=self.proxy, headers=headers, trust_env=False, ) self._eapi = httpx.Client( base_url=self.eapi_base, timeout=timeout, proxy=self.proxy, headers=headers, trust_env=False, ) self._exchange_info: dict[str, Any] | None = None def close(self) -> None: self._fapi.close() self._eapi.close() def _get_json(self, client: httpx.Client, path: str, params: dict[str, Any] | None = None) -> Any: r = client.get(path, params=params or {}) r.raise_for_status() return r.json() def fetch_option_exchange_info(self) -> dict[str, Any]: if self._exchange_info is None: body = self._get_json(self._eapi, "/eapi/v1/exchangeInfo") self._exchange_info = body if isinstance(body, dict) else {} return self._exchange_info def fetch_option_instruments(self, underlying: str) -> list[dict[str, Any]]: """underlying 如 ETH / ETHUSDT。误传 OKX family（ETH-USD_UM）时映射到 ETHUSDT。""" info = self.fetch_option_exchange_info() rows = info.get("optionSymbols") or info.get("symbols") or [] want = (underlying or "ETHUSDT").strip().upper() # 兼容误用 OKX 期权族名 if "USD_UM" in want or want in ("ETH-USD", "ETH-USDT", "ETHUSD"): want = "ETHUSDT" eth_mode = want in ("ETH", "ETHUSDT") or ( want.startswith("ETH") and "-" not in want ) out: list[dict[str, Any]] = [] for row in rows: if not isinstance(row, dict): continue u = str(row.get("underlying") or row.get("underlyingAsset") or "").upper() sym = str(row.get("symbol") or "").upper() # 只收币安格式 ETH-YYMMDD-STRIKE-C/P，避免脏符号进缓存 if eth_mode: if sym.startswith("ETH-") and u.startswith("ETH"): out.append(row) continue base = want.replace("USDT", "") if want.endswith("USDT") else want if u == want or u == base or sym.startswith(f"{base}-"): out.append(row) return out def list_option_contracts(self, family: str) -> list[dict[str, Any]]: return rows_to_option_contracts(self.fetch_option_instruments(family)) def fetch_index(self, underlying: str) -> float | None: """期权指数：underlying=ETHUSDT。""" u = (underlying or "ETHUSDT").strip().upper() if not u.endswith("USDT") and u.isalpha(): u = f"{u}USDT" try: body = self._get_json(self._eapi, "/eapi/v1/index", {"underlying": u}) if isinstance(body, dict): return safe_float(body.get("indexPrice") or body.get("price")) except Exception: pass # 回退永续标记 return self.fetch_mark_perp(u if u.endswith("USDT") else "ETHUSDT") def fetch_mark_perp(self, symbol: str) -> float | None: body = self._get_json( self._fapi, "/fapi/v1/premiumIndex", {"symbol": (symbol or "ETHUSDT").upper()} ) if isinstance(body, dict): return safe_float(body.get("markPrice")) or safe_float(body.get("indexPrice")) return None def fetch_mark_option(self, symbol: str) -> float | None: body = self._get_json(self._eapi, "/eapi/v1/mark", {"symbol": symbol}) if isinstance(body, list) and body: return safe_float(body[0].get("markPrice")) if isinstance(body, dict): return safe_float(body.get("markPrice")) return None def fetch_mark(self, inst_id: str) -> float | None: from .parse import is_option_symbol if is_option_symbol(inst_id): return self.fetch_mark_option(inst_id) return self.fetch_mark_perp(inst_id) def fetch_option_ticker(self, symbol: str) -> dict[str, Any] | None: body = self._get_json(self._eapi, "/eapi/v1/ticker", {"symbol": symbol}) if isinstance(body, list) and body: row = body[0] return row if isinstance(row, dict) else None if isinstance(body, dict): return body return None def fetch_books( self, inst_id: str, sz: int = 5 ) -> tuple[list[BookLevel], list[BookLevel], int | None]: from .parse import is_option_symbol if is_option_symbol(inst_id): # 币安期权 depth 的 limit 仅支持 10/20/50/100 等，5 会失败 limit = 10 if int(sz) < 10 else min(int(sz), 100) try: body = self._get_json( self._eapi, "/eapi/v1/depth", {"symbol": inst_id, "limit": limit} ) except Exception: body = None bids: list[BookLevel] = [] asks: list[BookLevel] = [] ts_ms = None if isinstance(body, dict): ts = safe_float(body.get("T") or body.get("E") or body.get("time")) ts_ms = int(ts) if ts is not None else None bids = _levels(body.get("bids") or body.get("b") or []) asks = _levels(body.get("asks") or body.get("a") or []) # depth 空盘时回退 ticker 买卖一 if not bids or not asks: tick = self.fetch_option_ticker(inst_id) if tick: bid = safe_float(tick.get("bidPrice") or tick.get("b")) ask = safe_float(tick.get("askPrice") or tick.get("a")) bid_sz = safe_float(tick.get("bidQty") or tick.get("B")) or 1.0 ask_sz = safe_float(tick.get("askQty") or tick.get("A")) or 1.0 ts = safe_float(tick.get("time") or tick.get("E") or tick.get("T")) ts_ms = int(ts) if ts is not None else ts_ms if bid is not None and bid > 0 and not bids: bids = [BookLevel(px=bid, sz=bid_sz)] if ask is not None and ask > 0 and not asks: asks = [BookLevel(px=ask, sz=ask_sz)] return bids, asks, ts_ms limit = max(5, min(int(sz), 20)) body = self._get_json( self._fapi, "/fapi/v1/depth", {"symbol": inst_id.upper(), "limit": limit}, ) if not isinstance(body, dict): return [], [], None ts = safe_float(body.get("T") or body.get("E") or body.get("time")) ts_ms = int(ts) if ts is not None else None return ( _levels(body.get("bids") or []), _levels(body.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