""" HYPERION — Hyperliquid Info client (Phase 1: read-only) All requests are POST https://api.hyperliquid.xyz/info with a JSON body. Multi-clearinghouse aware: main perps dex ("") plus HIP-3 builder dexes (stocks/commodities/fx, symbols prefixed like "xyz:NVDA"). No signing needed for any of this. Exchange endpoint arrives in Phase 3. """ from __future__ import annotations import time import requests MAINNET_INFO = "https://api.hyperliquid.xyz/info" TESTNET_INFO = "https://api.hyperliquid-testnet.xyz/info" class HLInfo: def __init__(self, mainnet: bool = True, timeout: int = 10): self.url = MAINNET_INFO if mainnet else TESTNET_INFO self.timeout = timeout self.s = requests.Session() self._meta_cache: dict[str, dict] = {} # dex -> meta self._meta_cache_ts: float = 0.0 # ---------- core ---------- def _post(self, body: dict) -> dict | list: r = self.s.post(self.url, json=body, timeout=self.timeout) r.raise_for_status() return r.json() # ---------- dex discovery ---------- def perp_dexs(self) -> list[dict]: """List all perp dexes. First entry is the main clearinghouse (name '').""" try: out = self._post({"type": "perpDexs"}) # API returns list where index 0 may be null for the main dex dexs = [] for i, d in enumerate(out): if d is None: dexs.append({"name": "", "full_name": "Main"}) else: dexs.append({"name": d.get("name", ""), "full_name": d.get("full_name") or d.get("name", "")}) return dexs except Exception: return [{"name": "", "full_name": "Main"}] def meta(self, dex: str = "") -> dict: """Perp universe + margin tables for a dex. Cached 5 min.""" now = time.time() if dex in self._meta_cache and now - self._meta_cache_ts < 300: return self._meta_cache[dex] body = {"type": "meta"} if dex: body["dex"] = dex m = self._post(body) self._meta_cache[dex] = m self._meta_cache_ts = now return m def universe(self) -> list[dict]: """Flattened tradable perp list across all dexes with leverage caps.""" assets = [] for d in self.perp_dexs(): dex = d["name"] try: m = self.meta(dex) except Exception: continue for a in m.get("universe", []): if a.get("isDelisted"): continue name = a["name"] assets.append({ "coin": name if not dex else name, # builder assets already carry prefix in name per meta "dex": dex, "dex_label": d["full_name"], "max_leverage": a.get("maxLeverage"), "sz_decimals": a.get("szDecimals"), "only_isolated": a.get("onlyIsolated", False), }) return assets # ---------- prices ---------- def all_mids(self, dex: str = "") -> dict: body = {"type": "allMids"} if dex: body["dex"] = dex return self._post(body) def all_mids_all_dexs(self) -> dict: mids = {} for d in self.perp_dexs(): try: mids.update(self.all_mids(d["name"])) except Exception: pass return mids # ---------- account ---------- def clearinghouse_state(self, address: str, dex: str = "") -> dict: body = {"type": "clearinghouseState", "user": address} if dex: body["dex"] = dex return self._post(body) def account_summary(self, address: str) -> dict: """Aggregate positions + margin across all clearinghouses.""" out = {"dexes": [], "total_account_value": 0.0, "total_upnl": 0.0, "total_margin_used": 0.0, "withdrawable": 0.0, "positions": []} for d in self.perp_dexs(): try: st = self.clearinghouse_state(address, d["name"]) except Exception: continue ms = st.get("marginSummary", {}) av = float(ms.get("accountValue", 0) or 0) mu = float(ms.get("totalMarginUsed", 0) or 0) wd = float(st.get("withdrawable", 0) or 0) dex_pos = [] for p in st.get("assetPositions", []): pos = p.get("position", {}) szi = float(pos.get("szi", 0) or 0) if szi == 0: continue upnl = float(pos.get("unrealizedPnl", 0) or 0) out["total_upnl"] += upnl dex_pos.append({ "coin": pos.get("coin"), "dex": d["name"], "side": "long" if szi > 0 else "short", "size": abs(szi), "entry_px": float(pos.get("entryPx", 0) or 0), "position_value": float(pos.get("positionValue", 0) or 0), "upnl": upnl, "roe": float(pos.get("returnOnEquity", 0) or 0), "liq_px": float(pos["liquidationPx"]) if pos.get("liquidationPx") else None, "leverage": pos.get("leverage", {}), "margin_used": float(pos.get("marginUsed", 0) or 0), "funding_all_time": float((pos.get("cumFunding") or {}).get("allTime", 0) or 0), }) out["dexes"].append({"dex": d["name"], "label": d["full_name"], "account_value": av, "margin_used": mu, "withdrawable": wd, "positions": len(dex_pos)}) out["total_account_value"] += av out["total_margin_used"] += mu out["withdrawable"] += wd out["positions"].extend(dex_pos) return out def user_fills(self, address: str, start_ms: int | None = None, end_ms: int | None = None) -> list: """Fills; time-ranged variant pages at 500 elements — caller loops.""" if start_ms is not None: body = {"type": "userFillsByTime", "user": address, "startTime": start_ms} if end_ms: body["endTime"] = end_ms return self._post(body) return self._post({"type": "userFills", "user": address}) def user_fills_paginated(self, address: str, start_ms: int, end_ms: int | None = None) -> list: """Walk forward in 500-element pages until exhausted.""" fills, cursor = [], start_ms for _ in range(200): # hard stop page = self.user_fills(address, cursor, end_ms) if not page: break fills.extend(page) if len(page) < 500: break cursor = int(page[-1]["time"]) + 1 return fills def funding_history(self, coin: str, start_ms: int, end_ms: int | None = None) -> list: body = {"type": "fundingHistory", "coin": coin, "startTime": start_ms} if end_ms: body["endTime"] = end_ms return self._post(body) # ---------- candles ---------- def candles(self, coin: str, interval: str, start_ms: int, end_ms: int) -> list: return self._post({"type": "candleSnapshot", "req": { "coin": coin, "interval": interval, "startTime": start_ms, "endTime": end_ms}})