Wire up real Hyperliquid integration and funding rate API

Replaced the placeholder live node with a proper NautilusTrader
TradingNode that connects to Hyperliquid Testnet using the
official adapter. Added:

- common/hyperliquid_api.py: direct REST calls to Hyperliquid's
  info endpoint for funding rates, predicted fundings, and
  asset contexts
- backtests/run_backtest.py: CLI runner for strategy backtests
- Updated funding_rate_arb.py to fetch real funding rates
  instead of using a hardcoded placeholder
- Added requests to requirements.txt
This commit is contained in:
ramseshk
2026-08-03 11:37:47 +00:00
parent b59dcc3629
commit c1da0cbe65
5 changed files with 366 additions and 25 deletions
+75
View File
@@ -0,0 +1,75 @@
"""
Hyperliquid API utilities.
Direct REST calls to Hyperliquid info endpoint for data
not yet covered by the NautilusTrader adapter (funding rates,
predicted fundings, asset contexts).
"""
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) -> list[dict]:
"""
Fetch asset contexts including current funding rates.
Returns list of per-asset dicts with keys:
funding, openInterest, markPx, oraclePx, premium, dayNtlVlm, etc.
"""
api = TESTNET_API if testnet else MAINNET_API
data = _post(api, {"type": "metaAndAssetCtxs"})
# data[0] = universe, data[1] = asset contexts
if isinstance(data, list) and len(data) >= 2:
return data[1]
return []
def get_funding_rate(asset_name: str, testnet: bool = True) -> float | None:
"""
Get the current funding rate for a specific asset.
Funding is paid every 8 hours. Positive = longs pay shorts.
"""
ctxs = get_asset_contexts(testnet=testnet)
for ctx in ctxs:
if isinstance(ctx, dict) and ctx.get("name") == asset_name.upper():
funding_str = ctx.get("funding", "0")
return float(funding_str)
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)
rates = {}
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:
"""
Get the predicted funding rate for the next interval.
Uses the predictedFundings endpoint.
"""
api = TESTNET_API if testnet else MAINNET_API
data = _post(api, {"type": "predictedFundings"})
if isinstance(data, list):
for item in data:
if isinstance(item, dict) and item.get("name") == asset_name.upper():
# Return the Hyperliquid-specific prediction
predicted = item.get("funding", "0")
return float(predicted)
return None