"""

Hyperliquid Client -- Direct REST API wrapper.

Covers crypto perps, HIP-3 stock perps (via Trade.xyz DEX), candles,

funding rates, order books, and whale-wall detection. Uses httpx for

HTTP with automatic 429 back-off and retry.

Usage (library):

from hl_client import HyperliquidClient

with HyperliquidClient() as client:

data = client.get_market_data(["BTC", "ETH", "SOL"])

print(data)

Usage (CLI):

python hl_client.py market BTC ETH SOL

python hl_client.py funding BTC --hours 48

python hl_client.py book BTC --depth 20

python hl_client.py whales BTC

python hl_client.py universe

python hl_client.py candles BTC --interval 1h --hours 168

python hl_client.py stock-perps

python hl_client.py stock-book TSLA NVDA

python hl_client.py stock-funding TSLA --hours 48

"""

from __future__ import annotations

import argparse

import json

import time

from datetime import datetime, timezone

import httpx

API_URL = "https://api.hyperliquid.xyz"

# HIP-3 Stock Perpetuals via the Trade.xyz DEX on Hyperliquid

STOCK_PERP_DEX = "xyz"

# Default stock perp tickers (HIP-3)

DEFAULT_STOCK_PERP_TICKERS: list[str] = [

"xyz:TSLA", "xyz:NVDA", "xyz:AMZN",

"xyz:AAPL", "xyz:MSFT", "xyz:COIN", "xyz:HOOD",

"xyz:PLTR", "xyz:MSTR", "xyz:META", "xyz:NFLX",

"xyz:AMD", "xyz:XYZ100",

]

# ---------------------------------------------------------------------------

# Helpers

# ---------------------------------------------------------------------------

def is_stock_perp(coin: str) -> bool:

"""Check if a coin identifier refers to a HIP-3 stock perpetual."""

return coin.startswith("xyz:")

def stock_perp_display_name(coin: str) -> str:

"""Strip the DEX prefix for display: ``'xyz:TSLA'`` -> ``'TSLA'``."""

return coin.split(":", 1)[1] if ":" in coin else coin

# ---------------------------------------------------------------------------

# Client

# ---------------------------------------------------------------------------

class HyperliquidClient:

"""Lightweight Hyperliquid REST API client.

Supports both crypto perpetuals and HIP-3 stock perpetuals. All

network calls go through :pymethod:`_post` which handles retries and

429 back-off automatically.

Use as a context manager for clean resource handling::

with HyperliquidClient() as client:

print(client.get_market_data(["BTC"]))

"""

def __init__(self, timeout: float = 15.0, cache_ttl: float = 5.0):

"""

Args:

timeout: HTTP request timeout in seconds.

cache_ttl: How long (seconds) to cache ``metaAndAssetCtxs``

responses. Set to ``0`` to disable caching.

"""

self.client = httpx.Client(timeout=timeout)

self._cache_ttl = cache_ttl

self._meta_cache = None

self._meta_time: float = 0

self._stock_meta_cache = None

self._stock_meta_time: float = 0

# -- lifecycle ----------------------------------------------------------

def close(self) -> None:

"""Close the underlying HTTP client."""

self.client.close()

def __enter__(self) -> "HyperliquidClient":

return self

def __exit__(self, exc_type, exc_val, exc_tb) -> bool:

self.close()

return False

def __del__(self) -> None:

try:

self.client.close()

except Exception:

pass

# -- low-level ----------------------------------------------------------

def _post(self, payload: dict, retries: int = 3) -> dict | list | None:

"""POST to ``/info`` with automatic retry and 429 back-off.

Returns the parsed JSON response, or ``None`` on total failure.

"""

for attempt in range(retries):

try:

r = self.client.post(f"{API_URL}/info", json=payload)

if r.status_code == 429:

wait = min(2 ** attempt * 2, 10)

time.sleep(wait)

continue

return r.json()

except Exception:

if attempt < retries - 1:

time.sleep(2)

continue

return None

return None

def _safe_unpack(self, data) -> tuple:

"""Safely unpack a ``metaAndAssetCtxs`` response.

Returns ``(meta_dict, asset_contexts_list)`` or ``(None, None)``.

"""

if not data or not isinstance(data, list) or len(data) < 2:

return None, None

return data[0], data[1]

# -- crypto perps: metadata & prices ------------------------------------

def meta_and_contexts(self) -> list | None:

"""Metadata + asset contexts for all crypto perps (cached)."""

now = time.time()

if self._meta_cache and now - self._meta_time < self._cache_ttl:

return self._meta_cache

