dekun/eth_hedge_sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Throttle Binance eapi REST to avoid 429 on Plan snapshot polls.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-25 12:44:41 +08:00
parent fff285c1f9
commit 867deba1bb
2 changed files with 157 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/exchange/binance/adapter.py
+25 -11
View File
@@ -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:
backend/app/exchange/binance/rest.py
+132 -36
View File
@@ -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