ftdt-quant-lab/live/monitors/hlp_vault.py

"""
HLP (Hyperliquidity Provider) Vault monitoring.

Tracks Hyperliquid's native protocol-level market-making vault at
address 0xfefefefefefefefefefefefefefefefefefefefe.

HLP acts as counterparty to all user trades. When it absorbs toxic flow
or becomes directionally overextended, it must rebalance — creating
predictable market impact that can be traded.

Signals:
  - fade_short: HLP is too short → expect buying rebalance → go long
  - fade_long:  HLP is too long → expect selling rebalance → go short
  - neutral: HLP delta is balanced, safe to provide liquidity alongside
"""

from __future__ import annotations

import logging
import time
from collections import deque
from typing import Optional

import requests

logger = logging.getLogger(__name__)

HLP_ADDRESS = "0xfefefefefefefefefefefefefefefefefefefefe"
TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
MAINNET_API = "https://api.hyperliquid.xyz/info"


class HlpVaultMonitor:
    """Monitor HLP vault state — delta, PnL, rebalancing pressure, toxicity."""

    def __init__(
        self,
        testnet: bool = True,
        overextended_threshold: float = 5.0,    # notional in $M before overextended
        history_window: int = 1000,
    ):
        self._api_url = TESTNET_API if testnet else MAINNET_API
        self._overextended_threshold = overextended_threshold * 1_000_000  # convert to USD
        self._testnet = testnet

        # Per-coin state
        self._positions: dict[str, dict] = {}      # coin → {side, szi, entry_px, upnl}
        self._mark_prices: dict[str, float] = {}    # coin → mark price
        self._oracle_prices: dict[str, float] = {}  # coin → oracle price
        self._funding_rates: dict[str, float] = {}  # coin → funding rate
        self._open_interest: dict[str, float] = {}  # coin → OI

        # History
        self._delta_history: dict[str, deque] = {}  # coin → deque of (time, notional)
        self._pnl_history: list[dict] = []
        self._history_window = history_window

        self._last_update: float = 0
        self._update_count: int = 0

    # ── Core update ──────────────────────────────────────────

    def _api_post(self, payload: dict) -> dict:
        """Make a POST request to HL info API (testable via mock)."""
        resp = requests.post(self._api_url, json=payload, timeout=10)
        resp.raise_for_status()
        return resp.json()

    def update(self):
        """Fetch latest HLP state from Hyperliquid API.

        Makes two calls:
          1. metaAndAssetCtxs → mark prices, funding, OI
          2. clearinghouseState(HLP_ADDRESS) → positions, PnL
        """
        now = time.time()

        # Fetch market data
        try:
            meta_and_ctx = self._api_post({"type": "metaAndAssetCtxs"})
            if isinstance(meta_and_ctx, list) and len(meta_and_ctx) >= 2:
                universe = meta_and_ctx[0].get("universe", [])
                ctxs = meta_and_ctx[1]
                for i, asset in enumerate(universe):
                    name = asset.get("name", "")
                    if name and i < len(ctxs):
                        self._mark_prices[name] = float(ctxs[i].get("markPx", 0))
                        self._oracle_prices[name] = float(ctxs[i].get("oraclePx", 0))
                        self._funding_rates[name] = float(ctxs[i].get("funding", 0))
                        self._open_interest[name] = float(ctxs[i].get("openInterest", 0))
        except Exception as e:
            logger.warning("HLP meta fetch error: %s", e)

        # Fetch HLP positions
        try:
            ch_state = self._api_post({
                "type": "clearinghouseState",
                "user": HLP_ADDRESS,
            })
            self._parse_positions(ch_state, now)
        except Exception as e:
            logger.warning("HLP clearinghouse fetch error: %s", e)

        self._last_update = now
        self._update_count += 1

    def _parse_positions(self, data: dict, now: float):
        """Parse clearinghouseState response into per-coin positions."""
        asset_positions = data.get("assetPositions", [])
        new_positions: dict[str, dict] = {}

        for ap in asset_positions:
            pos = ap.get("position", {})
            if not pos:
                continue
            coin = pos.get("coin", "")
            if not coin:
                continue
            szi = float(pos.get("szi", 0))
            entry_px = float(pos.get("entryPx", 0))
            upnl = float(pos.get("unrealizedPnl", 0))
            side = pos.get("side", "")  # "A" = short, "B" = long

            # HLP short is side="A", long is side="B"
            signed_szi = -szi if side == "A" else szi

            new_positions[coin] = {
                "side": side,
                "szi": szi,
                "signed_szi": signed_szi,
                "entry_px": entry_px,
                "unrealized_pnl": upnl,
                "mark_px": self._mark_prices.get(coin, entry_px),
                "notional_usd": abs(szi) * self._mark_prices.get(coin, entry_px),
            }

            # Update delta history
            if coin not in self._delta_history:
                self._delta_history[coin] = deque(maxlen=self._history_window)
            self._delta_history[coin].append({
                "t": now,
                "signed_szi": signed_szi,
                "notional_usd": new_positions[coin]["notional_usd"],
                "upnl": upnl,
            })

        self._positions = new_positions
        self._pnl_history.append({
            "t": now,
            "total_upnl": sum(p["unrealized_pnl"] for p in new_positions.values()),
            "asset_count": len(new_positions),
        })
        if len(self._pnl_history) > self._history_window:
            self._pnl_history = self._pnl_history[-self._history_window:]

    # ── Queries ──────────────────────────────────────────────

    def position(self, coin: str) -> float:
        """Signed position for a coin (positive = long)."""
        pos = self._positions.get(coin.upper(), {})
        return pos.get("signed_szi", 0.0)

    def delta_exposure(self) -> dict:
        """Delta exposure per coin with notional and PnL."""
        result = {}
        for coin, pos in self._positions.items():
            result[coin] = {
                "signed_size": pos["signed_szi"],
                "notional_usd": round(pos["notional_usd"], 2),
                "unrealized_pnl": round(pos["unrealized_pnl"], 2),
                "side": "long" if pos["side"] == "B" else "short",
            }
        return result

    def is_overextended(self, coin: str) -> bool:
        """Check if HLP delta on this coin exceeds the threshold."""
        pos = self._positions.get(coin.upper(), {})
        notional = pos.get("notional_usd", 0)
        return notional > self._overextended_threshold

    def overextended_assets(self) -> list[str]:
        """List of assets where HLP is overextended."""
        return [c for c in self._positions if self.is_overextended(c)]

    def rebalancing_signal(self, coin: str) -> dict:
        """Generate a trading signal based on HLP rebalancing pressure.

        Returns:
            signal: "fade_short" | "fade_long" | "neutral"
            direction: -1 (short) | 0 | 1 (long) for the TRADE direction
            overextended: whether HLP is over capacity
        """
        pos = self._positions.get(coin.upper(), {})
        if not pos:
            return {"signal": "neutral", "direction": 0, "overextended": False, "reason": "no_position"}

        signed = pos["signed_szi"]
        is_over = self.is_overextended(coin)

        if not is_over:
            return {"signal": "neutral", "direction": 0, "overextended": False, "reason": "balanced"}

        # HLP is short (side=A, signed_szi negative) → it will need to buy to rebalance
        # We should fade the short (go long)
        if signed < -0.001:
            return {
                "signal": "fade_short",
                "direction": 1,
                "overextended": True,
                "reason": f"HLP short {abs(signed):.2f} units, expect buying rebalance",
                "notional_usd": round(pos["notional_usd"], 2),
            }
        # HLP is long (side=B, signed_szi positive) → it will need to sell to rebalance
        # We should fade the long (go short)
        elif signed > 0.001:
            return {
                "signal": "fade_long",
                "direction": -1,
                "overextended": True,
                "reason": f"HLP long {signed:.2f} units, expect selling rebalance",
                "notional_usd": round(pos["notional_usd"], 2),
            }
        else:
            return {"signal": "neutral", "direction": 0, "overextended": False, "reason": "flat"}

    def toxicity_score(self) -> float:
        """Estimate how much toxic flow HLP is absorbing.

        Higher score = HLP is losing money = informed traders are beating it.
        Range: 0 (healthy) to 1 (toxic).

        Uses: unrealized PnL / total notional as a proxy.
        """
        total_notional = sum(p["notional_usd"] for p in self._positions.values())
        total_upnl = sum(p["unrealized_pnl"] for p in self._positions.values())

        if total_notional <= 0:
            return 0.0

        # Negative PnL → toxic score > 0
        # Positive PnL → toxic score 0 (healthy)
        loss_ratio = max(0.0, -total_upnl / total_notional)
        toxicity = min(1.0, loss_ratio * 10)  # scale: 10% loss = 1.0 toxicity
        return round(toxicity, 4)

    def delta_history(self, coin: str) -> list[dict]:
        """Historical delta trace for a coin."""
        return list(self._delta_history.get(coin.upper(), []))

    # ── Summary ──────────────────────────────────────────────

    def summary(self) -> dict:
        """One-shot summary of HLP state for dashboard/monitoring."""
        assets = list(self._positions.keys())
        total_delta = sum(p["notional_usd"] for p in self._positions.values())
        overextended = self.overextended_assets()
        signals = {coin: self.rebalancing_signal(coin) for coin in assets}

        return {
            "assets_tracked": len(assets),
            "total_delta_usd": round(total_delta, 2),
            "total_delta_m": round(total_delta / 1_000_000, 2),
            "toxicity_score": self.toxicity_score(),
            "overextended_assets": overextended,
            "signals": signals,
            "positions": self.delta_exposure(),
            "last_update": self._last_update,
            "update_count": self._update_count,
        }