""" Spot-Perp Basis Arbitrage — delta-neutral strategy exploiting spot/perp price gaps. Hyperliquid has both spot and perpetual futures markets for most assets. The spot price and perp price should converge at settlement, but the perp can trade at a premium (contango) or discount (backwardation) to spot. Strategy: - When perp > spot + threshold: short perp, long spot - When perp < spot - threshold: long perp, short spot - Hold until basis converges or funding arbitrage overwhelms Key advantages: - Delta-neutral (market-neutral) - Low correlation to other strategies - Capital-efficient (spot + perp use separate collateral on HL) - Combinable with funding arb (earn funding while holding basis trade) Data needed: - Hyperliquid spot mid price - Hyperliquid perp mark price - Perp funding rate (for combined carry trade) """ from __future__ import annotations import logging from collections import deque import numpy as np logger = logging.getLogger(__name__) class SpotPerpBasisArb: """Delta-neutral spot-perpetual basis arbitrage. Parameters ---------- entry_threshold_bps: float — minimum basis (in bps) to enter a trade exit_threshold_bps: float — basis below which to exit max_hold_hours: float — maximum trade duration size_usd: float — notional size per leg in USD spot_fee_rate: float — spot taker fee rate (decimal) perp_fee_rate: float — perp taker fee rate (decimal) min_expected_profit_bps: float — minimum expected profit after fees """ def __init__( self, entry_threshold_bps: float = 3.0, exit_threshold_bps: float = 1.0, max_hold_hours: float = 48.0, size_usd: float = 1000.0, spot_fee_rate: float = 0.0010, perp_fee_rate: float = 0.0007, min_expected_profit_bps: float = 1.5, ): self.entry_threshold = entry_threshold_bps / 10000 # bps → decimal self.exit_threshold = exit_threshold_bps / 10000 self.max_hold_hours = max_hold_hours self.size_usd = size_usd self.spot_fee_rate = spot_fee_rate self.perp_fee_rate = perp_fee_rate self.min_expected_profit = min_expected_profit_bps / 10000 self._position: dict = {} self._trades: list[dict] = [] self._basis_history: deque = deque(maxlen=500) def signal( self, spot_price: float, perp_price: float, funding_rate: float = 0.0, ) -> dict: """Compute basis trade signal. Args: spot_price: current spot mid price perp_price: current perp mark/index price funding_rate: current 8h funding rate (decimal) Returns: dict with action, basis_bps, expected_profit_bps, funding_apr """ if spot_price <= 0 or perp_price <= 0: return {"action": "HOLD", "basis_bps": 0.0} basis = (perp_price - spot_price) / spot_price basis_bps = basis * 10000 self._basis_history.append(basis_bps) in_position = bool(self._position) if in_position: # Exit if basis narrows enough or max hold exceeded pos_basis = abs(self._position.get("entry_basis", 0)) current_basis_abs = abs(basis) if current_basis_abs < self.exit_threshold: return { "action": "EXIT", "basis_bps": round(basis_bps, 2), "reason": f"basis_converged_{current_basis_abs*10000:.0f}bps", } # Compute expected profit after fees round_trip_fee = self.spot_fee_rate * 2 + self.perp_fee_rate * 2 expected_profit = abs(basis) - round_trip_fee # Add funding carry expectation funding_apr = abs(funding_rate) * 365 * 3 # 8h → annual funding_profit = funding_apr * (self.max_hold_hours / (365 * 24)) if not in_position and abs(basis) > self.entry_threshold and expected_profit > self.min_expected_profit: if basis > 0: return { "action": "SELL_PERP_BUY_SPOT", "basis_bps": round(basis_bps, 2), "expected_profit_bps": round(expected_profit * 10000, 2), "funding_apr_pct": round(funding_apr * 100, 2), "reason": f"perp_premium_{basis_bps:.0f}bps", } else: return { "action": "BUY_PERP_SELL_SPOT", "basis_bps": round(basis_bps, 2), "expected_profit_bps": round(expected_profit * 10000, 2), "funding_apr_pct": round(funding_apr * 100, 2), "reason": f"perp_discount_{abs(basis_bps):.0f}bps", } return { "action": "HOLD", "basis_bps": round(basis_bps, 2), "expected_profit_bps": round(expected_profit * 10000, 2) if expected_profit > 0 else 0, "funding_apr_pct": round(funding_apr * 100, 2), } def enter(self, action: str, spot_price: float, perp_price: float): """Record a trade entry.""" basis = (perp_price - spot_price) / spot_price self._position = { "action": action, "entry_spot": spot_price, "entry_perp": perp_price, "entry_basis": basis, "entry_time": __import__('time').time(), "spot_size": self.size_usd / spot_price, "perp_size": self.size_usd / perp_price, } def exit(self, spot_price: float, perp_price: float) -> dict: """Exit the current position and compute PnL.""" if not self._position: return {"pnl": 0.0, "pnl_pct": 0.0} entry = self._position action = entry["action"] if action == "SELL_PERP_BUY_SPOT": perp_pnl = (entry["entry_perp"] - perp_price) * entry["perp_size"] spot_pnl = (spot_price - entry["entry_spot"]) * entry["spot_size"] elif action == "BUY_PERP_SELL_SPOT": perp_pnl = (perp_price - entry["entry_perp"]) * entry["perp_size"] spot_pnl = (entry["entry_spot"] - spot_price) * entry["spot_size"] else: perp_pnl = 0.0 spot_pnl = 0.0 gross_pnl = perp_pnl + spot_pnl spot_fee = entry["spot_size"] * entry["entry_spot"] * self.spot_fee_rate + \ entry["spot_size"] * spot_price * self.spot_fee_rate perp_fee = entry["perp_size"] * entry["entry_perp"] * self.perp_fee_rate + \ entry["perp_size"] * perp_price * self.perp_fee_rate total_fee = spot_fee + perp_fee net_pnl = gross_pnl - total_fee pnl_pct = net_pnl / (self.size_usd * 2) if self.size_usd > 0 else 0 trade = { "action": action, "entry_spot": round(entry["entry_spot"], 2), "entry_perp": round(entry["entry_perp"], 2), "exit_spot": round(spot_price, 2), "exit_perp": round(perp_price, 2), "entry_basis_bps": round(entry["entry_basis"] * 10000, 2), "exit_basis_bps": round((perp_price - spot_price) / spot_price * 10000, 2), "gross_pnl": round(gross_pnl, 4), "fee": round(total_fee, 4), "net_pnl": round(net_pnl, 4), "pnl_pct": round(pnl_pct * 100, 3), } self._trades.append(trade) self._position = {} return trade def is_in_position(self) -> bool: return bool(self._position) def summary(self) -> dict: """Return strategy summary.""" return { "in_position": self.is_in_position(), "position": self._position if self._position else None, "total_trades": len(self._trades), "total_pnl": round(sum(t["net_pnl"] for t in self._trades), 4), "recent_basis_bps": list(self._basis_history)[-10:], } # Multi-asset basis arb across coins BASIS_ARB_COINS = ["BTC", "ETH", "SOL", "HYPE", "ARB"] def multi_asset_basis_scan( spot_prices: dict[str, float], perp_prices: dict[str, float], funding_rates: dict[str, float], entry_threshold_bps: float = 3.0, ) -> list[dict]: """Scan all assets for basis arb opportunities. Returns sorted list of opportunities, best first. """ opportunities = [] for coin in perp_prices: if coin not in spot_prices: continue spot = spot_prices.get(coin, 0) perp = perp_prices.get(coin, 0) fr = funding_rates.get(coin, 0) if spot <= 0 or perp <= 0: continue basis = (perp - spot) / spot if abs(basis) > entry_threshold_bps / 10000: opportunities.append({ "coin": coin, "spot": spot, "perp": perp, "basis_bps": round(basis * 10000, 2), "funding_rate": round(fr, 8), "direction": "SELL_PERP_BUY_SPOT" if basis > 0 else "BUY_PERP_SELL_SPOT", }) opportunities.sort(key=lambda x: abs(x["basis_bps"]), reverse=True) return opportunities