diff --git a/backend/app/exchange/binance/adapter.py b/backend/app/exchange/binance/adapter.py index e7d80bf..93eb0c9 100644 --- a/backend/app/exchange/binance/adapter.py +++ b/backend/app/exchange/binance/adapter.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +import time from typing import Any, Sequence from ...config import Settings, get_settings @@ -14,6 +15,9 @@ from .ws import BinancePublicWs logger = logging.getLogger(__name__) +# snapshot 轮询勿每 1.5s 打 REST；缺盘口时最少间隔再补 +_REST_FILL_MIN_INTERVAL_SEC = 5.0 + class BinanceExchange: name = "binance" @@ -39,6 +43,7 @@ class BinanceExchange: ) self._started = False self._ct_cache: dict[str, float] = {} + self._rest_fill_at: dict[str, float] = {} async def start(self) -> None: if self._started: @@ -91,19 +96,23 @@ class BinanceExchange: logger.warning("binance warm book empty: %s", inst) except Exception as e: logger.warning("binance warm book %s failed: %s", inst, e) - try: - mp = self.rest.fetch_mark(inst) - if mp: - self.cache.set_mark_px(inst, mp) - except Exception: - pass - # 指数 + # 期权 mark 易触发 eapi 限流；有盘口即可，永续再拉 mark + from .parse import is_option_symbol + + if not is_option_symbol(inst): + try: + mp = self.rest.fetch_mark(inst) + if mp: + self.cache.set_mark_px(inst, mp) + except Exception: + pass + # 指数（失败则跳过，勿连环打 eapi） try: idx = self.rest.fetch_index(self.settings.index_inst_id) if idx: self.cache.set_index_px(idx) except Exception as e: - logger.warning("binance index failed: %s", e) + logger.debug("binance index failed: %s", e) keep = set(ids) self.cache.drop_except(keep) self.ws.set_instruments(ids) @@ -115,11 +124,17 @@ class BinanceExchange: return self.cache.get(inst_id) def _refresh_quote_if_stale(self, inst_id: str) -> None: + """仅在买卖一都缺失时补 REST，且有最短间隔，避免 429。""" if not inst_id: return q = self.cache.get(inst_id) - if q is not None and (q.bid is not None or q.ask is not None): + if q is not None and q.bid is not None and q.ask is not None: return + now = time.monotonic() + last = self._rest_fill_at.get(inst_id, 0.0) + if now - last < _REST_FILL_MIN_INTERVAL_SEC: + return + self._rest_fill_at[inst_id] = now try: bids, asks, ts = self.rest.fetch_books(inst_id, sz=10) if bids or asks: @@ -128,11 +143,10 @@ class BinanceExchange: logger.debug("binance refresh quote %s: %s", inst_id, e) def snapshot(self, perp_inst_id: str) -> MarketSnapshot: - # 期权 WS 偶发无推送时，用 REST/ticker 补买卖一，避免 UI 一直是 - + # 优先吃 WS 缓存；仅缺盘口时低频 REST 补齐 self._refresh_quote_if_stale(perp_inst_id) pair = None try: - # BookCache 无公开 getter；从 snapshot 前先按 pair 刷新 with self.cache._lock: # noqa: SLF001 pair = self.cache._pair except Exception: diff --git a/backend/app/exchange/binance/rest.py b/backend/app/exchange/binance/rest.py index d991da4..eb93913 100644 --- a/backend/app/exchange/binance/rest.py +++ b/backend/app/exchange/binance/rest.py @@ -1,7 +1,10 @@ -"""币安只读 REST：USDT 永续 (fapi) + 欧洲期权 (eapi)。""" +"""币安只读 REST：USDT 永续 (fapi) + 欧洲期权 (eapi)。含节流与 429 退避。""" from __future__ import annotations +import logging +import threading +import time from typing import Any import httpx @@ -9,6 +12,18 @@ import httpx from ..types import BookLevel from .parse import rows_to_option_contracts, safe_float +logger = logging.getLogger(__name__) + +# eapi 公共接口较严；UI 1.5s 轮询时必须节流 +_BOOK_TTL_SEC = 2.5 +_CONTRACTS_TTL_SEC = 120.0 +_MIN_EAPI_INTERVAL_SEC = 0.12 +_DEFAULT_429_COOLDOWN_SEC = 8.0 + + +class EapiCooldownError(RuntimeError): + """eapi 限流冷却中，调用方应使用缓存或跳过。""" + class BinanceRestClient: def __init__( @@ -38,19 +53,60 @@ class BinanceRestClient: trust_env=False, ) self._exchange_info: dict[str, Any] | None = None + self._lock = threading.Lock() + self._eapi_cool_until = 0.0 + self._eapi_last_at = 0.0 + self._book_ttl: dict[str, tuple[list[BookLevel], list[BookLevel], int | None, float]] = {} + self._contracts_ttl: dict[str, tuple[float, list[dict[str, Any]]]] = {} 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 _eapi_cooling(self) -> bool: + return time.monotonic() < self._eapi_cool_until + + def _mark_eapi_429(self, retry_after: float | None = None) -> None: + wait = float(retry_after) if retry_after and retry_after > 0 else _DEFAULT_429_COOLDOWN_SEC + wait = max(wait, _DEFAULT_429_COOLDOWN_SEC) + self._eapi_cool_until = time.monotonic() + wait + logger.warning("binance eapi 429, cooldown %.1fs", wait) + + def _throttle_eapi(self) -> None: + """简单串行节流，避免并发 snapshot/选约打爆 eapi。""" + now = time.monotonic() + if now < self._eapi_cool_until: + raise EapiCooldownError( + f"eapi cooldown {self._eapi_cool_until - now:.1f}s left" + ) + gap = now - self._eapi_last_at + if gap < _MIN_EAPI_INTERVAL_SEC: + time.sleep(_MIN_EAPI_INTERVAL_SEC - gap) + self._eapi_last_at = time.monotonic() + + def _get_json( + self, + client: httpx.Client, + path: str, + params: dict[str, Any] | None = None, + *, + is_eapi: bool = False, + ) -> Any: + with self._lock: + if is_eapi: + self._throttle_eapi() + r = client.get(path, params=params or {}) + if r.status_code == 429: + if is_eapi: + ra = safe_float(r.headers.get("Retry-After")) + self._mark_eapi_429(ra) + r.raise_for_status() + 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") + body = self._get_json(self._eapi, "/eapi/v1/exchangeInfo", is_eapi=True) self._exchange_info = body if isinstance(body, dict) else {} return self._exchange_info @@ -82,7 +138,14 @@ class BinanceRestClient: return out def list_option_contracts(self, family: str) -> list[dict[str, Any]]: - return rows_to_option_contracts(self.fetch_option_instruments(family)) + key = (family or "").strip().upper() or "ETHUSDT" + now = time.monotonic() + hit = self._contracts_ttl.get(key) + if hit and now - hit[0] < _CONTRACTS_TTL_SEC: + return hit[1] + contracts = rows_to_option_contracts(self.fetch_option_instruments(family)) + self._contracts_ttl[key] = (now, contracts) + return contracts def fetch_index(self, underlying: str) -> float | None: """期权指数：underlying=ETHUSDT。""" @@ -90,7 +153,9 @@ class BinanceRestClient: if not u.endswith("USDT") and u.isalpha(): u = f"{u}USDT" try: - body = self._get_json(self._eapi, "/eapi/v1/index", {"underlying": u}) + body = self._get_json( + self._eapi, "/eapi/v1/index", {"underlying": u}, is_eapi=True + ) if isinstance(body, dict): return safe_float(body.get("indexPrice") or body.get("price")) except Exception: @@ -100,14 +165,19 @@ class BinanceRestClient: def fetch_mark_perp(self, symbol: str) -> float | None: body = self._get_json( - self._fapi, "/fapi/v1/premiumIndex", {"symbol": (symbol or "ETHUSDT").upper()} + self._fapi, + "/fapi/v1/premiumIndex", + {"symbol": (symbol or "ETHUSDT").upper()}, + is_eapi=False, ) 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}) + body = self._get_json( + self._eapi, "/eapi/v1/mark", {"symbol": symbol}, is_eapi=True + ) if isinstance(body, list) and body: return safe_float(body[0].get("markPrice")) if isinstance(body, dict): @@ -122,7 +192,9 @@ class BinanceRestClient: 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}) + body = self._get_json( + self._eapi, "/eapi/v1/ticker", {"symbol": symbol}, is_eapi=True + ) if isinstance(body, list) and body: row = body[0] return row if isinstance(row, dict) else None @@ -136,36 +208,59 @@ class BinanceRestClient: from .parse import is_option_symbol if is_option_symbol(inst_id): + now = time.monotonic() + cached = self._book_ttl.get(inst_id) + if cached and now - cached[3] < _BOOK_TTL_SEC: + return cached[0], cached[1], cached[2] + if self._eapi_cooling(): + if cached: + return cached[0], cached[1], cached[2] + return [], [], None + # 币安期权 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 + depth_ok = False 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)] + try: + body = self._get_json( + self._eapi, + "/eapi/v1/depth", + {"symbol": inst_id, "limit": limit}, + is_eapi=True, + ) + depth_ok = isinstance(body, dict) + if depth_ok and 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 []) + except Exception as e: + logger.debug("binance option depth %s: %s", inst_id, e) + if self._eapi_cooling() and cached: + return cached[0], cached[1], cached[2] + + # depth 缺买卖一时回退 ticker；冷却中不再打 + if (not bids or not asks) and not self._eapi_cooling(): + try: + 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)] + except Exception as e: + logger.debug("binance option ticker %s: %s", inst_id, e) + + if bids or asks or depth_ok: + self._book_ttl[inst_id] = (bids, asks, ts_ms, time.monotonic()) return bids, asks, ts_ms limit = max(5, min(int(sz), 20)) @@ -173,6 +268,7 @@ class BinanceRestClient: self._fapi, "/fapi/v1/depth", {"symbol": inst_id.upper(), "limit": limit}, + is_eapi=False, ) if not isinstance(body, dict): return [], [], None