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

Fix nav spot tickers: register JS in static allowlist and harden price fetch.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-17 11:46:35 +08:00
parent 08029ccb15
commit f03d5d3704
4 changed files with 93 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/market/nav_spot_tickers_lib.py
+90 -23
View File
@@ -10,10 +10,10 @@ from zoneinfo import ZoneInfo
BJ = ZoneInfo("Asia/Shanghai")
# (base, OKX spot instId, display symbol)
# (base, OKX spot instId, display symbol, swap ccxt symbol, index instId)
NAV_SPOT_SPECS = (
("ETH", "ETH-USDT", "ETHUSDT"),
("BTC", "BTC-USDT", "BTCUSDT"),
("ETH", "ETH-USDT", "ETHUSDT", "ETH/USDT:USDT", "ETH-USD"),
("BTC", "BTC-USDT", "BTCUSDT", "BTC/USDT:USDT", "BTC-USD"),
)
_OPEN_CACHE_LOCK = threading.Lock()
@@ -42,38 +42,82 @@ def session_day_key(now: Optional[datetime] = None, reset_hour: int = 8) -> str:
return dt.strftime("%Y-%m-%d")
def _fetch_spot_last(ex: Any, inst_id: str) -> Optional[float]:
def _public_last(ex: Any, inst_id: str, last_key: str = "last") -> Optional[float]:
try:
rows = ex.public_get_market_ticker({"instId": inst_id}).get("data") or []
if rows and isinstance(rows[0], dict):
return _safe_float(rows[0].get("last"))
return _safe_float(rows[0].get(last_key))
except Exception:
pass
return None
def _public_index_last(ex: Any, inst_id: str) -> Optional[float]:
try:
# ccxt spot id: BTC-USDT → BTC/USDT
sym = inst_id.replace("-", "/")
t = ex.fetch_ticker(sym)
return _safe_float(t.get("last"))
rows = ex.public_get_market_index_tickers({"instId": inst_id}).get("data") or []
if rows and isinstance(rows[0], dict):
return _safe_float(rows[0].get("idxPx"))
except Exception:
pass
return None
def _ccxt_last(ex: Any, symbol: str) -> Optional[float]:
try:
t = ex.fetch_ticker(symbol)
return _safe_float(t.get("last") or t.get("close"))
except Exception:
return None
def _fetch_day_open_bj8(ex: Any, inst_id: str) -> Optional[float]:
"""OKX 1D candle open is UTC 00:00 = Beijing 08:00."""
def _fetch_spot_last(ex: Any, spot_inst: str, swap_sym: str, index_inst: str) -> Optional[float]:
# 1) 现货 2) 永续(本站常用) 3) 指数
last = _public_last(ex, spot_inst, "last")
if last is not None:
return last
last = _public_last(ex, f"{spot_inst}-SWAP", "last")
if last is not None:
return last
last = _ccxt_last(ex, swap_sym)
if last is not None:
return last
last = _ccxt_last(ex, spot_inst.replace("-", "/"))
if last is not None:
return last
return _public_index_last(ex, index_inst)
def _public_candle_open(ex: Any, inst_id: str, bar: str = "1D") -> Optional[float]:
try:
rows = ex.public_get_market_candles(
{"instId": inst_id, "bar": "1D", "limit": "1"}
{"instId": inst_id, "bar": bar, "limit": "1"}
).get("data") or []
if rows:
row = rows[0]
# OKX: [ts, o, h, l, c, vol, volCcy, volCcyQuote, confirm]
if isinstance(row, (list, tuple)) and len(row) > 1:
return _safe_float(row[1])
except Exception:
pass
return None
def _public_index_candle_open(ex: Any, inst_id: str) -> Optional[float]:
try:
sym = inst_id.replace("-", "/")
ohlcv = ex.fetch_ohlcv(sym, timeframe="1d", limit=1) or []
rows = ex.public_get_market_index_candles(
{"instId": inst_id, "bar": "1D", "limit": "1"}
).get("data") or []
if rows:
row = rows[0]
if isinstance(row, (list, tuple)) and len(row) > 1:
return _safe_float(row[1])
except Exception:
pass
return None
def _ccxt_day_open(ex: Any, symbol: str) -> Optional[float]:
try:
ohlcv = ex.fetch_ohlcv(symbol, timeframe="1d", limit=1) or []
if ohlcv:
return _safe_float(ohlcv[-1][1])
except Exception:
@@ -81,10 +125,31 @@ def _fetch_day_open_bj8(ex: Any, inst_id: str) -> Optional[float]:
return None
def _cached_day_open(ex: Any, inst_id: str, day_key: str) -> Optional[float]:
def _fetch_day_open_bj8(
ex: Any, spot_inst: str, swap_sym: str, index_inst: str
) -> Optional[float]:
"""OKX 1D candle open is UTC 00:00 = Beijing 08:00."""
open_px = _public_candle_open(ex, spot_inst)
if open_px is not None:
return open_px
open_px = _public_candle_open(ex, f"{spot_inst}-SWAP")
if open_px is not None:
return open_px
open_px = _ccxt_day_open(ex, swap_sym)
if open_px is not None:
return open_px
open_px = _ccxt_day_open(ex, spot_inst.replace("-", "/"))
if open_px is not None:
return open_px
return _public_index_candle_open(ex, index_inst)
def _cached_day_open(
ex: Any, cache_key: str, spot_inst: str, swap_sym: str, index_inst: str, day_key: str
) -> Optional[float]:
now_ts = time.time()
with _OPEN_CACHE_LOCK:
entry = _OPEN_CACHE.get(inst_id)
entry = _OPEN_CACHE.get(cache_key)
if (
entry
and entry.get("day_key") == day_key
@@ -93,10 +158,10 @@ def _cached_day_open(ex: Any, inst_id: str, day_key: str) -> Optional[float]:
):
return _safe_float(entry["open"])
open_px = _fetch_day_open_bj8(ex, inst_id)
open_px = _fetch_day_open_bj8(ex, spot_inst, swap_sym, index_inst)
if open_px is not None:
with _OPEN_CACHE_LOCK:
_OPEN_CACHE[inst_id] = {
_OPEN_CACHE[cache_key] = {
"day_key": day_key,
"open": open_px,
"fetched_at": now_ts,
@@ -114,9 +179,11 @@ def build_nav_spot_tickers(
now = now_fn() if now_fn else datetime.now(BJ)
day_key = session_day_key(now, reset_hour=reset_hour)
out: list[dict[str, Any]] = []
for base, inst_id, display in NAV_SPOT_SPECS:
last = _fetch_spot_last(exchange, inst_id)
open_px = _cached_day_open(exchange, inst_id, day_key)
for base, spot_inst, display, swap_sym, index_inst in NAV_SPOT_SPECS:
last = _fetch_spot_last(exchange, spot_inst, swap_sym, index_inst)
open_px = _cached_day_open(
exchange, spot_inst, spot_inst, swap_sym, index_inst, day_key
)
change = None
change_pct = None
if last is not None and open_px is not None and open_px != 0:
@@ -125,7 +192,7 @@ def build_nav_spot_tickers(
out.append(
{
"base": base,
"inst_id": inst_id,
"inst_id": spot_inst,
"symbol": display,
"last": round(last, 8) if last is not None else None,
"open": round(open_px, 8) if open_px is not None else None,