Fix Hyperliquid API: asset names come from universe array, not context dicts

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.
This commit is contained in:
ramseshk
2026-08-03 11:39:11 +00:00
parent c1da0cbe65
commit 3520056313
+45 -26
View File
@@ -1,14 +1,19 @@
""" """
Hyperliquid API utilities. Hyperliquid API utilities.
Direct REST calls to Hyperliquid info endpoint for data Direct REST calls to Hyperliquid's info endpoint for data
not yet covered by the NautilusTrader adapter (funding rates, not yet covered by the NautilusTrader adapter (funding rates,
predicted fundings, asset contexts). 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 import requests
from typing import Any from typing import Any
TESTNET_API = "https://api.hyperliquid-testnet.xyz/info" TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"
MAINNET_API = "https://api.hyperliquid.xyz/info" MAINNET_API = "https://api.hyperliquid.xyz/info"
@@ -19,44 +24,47 @@ def _post(api_url: str, payload: dict) -> Any:
return resp.json() return resp.json()
def get_asset_contexts(testnet: bool = True) -> list[dict]: def get_asset_contexts(testnet: bool = True) -> dict[str, dict]:
""" """
Fetch asset contexts including current funding rates. Fetch asset contexts including current funding rates.
Returns list of per-asset dicts with keys: Returns dict keyed by asset name (BTC, ETH, SOL, etc.)
funding, openInterest, markPx, oraclePx, premium, dayNtlVlm, etc. with values containing funding, markPx, openInterest, etc.
""" """
api = TESTNET_API if testnet else MAINNET_API api = TESTNET_API if testnet else MAINNET_API
data = _post(api, {"type": "metaAndAssetCtxs"}) data = _post(api, {"type": "metaAndAssetCtxs"})
# data[0] = universe, data[1] = asset contexts
if isinstance(data, list) and len(data) >= 2: if not isinstance(data, list) or len(data) < 2:
return data[1] return {}
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: def get_funding_rate(asset_name: str, testnet: bool = True) -> float | None:
""" """
Get the current funding rate for a specific asset. Get the current funding rate for a specific asset.
Funding is paid every 8 hours. Positive = longs pay shorts. 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) ctxs = get_asset_contexts(testnet=testnet)
for ctx in ctxs: ctx = ctxs.get(asset_name.upper())
if isinstance(ctx, dict) and ctx.get("name") == asset_name.upper(): if ctx:
funding_str = ctx.get("funding", "0") return float(ctx.get("funding", 0))
return float(funding_str)
return None return None
def get_all_funding_rates(testnet: bool = True) -> dict[str, float]: def get_all_funding_rates(testnet: bool = True) -> dict[str, float]:
"""Get funding rates for all assets on Hyperliquid.""" """Get funding rates for all assets on Hyperliquid."""
ctxs = get_asset_contexts(testnet=testnet) ctxs = get_asset_contexts(testnet=testnet)
rates = {} return {name: float(ctx["funding"]) for name, ctx in ctxs.items()}
for ctx in ctxs:
if isinstance(ctx, dict):
name = ctx.get("name", "")
funding_str = ctx.get("funding", "0")
if name:
rates[name] = float(funding_str)
return rates
def get_predicted_funding(asset_name: str, testnet: bool = True) -> float | None: def get_predicted_funding(asset_name: str, testnet: bool = True) -> float | None:
@@ -65,11 +73,22 @@ def get_predicted_funding(asset_name: str, testnet: bool = True) -> float | None
Uses the predictedFundings endpoint. Uses the predictedFundings endpoint.
""" """
api = TESTNET_API if testnet else MAINNET_API api = TESTNET_API if testnet else MAINNET_API
try:
data = _post(api, {"type": "predictedFundings"}) data = _post(api, {"type": "predictedFundings"})
except Exception:
return None
if isinstance(data, list): if isinstance(data, list):
for item in data: for item in data:
if isinstance(item, dict) and item.get("name") == asset_name.upper(): if isinstance(item, dict) and item.get("name", "").upper() == asset_name.upper():
# Return the Hyperliquid-specific prediction return float(item.get("funding", 0))
predicted = item.get("funding", "0") return None
return float(predicted)
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 return None