diff --git a/lib/common/static/options_panel.js b/lib/common/static/options_panel.js
index fffd075..b8504d3 100644
--- a/lib/common/static/options_panel.js
+++ b/lib/common/static/options_panel.js
@@ -7,10 +7,6 @@
root.setAttribute("data-options-booted", "1");
const panelCache = (window.__optionsPanelCache = window.__optionsPanelCache || {});
- if (!panelCache.quoteWatcherId) {
- panelCache.quoteWatcherId =
- "w" + Date.now().toString(36) + Math.random().toString(36).slice(2, 8);
- }
const state = {
underlying: root.dataset.defaultUnderly || "ETH",
@@ -44,17 +40,11 @@
let chainSoftTimer = null;
let lastChainSoftAt = 0;
let chainQuotedAt = 0;
- let quoteLiveEs = null;
- let quoteLiveReconnectTimer = null;
- let quoteLiveOk = false;
- let quoteLiveWsOk = false;
- let lastOrderQuoteLiveAt = 0;
let pendingTtlSeconds = 600;
const POSITIONS_STALE_MS = 45000;
const PENDING_POLL_MS = 8000;
- /** SSE/WS 断开时的 REST 兜底;连上后停用 */
- const CHAIN_SOFT_POLL_MS = 30000;
- const ORDER_QUOTE_LIVE_MIN_MS = 800;
+ /** 链卖一/买一静默刷新节流:无推送,靠拉;过密会撞 OKX 50011 */
+ const CHAIN_SOFT_POLL_MS = 15000;
const orderPanelHome = (function () {
const host = document.getElementById("opt-order-panel-host");
return host ? host.parentElement : null;
@@ -669,214 +659,16 @@
const line = document.getElementById("opt-index-line");
if (line) {
const liqHint = askLiqFilterOn() ? "仅显示卖一深度≥1张" : "显示全部卖一(含估算~)";
- let liveHint = "";
- if (quoteLiveOk && quoteLiveWsOk) {
- liveHint = chainQuotedAt
- ? " · WS实时 " + fmtChainQuotedAt()
- : " · WS实时";
- } else if (quoteLiveOk) {
- liveHint = " · 推送已连,等待 OKX WS…";
- } else if (chainQuotedAt) {
- liveHint = " · 链报价 " + fmtChainQuotedAt() + "（REST兜底）";
- }
+ const ageHint = chainQuotedAt ? " · 链报价 " + fmtChainQuotedAt() + "（约每15s静默刷新）" : "";
line.textContent =
"指数 " + state.underlying + " ≈ " + fmt(idx, 2) +
- " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + liveHint;
+ " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + ageHint;
}
}
- function findChainContract(instId) {
- if (!state.chain || !instId) return null;
- const exps = state.chain.expiries || [];
- for (let i = 0; i < exps.length; i++) {
- const contracts = exps[i].contracts || [];
- for (let j = 0; j < contracts.length; j++) {
- if (String(contracts[j].inst_id) === String(instId)) return contracts[j];
- }
- }
- return null;
- }
-
- function currentExpiryContracts() {
- if (!state.chain) return [];
- const expMs = (document.getElementById("opt-exp-select") || {}).value;
- const exp = (state.chain.expiries || []).find(function (e) {
- return String(e.exp_time) === String(expMs);
- });
- return (exp && exp.contracts) || [];
- }
-
- async function watchCurrentExpiryQuotes() {
- if (!state.chain) return;
- const expMs = (document.getElementById("opt-exp-select") || {}).value;
- const contracts = currentExpiryContracts().map(function (c) {
- return {
- inst_id: c.inst_id,
- opt_type: c.opt_type,
- strike: c.strike,
- tick_sz: c.tick_sz,
- };
- });
- if (!contracts.length) return;
- try {
- await apiJson("/api/options/quotes/watch", {
- method: "POST",
- headers: { "Content-Type": "application/json" },
- body: JSON.stringify({
- underlying: state.underlying,
- exp_time: expMs,
- contracts: contracts,
- index_inst_id: state.underlying + "-USD",
- watcher_id: panelCache.quoteWatcherId,
- }),
- });
- } catch (_) {
- /* ignore watch errors; soft poll fallback remains */
- }
- }
-
- function patchListRowDom(instId, c) {
- const tr = document.querySelector('#opt-strike-tbody tr.opt-strike-row[data-inst="' + instId + '"]');
- if (!tr || !c) return;
- const indexPx = state.chain && state.chain.index_px;
- const tds = tr.children;
- if (tds.length < 8) return;
- tds[3].textContent = "";
- tds[3].className = "opt-px-sz";
- tds[3].innerHTML = fmtPxSz(c.ask, c.ask_sz, c.ask_estimated);
- tds[4].className = "opt-chain-lev";
- tds[4].textContent = fmtChainLeverage(calcAskLeverage(indexPx, c.ask));
- tds[5].className = "opt-px-sz";
- tds[5].innerHTML = fmtPxSz(c.bid, c.bid_sz);
- tds[6].textContent = c.expiry_be_px != null ? fmt(c.expiry_be_px, 0) : "—";
- tds[7].className = distBeClass(c.dist_expiry_be);
- tds[7].textContent = fmtDist(c.dist_expiry_be);
- }
-
- function patchTRowDom(instId, c) {
- if (!c) return;
- const callTr = document.querySelector(
- '#opt-strike-tbody tr.opt-strike-row-t[data-call-inst="' + instId + '"]'
- );
- const putTr = document.querySelector(
- '#opt-strike-tbody tr.opt-strike-row-t[data-put-inst="' + instId + '"]'
- );
- const tr = callTr || putTr;
- if (!tr) return;
- const callInst = tr.getAttribute("data-call-inst");
- const putInst = tr.getAttribute("data-put-inst");
- const call = callInst ? findChainContract(callInst) : null;
- const put = putInst ? findChainContract(putInst) : null;
- const callOk = call && (!askLiqFilterOn() || hasAskLiquidity(call)) ? call : null;
- const putOk = put && (!askLiqFilterOn() || hasAskLiquidity(put)) ? put : null;
- const tds = tr.children;
- if (tds.length < 9) return;
- tds[0].innerHTML = callOk ? fmtPxSz(callOk.ask, callOk.ask_sz, callOk.ask_estimated) : "—";
- tds[7].innerHTML = putOk ? fmtPxSz(putOk.ask, putOk.ask_sz, putOk.ask_estimated) : "—";
- const combined = straddleAskPerUnit(callOk && callOk.ask, putOk && putOk.ask);
- tds[4].innerHTML = formatStraddlePremiumCell(callOk && callOk.ask, putOk && putOk.ask);
- tds[5].innerHTML = formatStraddleBand(tr.getAttribute("data-strike"), combined);
- }
-
- function applyLiveQuotes(payload) {
- if (!payload || !state.chain) return;
- const uly = String(state.underlying || "").toUpperCase();
- if (payload.indexes && payload.indexes[uly] != null && Number.isFinite(Number(payload.indexes[uly]))) {
- state.chain.index_px = Number(payload.indexes[uly]);
- } else if (payload.underlying && String(payload.underlying).toUpperCase() === uly) {
- if (payload.index_px != null && Number.isFinite(Number(payload.index_px))) {
- state.chain.index_px = Number(payload.index_px);
- }
- } else if (payload.underlying && String(payload.underlying).toUpperCase() !== uly) {
- // 别的标的推送:仍可 patch 本页已有合约
- }
- const quotes = payload.quotes || [];
- quotes.forEach(function (q) {
- const instId = q && q.inst_id;
- if (!instId) return;
- if (q.underlying && String(q.underlying).toUpperCase() !== uly) return;
- const c = findChainContract(instId);
- if (!c) return;
- if (q.ask !== undefined) c.ask = q.ask;
- if (q.bid !== undefined) c.bid = q.bid;
- if (q.ask_sz !== undefined) c.ask_sz = q.ask_sz;
- if (q.bid_sz !== undefined) c.bid_sz = q.bid_sz;
- if (q.mark_px !== undefined) c.mark_px = q.mark_px;
- if (q.ask_estimated !== undefined) c.ask_estimated = !!q.ask_estimated;
- if (q.expiry_be_px !== undefined) c.expiry_be_px = q.expiry_be_px;
- if (q.dist_expiry_be !== undefined) c.dist_expiry_be = q.dist_expiry_be;
- if (state.chainView === "t") patchTRowDom(instId, c);
- else patchListRowDom(instId, c);
- });
- if (payload.ts) chainQuotedAt = Number(payload.ts) || Date.now();
- else if (quotes.length || payload.index_px != null) chainQuotedAt = Date.now();
- quoteLiveWsOk = payload.ws_ok !== false;
- renderIndexLine();
- if (state.selectedInst && quotes.some(function (q) { return q && q.inst_id === state.selectedInst; })) {
- const now = Date.now();
- if (now - lastOrderQuoteLiveAt >= ORDER_QUOTE_LIVE_MIN_MS) {
- lastOrderQuoteLiveAt = now;
- void selectContract(state.selectedInst, null, true);
- }
- }
- }
-
- function stopQuoteLiveStream() {
- if (quoteLiveReconnectTimer) {
- clearTimeout(quoteLiveReconnectTimer);
- quoteLiveReconnectTimer = null;
- }
- if (quoteLiveEs) {
- try { quoteLiveEs.close(); } catch (_) {}
- quoteLiveEs = null;
- }
- quoteLiveOk = false;
- quoteLiveWsOk = false;
- }
-
- function startQuoteLiveStream() {
- if (quoteLiveEs) return;
- if (typeof EventSource === "undefined") return;
- try {
- quoteLiveEs = new EventSource("/api/options/quotes/stream");
- } catch (_) {
- quoteLiveOk = false;
- return;
- }
- quoteLiveEs.addEventListener("quotes", function (ev) {
- try {
- const data = JSON.parse(ev.data || "{}");
- quoteLiveOk = true;
- if (data.reason === "connect") {
- quoteLiveWsOk = !!data.ws_ok;
- renderIndexLine();
- return;
- }
- applyLiveQuotes(data);
- } catch (_) {}
- });
- quoteLiveEs.onopen = function () {
- quoteLiveOk = true;
- renderIndexLine();
- void watchCurrentExpiryQuotes();
- };
- quoteLiveEs.onerror = function () {
- quoteLiveOk = false;
- quoteLiveWsOk = false;
- renderIndexLine();
- stopQuoteLiveStream();
- quoteLiveReconnectTimer = setTimeout(function () {
- quoteLiveReconnectTimer = null;
- startQuoteLiveStream();
- }, 8000);
- };
- }
-
function softRefreshChainThrottled(force) {
if (document.hidden) return;
if (!document.getElementById("options-root")) return;
- // WS 推送正常时不靠 REST 刷卖一,避免 50011;仅结构兜底可 force
- if (!force && quoteLiveOk && quoteLiveWsOk) return;
const now = Date.now();
if (!force && now - lastChainSoftAt < CHAIN_SOFT_POLL_MS) return;
lastChainSoftAt = now;
@@ -1536,8 +1328,6 @@
}
// soft 时保留 selectedInst；renderStrikes 会先 park 再按 prevSelected 静默重挂下单面板
renderStrikes();
- void watchCurrentExpiryQuotes();
- startQuoteLiveStream();
} catch (e) {
if (seq !== chainLoadSeq || soft) return;
setExpirySelectStatus("选择到期日");
@@ -2446,7 +2236,6 @@
const expandCb = document.getElementById("opt-strike-expand-all");
if (expandCb) expandCb.checked = false;
renderStrikes();
- void watchCurrentExpiryQuotes();
}
function bootOptionsPanel() {
@@ -2458,7 +2247,6 @@
refreshPendingOrders();
startPendingOrdersPoll();
startChainSoftPoll();
- startQuoteLiveStream();
const hasCache =
chainHasExpiries(panelCache.chain) &&
panelCache.underlying === state.underlying &&
@@ -2468,8 +2256,7 @@
renderExpiries();
renderStrikes();
refreshAllPositions();
- void watchCurrentExpiryQuotes();
- // 后台静默刷新结构;卖一优先走 WS
+ // 后台静默刷新,避免缓存过期后到期日变空 / 卖一过期
softRefreshChainThrottled(true);
return;
}
@@ -2594,7 +2381,7 @@
window.OptionsPanelLive = {
refreshSoft: function () {
refreshAllPositions();
- // 有 WS 实时报价时不再 REST 刷链;断开时才兜底
+ // embed SSE 只通知「该拉了」,不推送链报价;这里节流拉新鲜卖一/买一
softRefreshChainThrottled(false);
},
refreshChain: loadChain,
diff --git a/lib/exchange/okx_public_ws_lib.py b/lib/exchange/okx_public_ws_lib.py
deleted file mode 100644
index d6b1865..0000000
--- a/lib/exchange/okx_public_ws_lib.py
+++ /dev/null
@@ -1,199 +0,0 @@
-"""OKX 公共 WebSocket(同步线程):订阅 tickers / index-tickers,自动重连."""
-from __future__ import annotations
-
-import json
-import logging
-import threading
-import time
-from collections.abc import Callable
-from typing import Any
-
-logger = logging.getLogger(__name__)
-
-OKX_PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
-_SUBSCRIBE_CHUNK = 40
-_APP_PING_SEC = 20.0
-
-
-class OkxPublicWs:
- """单连接公共 WS;set_subscriptions 全量对齐目标频道."""
-
- def __init__(
- self,
- *,
- on_data: Callable[[dict[str, Any]], None],
- url: str = OKX_PUBLIC_WS_URL,
- name: str = "okx-public-ws",
- ) -> None:
- self._on_data = on_data
- self._url = url
- self._name = name
- self._lock = threading.RLock()
- self._desired: dict[str, dict[str, str]] = {}
- self._active: set[str] = set()
- self._stop = threading.Event()
- self._thread: threading.Thread | None = None
- self._ws: Any = None
- self._connected = False
- self._last_msg_at = 0.0
-
- @property
- def connected(self) -> bool:
- return self._connected
-
- @property
- def last_msg_at(self) -> float:
- return self._last_msg_at
-
- def start(self) -> None:
- if self._thread and self._thread.is_alive():
- return
- self._stop.clear()
- self._thread = threading.Thread(target=self._run_loop, name=self._name, daemon=True)
- self._thread.start()
-
- def stop(self) -> None:
- self._stop.set()
- ws = self._ws
- if ws is not None:
- try:
- ws.close()
- except Exception:
- pass
- if self._thread and self._thread.is_alive():
- self._thread.join(timeout=3.0)
-
- def set_subscriptions(self, args: list[dict[str, str]]) -> None:
- desired: dict[str, dict[str, str]] = {}
- for raw in args:
- if not isinstance(raw, dict):
- continue
- channel = str(raw.get("channel") or "").strip()
- inst_id = str(raw.get("instId") or "").strip()
- if not channel or not inst_id:
- continue
- key = f"{channel}:{inst_id}"
- desired[key] = {"channel": channel, "instId": inst_id}
- with self._lock:
- self._desired = desired
- ws = self._ws
- connected = self._connected
- active = set(self._active)
- if connected and ws is not None:
- self._sync_subs(ws, active, desired)
-
- def _sync_subs(
- self,
- ws: Any,
- active: set[str],
- desired: dict[str, dict[str, str]],
- ) -> None:
- unsub_args: list[dict[str, str]] = []
- for key in active - set(desired.keys()):
- channel, _, inst_id = key.partition(":")
- if channel and inst_id:
- unsub_args.append({"channel": channel, "instId": inst_id})
- sub_args = [desired[k] for k in (set(desired.keys()) - active)]
- if unsub_args:
- self._send_op(ws, "unsubscribe", unsub_args)
- if sub_args:
- self._send_op(ws, "subscribe", sub_args)
- with self._lock:
- self._active = set(desired.keys())
-
- def _send_op(self, ws: Any, op: str, args: list[dict[str, str]]) -> None:
- for i in range(0, len(args), _SUBSCRIBE_CHUNK):
- chunk = args[i : i + _SUBSCRIBE_CHUNK]
- try:
- ws.send(json.dumps({"op": op, "args": chunk}, ensure_ascii=False))
- except Exception as e:
- logger.warning("%s %s failed: %s", self._name, op, e)
- return
- if i + _SUBSCRIBE_CHUNK < len(args):
- time.sleep(0.08)
-
- def _run_loop(self) -> None:
- try:
- import websocket
- except ImportError:
- logger.error("%s: websocket-client not installed", self._name)
- return
- backoff = 1.0
- while not self._stop.is_set():
- opened = False
- try:
- self._connected = False
- with self._lock:
- self._active.clear()
-
- def on_open(ws: Any) -> None:
- nonlocal opened
- opened = True
- self._connected = True
- self._last_msg_at = time.time()
- with self._lock:
- desired = dict(self._desired)
- self._sync_subs(ws, set(), desired)
-
- def on_message(_ws: Any, message: str) -> None:
- self._last_msg_at = time.time()
- if message == "pong":
- return
- try:
- payload = json.loads(message)
- except Exception:
- return
- if not isinstance(payload, dict):
- return
- if payload.get("event") in ("subscribe", "unsubscribe", "error"):
- if payload.get("event") == "error":
- logger.warning("%s event error: %s", self._name, payload)
- return
- if payload.get("arg") and payload.get("data") is not None:
- try:
- self._on_data(payload)
- except Exception:
- logger.exception("%s on_data failed", self._name)
-
- def on_error(_ws: Any, error: Any) -> None:
- logger.warning("%s error: %s", self._name, error)
-
- def on_close(_ws: Any, *_args: Any) -> None:
- self._connected = False
-
- self._ws = websocket.WebSocketApp(
- self._url,
- on_open=on_open,
- on_message=on_message,
- on_error=on_error,
- on_close=on_close,
- )
- ping_stop = threading.Event()
-
- def ping_loop() -> None:
- while not self._stop.is_set() and not ping_stop.is_set():
- ws = self._ws
- if ws is not None and self._connected:
- try:
- ws.send("ping")
- except Exception:
- pass
- if ping_stop.wait(_APP_PING_SEC):
- break
-
- ping_thread = threading.Thread(
- target=ping_loop, name=f"{self._name}-ping", daemon=True
- )
- ping_thread.start()
- self._ws.run_forever(ping_interval=0)
- ping_stop.set()
- except Exception as e:
- logger.warning("%s run failed: %s", self._name, e)
- finally:
- self._connected = False
- self._ws = None
- if self._stop.is_set():
- break
- time.sleep(backoff)
- backoff = 1.0 if opened else min(30.0, backoff * 1.7)
-
diff --git a/lib/options/options_quote_live_lib.py b/lib/options/options_quote_live_lib.py
deleted file mode 100644
index 22a9e66..0000000
--- a/lib/options/options_quote_live_lib.py
+++ /dev/null
@@ -1,458 +0,0 @@
-"""期权链实时报价:OKX 公共 WS tickers → 内存缓存 → SSE 推前端."""
-from __future__ import annotations
-
-import json
-import logging
-import os
-import queue
-import threading
-import time
-from collections.abc import Iterator
-from typing import Any, Callable
-
-from lib.exchange.okx_public_ws_lib import OkxPublicWs
-from lib.options.options_pricing_lib import (
- expiry_breakeven_from_ask,
- idx_distance_to_be,
-)
-
-logger = logging.getLogger(__name__)
-
-OPTIONS_QUOTE_SSE_HEARTBEAT_SEC = float(os.getenv("OKX_OPTIONS_QUOTE_SSE_HEARTBEAT_SEC", "20"))
-OPTIONS_QUOTE_FLUSH_MS = float(os.getenv("OKX_OPTIONS_QUOTE_FLUSH_MS", "120"))
-# OKX 单连接约 240 频道;当前到期日合约 + 指数通常够用
-OPTIONS_QUOTE_MAX_INST = int(os.getenv("OKX_OPTIONS_QUOTE_MAX_INST", "220"))
-
-
-def _safe_float(v: Any) -> float | None:
- try:
- if v is None or v == "":
- return None
- return float(v)
- except (TypeError, ValueError):
- return None
-
-
-class OptionsQuoteLive:
- def __init__(self) -> None:
- self._lock = threading.RLock()
- self._watchers: dict[str, dict[str, Any]] = {}
- self._meta: dict[str, dict[str, Any]] = {}
- self._tickers: dict[str, dict[str, Any]] = {}
- self._index_by_uly: dict[str, float] = {}
- self._index_insts: set[str] = set()
- self._dirty_inst: set[str] = set()
- self._dirty_index: set[str] = set()
- self._version = 0
- self._subscribers: list[queue.Queue[str | None]] = []
- self._stop = threading.Event()
- self._flush_thread: threading.Thread | None = None
- ws_url = (os.getenv("OKX_PUBLIC_WS_URL") or "").strip() or None
- self._ws = OkxPublicWs(
- on_data=self._on_ws_data,
- name="okx-options-quote-ws",
- **({"url": ws_url} if ws_url else {}),
- )
- self._started = False
-
- def start(self) -> None:
- if self._started:
- return
- self._started = True
- self._stop.clear()
- self._ws.start()
- self._flush_thread = threading.Thread(
- target=self._flush_loop, name="options-quote-flush", daemon=True
- )
- self._flush_thread.start()
-
- def stop(self) -> None:
- self._stop.set()
- self._ws.stop()
- self._broadcast(close=True)
- self._started = False
-
- def status(self) -> dict[str, Any]:
- with self._lock:
- uly = ""
- exp = ""
- index_inst = ""
- if self._watchers:
- last = next(reversed(list(self._watchers.values())))
- uly = str(last.get("underlying") or "")
- exp = str(last.get("exp_time") or "")
- index_inst = str(last.get("index_inst") or "")
- return {
- "ok": True,
- "started": self._started,
- "ws_ok": self._ws.connected,
- "underlying": uly,
- "index_inst": index_inst,
- "index_px": self._index_by_uly.get(uly),
- "watch_exp": exp,
- "watch_count": len(self._meta),
- "watcher_count": len(self._watchers),
- "version": self._version,
- "last_msg_at": self._ws.last_msg_at,
- }
-
- def watch(
- self,
- *,
- underlying: str,
- exp_time: str | int | None,
- contracts: list[dict[str, Any]],
- index_inst_id: str | None = None,
- watcher_id: str | None = None,
- ) -> dict[str, Any]:
- u = (underlying or "ETH").upper()
- index_id = (index_inst_id or f"{u}-USD").strip()
- wid = (watcher_id or "default").strip() or "default"
- meta: dict[str, dict[str, Any]] = {}
- for c in contracts or []:
- if not isinstance(c, dict):
- continue
- inst_id = str(c.get("inst_id") or c.get("instId") or "").strip()
- if not inst_id:
- continue
- meta[inst_id] = {
- "inst_id": inst_id,
- "opt_type": str(c.get("opt_type") or c.get("optType") or "").upper(),
- "strike": _safe_float(c.get("strike")),
- "tick_sz": c.get("tick_sz") or c.get("tickSz"),
- "underlying": u,
- }
- if len(meta) >= max(1, OPTIONS_QUOTE_MAX_INST):
- break
- with self._lock:
- self._watchers[wid] = {
- "underlying": u,
- "exp_time": str(exp_time or ""),
- "index_inst": index_id,
- "meta": meta,
- }
- self._rebuild_subscriptions_locked()
- if not self._started:
- self.start()
- return self.status()
-
- def _rebuild_subscriptions_locked(self) -> None:
- merged: dict[str, dict[str, Any]] = {}
- index_insts: set[str] = set()
- for w in self._watchers.values():
- index_insts.add(str(w.get("index_inst") or ""))
- for inst_id, m in (w.get("meta") or {}).items():
- if inst_id not in merged:
- merged[inst_id] = dict(m)
- if len(merged) >= max(1, OPTIONS_QUOTE_MAX_INST):
- break
- if len(merged) >= max(1, OPTIONS_QUOTE_MAX_INST):
- break
- index_insts = {x for x in index_insts if x}
- self._meta = merged
- self._index_insts = index_insts
- keep = set(merged.keys())
- for k in list(self._tickers.keys()):
- if k not in keep:
- self._tickers.pop(k, None)
- args = [{"channel": "tickers", "instId": iid} for iid in merged]
- for iid in sorted(index_insts):
- args.append({"channel": "index-tickers", "instId": iid})
- self._ws.set_subscriptions(args)
-
- def seed_from_chain(
- self,
- chain: dict[str, Any],
- *,
- exp_time: str | int | None = None,
- watcher_id: str | None = None,
- ) -> None:
- """REST 拉链后预填报价,并默认监视指定/最近到期."""
- if not isinstance(chain, dict):
- return
- u = str(chain.get("underlying") or "ETH").upper()
- index_px = _safe_float(chain.get("index_px"))
- expiries = chain.get("expiries") or []
- target = None
- if exp_time is not None and str(exp_time):
- for e in expiries:
- if str(e.get("exp_time")) == str(exp_time):
- target = e
- break
- if target is None and expiries:
- target = expiries[0]
- contracts = list((target or {}).get("contracts") or [])
- if index_px is not None:
- with self._lock:
- self._index_by_uly[u] = index_px
- self._dirty_index.add(u)
- for c in contracts:
- inst_id = str(c.get("inst_id") or "").strip()
- if not inst_id:
- continue
- patch = {
- "inst_id": inst_id,
- "ask": c.get("ask"),
- "bid": c.get("bid"),
- "ask_sz": c.get("ask_sz"),
- "bid_sz": c.get("bid_sz"),
- "mark_px": c.get("mark_px"),
- "ask_estimated": bool(c.get("ask_estimated")),
- "expiry_be_px": c.get("expiry_be_px"),
- "dist_expiry_be": c.get("dist_expiry_be"),
- "underlying": u,
- }
- with self._lock:
- self._tickers[inst_id] = patch
- self._dirty_inst.add(inst_id)
- self.watch(
- underlying=u,
- exp_time=(target or {}).get("exp_time"),
- contracts=contracts,
- index_inst_id=f"{u}-USD",
- watcher_id=watcher_id or f"seed:{u}",
- )
-
- def _on_ws_data(self, payload: dict[str, Any]) -> None:
- arg = payload.get("arg") or {}
- channel = str(arg.get("channel") or "")
- rows = payload.get("data") or []
- if not isinstance(rows, list) or not rows:
- return
- if channel == "index-tickers":
- row = rows[0] if isinstance(rows[0], dict) else {}
- px = _safe_float(row.get("idxPx"))
- inst = str(row.get("instId") or arg.get("instId") or "")
- uly = inst.split("-")[0].upper() if inst else ""
- if px is None or not uly:
- return
- with self._lock:
- if self._index_by_uly.get(uly) == px:
- return
- self._index_by_uly[uly] = px
- self._dirty_index.add(uly)
- return
- if channel != "tickers":
- return
- for row in rows:
- if not isinstance(row, dict):
- continue
- inst_id = str(row.get("instId") or arg.get("instId") or "").strip()
- if not inst_id:
- continue
- patch = self._ticker_to_patch(inst_id, row)
- with self._lock:
- prev = self._tickers.get(inst_id) or {}
- if (
- prev.get("ask") == patch.get("ask")
- and prev.get("bid") == patch.get("bid")
- and prev.get("ask_sz") == patch.get("ask_sz")
- and prev.get("bid_sz") == patch.get("bid_sz")
- and prev.get("mark_px") == patch.get("mark_px")
- ):
- continue
- self._tickers[inst_id] = patch
- self._dirty_inst.add(inst_id)
-
- def _ticker_to_patch(self, inst_id: str, row: dict[str, Any]) -> dict[str, Any]:
- ask = _safe_float(row.get("askPx"))
- bid = _safe_float(row.get("bidPx"))
- ask_sz = _safe_float(row.get("askSz"))
- bid_sz = _safe_float(row.get("bidSz"))
- mark = _safe_float(row.get("markPx"))
- ask_estimated = False
- with self._lock:
- meta = dict(self._meta.get(inst_id) or {})
- uly = str(meta.get("underlying") or inst_id.split("-")[0] or "").upper()
- index_px = self._index_by_uly.get(uly)
- if ask is None and mark is not None and mark > 0:
- ask = mark
- ask_estimated = True
- ask_sz = None
- if bid is None and mark is not None and mark > 0:
- bid = mark
- be = expiry_breakeven_from_ask(
- opt_type=str(meta.get("opt_type") or ""),
- strike=meta.get("strike"),
- ask_px=None if ask_estimated else ask,
- mark_px=mark,
- )
- dist = idx_distance_to_be(index_px, be)
- return {
- "inst_id": inst_id,
- "underlying": uly,
- "ask": ask,
- "bid": bid,
- "ask_sz": ask_sz,
- "bid_sz": bid_sz,
- "mark_px": mark,
- "ask_estimated": ask_estimated,
- "expiry_be_px": be,
- "dist_expiry_be": dist,
- }
-
- def _flush_loop(self) -> None:
- interval = max(0.05, OPTIONS_QUOTE_FLUSH_MS / 1000.0)
- while not self._stop.is_set():
- if self._stop.wait(interval):
- break
- event = self._build_flush_event()
- if event is None:
- continue
- self._broadcast(event)
-
- def _build_flush_event(self) -> str | None:
- with self._lock:
- if not self._dirty_inst and not self._dirty_index:
- return None
- dirty_uly = set(self._dirty_index)
- self._dirty_index.clear()
- quotes: list[dict[str, Any]] = []
- for inst_id in list(self._dirty_inst):
- q = self._tickers.get(inst_id)
- if q:
- quotes.append(dict(q))
- self._dirty_inst.clear()
- for uly in dirty_uly:
- index_px = self._index_by_uly.get(uly)
- if index_px is None:
- continue
- for inst_id, q in list(self._tickers.items()):
- if str(q.get("underlying") or "").upper() != uly:
- continue
- be = q.get("expiry_be_px")
- dist = idx_distance_to_be(index_px, be if be is not None else None)
- if q.get("dist_expiry_be") != dist:
- q2 = dict(q)
- q2["dist_expiry_be"] = dist
- self._tickers[inst_id] = q2
- quotes.append(q2)
- self._version += 1
- # 多标的时 index_px 取「最近一次 watch」的标的,前端仍以 payload.underlying 过滤
- uly = ""
- exp = ""
- if self._watchers:
- last = next(reversed(list(self._watchers.values())))
- uly = str(last.get("underlying") or "")
- exp = str(last.get("exp_time") or "")
- # 若本批只有单一 underlying 的 quotes/index,优先用它
- quote_ulys = {str(q.get("underlying") or "").upper() for q in quotes if q.get("underlying")}
- if len(dirty_uly) == 1:
- uly = next(iter(dirty_uly))
- elif len(quote_ulys) == 1:
- uly = next(iter(quote_ulys))
- payload = {
- "ok": True,
- "live": True,
- "ws_ok": self._ws.connected,
- "version": self._version,
- "underlying": uly,
- "watch_exp": exp,
- "index_px": self._index_by_uly.get(uly),
- "indexes": dict(self._index_by_uly),
- "quotes": quotes,
- "ts": int(time.time() * 1000),
- }
- return json.dumps(payload, ensure_ascii=False)
-
- def _broadcast(self, event: str | None = None, *, close: bool = False) -> None:
- with self._lock:
- subs = list(self._subscribers)
- dead: list[queue.Queue[str | None]] = []
- for q in subs:
- try:
- q.put_nowait(None if close else event)
- except Exception:
- dead.append(q)
- if dead:
- with self._lock:
- for q in dead:
- if q in self._subscribers:
- self._subscribers.remove(q)
-
- def _subscribe(self) -> queue.Queue[str | None]:
- q: queue.Queue[str | None] = queue.Queue(maxsize=64)
- with self._lock:
- self._subscribers.append(q)
- return q
-
- def _unsubscribe(self, q: queue.Queue[str | None]) -> None:
- with self._lock:
- if q in self._subscribers:
- self._subscribers.remove(q)
-
- def iter_sse(self) -> Iterator[str]:
- q = self._subscribe()
- try:
- yield self._format_event(
- {
- "ok": True,
- "reason": "connect",
- **self.status(),
- "quotes": [],
- "ts": int(time.time() * 1000),
- }
- )
- while True:
- try:
- raw = q.get(timeout=OPTIONS_QUOTE_SSE_HEARTBEAT_SEC)
- except queue.Empty:
- yield ": heartbeat\n\n"
- continue
- if raw is None:
- break
- yield f"event: quotes\ndata: {raw}\n\n"
- finally:
- self._unsubscribe(q)
-
- @staticmethod
- def _format_event(data: dict[str, Any]) -> str:
- return "event: quotes\ndata: " + json.dumps(data, ensure_ascii=False) + "\n\n"
-
-
-options_quote_live = OptionsQuoteLive()
-
-
-def start_options_quote_live() -> OptionsQuoteLive:
- options_quote_live.start()
- return options_quote_live
-
-
-def register_options_quote_live_routes(app: Any, login_required: Callable) -> None:
- from flask import Response, jsonify, request, stream_with_context
-
- start_options_quote_live()
-
- @app.route("/api/options/quotes/stream")
- @login_required
- def api_options_quotes_stream():
- return Response(
- stream_with_context(options_quote_live.iter_sse()),
- mimetype="text/event-stream",
- headers={
- "Cache-Control": "no-cache",
- "Connection": "keep-alive",
- "X-Accel-Buffering": "no",
- },
- )
-
- @app.route("/api/options/quotes/watch", methods=["POST"])
- @login_required
- def api_options_quotes_watch():
- data = request.get_json(silent=True) or {}
- contracts = data.get("contracts") or []
- if not contracts and data.get("inst_ids"):
- contracts = [{"inst_id": x} for x in (data.get("inst_ids") or [])]
- st = options_quote_live.watch(
- underlying=str(data.get("underlying") or "ETH"),
- exp_time=data.get("exp_time"),
- contracts=contracts,
- index_inst_id=data.get("index_inst_id"),
- watcher_id=str(data.get("watcher_id") or "default"),
- )
- return jsonify({"ok": True, **st})
-
- @app.route("/api/options/quotes/status")
- @login_required
- def api_options_quotes_status():
- return jsonify(options_quote_live.status())
diff --git a/lib/options/options_register.py b/lib/options/options_register.py
index c57c37c..97865e0 100644
--- a/lib/options/options_register.py
+++ b/lib/options/options_register.py
@@ -61,14 +61,6 @@ def install_options_trading(app: Flask, repo_root: str, app_module: Any) -> None
register_options_routes(app, cfg)
_register_options_hub_bridge(app, cfg)
if enabled:
- try:
- from lib.options.options_quote_live_lib import register_options_quote_live_routes
-
- register_options_quote_live_routes(app, cfg["login_required"])
- except Exception as e:
- import logging
-
- logging.getLogger(__name__).exception("options quote live init failed: %s", e)
_start_monitor_thread(app, cfg)
@@ -389,14 +381,6 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
# 热更新:每次读 env,保存配置后刷新链即可生效
ask_liq_filter = _env_bool("OKX_OPTIONS_CHAIN_ASK_LIQ_FILTER_ENABLED", True)
budget_buffer = _env_float("OKX_OPTIONS_BUDGET_BUFFER", 0.95)
- if expiries:
- try:
- from lib.options.options_quote_live_lib import options_quote_live
-
- watch_exp = (request.args.get("exp_time") or "").strip() or None
- options_quote_live.seed_from_chain(chain, exp_time=watch_exp)
- except Exception:
- pass
if not expiries:
return jsonify(
{
@@ -417,7 +401,6 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None:
"ask_liq_filter_enabled": ask_liq_filter,
"budget_buffer": budget_buffer,
"trade_budget": cfg["trade_budget"],
- "quote_live": True,
}
)
diff --git a/lib/options/templates/options_panel.html b/lib/options/templates/options_panel.html
index 84306f7..7ba1c7b 100644
--- a/lib/options/templates/options_panel.html
+++ b/lib/options/templates/options_panel.html
@@ -322,4 +322,4 @@
-
+
diff --git a/requirements.txt b/requirements.txt
index 6f01d27..c88f68a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,7 +4,6 @@
flask>=3.0,<4
requests>=2.31,<3
ccxt>=4.2,<5
-websocket-client>=1.6,<2
werkzeug>=3.0,<4
PySocks>=1.7,<2
Pillow>=10.0,<12
diff --git a/tests/test_options_quote_live_lib.py b/tests/test_options_quote_live_lib.py
deleted file mode 100644
index 007164b..0000000
--- a/tests/test_options_quote_live_lib.py
+++ /dev/null
@@ -1,64 +0,0 @@
-"""options_quote_live_lib 单元测试."""
-from __future__ import annotations
-
-import json
-
-from lib.options.options_quote_live_lib import OptionsQuoteLive
-
-
-class _FakeWs:
- connected = True
- last_msg_at = 0.0
-
- def start(self) -> None:
- return None
-
- def stop(self) -> None:
- return None
-
- def set_subscriptions(self, args) -> None:
- self.last_args = list(args)
-
-
-def test_ticker_patch_and_flush():
- live = OptionsQuoteLive()
- live._ws = _FakeWs() # type: ignore[assignment]
- live._started = True
- live.watch(
- underlying="ETH",
- exp_time="1",
- contracts=[{"inst_id": "ETH-USD-260811-2500-C", "opt_type": "C", "strike": 2500}],
- index_inst_id="ETH-USD",
- )
- live._on_ws_data(
- {
- "arg": {"channel": "tickers", "instId": "ETH-USD-260811-2500-C"},
- "data": [
- {
- "instId": "ETH-USD-260811-2500-C",
- "askPx": "12.5",
- "askSz": "3",
- "bidPx": "11.0",
- "bidSz": "2",
- "markPx": "12.0",
- }
- ],
- }
- )
- live._on_ws_data(
- {
- "arg": {"channel": "index-tickers", "instId": "ETH-USD"},
- "data": [{"idxPx": "2600"}],
- }
- )
- raw = live._build_flush_event()
- assert raw is not None
- payload = json.loads(raw)
- assert payload["index_px"] == 2600.0
- assert payload["quotes"]
- q = next(x for x in payload["quotes"] if x["inst_id"] == "ETH-USD-260811-2500-C")
- assert q["ask"] == 12.5
- assert q["ask_sz"] == 3.0
- assert q["expiry_be_px"] == 2512.5
- st = live.status()
- assert st["watch_count"] == 1