data = self._post({"type": "metaAndAssetCtxs"})

self._meta_cache = data

self._meta_time = now

return data

def all_mids(self) -> dict[str, str]:

"""Mid prices for every crypto perp.

Returns ``{"BTC": "95123.5", "ETH": "3412.1", ...}``.

"""

return self._post({"type": "allMids"}) or {}

# -- crypto perps: rich endpoints ---------------------------------------

def get_market_data(self, symbols: list[str] | None = None) -> dict:

"""Comprehensive market data for the given crypto symbols.

Args:

symbols: List of tickers, e.g. ``["BTC", "ETH"]``.

Defaults to ``["BTC", "ETH", "SOL"]``.

Returns:

``{symbol: {price, funding_rate, funding_apr, open_interest,

volume_24h, price_change_24h}}``

"""

if symbols is None:

symbols = ["BTC", "ETH", "SOL"]

data = self.meta_and_contexts()

meta, asset_ctxs = self._safe_unpack(data)

if meta is None:

return {}

universe = meta.get("universe", [])

prices = self.all_mids()

results: dict = {}

for i, ctx in enumerate(asset_ctxs):

if i >= len(universe):

break

coin = universe[i].get("name", "")

if symbols and coin not in symbols:

continue

funding = float(ctx.get("funding", 0))

price = float(prices.get(coin, 0))

# OI from API is in contract units -- convert to USD notional

oi_contracts = float(ctx.get("openInterest", 0))

oi_usd = oi_contracts * price if price > 0 else 0

# 24h price change (prefer API field, fall back to prevDayPx)

pct_chg = 0.0

if ctx.get("dayPxPctChg"):

pct_chg = float(ctx["dayPxPctChg"])

else:

prev_day_px = float(ctx.get("prevDayPx", 0)) if ctx.get("prevDayPx") else 0

if prev_day_px > 0 and price > 0:

pct_chg = ((price - prev_day_px) / prev_day_px) * 100

results[coin] = {

"price": price,

"funding_rate": funding,

"funding_apr": round(funding * 24 * 365 * 100, 1),

"open_interest": round(oi_usd, 2),

"volume_24h": float(ctx.get("dayNtlVlm", 0)),

"price_change_24h": round(pct_chg, 2),

}

return results

def get_universe(self) -> list[dict]:

"""All available crypto perps, sorted by 24h volume descending.

Returns a list of dicts with keys: ``coin, price, volume_24h, oi,

funding_apr``.

"""

data = self.meta_and_contexts()

meta, asset_ctxs = self._safe_unpack(data)

if meta is None:

return []

universe = meta.get("universe", [])

prices = self.all_mids()

result: list[dict] = []

for i, ctx in enumerate(asset_ctxs):

if i >= len(universe):

break

coin = universe[i].get("name", "")

if not coin or coin.startswith("@"):

continue

price = float(prices.get(coin, 0))

volume = float(ctx.get("dayNtlVlm", 0))

oi_contracts = float(ctx.get("openInterest", 0))

oi_usd = oi_contracts * price if price > 0 else 0

result.append({

"coin": coin,

"price": price,

"volume_24h": round(volume, 0),

"oi": round(oi_usd, 2),

"funding_apr": round(float(ctx.get("funding", 0)) * 24 * 365 * 100, 1),

})

result.sort(key=lambda x: x["volume_24h"], reverse=True)

return result

def get_candles(

self,

symbol: str,

interval: str = "1h",

lookback_hours: int = 168,

) -> list[dict]:

"""Historical OHLCV candles.

Args:

symbol: Ticker (e.g. ``"BTC"``). For stock perps, use the

full ``"xyz:TSLA"`` form.

interval: Candle interval -- ``"1m"``, ``"5m"``, ``"15m"``,

``"1h"``, ``"4h"``, ``"1d"``.

lookback_hours: How far back to fetch.

Returns:

List of ``{time, open, high, low, close}`` dicts.

"""

end_time = int(time.time() * 1000)

start_time = end_time - (lookback_hours * 3600 * 1000)

data = self._post({

"type": "candleSnapshot",

"req": {

"coin": symbol,

"interval": interval,

"startTime": start_time,

"endTime": end_time,

},

})

if not isinstance(data, list):

return []

