diff --git a/lib/common/static/options_panel.js b/lib/common/static/options_panel.js
index b8504d3..fffd075 100644
--- a/lib/common/static/options_panel.js
+++ b/lib/common/static/options_panel.js
@@ -7,6 +7,10 @@
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",
@@ -40,11 +44,17 @@
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;
- /** 链卖一/买一静默刷新节流:无推送,靠拉;过密会撞 OKX 50011 */
- const CHAIN_SOFT_POLL_MS = 15000;
+ /** SSE/WS 断开时的 REST 兜底;连上后停用 */
+ const CHAIN_SOFT_POLL_MS = 30000;
+ const ORDER_QUOTE_LIVE_MIN_MS = 800;
const orderPanelHome = (function () {
const host = document.getElementById("opt-order-panel-host");
return host ? host.parentElement : null;
@@ -659,16 +669,214 @@
const line = document.getElementById("opt-index-line");
if (line) {
const liqHint = askLiqFilterOn() ? "仅显示卖一深度≥1张" : "显示全部卖一(含估算~)";
- const ageHint = chainQuotedAt ? " · 链报价 " + fmtChainQuotedAt() + "（约每15s静默刷新）" : "";
+ let liveHint = "";
+ if (quoteLiveOk && quoteLiveWsOk) {
+ liveHint = chainQuotedAt
+ ? " · WS实时 " + fmtChainQuotedAt()
+ : " · WS实时";
+ } else if (quoteLiveOk) {
+ liveHint = " · 推送已连,等待 OKX WS…";
+ } else if (chainQuotedAt) {
+ liveHint = " · 链报价 " + fmtChainQuotedAt() + "（REST兜底）";
+ }
line.textContent =
"指数 " + state.underlying + " ≈ " + fmt(idx, 2) +
- " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + ageHint;
+ " · 默认最近一期 · " + liqHint + " · 实值含平值 · 虚值=价外" + liveHint;
}
}
+ 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;
@@ -1328,6 +1536,8 @@
}
// soft 时保留 selectedInst；renderStrikes 会先 park 再按 prevSelected 静默重挂下单面板
renderStrikes();
+ void watchCurrentExpiryQuotes();
+ startQuoteLiveStream();
} catch (e) {
if (seq !== chainLoadSeq || soft) return;
setExpirySelectStatus("选择到期日");
@@ -2236,6 +2446,7 @@
const expandCb = document.getElementById("opt-strike-expand-all");
if (expandCb) expandCb.checked = false;
renderStrikes();
+ void watchCurrentExpiryQuotes();
}
function bootOptionsPanel() {
@@ -2247,6 +2458,7 @@
refreshPendingOrders();
startPendingOrdersPoll();
startChainSoftPoll();
+ startQuoteLiveStream();
const hasCache =
chainHasExpiries(panelCache.chain) &&
panelCache.underlying === state.underlying &&
@@ -2256,7 +2468,8 @@
renderExpiries();
renderStrikes();
refreshAllPositions();
- // 后台静默刷新,避免缓存过期后到期日变空 / 卖一过期
+ void watchCurrentExpiryQuotes();
+ // 后台静默刷新结构;卖一优先走 WS
softRefreshChainThrottled(true);
return;
}
@@ -2381,7 +2594,7 @@
window.OptionsPanelLive = {
refreshSoft: function () {
refreshAllPositions();
- // embed SSE 只通知「该拉了」,不推送链报价;这里节流拉新鲜卖一/买一
+ // 有 WS 实时报价时不再 REST 刷链;断开时才兜底
softRefreshChainThrottled(false);
},
refreshChain: loadChain,
diff --git a/lib/exchange/okx_public_ws_lib.py b/lib/exchange/okx_public_ws_lib.py
new file mode 100644
index 0000000..d6b1865
--- /dev/null
+++ b/lib/exchange/okx_public_ws_lib.py
@@ -0,0 +1,199 @@
+"""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
new file mode 100644
index 0000000..22a9e66
--- /dev/null
+++ b/lib/options/options_quote_live_lib.py
@@ -0,0 +1,458 @@
+"""期权链实时报价: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 97865e0..c57c37c 100644
--- a/lib/options/options_register.py
+++ b/lib/options/options_register.py
@@ -61,6 +61,14 @@ 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)
@@ -381,6 +389,14 @@ 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(
{
@@ -401,6 +417,7 @@ 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 7ba1c7b..84306f7 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 c88f68a..6f01d27 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,6 +4,7 @@
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
new file mode 100644
index 0000000..007164b
--- /dev/null
+++ b/tests/test_options_quote_live_lib.py
@@ -0,0 +1,64 @@
+"""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