ftdt-quant-lab/strategies/funding_arb.py

"""
Funding Rate Arb — Complete Implementation.

Strategy:
  Funding rates on perpetual futures represent the cost of leverage.
  When funding is positive (longs pay shorts), short the perp and collect.
  When funding is negative (shorts pay longs), go long the perp and collect.

  The Hyperliquid API provides predicted funding rates via:
    - predictedFundings: current predicted rate for each interval
    - metaAndAssetCtxs: asset context including current funding

  Entry: |annualized_funding_rate| > threshold (5-10% APR)
  Exit:  |annualized_funding_rate| < threshold/2 or after N hours
  Size:  scales with rate — higher rate = larger size
"""

import requests
import time
import math
from typing import Optional

MAINNET_API = "https://api.hyperliquid.xyz/info"
TESTNET_API = "https://api.hyperliquid-testnet.xyz/info"

# Cache funding rates to avoid hitting API every tick
_funding_cache: dict = {}
_last_funding_fetch: float = 0
FUNDING_CACHE_TTL = 30  # seconds


def get_funding_rates(use_testnet: bool = False) -> dict[str, float]:
    """
    Fetch current predicted funding rates for supported coins.

    Uses Hyperliquid's predictedFundings endpoint which returns
    the current projected funding rate for each perpetual.

    Returns: {coin: funding_rate_annualized}
    """
    global _funding_cache, _last_funding_fetch

    now = time.time()
    if now - _last_funding_fetch < FUNDING_CACHE_TTL and _funding_cache:
        return _funding_cache

    api = TESTNET_API if use_testnet else MAINNET_API
    rates: dict[str, float] = {}

    # Method 1: Try metaAndAssetCtxs (most reliable)
    try:
        r = requests.post(MAINNET_API, json={"type": "metaAndAssetCtxs"}, timeout=10)
        data = r.json()
        if isinstance(data, list) and len(data) >= 2:
            universe = data[0].get("universe", [])
            ctxs = data[1]
            for i, u in enumerate(universe):
                name = u.get("name", "")
                if name in ("BTC", "ETH", "HYPE", "VVV", "SOL"):
                    try:
                        funding = float(ctxs[i].get("funding", 0))
                        # funding is the 8h rate; annualize: × 365 × (24/8) = × 1095
                        annual = funding * 1095
                        rates[name] = annual
                    except (IndexError, ValueError, TypeError):
                        pass
    except Exception:
        pass

    # Method 2: Fallback to predictedFundings
    if not rates:
        try:
            r = requests.post(MAINNET_API, json={"type": "predictedFundings"}, timeout=10)
            data = r.json()
            if isinstance(data, list):
                for coin_entry in data:
                    coin = coin_entry[0]
                    if coin not in ("BTC", "ETH", "HYPE", "VVV", "SOL"):
                        continue
                    for venue_entry in coin_entry[1]:
                        venue = venue_entry[0]
                        info = venue_entry[1]
                        rate_str = info.get("fundingRate", "0")
                        try:
                            rate = float(rate_str)
                        except (ValueError, TypeError):
                            rate = 0.0
                        interval_hours = info.get("fundingIntervalHours", 8)
                        annual = rate * (365 * 24 / interval_hours)
                        if coin not in rates or "HlPerp" in venue:
                            rates[coin] = annual
        except Exception:
            pass

    _funding_cache = rates
    _last_funding_fetch = now
    return rates


def funding_arb_signal(
    coin: str = "BTC",
    apr_threshold: float = 0.05,  # 5% APR minimum
    apr_exit: float = 0.02,       # 2% APR to exit
    current_position: int = 0,
) -> dict:
    """
    Generate funding rate arbitrage signal.

    Args:
        coin: Ticker to check.
        apr_threshold: Minimum annualized funding rate to enter (>0.05 = 5%).
        apr_exit: Rate below which to exit position.
        current_position: -1 (short), 0 (none), +1 (long).

    Returns:
        dict with signal, rate, annual_apr, reason.
    """
    rates = get_funding_rates()
    annual = rates.get(coin, 0)
    rate_8h = annual / 1095  # de-annualize

    signal = 0
    reason = ""

    if abs(annual) > apr_threshold and current_position == 0:
        signal = -1 if annual > 0 else +1  # short if funding positive, long if negative
        reason = f"funding_{annual*100:.1f}pct_apr"
    elif current_position != 0:
        # Exit condition: rate has dropped below exit threshold
        if abs(annual) < apr_exit:
            signal = -current_position
            reason = f"exit_funding_{annual*100:.2f}pct_apr"
        # Also exit if funding flips sign (we'd be paying instead of collecting)
        elif (current_position == -1 and annual < 0) or (current_position == 1 and annual > 0):
            signal = -current_position
            reason = f"exit_funding_flipped_{annual*100:.2f}pct_apr"

    return {
        "signal": signal,
        "rate_8h": rate_8h,
        "annual_apr": annual,
        "reason": reason,
    }