"""Read-only Hyperliquid websocket orderflow recorder."""
from __future__ import annotations

import json
import logging
import threading
import time
from collections import defaultdict, deque
from pathlib import Path
from statistics import median
from typing import Any, Iterable, Optional

log = logging.getLogger(__name__)


def websocket_url(api_url: str) -> str:
    return api_url.replace("https://", "wss://").replace("http://", "ws://").rstrip("/") + "/ws"


def market_subscriptions(coins: Iterable[str], intervals: Iterable[str]) -> list[dict]:
    subs: list[dict] = []
    seen: set[tuple[str, str, str]] = set()
    for coin in coins:
        coin = coin.strip().upper()
        if not coin:
            continue
        for typ in ("l2Book", "bbo", "trades", "activeAssetCtx"):
            key = (typ, coin, "")
            if key not in seen:
                subs.append({"type": typ, "coin": coin})
                seen.add(key)
        for interval in intervals:
            key = ("candle", coin, interval)
            if key not in seen:
                subs.append({"type": "candle", "coin": coin, "interval": interval})
                seen.add(key)
    return subs


def user_subscriptions(user: Optional[str]) -> list[dict]:
    if not user:
        return []
    return [
        {"type": "orderUpdates", "user": user},
        {"type": "userEvents", "user": user},
        {"type": "userFills", "user": user},
        {"type": "clearinghouseState", "user": user},
        {"type": "openOrders", "user": user},
    ]