return [

{

"time": datetime.fromtimestamp(c["t"] / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M"),

"open": float(c["o"]),

"high": float(c["h"]),

"low": float(c["l"]),

"close": float(c["c"]),

}

for c in data

]

def get_funding_history(self, symbol: str, hours: int = 24) -> list[dict]:

"""Funding rate history for a crypto perp.

Returns list of ``{time, rate, apr}`` dicts.

"""

start_time = int(time.time() * 1000) - (hours * 3600 * 1000)

data = self._post({

"type": "fundingHistory",

"coin": symbol,

"startTime": start_time,

})

if not isinstance(data, list):

return []

return [

{

"time": datetime.fromtimestamp(r["time"] / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M"),

"rate": float(r["fundingRate"]),

"apr": round(float(r["fundingRate"]) * 24 * 365 * 100, 1),

}

for r in data

]

def get_orderbook(self, symbol: str, depth: int = 20) -> dict:

"""L2 order book for a crypto perp.

Returns ``{bids: [...], asks: [...], spread_bps}`` where each

level has ``price``, ``size``, and ``usd`` fields.

"""

data = self._post({"type": "l2Book", "coin": symbol})

if not data or not isinstance(data, dict):

return {"bids": [], "asks": [], "spread_bps": 0}

levels = data.get("levels", [[], []])

bids = levels[0][:depth] if len(levels) > 0 else []

asks = levels[1][:depth] if len(levels) > 1 else []

return {

"bids": [

{"price": float(b["px"]), "size": float(b["sz"]), "usd": float(b["px"]) * float(b["sz"])}

for b in bids

],

"asks": [

{"price": float(a["px"]), "size": float(a["sz"]), "usd": float(a["px"]) * float(a["sz"])}

for a in asks

],

"spread_bps": round(

(float(asks[0]["px"]) - float(bids[0]["px"]))

/ ((float(asks[0]["px"]) + float(bids[0]["px"])) / 2)

* 10000,

2,

)

if bids and asks

else 0,

}

def get_whale_walls(

self, symbol: str, min_size_usd: float = 100_000, depth: int = 50

) -> list[dict]:

"""Detect large resting orders (whale walls) in the order book.

Args:

symbol: Ticker (e.g. ``"BTC"``).

min_size_usd: Minimum notional size to qualify as a wall.

depth: Order book depth to scan.

Returns:

List of ``{side, price, size_usd}`` dicts sorted by size

descending.

"""

book = self.get_orderbook(symbol, depth=depth)

if not book:

return []

walls: list[dict] = []

for bid in book["bids"]:

if bid["usd"] >= min_size_usd:

walls.append({"side": "BUY", "price": bid["price"], "size_usd": round(bid["usd"], 0)})

for ask in book["asks"]:

if ask["usd"] >= min_size_usd:

walls.append({"side": "SELL", "price": ask["price"], "size_usd": round(ask["usd"], 0)})

walls.sort(key=lambda w: w["size_usd"], reverse=True)

return walls

# -- HIP-3 stock perps: metadata & prices --------------------------------

def stock_perp_meta_and_contexts(self) -> list | None:

"""Metadata + asset contexts for HIP-3 stock perps (cached)."""

now = time.time()

if self._stock_meta_cache and now - self._stock_meta_time < self._cache_ttl:

return self._stock_meta_cache

data = self._post({"type": "metaAndAssetCtxs", "dex": STOCK_PERP_DEX})

self._stock_meta_cache = data

self._stock_meta_time = now

return data

def stock_perp_mids(self) -> dict[str, str]:

"""Mid prices for all HIP-3 stock perps."""

return self._post({"type": "allMids", "dex": STOCK_PERP_DEX}) or {}

# -- HIP-3 stock perps: rich endpoints -----------------------------------

def get_stock_market_data(self, symbols: list[str] | None = None) -> dict:

"""Comprehensive market data for HIP-3 stock perpetuals.

Args:

symbols: List of stock tickers (e.g. ``["TSLA", "NVDA"]``).

Accepts both plain (``"TSLA"``) and prefixed

(``"xyz:TSLA"``) forms. Defaults to all tickers in

:data:`DEFAULT_STOCK_PERP_TICKERS`.

Returns:

``{ticker: {api_name, price, funding_rate, funding_apr,

open_interest, volume_24h, price_change_24h}}``

"""

if symbols is None:

symbols = [stock_perp_display_name(t) for t in DEFAULT_STOCK_PERP_TICKERS]

want = set()

for s in symbols:

want.add(s if not s.startswith("xyz:") else s.split(":", 1)[1])

data = self.stock_perp_meta_and_contexts()

meta, asset_ctxs = self._safe_unpack(data)

if meta is None:

return {}

universe = meta.get("universe", [])

prices = self.stock_perp_mids() or {}

results: dict = {}

for i, ctx in enumerate(asset_ctxs):

if i >= len(universe):

break

raw_name = universe[i].get("name", "")

display = stock_perp_display_name(raw_name)

if want and display not in want:

continue

funding = float(ctx.get("funding", 0))

price = float(prices.get(raw_name, 0))

oi_contracts = float(ctx.get("openInterest", 0))

oi_usd = oi_contracts * price if price > 0 else 0

prev_day_px = float(ctx.get("prevDayPx", 0)) if ctx.get("prevDayPx") else 0

price_change_24h = (

((price - prev_day_px) / prev_day_px) * 100

if prev_day_px > 0 and price > 0

else 0

)

results[display] = {

"api_name": raw_name,

"price": price,

"funding_rate": funding,

"funding_apr": round(funding * 24 * 365 * 100, 1),

"open_interest": round(oi_usd, 2),

"volume_24h": float(ctx.get("dayNtlVlm", 0)),

"price_change_24h": round(price_change_24h, 2),

}

return results

def get_stock_universe(self) -> list[dict]:

"""All available HIP-3 stock perps, sorted by volume descending."""

data = self.stock_perp_meta_and_contexts()

meta, asset_ctxs = self._safe_unpack(data)

if meta is None:

return []

universe = meta.get("universe", [])

prices = self.stock_perp_mids()

result: list[dict] = []

for i, ctx in enumerate(asset_ctxs):

if i >= len(universe):

break

raw_name = universe[i].get("name", "")

if not raw_name:

continue

display = stock_perp_display_name(raw_name)

price = float(prices.get(raw_name, 0))

volume = float(ctx.get("dayNtlVlm", 0))

oi_contracts = float(ctx.get("openInterest", 0))

oi_usd = oi_contracts * price if price > 0 else 0

result.append({

"coin": display,

"api_name": raw_name,

"price": price,

"volume_24h": round(volume, 0),

"oi": round(oi_usd, 2),

"funding_apr": round(float(ctx.get("funding", 0)) * 24 * 365 * 100, 1),

})

result.sort(key=lambda x: x["volume_24h"], reverse=True)

return result

def get_stock_orderbook(self, symbol: str, depth: int = 20) -> dict:

"""L2 order book for a HIP-3 stock perp.

Accepts plain ``"TSLA"`` or prefixed ``"xyz:TSLA"`` form.

"""

coin = symbol if symbol.startswith("xyz:") else f"xyz:{symbol}"

data = self._post({"type": "l2Book", "coin": coin})

if not data or not isinstance(data, dict):

return {"bids": [], "asks": [], "spread_bps": 0}

levels = data.get("levels", [[], []])

bids = levels[0][:depth] if len(levels) > 0 else []

asks = levels[1][:depth] if len(levels) > 1 else []

return {

"bids": [

{"price": float(b["px"]), "size": float(b["sz"]), "usd": float(b["px"]) * float(b["sz"])}

for b in bids

],

"asks": [

{"price": float(a["px"]), "size": float(a["sz"]), "usd": float(a["px"]) * float(a["sz"])}

for a in asks

],

"spread_bps": round(

(float(asks[0]["px"]) - float(bids[0]["px"]))

/ ((float(asks[0]["px"]) + float(bids[0]["px"])) / 2)

* 10000,

2,

)

if bids and asks

else 0,

}

def get_stock_funding_history(self, symbol: str, hours: int = 24) -> list[dict]:

"""Funding rate history for a HIP-3 stock perp."""

coin = symbol if symbol.startswith("xyz:") else f"xyz:{symbol}"

start_time = int(time.time() * 1000) - (hours * 3600 * 1000)

data = self._post({

"type": "fundingHistory",

"coin": coin,

"startTime": start_time,

})

if not isinstance(data, list):

return []

return [

{

"time": datetime.fromtimestamp(r["time"] / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M"),

"rate": float(r["fundingRate"]),

"apr": round(float(r["fundingRate"]) * 24 * 365 * 100, 1),

}

for r in data

]

# ---------------------------------------------------------------------------

# CLI

# ---------------------------------------------------------------------------

def main() -> None:

parser = argparse.ArgumentParser(

description="Hyperliquid REST API Client",

formatter_class=argparse.RawDescriptionHelpFormatter,

epilog=(

"examples:\n"

" python hl_client.py market BTC ETH SOL\n"

" python hl_client.py candles BTC --interval 4h --hours 72\n"

" python hl_client.py stock-perps\n"

" python hl_client.py whales BTC --min-size 500000\n"

),

)

parser.add_argument(

"command",

choices=[

"market", "funding", "book", "whales", "universe", "candles",

"stock-perps", "stock-book", "stock-funding",

],

help="Command to run",

)

parser.add_argument("symbols", nargs="*", default=["BTC", "ETH", "SOL"], help="Symbols")

parser.add_argument("--hours", type=int, default=24, help="Lookback hours (default: 24)")

parser.add_argument("--depth", type=int, default=20, help="Order book depth (default: 20)")

parser.add_argument("--min-size", type=float, default=100_000, help="Min whale wall size in USD")

parser.add_argument("--interval", type=str, default="1h", help="Candle interval (default: 1h)")

args = parser.parse_args()

with HyperliquidClient() as client:

if args.command == "market":

print(json.dumps(client.get_market_data(args.symbols), indent=2))

elif args.command == "funding":

for sym in args.symbols:

data = client.get_funding_history(sym, hours=args.hours)

print(f"\n=== {sym} Funding History ({args.hours}h) ===")

for r in data[-10:]:

print(f" {r['time']} | {r['apr']:+.1f}% APR")

elif args.command == "book":

for sym in args.symbols:

book = client.get_orderbook(sym, depth=args.depth)

print(f"\n=== {sym} Orderbook (spread: {book['spread_bps']:.2f} bps) ===")

print(" BIDS:")

for b in book["bids"][:5]:

print(f" ${b['price']:,.2f} {b['size']:.4f} (${b['usd']:,.0f})")

print(" ASKS:")

for a in book["asks"][:5]:

print(f" ${a['price']:,.2f} {a['size']:.4f} (${a['usd']:,.0f})")

elif args.command == "whales":

for sym in args.symbols:

walls = client.get_whale_walls(sym, min_size_usd=args.min_size)

print(f"\n=== {sym} Whale Walls (>${args.min_size / 1000:.0f}k) ===")

if walls:

for w in walls[:10]:

print(f" {w['side']:<4} ${w['price']:,.2f} ${w['size_usd']:,.0f}")

else:

print(" No whale walls found")

elif args.command == "universe":

coins = client.get_universe()

print(f"=== Hyperliquid Perps ({len(coins)} coins) ===")

print(f"{'Coin':<8} {'Price':>12} {'Volume 24h':>14} {'OI':>12} {'Funding':>10}")

print("-" * 60)

for c in coins[:30]:

print(

f"{c['coin']:<8} ${c['price']:>10,.2f} "

f"${c['volume_24h'] / 1e6:>10.1f}M "

f"${c['oi'] / 1e6:>8.1f}M "

f"{c['funding_apr']:>+8.1f}%"

)

elif args.command == "candles":

for sym in args.symbols:

candles = client.get_candles(sym, interval=args.interval, lookback_hours=args.hours)

print(f"\n=== {sym} {args.interval} Candles ===")

for c in candles[-10:]:

print(

f" {c['time']} | O:{c['open']:,.2f} H:{c['high']:,.2f} "

f"L:{c['low']:,.2f} C:{c['close']:,.2f}"

)

elif args.command == "stock-perps":

stocks = client.get_stock_universe()

print(f"=== HIP-3 Stock Perpetuals ({len(stocks)} assets) ===")

print(f"{'Ticker':<12} {'Price':>12} {'Volume 24h':>14} {'OI':>12} {'Funding':>10}")

print("-" * 64)

for s in stocks:

print(

f"{s['coin']:<12} ${s['price']:>10,.2f} "

f"${s['volume_24h'] / 1e6:>10.1f}M "

f"${s['oi'] / 1e6:>8.1f}M "

f"{s['funding_apr']:>+8.1f}%"

)

elif args.command == "stock-book":

syms = args.symbols if args.symbols != ["BTC", "ETH", "SOL"] else ["TSLA", "NVDA"]

for sym in syms:

book = client.get_stock_orderbook(sym, depth=args.depth)

print(f"\n=== {sym} Stock Perp Book (spread: {book['spread_bps']:.2f} bps) ===")

print(" BIDS:")

for b in book["bids"][:5]:

print(f" ${b['price']:,.2f} {b['size']:.4f} (${b['usd']:,.0f})")

print(" ASKS:")

for a in book["asks"][:5]:

print(f" ${a['price']:,.2f} {a['size']:.4f} (${a['usd']:,.0f})")

elif args.command == "stock-funding":

syms = args.symbols if args.symbols != ["BTC", "ETH", "SOL"] else ["TSLA", "NVDA"]

for sym in syms:

data = client.get_stock_funding_history(sym, hours=args.hours)

print(f"\n=== {sym} Stock Perp Funding ({args.hours}h) ===")

for r in data[-10:]:

print(f" {r['time']} | {r['apr']:+.1f}% APR")

if __name__ == "__main__":

main()