ramseshk
254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""
|
|
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__)
|
|
|
|
EPS = 1e-10
|
|
|
|
|
|
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 = 5.0,
|
|
exit_threshold_bps: float = 1.0,
|
|
max_hold_hours: float = 48.0,
|
|
size_usd: float = 1000.0,
|
|
spot_fee_rate: float = 0.0004,
|
|
perp_fee_rate: float = 0.00015,
|
|
min_expected_profit_bps: float = 1.0,
|
|
):
|
|
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 - EPS:
|
|
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 - EPS:
|
|
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
|