ramseshk
3520056313
The metaAndAssetCtxs endpoint returns asset names in a parallel universe array — asset contexts don't have a "name" field. Fixed the API module to index by the universe array properly. Also added get_mark_price() helper. Tested against testnet: BTC funding 0.00125% per 8h, mark $62,873.
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""
|
|
Hyperliquid API utilities.
|
|
|
|
Direct REST calls to Hyperliquid's info endpoint for data
|
|
not yet covered by the NautilusTrader adapter (funding rates,
|
|
predicted fundings, asset contexts).
|
|
|
|
API structure:
|
|
POST /info {"type": "metaAndAssetCtxs"}
|
|
Returns [universe_meta, asset_contexts]
|
|
- universe_meta["universe"] = list of {"name": "BTC", ...}
|
|
- asset_contexts[i] corresponds to universe[i]
|
|
"""
|
|
import requests
|
|
from typing import Any
|
|
|
|
TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
|
|
MAINNET_API = "https://api.hyperliquid.xyz/info"
|
|
|
|
|
|
def _post(api_url: str, payload: dict) -> Any:
|
|
resp = requests.post(api_url, json=payload, timeout=10)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def get_asset_contexts(testnet: bool = True) -> dict[str, dict]:
|
|
"""
|
|
Fetch asset contexts including current funding rates.
|
|
Returns dict keyed by asset name (BTC, ETH, SOL, etc.)
|
|
with values containing funding, markPx, openInterest, etc.
|
|
"""
|
|
api = TESTNET_API if testnet else MAINNET_API
|
|
data = _post(api, {"type": "metaAndAssetCtxs"})
|
|
|
|
if not isinstance(data, list) or len(data) < 2:
|
|
return {}
|
|
|
|
universe = data[0].get("universe", [])
|
|
ctxs = data[1]
|
|
|
|
result = {}
|
|
for i, asset_info in enumerate(universe):
|
|
if i < len(ctxs):
|
|
name = asset_info.get("name", "")
|
|
if name:
|
|
result[name] = ctxs[i]
|
|
return result
|
|
|
|
|
|
def get_funding_rate(asset_name: str, testnet: bool = True) -> float | None:
|
|
"""
|
|
Get the current funding rate for a specific asset.
|
|
Funding is a per-8-hour rate. Positive = longs pay shorts.
|
|
Multiply by 3 * 365 for approximate annualized rate.
|
|
"""
|
|
ctxs = get_asset_contexts(testnet=testnet)
|
|
ctx = ctxs.get(asset_name.upper())
|
|
if ctx:
|
|
return float(ctx.get("funding", 0))
|
|
return None
|
|
|
|
|
|
def get_all_funding_rates(testnet: bool = True) -> dict[str, float]:
|
|
"""Get funding rates for all assets on Hyperliquid."""
|
|
ctxs = get_asset_contexts(testnet=testnet)
|
|
return {name: float(ctx["funding"]) for name, ctx in ctxs.items()}
|
|
|
|
|
|
def get_predicted_funding(asset_name: str, testnet: bool = True) -> float | None:
|
|
"""
|
|
Get the predicted funding rate for the next interval.
|
|
Uses the predictedFundings endpoint.
|
|
"""
|
|
api = TESTNET_API if testnet else MAINNET_API
|
|
try:
|
|
data = _post(api, {"type": "predictedFundings"})
|
|
except Exception:
|
|
return None
|
|
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
if isinstance(item, dict) and item.get("name", "").upper() == asset_name.upper():
|
|
return float(item.get("funding", 0))
|
|
return None
|
|
|
|
|
|
def get_mark_price(asset_name: str, testnet: bool = True) -> float | None:
|
|
"""Get the current mark price for an asset."""
|
|
ctxs = get_asset_contexts(testnet=testnet)
|
|
ctx = ctxs.get(asset_name.upper())
|
|
if ctx:
|
|
return float(ctx.get("markPx", 0))
|
|
return None
|