class LiveOrderflowCache:
    """Background websocket cache for compact live orderflow features.

    The cache is read-only and best-effort. If the websocket drops or data is stale,
    callers receive a summary with `available=False` and can fall back to REST data.
    """

    def __init__(
        self,
        *,
        api_url: str,
        coins: Iterable[str],
        intervals: Iterable[str] = ("1m", "5m", "15m"),
        user: Optional[str] = None,
        max_trades: int = 5000,
        max_books: int = 600,
        max_ctx: int = 600,
        max_silent_s: float = 45.0,
    ) -> None:
        self.api_url = api_url
        self.coins = [coin.strip().upper() for coin in coins if coin.strip()]
        self.intervals = list(intervals)
        self.user = user
        self.max_trades = max_trades
        self.max_books = max_books
        self.max_ctx = max_ctx
        self.max_silent_s = float(max_silent_s)
        self._lock = threading.RLock()
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._last_error = ""
        self._last_msg_ms = 0

        self._trades: dict[str, deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=self.max_trades))
        self._seen_tids: dict[str, deque[int]] = defaultdict(lambda: deque(maxlen=self.max_trades))
        self._books: dict[str, deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=self.max_books))
        self._bbos: dict[str, deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=self.max_books))
        self._ctxs: dict[str, deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=self.max_ctx))
        self._candles: dict[tuple[str, str], deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=240))

    def start(self) -> None:
        if self._thread and self._thread.is_alive():
            return
        self._stop.clear()
        self._thread = threading.Thread(target=self._run, name="hl-ws-orderflow", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread:
            self._thread.join(timeout=3)

    def summary(self, coin: str) -> dict[str, Any]:
        coin = coin.upper()
        now_ms = int(time.time() * 1000)
        with self._lock:
            trades = list(self._trades.get(coin, ()))
            books = list(self._books.get(coin, ()))
            bbos = list(self._bbos.get(coin, ()))
            ctxs = list(self._ctxs.get(coin, ()))
            last_msg_ms = self._last_msg_ms
            last_error = self._last_error

        book_ts = max([row.get("recv_time_ms", 0) for row in books[-1:]], default=0)
        bbo_ts = max([row.get("recv_time_ms", 0) for row in bbos[-1:]], default=0)
        tape_ts = max([row.get("recv_time_ms", 0) for row in trades[-1:]], default=0)
        ctx_ts = max([row.get("recv_time_ms", 0) for row in ctxs[-1:]], default=0)
        latest_ts = max(book_ts, bbo_ts, tape_ts, ctx_ts)
        stale_s = (now_ms - latest_ts) / 1000 if latest_ts else 9999.0
        book_stale_s = (now_ms - book_ts) / 1000 if book_ts else 9999.0
        bbo_stale_s = (now_ms - bbo_ts) / 1000 if bbo_ts else 9999.0
        tape_stale_s = (now_ms - tape_ts) / 1000 if tape_ts else 9999.0
        ctx_stale_s = (now_ms - ctx_ts) / 1000 if ctx_ts else 9999.0
        summary = {
            "available": book_ts > 0 and tape_ts > 0 and book_stale_s <= 20 and tape_stale_s <= 20,
            "stale_s": round(stale_s, 2),
            "book_stale_s": round(book_stale_s, 2),
            "bbo_stale_s": round(bbo_stale_s, 2),
            "tape_stale_s": round(tape_stale_s, 2),
            "ctx_stale_s": round(ctx_stale_s, 2),
            "last_ws_msg_age_s": round((now_ms - last_msg_ms) / 1000, 2) if last_msg_ms else 9999.0,
            "last_error": last_error,
        }
        summary.update(self._book_summary(books, now_ms))
        summary.update(self._tape_summary(trades, now_ms))
        summary.update(self._ctx_summary(ctxs, now_ms))
        return summary

    def _run(self) -> None:
        try:
            import websocket
        except ImportError:
            self._last_error = "websocket-client unavailable"
            log.warning("LiveOrderflowCache disabled: websocket-client unavailable")
            return

        subs = market_subscriptions(self.coins, self.intervals) + user_subscriptions(self.user)
        if not subs:
            self._last_error = "no subscriptions"
            return

        url = websocket_url(self.api_url)
        while not self._stop.is_set():
            ws = None
            try:
                ws = websocket.create_connection(url, timeout=10)
                ws.settimeout(1.0)
                last_recv = time.time()
                for sub in subs:
                    ws.send(json.dumps({"method": "subscribe", "subscription": sub}))
                self._last_error = ""
                log.info("LiveOrderflowCache subscribed to %s streams for %s", len(subs), ",".join(self.coins))
                while not self._stop.is_set():
                    try:
                        raw = ws.recv()
                    except websocket.WebSocketTimeoutException:
                        if time.time() - last_recv > self.max_silent_s:
                            raise TimeoutError(f"websocket silent for >{self.max_silent_s:.0f}s")
                        continue
                    except Exception:
                        raise
                    if not raw:
                        if time.time() - last_recv > self.max_silent_s:
                            raise TimeoutError(f"websocket empty/silent for >{self.max_silent_s:.0f}s")
                        continue
                    last_recv = time.time()
                    try:
                        msg = json.loads(raw)
                    except json.JSONDecodeError:
                        continue
                    self._on_message(msg)
            except Exception as exc:
                self._last_error = str(exc)
                log.warning("LiveOrderflowCache reconnecting after error: %s", exc)
                time.sleep(3)
            finally:
                if ws is not None:
                    try:
                        ws.close()
                    except Exception:
                        pass

    def _on_message(self, msg: dict[str, Any]) -> None:
        channel = msg.get("channel")
        data = msg.get("data")
        recv_ms = int(time.time() * 1000)
        with self._lock:
            self._last_msg_ms = recv_ms
            if channel == "trades" and isinstance(data, list):
                for trade in data:
                    if not isinstance(trade, dict):
                        continue
                    coin = str(trade.get("coin", "")).upper()
                    if not coin:
                        continue
                    tid = _to_int(trade.get("tid"))
                    if tid is not None and tid in self._seen_tids[coin]:
                        continue
                    row = dict(trade)
                    row["recv_time_ms"] = recv_ms
                    self._trades[coin].append(row)
                    if tid is not None:
                        self._seen_tids[coin].append(tid)
            elif channel == "l2Book" and isinstance(data, dict):
                coin = str(data.get("coin", "")).upper()
                row = dict(data)
                row["recv_time_ms"] = recv_ms
                self._books[coin].append(row)
            elif channel == "bbo" and isinstance(data, dict):
                coin = str(data.get("coin", "")).upper()
                row = dict(data)
                row["recv_time_ms"] = recv_ms
                self._bbos[coin].append(row)
            elif channel == "activeAssetCtx" and isinstance(data, dict):
                coin = str(data.get("coin", "")).upper()
                row = dict(data)
                row["recv_time_ms"] = recv_ms
                self._ctxs[coin].append(row)
            elif channel == "candle" and isinstance(data, dict):
                coin = str(data.get("s", "")).upper()
                interval = str(data.get("i", ""))
                row = dict(data)
                row["recv_time_ms"] = recv_ms
                self._candles[(coin, interval)].append(row)

    def _book_summary(self, books: list[dict[str, Any]], now_ms: int) -> dict[str, Any]:
        if not books:
            return {
                "book_available": False,
                "ws_mid": 0.0,
                "ws_spread_bps": 999.0,
                "bid_depth_change_15s": 0.0,
                "ask_depth_change_15s": 0.0,
                "book_velocity": "stale",
            }
        latest = books[-1]
        current = _book_depth(latest)
        older_15 = _older_or_first(books, now_ms - 15_000)
        older_60 = _older_or_first(books, now_ms - 60_000)
        d15 = _book_depth(older_15)
        d60 = _book_depth(older_60)
        bid_chg_15 = _ratio(current["bid_depth_usd"] - d15["bid_depth_usd"], d15["bid_depth_usd"])
        ask_chg_15 = _ratio(current["ask_depth_usd"] - d15["ask_depth_usd"], d15["ask_depth_usd"])
        bid_chg_60 = _ratio(current["bid_depth_usd"] - d60["bid_depth_usd"], d60["bid_depth_usd"])
        ask_chg_60 = _ratio(current["ask_depth_usd"] - d60["ask_depth_usd"], d60["ask_depth_usd"])

        if bid_chg_15 > 0.2 and ask_chg_15 < -0.15:
            velocity = "bullish_liquidity_shift"
        elif ask_chg_15 > 0.2 and bid_chg_15 < -0.15:
            velocity = "bearish_liquidity_shift"
        elif bid_chg_15 < -0.25 or ask_chg_15 < -0.25:
            velocity = "liquidity_pulled"
        else:
            velocity = "stable"

        return {
            "book_available": True,
            "ws_mid": current["mid"],
            "ws_spread_bps": current["spread_bps"],
            "ws_bid_depth_usd": round(current["bid_depth_usd"], 2),
            "ws_ask_depth_usd": round(current["ask_depth_usd"], 2),
            "bid_depth_change_15s": round(bid_chg_15, 4),
            "ask_depth_change_15s": round(ask_chg_15, 4),
            "bid_depth_change_60s": round(bid_chg_60, 4),
            "ask_depth_change_60s": round(ask_chg_60, 4),
            "book_velocity": velocity,
        }

    def _tape_summary(self, trades: list[dict[str, Any]], now_ms: int) -> dict[str, Any]:
        if not trades:
            return {
                "tape_available": False,
                "buy_usd_30s": 0.0,
                "sell_usd_30s": 0.0,
                "delta_usd_30s": 0.0,
                "delta_ratio_30s": 0.0,
                "tape_windows": {},
                "cvd_state": {},
                "cvd_5m": 0.0,
                "cvd_divergence_2m": "none",
                "cvd_divergence_5m": "none",
                "tape_state": "stale",
            }
        w10 = _trade_stats(trades, now_ms - 10_000, now_ms)
        w30 = _trade_stats(trades, now_ms - 30_000, now_ms)
        w30_prev = _trade_stats(trades, now_ms - 60_000, now_ms - 30_000)
        w60 = _trade_stats(trades, now_ms - 60_000, now_ms)
        w120 = _trade_stats(trades, now_ms - 120_000, now_ms)
        w300 = _trade_stats(trades, now_ms - 300_000, now_ms)
        delta_accel = w30["delta_usd"] - w30_prev["delta_usd"]
        cvd_div_2m = _cvd_divergence(w120)
        cvd_div_5m = _cvd_divergence(w300)

        if w30["delta_ratio"] > 0.35 and delta_accel > 0:
            state = "buy_acceleration"
        elif w30["delta_ratio"] < -0.35 and delta_accel < 0:
            state = "sell_acceleration"
        elif abs(w30["delta_ratio"]) < 0.1:
            state = "balanced"
        elif w30["delta_ratio"] > 0:
            state = "buy_aggression"
        else:
            state = "sell_aggression"

        return {
            "tape_available": True,
            "buy_usd_30s": round(w30["buy_usd"], 2),
            "sell_usd_30s": round(w30["sell_usd"], 2),
            "delta_usd_30s": round(w30["delta_usd"], 2),
            "delta_ratio_30s": round(w30["delta_ratio"], 4),
            "total_usd_120s": round(w120["total_usd"], 2),
            "tape_windows": {
                "10s": _compact_trade_stats(w10, 10.0),
                "30s": _compact_trade_stats(w30, 30.0),
                "1m": _compact_trade_stats(w60, 60.0),
                "2m": _compact_trade_stats(w120, 120.0),
                "5m": _compact_trade_stats(w300, 300.0),
            },
            "cvd_state": {
                "cvd_1m": round(w60["delta_usd"], 2),
                "cvd_2m": round(w120["delta_usd"], 2),
                "cvd_5m": round(w300["delta_usd"], 2),
                "cvd_slope_30s": round(delta_accel, 2),
                "divergence_2m": cvd_div_2m,
                "divergence_5m": cvd_div_5m,
            },
            "cvd_5m": round(w300["delta_usd"], 2),
            "cvd_divergence_2m": cvd_div_2m,
            "cvd_divergence_5m": cvd_div_5m,
            "delta_accel_30s": round(delta_accel, 2),
            "large_buy_prints_2m": w120["large_buy_count"],
            "large_sell_prints_2m": w120["large_sell_count"],
            "trades_per_sec_30s": round(w30["trade_count"] / 30.0, 2),
            "tape_state": state,
        }

    def _ctx_summary(self, ctxs: list[dict[str, Any]], now_ms: int) -> dict[str, Any]:
        if not ctxs:
            return {
                "ctx_available": False,
                "oi_delta_60s": 0.0,
                "oi_delta_5m": 0.0,
                "oi_delta_15m": 0.0,
                "funding_ws": 0.0,
                "funding_delta_5m": 0.0,
                "funding_delta_15m": 0.0,
                "premium_ws": 0.0,
                "premium_delta_5m": 0.0,
                "premium_delta_15m": 0.0,
                "crowding_state": "unknown",
            }
        latest = ctxs[-1]
        older = _older_or_first(ctxs, now_ms - 60_000)
        older_5m = _older_or_first(ctxs, now_ms - 300_000)
        older_15m = _older_or_first(ctxs, now_ms - 900_000)
        ctx = latest.get("ctx", {}) if isinstance(latest.get("ctx"), dict) else {}
        old_ctx = older.get("ctx", {}) if isinstance(older.get("ctx"), dict) else {}
        old_ctx_5m = older_5m.get("ctx", {}) if isinstance(older_5m.get("ctx"), dict) else {}
        old_ctx_15m = older_15m.get("ctx", {}) if isinstance(older_15m.get("ctx"), dict) else {}
        oi = _to_float(ctx.get("openInterest")) or 0.0
        old_oi = _to_float(old_ctx.get("openInterest")) or oi
        old_oi_5m = _to_float(old_ctx_5m.get("openInterest")) or oi
        old_oi_15m = _to_float(old_ctx_15m.get("openInterest")) or oi
        oi_delta = _ratio(oi - old_oi, old_oi)
        oi_delta_5m = _ratio(oi - old_oi_5m, old_oi_5m)
        oi_delta_15m = _ratio(oi - old_oi_15m, old_oi_15m)
        funding = _to_float(ctx.get("funding")) or 0.0
        old_funding_5m = _to_float(old_ctx_5m.get("funding")) or funding
        old_funding_15m = _to_float(old_ctx_15m.get("funding")) or funding
        premium = _to_float(ctx.get("premium")) or 0.0
        old_premium_5m = _to_float(old_ctx_5m.get("premium")) or premium
        old_premium_15m = _to_float(old_ctx_15m.get("premium")) or premium

        if funding > 0.00008 or premium > 0.001:
            crowding = "crowded_longs"
        elif funding < -0.00008 or premium < -0.001:
            crowding = "crowded_shorts"
        elif oi_delta > 0.03:
            crowding = "oi_expanding"
        elif oi_delta < -0.03:
            crowding = "oi_unwinding"
        else:
            crowding = "neutral"

        return {
            "ctx_available": True,
            "oi_ws": round(oi, 6),
            "oi_delta_60s": round(oi_delta, 4),
            "oi_delta_5m": round(oi_delta_5m, 4),
            "oi_delta_15m": round(oi_delta_15m, 4),
            "funding_ws": round(funding, 8),
            "funding_delta_5m": round(funding - old_funding_5m, 8),
            "funding_delta_15m": round(funding - old_funding_15m, 8),
            "premium_ws": round(premium, 8),
            "premium_delta_5m": round(premium - old_premium_5m, 8),
            "premium_delta_15m": round(premium - old_premium_15m, 8),
            "crowding_state": crowding,
        }


def record_orderflow(
    *,
    api_url: str,
    out_path: str | Path,
    seconds: int,
    coins: Iterable[str],
    intervals: Iterable[str] = ("1m", "5m", "15m"),
    user: Optional[str] = None,
    max_messages: int = 100_000,
) -> dict:
    """Record Hyperliquid websocket messages to JSONL.

    This is intentionally read-only: it does not need a private key and cannot place orders.
    """
    try:
        import websocket
    except ImportError as exc:
        raise RuntimeError("websocket-client is required for websocket recording") from exc

    out = Path(out_path)
    out.parent.mkdir(parents=True, exist_ok=True)

    url = websocket_url(api_url)
    subs = market_subscriptions(coins, intervals) + user_subscriptions(user)
    if not subs:
        raise ValueError("no websocket subscriptions requested")

    counts: dict[str, int] = {}
    started_ms = int(time.time() * 1000)
    deadline = time.time() + max(1, int(seconds))
    ws = websocket.create_connection(url, timeout=10)
    ws.settimeout(1.0)
    count = 0
    try:
        for sub in subs:
            ws.send(json.dumps({"method": "subscribe", "subscription": sub}))

        with out.open("a", encoding="utf-8") as fh:
            while time.time() < deadline and count < max_messages:
                try:
                    raw = ws.recv()
                except Exception:
                    continue
                recv_ms = int(time.time() * 1000)
                try:
                    msg = json.loads(raw)
                except json.JSONDecodeError:
                    continue
                channel = str(msg.get("channel", "unknown"))
                counts[channel] = counts.get(channel, 0) + 1
                fh.write(json.dumps({"recv_time_ms": recv_ms, "message": msg}, separators=(",", ":")) + "\n")
                count += 1
    finally:
        try:
            ws.close()
        except Exception:
            pass

    return {
        "out_path": str(out),
        "started_ms": started_ms,
        "ended_ms": int(time.time() * 1000),
        "seconds": seconds,
        "subscriptions": len(subs),
        "messages": count,
        "channels": counts,
    }


def _book_depth(book: dict[str, Any], depth_levels: int = 10) -> dict[str, float]:
    levels = book.get("levels") or [[], []]
    bids = levels[0] if len(levels) > 0 else []
    asks = levels[1] if len(levels) > 1 else []
    best_bid = _to_float(bids[0].get("px")) if bids else None
    best_ask = _to_float(asks[0].get("px")) if asks else None
    mid = ((best_bid + best_ask) / 2.0) if best_bid and best_ask else 0.0
    bid_depth = sum((_to_float(row.get("px")) or 0.0) * (_to_float(row.get("sz")) or 0.0) for row in bids[:depth_levels])
    ask_depth = sum((_to_float(row.get("px")) or 0.0) * (_to_float(row.get("sz")) or 0.0) for row in asks[:depth_levels])
    spread_bps = ((best_ask - best_bid) / mid * 10_000.0) if best_bid and best_ask and mid else 999.0
    return {
        "mid": mid,
        "spread_bps": spread_bps,
        "bid_depth_usd": bid_depth,
        "ask_depth_usd": ask_depth,
    }


def _trade_stats(trades: list[dict[str, Any]], start_ms: int, end_ms: int) -> dict[str, Any]:
    rows = []
    prices = []
    for trade in trades:
        ts = _to_int(trade.get("time") or trade.get("t") or trade.get("recv_time_ms"))
        if ts is None or ts < start_ms or ts > end_ms:
            continue
        px = _to_float(trade.get("px"))
        sz = _to_float(trade.get("sz"))
        if px is None or sz is None:
            continue
        notional = px * sz
        side = str(trade.get("side", "")).upper()
        rows.append((side, notional))
        prices.append(px)

    buy_usd = sum(n for side, n in rows if side == "B")
    sell_usd = sum(n for side, n in rows if side == "A")
    total = buy_usd + sell_usd
    notionals = [n for _, n in rows]
    large_threshold = max(5_000.0, median(notionals) * 8.0 if notionals else 0.0)
    large_buy = sum(1 for side, n in rows if side == "B" and n >= large_threshold)
    large_sell = sum(1 for side, n in rows if side == "A" and n >= large_threshold)
    delta = buy_usd - sell_usd
    price_change_pct = _ratio(prices[-1] - prices[0], prices[0]) if len(prices) >= 2 else 0.0
    return {
        "trade_count": len(rows),
        "buy_usd": buy_usd,
        "sell_usd": sell_usd,
        "total_usd": total,
        "delta_usd": delta,
        "delta_ratio": delta / total if total else 0.0,
        "large_buy_count": large_buy,
        "large_sell_count": large_sell,
        "price_change_pct": price_change_pct,
    }


def _compact_trade_stats(stats: dict[str, Any], seconds: float) -> dict[str, Any]:
    return {
        "buy_usd": round(stats["buy_usd"], 2),
        "sell_usd": round(stats["sell_usd"], 2),
        "delta_usd": round(stats["delta_usd"], 2),
        "delta_ratio": round(stats["delta_ratio"], 4),
        "total_usd": round(stats["total_usd"], 2),
        "trades": int(stats["trade_count"]),
        "trades_per_sec": round(stats["trade_count"] / seconds, 2),
        "price_change_pct": round(stats["price_change_pct"], 5),
    }


def _cvd_divergence(stats: dict[str, Any]) -> str:
    price_change = float(stats.get("price_change_pct") or 0.0)
    delta_ratio = float(stats.get("delta_ratio") or 0.0)
    delta_usd = float(stats.get("delta_usd") or 0.0)
    if price_change > 0.001 and (delta_ratio < -0.10 or delta_usd < 0):
        return "bearish"
    if price_change < -0.001 and (delta_ratio > 0.10 or delta_usd > 0):
        return "bullish"
    return "none"


def _older_or_first(rows: list[dict[str, Any]], cutoff_ms: int) -> dict[str, Any]:
    older = rows[0]
    for row in rows:
        if int(row.get("recv_time_ms", 0)) <= cutoff_ms:
            older = row
        else:
            break
    return older


def _ratio(numerator: float, denominator: float) -> float:
    return numerator / max(abs(denominator), 1e-12)


def _to_float(value: Any) -> float | None:
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _to_int(value: Any) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None
