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

fix(hub): restore pre-WS quote path and relieve Gate/account contention

Browse Source 
Two audit rounds after WS rollback: lighten hub options snapshot, cache hub balances without extra fetch_balance, soft-poll single-flight, and document fixes in R1/R2 reports.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-11 12:30:27 +08:00
parent d592632834
commit bab42b1b53
12 changed files with 209 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/exchange/okx_options_lib.py
+30 -2
View File
@@ -30,6 +30,9 @@ _INSTRUMENTS_CACHE: dict[str, dict[str, Any]] = {}
_INSTRUMENTS_CACHE_LOCK = threading.Lock()
_INSTRUMENTS_CACHE_TTL_SEC = 90.0
_INSTRUMENTS_STALE_SEC = 600.0
_TICKERS_CACHE: dict[str, dict[str, Any]] = {}
_TICKERS_CACHE_LOCK = threading.Lock()
_TICKERS_CACHE_TTL_SEC = float(os.getenv("OKX_OPTIONS_TICKERS_CACHE_SEC", "10") or "10")
def invalidate_options_balance_cache() -> None:
@@ -712,17 +715,40 @@ def fetch_option_instruments(
return []
def fetch_option_tickers(ex: ccxt.okx, inst_family: str) -> dict[str, dict[str, Any]]:
def fetch_option_tickers(
ex: ccxt.okx,
inst_family: str,
*,
force: bool = False,
) -> dict[str, dict[str, Any]]:
family = (inst_family or "").strip()
if not family:
return {}
now = time.time()
with _TICKERS_CACHE_LOCK:
cached = _TICKERS_CACHE.get(family)
if (
not force
and cached
and now - float(cached.get("updated_at") or 0) < max(1.0, _TICKERS_CACHE_TTL_SEC)
and isinstance(cached.get("rows"), dict)
and cached["rows"]
):
return dict(cached["rows"])
out: dict[str, dict[str, Any]] = {}
last_err: BaseException | None = None
for attempt in range(3):
try:
rows = ex.public_get_market_tickers(
{"instType": "OPTION", "instFamily": inst_family}
{"instType": "OPTION", "instFamily": family}
).get("data") or []
for r in rows:
if isinstance(r, dict) and r.get("instId"):
out[str(r["instId"])] = r
if out:
with _TICKERS_CACHE_LOCK:
_TICKERS_CACHE[family] = {"updated_at": time.time(), "rows": dict(out)}
return out
except Exception as e:
last_err = e
@@ -730,6 +756,8 @@ def fetch_option_tickers(ex: ccxt.okx, inst_family: str) -> dict[str, dict[str,
time.sleep(0.6 * (attempt + 1))
continue
break
if cached and isinstance(cached.get("rows"), dict) and cached["rows"]:
return dict(cached["rows"])
if last_err is not None and _is_okx_rate_limit(last_err):
return out
return out