crypto_okx/lib/market/nav_spot_tickers_lib.py

"""Nav bar BTC/ETH spot tickers vs Beijing 08:00 open (OKX 1D = UTC 00:00)."""

from __future__ import annotations

import threading
import time
from datetime import datetime, timedelta
from typing import Any, Callable, Optional
from zoneinfo import ZoneInfo

BJ = ZoneInfo("Asia/Shanghai")

# (base, OKX spot instId, display symbol, swap ccxt symbol, index instId)
NAV_SPOT_SPECS = (
    ("ETH", "ETH-USDT", "ETHUSDT", "ETH/USDT:USDT", "ETH-USD"),
    ("BTC", "BTC-USDT", "BTCUSDT", "BTC/USDT:USDT", "BTC-USD"),
)

_OPEN_CACHE_LOCK = threading.Lock()
_OPEN_CACHE: dict[str, dict[str, Any]] = {}
_OPEN_CACHE_TTL_SEC = 120.0


def _safe_float(v: Any) -> Optional[float]:
    try:
        if v is None or v == "":
            return None
        return float(v)
    except (TypeError, ValueError):
        return None


def session_day_key(now: Optional[datetime] = None, reset_hour: int = 8) -> str:
    """Trading day key: before reset_hour BJ belongs to previous calendar day."""
    dt = now or datetime.now(BJ)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=BJ)
    else:
        dt = dt.astimezone(BJ)
    if dt.hour < int(reset_hour):
        dt = dt - timedelta(days=1)
    return dt.strftime("%Y-%m-%d")


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_key))
    except Exception:
        pass
    return None


def _public_index_last(ex: Any, inst_id: str) -> Optional[float]:
    try:
        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_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": bar, "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 _public_index_candle_open(ex: Any, inst_id: str) -> Optional[float]:
    try:
        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:
        return None
    return None


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(cache_key)
        if (
            entry
            and entry.get("day_key") == day_key
            and entry.get("open") is not None
            and now_ts - float(entry.get("fetched_at") or 0) < _OPEN_CACHE_TTL_SEC
        ):
            return _safe_float(entry["open"])

    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[cache_key] = {
                "day_key": day_key,
                "open": open_px,
                "fetched_at": now_ts,
            }
    return open_px


def build_nav_spot_tickers(
    exchange: Any,
    *,
    reset_hour: int = 8,
    now_fn: Optional[Callable[[], datetime]] = None,
) -> list[dict[str, Any]]:
    """Return ETH then BTC spot last/change vs Beijing 08:00 open."""
    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, 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:
            change = last - open_px
            change_pct = (change / open_px) * 100.0
        out.append(
            {
                "base": base,
                "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,
                "change": round(change, 8) if change is not None else None,
                "change_pct": round(change_pct, 4) if change_pct is not None else None,
                "session_day": day_key,
                "anchor": f"BJ {int(reset_hour):02d}:00",
            }
        )
    return out