hyperliquid_api.py

import requests
import json
import time
import websocket
import threading
import queue
from typing import Any

try:
    from urllib3.util.retry import Retry  # type: ignore
    from requests.adapters import HTTPAdapter
except Exception:  # pragma: no cover
    Retry = None  # type: ignore[assignment]
    HTTPAdapter = None  # type: ignore[assignment]
from eth_account import Account
from eth_account import messages as _eth_messages

# Avoid importing legacy symbols directly (Pylance stubs for modern eth-account
# don't expose them), but still support older eth-account at runtime.
if hasattr(_eth_messages, "encode_structured_data"):
    _encode_eip712 = getattr(_eth_messages, "encode_structured_data")  # type: ignore[attr-defined]
else:  # eth-account>=0.10
    _encode_typed_data = getattr(_eth_messages, "encode_typed_data")

    def _encode_eip712(typed_data: dict):
        # eth-account>=0.10 (incl 0.13.x) expects `full_message=`.
        # Older variants accepted positional; keep a small fallback.
        try:
            return _encode_typed_data(full_message=typed_data)
        except TypeError:
            return _encode_typed_data(typed_data)  # type: ignore[misc]
import pandas as pd
from config import Config

class HyperliquidAPI:
    def __init__(self):
        self.meta = None
        self.asset_map = {}  # symbol -> asset_id (index for default dex; offset for builder dexes)
        self.universe = None
        self.max_leverage_map = {}  # symbol -> int max leverage
        # Multi-dex support (HIP-3 builder dexes like 'flx', 'xyz').
        # A coin name with a 'dex:' prefix (e.g. 'flx:OIL') triggers builder-dex routing.
        # asset_id formula per HL docs: 100000 + perp_dex_index*10000 + index_in_meta.
        self._perp_dexs: list[dict | None] | None = None  # cache of /info perpDexs
        self._dex_meta: dict[str, dict] = {}  # dex_name -> meta (universe, etc.)
        self._dex_loaded: set[str] = set()    # builder dex names we've fetched + indexed
        self._last_set_leverage = {}  # (symbol, leverage_type) -> int
        self.ws = None
        self.candle_data = {}
        self.candle_data_h1 = {}
        self._candle_lock = threading.Lock()

        # WebSocket event handling: keep WS callbacks lightweight.
        # Heavy strategy work in the WS thread can block ping/pong handling and cause disconnects.
        self._ws_event_q: "queue.Queue[str]" = queue.Queue()
        self._ws_pending: set[str] = set()
        self._ws_pending_lock = threading.Lock()
        self._ws_worker_started = False
        self._ws_on_candle_callback = None
        self._ws_logger = None
        self._ws_reconnect_lock = threading.Lock()
        self._ws_reconnecting = False
        self._ws_last_restart_ts = 0.0

        # REST hygiene: reuse a session, set timeouts, and retry transient failures.
        self._session = requests.Session()
        if Retry is not None and HTTPAdapter is not None:
            retry = Retry(
                total=3,
                connect=3,
                read=3,
                backoff_factor=0.3,
                status_forcelist=(429, 500, 502, 503, 504),
                allowed_methods=frozenset(["GET", "POST"]),
                raise_on_status=False,
            )
            self._session.mount("https://", HTTPAdapter(max_retries=retry))

        self._price_cache: dict[str, tuple[float, float]] = {}
        self._price_last_req_ts: dict[str, float] = {}

        # Order flow tracker (set via set_flow_tracker before start_ws)
        self._flow_tracker = None

        # Separate session for Binance (external API) to avoid thread contention
        # with Hyperliquid calls.  requests.Session is NOT fully thread-safe,
        # so each external API gets its own session.
        self._binance_session = requests.Session()
        if Retry is not None and HTTPAdapter is not None:
            binance_retry = Retry(
                total=2,
                connect=2,
                read=2,
                backoff_factor=0.5,
                status_forcelist=(429, 500, 502, 503, 504),
                allowed_methods=frozenset(["GET"]),
                raise_on_status=False,
            )
            self._binance_session.mount("https://", HTTPAdapter(max_retries=binance_retry))

        self.fetch_meta_and_ctxs()  # Fixed name

    def _post_info(self, payload: dict[str, Any], *, timeout_s: float | None = None) -> Any:
        timeout = float(timeout_s) if timeout_s is not None else float(getattr(Config, "REST_TIMEOUT_SECONDS", 2.0))
        resp = self._session.post(Config.INFO_URL, json=payload, timeout=timeout)
        resp.raise_for_status()
        return resp.json()

    # ── Multi-dex (HIP-3 builder dex) helpers ────────────────────────────
    @staticmethod
    def split_dex(symbol: str) -> tuple[str, str]:
        """Return (dex_name, asset_name) for a coin string.

        Default dex coins ('BTC', 'ETH') -> ('', 'BTC').
        Builder dex coins ('flx:OIL') -> ('flx', 'OIL').
        """
        s = str(symbol or "").strip()
        if ":" in s:
            d, a = s.split(":", 1)
            return d.strip(), a.strip()
        return "", s

    def _ensure_perp_dexs(self) -> list[dict | None]:
        """Lazy-load /info perpDexs; index 0 is null (default dex)."""
        if self._perp_dexs is None:
            data = self._post_info({"type": "perpDexs"}, timeout_s=10.0)
            if not isinstance(data, list):
                raise Exception(f"Unexpected perpDexs response: {data}")
            self._perp_dexs = data
        return self._perp_dexs

    def get_perp_dex_index(self, dex_name: str) -> int | None:
        """Return the perp_dex_index for a builder dex name (e.g. 'flx' -> 2)."""
        if not dex_name:
            return 0
        for i, d in enumerate(self._ensure_perp_dexs()):
            if isinstance(d, dict) and d.get("name") == dex_name:
                return i
        return None

    def load_builder_dex(self, dex_name: str) -> dict:
        """Fetch metaAndAssetCtxs for a builder dex and index its assets.

        Indexes asset_ids (100000 + perp_dex_index*10000 + index_in_meta) and
        max-leverage into the global self.asset_map / self.max_leverage_map so
        the rest of the code doesn't need to know which dex a coin belongs to.
        Symbols are stored with the full 'dex:asset' name (e.g. 'flx:OIL').
        """
        if dex_name in self._dex_loaded:
            return self._dex_meta[dex_name]

        idx = self.get_perp_dex_index(dex_name)
        if idx is None:
            raise Exception(f"Unknown builder perp dex: {dex_name!r}")

        data = self._post_info({"type": "metaAndAssetCtxs", "dex": dex_name}, timeout_s=10.0)
        if not isinstance(data, list) or len(data) < 2:
            raise Exception(f"Unexpected metaAndAssetCtxs response for dex={dex_name!r}: {data}")

        meta = data[0]
        universe = meta.get("universe") or []
        for i, asset in enumerate(universe):
            full_name = asset.get("name")  # e.g. 'flx:OIL' (HL already prefixes)
            if not full_name:
                continue
            asset_id = 100_000 + int(idx) * 10_000 + i
            self.asset_map[full_name] = asset_id
            ml = asset.get("maxLeverage")
            if ml is not None:
                try:
                    self.max_leverage_map[full_name] = int(ml)
                except Exception:
                    pass

        self._dex_meta[dex_name] = meta
        self._dex_loaded.add(dex_name)
        return meta

    def _autoload_dexes_for_coins(self, coins: list[str]) -> None:
        """Inspect coin list for 'dex:' prefixes and load each referenced builder dex."""
        wanted: set[str] = set()
        for c in coins or []:
            d, _ = self.split_dex(c)
            if d:
                wanted.add(d)
        for d in wanted:
            try:
                self.load_builder_dex(d)
            except Exception as e:
                print(f"[multi-dex] Failed to load builder dex {d!r}: {e}")

    def _ensure_ws_worker(self):
        if self._ws_worker_started:
            return
        self._ws_worker_started = True

        def worker_loop():
            while True:
                coin = self._ws_event_q.get()
                with self._ws_pending_lock:
                    self._ws_pending.discard(coin)
                cb = self._ws_on_candle_callback
                if not cb:
                    continue
                try:
                    cb(coin)
                except Exception as e:
                    if self._ws_logger:
                        self._ws_logger.error(f"WS candle callback error (safe skip): {e}")

        t = threading.Thread(target=worker_loop, name="ws-candle-worker", daemon=True)
        t.start()

    def set_flow_tracker(self, tracker) -> None:
        """Attach an OrderFlowTracker so WS trades/l2Book are forwarded."""
        self._flow_tracker = tracker

    def _enqueue_ws_candle_event(self, coin: str) -> None:
        # Deduplicate per coin to avoid queue blow-ups during bursts.
        with self._ws_pending_lock:
            if coin in self._ws_pending:
                return
            self._ws_pending.add(coin)
        self._ws_event_q.put(coin)

    @staticmethod
    def _interval_ms(interval: str) -> int:
        # Hyperliquid intervals are typically like: 1m, 5m, 15m, 1h, 4h, 1d
        interval = interval.strip().lower()
        if interval.endswith("m"):
            return int(interval[:-1]) * 60_000
        if interval.endswith("h"):
            return int(interval[:-1]) * 60 * 60_000
        if interval.endswith("d"):
            return int(interval[:-1]) * 24 * 60 * 60_000
        raise ValueError(f"Unsupported interval: {interval}")

    def fetch_meta_and_ctxs(self):
        """Fetch meta + asset contexts (includes 24h volume & OI)"""
        payload = {"type": "metaAndAssetCtxs"}  # Correct endpoint (with 's'), no 'user'
        response = self._session.post(
            Config.INFO_URL,
            json=payload,
            timeout=float(getattr(Config, "REST_TIMEOUT_SECONDS", 2.0)),
        )
        
        if response.status_code != 200:
            raise Exception(f"metaAndAssetCtxs failed (status {response.status_code}): {response.text}")
        
        try:
            data = response.json()
        except json.JSONDecodeError:
            raise Exception(f"Invalid JSON response: {response.text}")
        
        if not isinstance(data, list) or len(data) < 2:
            raise Exception(f"Unexpected response format: {data}")
        
        self.meta = data[0]
        asset_ctxs = data[1]  # List of contexts in same order as universe
        
        self.universe = self.meta['universe']

        # Build max leverage map
        self.max_leverage_map = {}
        for asset in self.universe:
            name = asset.get("name")
            if not name:
                continue
            ml = asset.get("maxLeverage")
            if ml is None:
                continue
            try:
                self.max_leverage_map[name] = int(ml)
            except Exception:
                # Best-effort; ignore weird values
                continue
        
        # Build asset map (index = asset_id)
        for i, asset in enumerate(self.universe):
            symbol = asset['name']
            self.asset_map[symbol] = i
        
        # Dynamic top N selection
        top_n = Config.TOP_N_COINS
        if top_n > 0:
            sort_key = 'dayNtlVlm' if Config.VOLUME_SORT else 'openInterest'
            # Get indices sorted by volume/OI descending
            indices = list(range(len(asset_ctxs)))
            indices.sort(key=lambda i: float(asset_ctxs[i].get(sort_key, '0') or '0'), reverse=True)
            top_indices = indices[:top_n]
            top_coins = [self.universe[i]['name'] for i in top_indices]
            Config.COINS = top_coins
            print(f"Auto-selected top {top_n} coins by {'24h volume' if Config.VOLUME_SORT else 'OI'}: {Config.COINS}")
        else:
            print(f"Using manual COINS list (cleaned): {Config.COINS}")

        # Multi-dex (HIP-3): if any coin has a 'dex:' prefix, load that builder dex's
        # universe so asset_ids and max-leverage are indexed for live order routing.
        try:
            self._autoload_dexes_for_coins(list(Config.COINS))
        except Exception as e:
            print(f"[multi-dex] auto-load skipped (non-fatal): {e}")

        # Initialize candle DataFrames
        self.candle_data = {
            coin: pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            for coin in Config.COINS
        }

        self.candle_data_h1 = {
            coin: pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            for coin in Config.COINS
        }

        print("Meta + asset contexts fetched successfully!")
        print(f"Monitoring {len(Config.COINS)} coins: {Config.COINS}")

    def refresh_top_coins(self, logger=None):
        """Re-fetch metaAndAssetCtxs and refresh Config.COINS if TOP_N_COINS > 0."""
        top_n = Config.TOP_N_COINS
        if top_n <= 0:
            return False

        old_coins = list(Config.COINS)

        payload = {"type": "metaAndAssetCtxs"}
        response = self._session.post(
            Config.INFO_URL,
            json=payload,
            timeout=float(getattr(Config, "REST_TIMEOUT_SECONDS", 2.0)),
        )
        response.raise_for_status()
        data = response.json()
        if not isinstance(data, list) or len(data) < 2:
            raise Exception(f"Unexpected response format: {data}")

        meta = data[0]
        asset_ctxs = data[1]
        universe = meta.get("universe") or []

        sort_key = 'dayNtlVlm' if Config.VOLUME_SORT else 'openInterest'
        indices = list(range(len(asset_ctxs)))
        indices.sort(key=lambda i: float(asset_ctxs[i].get(sort_key, '0') or '0'), reverse=True)
        top_indices = indices[:top_n]
        top_coins = [universe[i]['name'] for i in top_indices]

        Config.COINS = top_coins
        # Ensure candle stores exist
        for coin in Config.COINS:
            if coin not in self.candle_data:
                self.candle_data[coin] = pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            if coin not in self.candle_data_h1:
                self.candle_data_h1[coin] = pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

        msg = f"Refreshed top {top_n} coins by {'24h volume' if Config.VOLUME_SORT else 'OI'}: {Config.COINS}"
        if logger:
            logger.info(msg)
        else:
            print(msg)

        return list(Config.COINS) != old_coins

    def request_ws_restart(self, *, reason: str, logger=None) -> bool:
        """Request a WS restart so subscriptions match the current Config.COINS.

        This is used when TOP_N_COINS changes the coin universe at runtime.
        Returns True if a restart was initiated.
        """
        now = time.time()
        cooldown_s = float(getattr(Config, "WS_RESTART_COOLDOWN_SECONDS", 10.0))
        if (now - float(self._ws_last_restart_ts or 0.0)) < cooldown_s:
            return False
        self._ws_last_restart_ts = float(now)

        if logger:
            logger.warning(f"[WS] Restart requested: {reason}")
        try:
            if self.ws is not None:
                self.ws.close()
                return True
        except Exception as e:
            if logger:
                logger.warning(f"[WS] Restart request failed (safe skip): {e}")
        return False

    def refresh_candles_for_coin(self, coin: str, logger=None):
        """Fetch fresh entry-tf + filter-tf candle snapshots and update internal caches."""
        try:
            df_15m = self.fetch_candles_snapshot(coin, interval=Config.CANDLE_INTERVAL, limit=Config.BOOTSTRAP_CANDLES)
            if not df_15m.empty:
                with self._candle_lock:
                    self.candle_data[coin] = df_15m
        except Exception as e:
            if logger:
                logger.warning(f"{Config.CANDLE_INTERVAL} candleSnapshot failed for {coin}: {e}")

    def get_candles_15m(self, coin: str) -> pd.DataFrame:
        with self._candle_lock:
            df = self.candle_data.get(coin)
            return df.copy() if isinstance(df, pd.DataFrame) else pd.DataFrame()

    def get_candles_1h(self, coin: str) -> pd.DataFrame:
        with self._candle_lock:
            df = self.candle_data_h1.get(coin)
            return df.copy() if isinstance(df, pd.DataFrame) else pd.DataFrame()

    def get_asset_id(self, symbol: str) -> int:
        if symbol not in self.asset_map:
            raise ValueError(f"Symbol {symbol} not found in universe")
        return self.asset_map[symbol]

    def get_max_leverage(self, symbol: str) -> int | None:
        """Return the exchange-defined maximum leverage for this perp symbol.

        Sourced from /info metaAndAssetCtxs -> meta.universe[].maxLeverage.
        """
        if not symbol:
            return None
        return self.max_leverage_map.get(symbol)

    def update_leverage(self, symbol: str, leverage: int, leverage_type: str = "isolated"):
        """Send an updateLeverage signed action.

        Docs: Exchange endpoint -> "Update leverage".
        leverage_type: "isolated" or "cross".
        """
        if not Config.PRIVATE_KEY:
            raise ValueError("PRIVATE_KEY is required for live actions")
        asset_id = self.get_asset_id(symbol)
        lev_type = (leverage_type or "isolated").strip().lower()
        is_cross = lev_type == "cross"

        # Hyperliquid action fields are JSON-encoded bytes in our signer.
        # Use strings for numeric fields to match other action conventions (orders use strings).
        action = {
            "type": "updateLeverage",
            "asset": asset_id,
            "isCross": is_cross,
            "leverage": str(int(leverage)),
        }
        signed = self.sign_order(action)
        return self.place_order(signed)

    def maybe_set_max_leverage(self, symbol: str, logger=None) -> bool:
        """If enabled for live trading, set max leverage for `symbol` before ordering.

        Returns True if leverage is set (or already set / not needed). False if it failed.
        """
        if Config.PAPER_TRADING:
            return True
        if not Config.LIVE_AUTO_SET_MAX_LEVERAGE:
            return True

        ex_lev = self.get_max_leverage(symbol)
        if ex_lev is None:
            if logger:
                logger.warning(f"[LEVERAGE] No maxLeverage found for {symbol}; skipping")
            return False

        lev_cap = max(1, int(getattr(Config, "LIVE_MAX_LEVERAGE_CAP", 25)))
        lev = min(int(ex_lev), int(lev_cap))

        lev_type = (Config.LIVE_LEVERAGE_TYPE or "isolated").strip().lower()
        if lev_type not in {"isolated", "cross"}:
            if logger:
                logger.warning(f"[LEVERAGE] Invalid LIVE_LEVERAGE_TYPE={Config.LIVE_LEVERAGE_TYPE}; using isolated")
            lev_type = "isolated"

        cache_key = (symbol, lev_type)
        if self._last_set_leverage.get(cache_key) == lev:
            return True

        try:
            resp = self.update_leverage(symbol, lev, leverage_type=lev_type)
            ok = bool(resp) and resp.get("status") == "ok"
            if ok:
                self._last_set_leverage[cache_key] = lev
                if logger:
                    logger.info(
                        f"[LEVERAGE] Set {symbol} leverage={lev}x type={lev_type} (exchange_max={int(ex_lev)} cap={int(lev_cap)})"
                    )
                return True
            if logger:
                logger.warning(f"[LEVERAGE] updateLeverage failed for {symbol}: {resp}")
            return False
        except Exception as e:
            if logger:
                logger.warning(f"[LEVERAGE] updateLeverage error for {symbol}: {e}")
            return False

    def sign_order(self, action: dict, nonce: int | None = None):
        if nonce is None:
            nonce = int(time.time() * 1000)

        if not Config.PRIVATE_KEY:
            raise ValueError("PRIVATE_KEY is required for signed actions")
        
        domain = {
            "name": "HyperliquidSignTransaction",
            "version": "1",
            "chainId": 1337 if Config.TESTNET else 42161,
            "verifyingContract": "0x0000000000000000000000000000000000000000"
        }
        
        types = {
            "EIP712Domain": [
                {"name": "name", "type": "string"},
                {"name": "version", "type": "string"},
                {"name": "chainId", "type": "uint256"},
                {"name": "verifyingContract", "type": "address"}
            ],
            "Transaction": [
                {"name": "action", "type": "bytes"},
                {"name": "nonce", "type": "uint64"},
                {"name": "vaultAddress", "type": "address"}
            ]
        }
        
        message = {
            "action": json.dumps(action).encode('utf-8'),
            "nonce": nonce,
            "vaultAddress": None
        }
        
        typed_data = {
            "types": types,
            "domain": domain,
            "primaryType": "Transaction",
            "message": message
        }
        
        encoded = _encode_eip712(typed_data)
        signed = Account.sign_message(encoded, Config.PRIVATE_KEY)
        
        return {
            "action": action,
            "nonce": nonce,
            "signature": {
                "r": hex(signed.r),
                "s": hex(signed.s),
                "v": signed.v
            },
            "vaultAddress": None
        }

    def place_order(self, signed_payload: dict):
        response = requests.post(Config.EXCHANGE_URL, json=signed_payload)
        if response.status_code != 200:
            print("Order failed:", response.text)
            return None
        return response.json()

    def fetch_funding_history(self, coin: str, start_time_ms: int, end_time_ms: int | None = None) -> pd.DataFrame:
        """Fetch hourly funding history via /info fundingHistory.

        HL pays funding every 1h. Each row: {timestamp_ms, funding_rate, premium}.
        Endpoint returns up to 500 rows per call → paginates by advancing startTime.
        """
        if end_time_ms is None:
            end_time_ms = int(time.time() * 1000)

        all_rows: list[dict] = []
        cursor = int(start_time_ms)
        # Hard safety cap to avoid runaway loops on edge cases.
        for _ in range(50):
            payload = {"type": "fundingHistory", "coin": coin, "startTime": cursor, "endTime": end_time_ms}
            try:
                resp = self._session.post(Config.INFO_URL, json=payload,
                                          timeout=float(getattr(Config, 'REST_TIMEOUT_SECONDS', 10.0)))
                resp.raise_for_status()
                batch = resp.json()
            except Exception:
                break
            if not isinstance(batch, list) or not batch:
                break
            for r in batch:
                if not isinstance(r, dict):
                    continue
                all_rows.append({
                    "timestamp": int(r.get("time", 0)),
                    "funding_rate": float(r.get("fundingRate", 0.0) or 0.0),
                    "premium": float(r.get("premium", 0.0) or 0.0),
                })
            last_ts = int(batch[-1].get("time", 0))
            if last_ts <= cursor or last_ts >= end_time_ms or len(batch) < 500:
                break
            cursor = last_ts + 1  # advance past last record

        df = pd.DataFrame(all_rows, columns=["timestamp", "funding_rate", "premium"])
        if df.empty:
            return df
        df = df.drop_duplicates(subset=["timestamp"]).sort_values("timestamp").reset_index(drop=True)
        return df

    def fetch_candles_snapshot(self, coin: str, interval: str, limit: int = 500) -> pd.DataFrame:
        """Fetch historical candles via /info candleSnapshot.

        Returns a DataFrame with columns: timestamp, open, high, low, close, volume.
        Timestamp is milliseconds since epoch.
        """
        end_time = int(time.time() * 1000)
        start_time = end_time - (limit * self._interval_ms(interval))
        payload = {
            "type": "candleSnapshot",
            "req": {
                "coin": coin,
                "interval": interval,
                "startTime": start_time,
                "endTime": end_time,
            },
        }
        resp = self._session.post(Config.INFO_URL, json=payload, timeout=float(getattr(Config, 'REST_TIMEOUT_SECONDS', 10.0)))
        resp.raise_for_status()
        data = resp.json()
        if not isinstance(data, list):
            raise Exception(f"Unexpected candleSnapshot response for {coin}: {data}")

        rows = []
        for c in data:
            if not isinstance(c, dict):
                continue
            rows.append(
                {
                    "timestamp": int(c.get("t", 0)),
                    "open": float(c.get("o", 0)),
                    "high": float(c.get("h", 0)),
                    "low": float(c.get("l", 0)),
                    "close": float(c.get("c", 0)),
                    "volume": float(c.get("v", 0)),
                }
            )

        df = pd.DataFrame(rows, columns=["timestamp", "open", "high", "low", "close", "volume"])
        if df.empty:
            return df
        df = df.drop_duplicates(subset=["timestamp"]).sort_values("timestamp").reset_index(drop=True)
        return df

    # Binance symbol mapping for spot order book
    _BINANCE_SYMBOL_MAP = {
        "BTC": "BTCUSDT",
        "ETH": "ETHUSDT",
        "SOL": "SOLUSDT",
        "XRP": "XRPUSDT",
    }
    _BINANCE_SPOT_URL = "https://api.binance.com/api/v3/depth"

    def fetch_l2_book(self, coin: str) -> dict:
        """Fetch Hyperliquid perp L2 order book snapshot (up to 20 levels per side).

        Returns: {"coin": str, "time": int,
                  "bids": [{"px": float, "sz": float, "n": int}, ...],
                  "asks": [{"px": float, "sz": float, "n": int}, ...]}
        """
        payload = {"type": "l2Book", "coin": coin}
        # (connect_timeout, read_timeout) — bounded to prevent thread hangs.
        timeout = (3, float(getattr(Config, "REST_TIMEOUT_SECONDS", 5.0)))
        resp = self._session.post(Config.INFO_URL, json=payload, timeout=timeout)
        resp.raise_for_status()
        data = resp.json()
        levels = data.get("levels", [[], []])

        def _parse_side(raw_levels):
            out = []
            for lv in raw_levels:
                out.append({
                    "px": float(lv.get("px", 0)),
                    "sz": float(lv.get("sz", 0)),
                    "n": int(lv.get("n", 0)),
                })
            return out

        return {
            "coin": data.get("coin", coin),
            "time": int(data.get("time", 0)),
            "bids": _parse_side(levels[0] if len(levels) > 0 else []),
            "asks": _parse_side(levels[1] if len(levels) > 1 else []),
        }

    def fetch_binance_spot_book(self, coin: str, limit: int = 20) -> dict:
        """Fetch Binance spot order book (no API key required).

        Returns same shape as fetch_l2_book for easy comparison:
          {"coin": str, "bids": [{"px": float, "sz": float}, ...],
           "asks": [{"px": float, "sz": float}, ...]}
        Uses a dedicated session to avoid thread contention with Hyperliquid calls.
        """
        symbol = self._BINANCE_SYMBOL_MAP.get(coin, f"{coin}USDT")
        # (connect_timeout, read_timeout) — generous but bounded.
        timeout = (5, 10)
        try:
            resp = self._binance_session.get(
                self._BINANCE_SPOT_URL,
                params={"symbol": symbol, "limit": limit},
                timeout=timeout,
            )
            resp.raise_for_status()
            data = resp.json()
        except Exception:
            return {"coin": coin, "bids": [], "asks": []}

        def _parse(levels):
            return [{"px": float(px), "sz": float(sz)} for px, sz in levels]

        return {
            "coin": coin,
            "bids": _parse(data.get("bids", [])),
            "asks": _parse(data.get("asks", [])),
        }

    def bootstrap_candles(self, interval: str | None = None, limit: int | None = None, logger=None):
        interval = interval or Config.CANDLE_INTERVAL
        limit = int(limit or Config.BOOTSTRAP_CANDLES)

        log = (logger.info if logger else print)
        log(f"Bootstrapping {interval} candles (limit={limit}) for {len(Config.COINS)} coins...")

        for coin in Config.COINS:
            try:
                df = self.fetch_candles_snapshot(coin, interval=interval, limit=limit)
                self.candle_data[coin] = df
                if logger:
                    logger.info(f"Loaded {len(df)} candles for {coin}")
            except Exception as e:
                if logger:
                    logger.warning(f"Failed to load candles for {coin}: {e}")
                else:
                    print(f"Failed to load candles for {coin}: {e}")

    def start_ws(self, on_candle_callback, logger=None):
        # Update callback/logger refs used by the worker thread (survives reconnects).
        self._ws_on_candle_callback = on_candle_callback
        self._ws_logger = logger
        self._ensure_ws_worker()

        def on_message(ws, message):
            try:
                data = json.loads(message)

                # Some messages may be non-dict JSON; ignore them.
                if not isinstance(data, dict):
                    return

                channel = data.get("channel")

                # Forward trades / l2Book to the order flow tracker.
                if channel in ("trades", "l2Book"):
                    ft = self._flow_tracker
                    if ft is not None:
                        ft.handle_hl_message(data)
                    return

                # Skip non-candle channels (subscription responses, pings, etc.)
                if channel != "candle":
                    return
                
                candle_payload = data.get("data")
                if not candle_payload:
                    return
                
                # Handle both single dict (live updates) and list (initial batch)
                candles = candle_payload if isinstance(candle_payload, list) else [candle_payload]
                
                for candle in candles:
                    if not isinstance(candle, dict):
                        continue
                    
                    coin = candle.get("s")  # Symbol key is "s"
                    interval = candle.get("i")  # Interval key is "i"
                    if not coin or interval != Config.CANDLE_INTERVAL or coin not in Config.COINS:
                        continue
                    
                    ts = int(candle.get("t", 0))
                    new_row = {
                        'timestamp': ts,
                        'open': float(candle.get('o', 0)),
                        'high': float(candle.get('h', 0)),
                        'low': float(candle.get('l', 0)),
                        'close': float(candle.get('c', 0)),
                        'volume': float(candle.get('v', 0))
                    }
                    
                    with self._candle_lock:
                        # Initialize if missing
                        if coin not in self.candle_data:
                            self.candle_data[coin] = pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

                        current_df = self.candle_data[coin]

                        # Append/update (avoid concat-on-empty warning)
                        if current_df.empty:
                            current_df = pd.DataFrame([new_row])
                        elif int(current_df.iloc[-1]['timestamp']) == ts:
                            # Update last row (live candle update)
                            current_df.loc[current_df.index[-1], ['timestamp', 'open', 'high', 'low', 'close', 'volume']] = [
                                new_row['timestamp'], new_row['open'], new_row['high'], new_row['low'], new_row['close'], new_row['volume']
                            ]
                        else:
                            current_df = pd.concat([current_df, pd.DataFrame([new_row])], ignore_index=True)

                        # Clean & limit
                        current_df = current_df.drop_duplicates(subset=['timestamp']).sort_values('timestamp').reset_index(drop=True)
                        if len(current_df) > 200:
                            current_df = current_df.tail(200)

                        self.candle_data[coin] = current_df
                    
                    # Trigger strategy only on new closed candle (timestamp change)
                    if len(current_df) > 1 and current_df.iloc[-2]['timestamp'] < ts:
                        # Do NOT run heavy strategy work in this WS callback thread.
                        self._enqueue_ws_candle_event(coin)
                
            except Exception as e:
                if logger:
                    snippet = message[:300] if isinstance(message, str) else str(message)[:300]
                    logger.error(f"WS processing error (safe skip): {e} | msg={snippet}")
                else:
                    print(f"WS processing error (safe skip): {e}")

        def on_error(ws, error):
            if logger:
                logger.error(f"WS Connection Error: {error}")
            else:
                print(f"WS Connection Error: {error}")

        def on_close(ws, *args):
            if logger:
                logger.warning("WS Closed – reconnecting in 5s...")
            else:
                print("WS Closed – reconnecting in 5s...")

            # Avoid multiple concurrent reconnect attempts.
            with self._ws_reconnect_lock:
                if self._ws_reconnecting:
                    return
                self._ws_reconnecting = True

            def _reconnect():
                try:
                    time.sleep(5)
                    self.start_ws(on_candle_callback, logger=logger)
                finally:
                    with self._ws_reconnect_lock:
                        self._ws_reconnecting = False

            threading.Thread(target=_reconnect, name="ws-reconnect", daemon=True).start()

        def on_open(ws):
            if logger:
                logger.info("WS Connected & Subscriptions Sent")
            else:
                print("WS Connected & Subscriptions Sent")
            for coin in Config.COINS:
                sub = {
                    "method": "subscribe",
                    "subscription": {
                        "type": "candle",
                        "coin": coin,
                        "interval": Config.CANDLE_INTERVAL
                    }
                }
                ws.send(json.dumps(sub))
                if logger:
                    logger.info(f"Subscribed → {coin} {Config.CANDLE_INTERVAL}")
                else:
                    print(f"Subscribed → {coin} {Config.CANDLE_INTERVAL}")

                # Order flow subscriptions (trades + l2Book)
                if self._flow_tracker is not None:
                    for sub_type in ("trades", "l2Book"):
                        flow_sub = {
                            "method": "subscribe",
                            "subscription": {"type": sub_type, "coin": coin},
                        }
                        ws.send(json.dumps(flow_sub))
                    if logger:
                        logger.info(f"Subscribed → {coin} trades + l2Book")

        self.ws = websocket.WebSocketApp(  # type: ignore[attr-defined]
            Config.WS_URL,
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
        
        # websocket-client requires ping_interval > ping_timeout.
        wst = threading.Thread(target=self.ws.run_forever, kwargs={"ping_interval": 30, "ping_timeout": 20})
        wst.daemon = True
        wst.start()

    def get_current_price(self, symbol: str) -> float | None:
        coin = str(symbol or "").strip()
        if not coin:
            return None

        now = time.time()
        ttl = float(getattr(Config, "PRICE_CACHE_TTL_SECONDS", 0.8))
        min_interval = float(getattr(Config, "REST_PRICE_MIN_INTERVAL_SECONDS", 0.2))

        cached = self._price_cache.get(coin)
        if cached is not None:
            ts, px = cached
            if (now - float(ts)) <= ttl and float(px) > 0:
                return float(px)

        last_req = float(self._price_last_req_ts.get(coin) or 0.0)
        if cached is not None and (now - last_req) < min_interval:
            # Too soon to hit REST again; return stale-ish cache rather than spamming /info.
            ts, px = cached
            return float(px) if float(px) > 0 else None

        self._price_last_req_ts[coin] = float(now)
        payload = {"type": "l2Book", "coin": coin}

        try:
            resp = self._post_info(payload)
            levels = (resp or {}).get("levels") if isinstance(resp, dict) else None
            if levels and levels[0] and levels[1]:
                bid = float(levels[0][0]["px"])
                ask = float(levels[1][0]["px"])
                mid = (bid + ask) / 2.0
                if mid > 0:
                    self._price_cache[coin] = (float(now), float(mid))
                return float(mid)
        except Exception:
            # Best-effort: if REST fails, return cached price if available.
            if cached is not None:
                ts, px = cached
                return float(px) if float(px) > 0 else None

        return None