Add Binance SIM market adapter and exchange switch in settings.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -0,0 +1,204 @@
|
||||
"""币安公共 WebSocket：USDT 永续 bookTicker + 期权 bookTicker。只读。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import websockets
|
||||
from websockets.asyncio.client import ClientConnection
|
||||
|
||||
from ..book_cache import BookCache
|
||||
from .parse import is_option_symbol, safe_float
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BinancePublicWs:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
futures_ws_base: str,
|
||||
options_ws_base: str,
|
||||
cache: BookCache,
|
||||
proxy: str | None = None,
|
||||
ping_interval: float = 20.0,
|
||||
) -> None:
|
||||
self.futures_ws_base = futures_ws_base.rstrip("/")
|
||||
self.options_ws_base = options_ws_base.rstrip("/")
|
||||
self.cache = cache
|
||||
self.proxy = (proxy or "").strip() or None
|
||||
self.ping_interval = ping_interval
|
||||
self._inst_ids: list[str] = []
|
||||
self._tasks: list[asyncio.Task[None]] = []
|
||||
self._stop = asyncio.Event()
|
||||
|
||||
def set_instruments(self, inst_ids: list[str]) -> None:
|
||||
self._inst_ids = [i for i in inst_ids if i]
|
||||
|
||||
def _split(self) -> tuple[list[str], list[str]]:
|
||||
perps: list[str] = []
|
||||
opts: list[str] = []
|
||||
for i in self._inst_ids:
|
||||
if is_option_symbol(i):
|
||||
opts.append(i)
|
||||
else:
|
||||
perps.append(i.upper())
|
||||
return perps, opts
|
||||
|
||||
async def start(self) -> None:
|
||||
if self._tasks and any(not t.done() for t in self._tasks):
|
||||
return
|
||||
self._stop.clear()
|
||||
await self._spawn()
|
||||
|
||||
async def stop(self) -> None:
|
||||
self._stop.set()
|
||||
for t in self._tasks:
|
||||
t.cancel()
|
||||
for t in self._tasks:
|
||||
try:
|
||||
await t
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._tasks = []
|
||||
self.cache.set_connected(False)
|
||||
|
||||
async def resubscribe(self, inst_ids: list[str]) -> None:
|
||||
self.set_instruments(inst_ids)
|
||||
await self.stop()
|
||||
self._stop.clear()
|
||||
await self._spawn()
|
||||
|
||||
async def _spawn(self) -> None:
|
||||
perps, opts = self._split()
|
||||
self._tasks = []
|
||||
if perps:
|
||||
url = self._combined_url(self.futures_ws_base, [f"{p.lower()}@bookTicker" for p in perps])
|
||||
self._tasks.append(
|
||||
asyncio.create_task(self._run_forever(url, kind="futures"), name="bn-fapi-ws")
|
||||
)
|
||||
if opts:
|
||||
streams = [f"{s}@bookTicker" for s in opts]
|
||||
url = self._combined_url(self.options_ws_base, streams)
|
||||
self._tasks.append(
|
||||
asyncio.create_task(self._run_forever(url, kind="options"), name="bn-eapi-ws")
|
||||
)
|
||||
if not self._tasks:
|
||||
self.cache.set_connected(False)
|
||||
|
||||
@staticmethod
|
||||
def _combined_url(base: str, streams: list[str]) -> str:
|
||||
# base like wss://fstream.binance.com/stream or .../eoptions/stream
|
||||
if "/stream" in base:
|
||||
root = base
|
||||
else:
|
||||
root = base.rstrip("/") + "/stream"
|
||||
return root + "?streams=" + "/".join(streams)
|
||||
|
||||
async def _open_connection(self, url: str) -> ClientConnection:
|
||||
if not self.proxy:
|
||||
return await websockets.connect(
|
||||
url,
|
||||
ping_interval=None,
|
||||
max_size=2**22,
|
||||
open_timeout=20,
|
||||
)
|
||||
from python_socks.async_.asyncio import Proxy
|
||||
|
||||
parsed = urlparse(url)
|
||||
host = parsed.hostname or "fstream.binance.com"
|
||||
port = parsed.port or (443 if parsed.scheme == "wss" else 80)
|
||||
sock = await Proxy.from_url(self.proxy).connect(dest_host=host, dest_port=port)
|
||||
return await websockets.connect(
|
||||
url,
|
||||
sock=sock,
|
||||
server_hostname=host,
|
||||
ping_interval=None,
|
||||
max_size=2**22,
|
||||
open_timeout=20,
|
||||
)
|
||||
|
||||
async def _run_forever(self, url: str, *, kind: str) -> None:
|
||||
backoff = 1.0
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
async with await self._open_connection(url) as ws:
|
||||
self.cache.set_connected(True)
|
||||
backoff = 1.0
|
||||
logger.info("Binance %s WS connected: %s", kind, url[:120])
|
||||
waiter = asyncio.create_task(self._stop.wait())
|
||||
reader = asyncio.create_task(self._read_loop(ws))
|
||||
pinger = asyncio.create_task(self._ping_loop(ws))
|
||||
done, pending = await asyncio.wait(
|
||||
{waiter, reader, pinger},
|
||||
return_when=asyncio.FIRST_COMPLETED,
|
||||
)
|
||||
for t in pending:
|
||||
t.cancel()
|
||||
for t in done:
|
||||
exc = t.exception()
|
||||
if exc and not isinstance(exc, asyncio.CancelledError):
|
||||
raise exc
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning("Binance %s WS disconnected: %s", kind, e)
|
||||
self.cache.set_connected(False)
|
||||
try:
|
||||
await asyncio.wait_for(self._stop.wait(), timeout=backoff)
|
||||
break
|
||||
except asyncio.TimeoutError:
|
||||
backoff = min(backoff * 2, 30.0)
|
||||
self.cache.set_connected(False)
|
||||
|
||||
async def _ping_loop(self, ws: ClientConnection) -> None:
|
||||
while True:
|
||||
await asyncio.sleep(self.ping_interval)
|
||||
try:
|
||||
await ws.ping()
|
||||
except Exception:
|
||||
return
|
||||
|
||||
async def _read_loop(self, ws: ClientConnection) -> None:
|
||||
try:
|
||||
async for raw in ws:
|
||||
if isinstance(raw, bytes):
|
||||
raw = raw.decode("utf-8", errors="ignore")
|
||||
try:
|
||||
msg = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
data = msg.get("data") if isinstance(msg, dict) and "stream" in msg else msg
|
||||
if isinstance(data, dict):
|
||||
self._handle_event(data)
|
||||
except websockets.exceptions.ConnectionClosed:
|
||||
return
|
||||
|
||||
def _handle_event(self, data: dict[str, Any]) -> None:
|
||||
et = str(data.get("e") or "")
|
||||
sym = str(data.get("s") or "")
|
||||
if not sym:
|
||||
return
|
||||
ts = safe_float(data.get("E") or data.get("T"))
|
||||
ts_ms = int(ts) if ts is not None else None
|
||||
if et in ("bookTicker", "") or ("b" in data and "a" in data and "s" in data):
|
||||
bid = safe_float(data.get("b"))
|
||||
ask = safe_float(data.get("a"))
|
||||
bid_sz = safe_float(data.get("B"))
|
||||
ask_sz = safe_float(data.get("A"))
|
||||
if bid is not None or ask is not None:
|
||||
self.cache.upsert_top(
|
||||
sym,
|
||||
bid=bid,
|
||||
ask=ask,
|
||||
bid_sz=bid_sz,
|
||||
ask_sz=ask_sz,
|
||||
ts_ms=ts_ms,
|
||||
)
|
||||
# 永续可用中间价近似 mark
|
||||
if not is_option_symbol(sym) and bid and ask:
|
||||
self.cache.set_mark_px(sym, (bid + ask) / 2.0, ts_ms=ts_ms)
|
||||
Reference in New Issue
Block a user