""" Funding Rate Arbitrage — backtestable strategy module. Delta-neutral carry trade on Hyperliquid perps. When funding rate is high: - Short the perpetual (collect funding payments) - The profit is the funding rate, not price direction Features: - Configurable entry/exit thresholds - Position sizing proportional to funding rate - Funding payment tracking with accurate Hyperliquid 8h schedule - Max hold time (exit after N hours regardless) - Stop-loss if basis widens (mark price moves against funding direction) - Per-trade PnL accounting with fees, funding, and mark PnL Usage (backtest): arb = FundingArb(apr_threshold=0.30, apr_exit=0.10, size=0.001) for hourly_funding in funding_history: trade = arb.tick(funding_rate, mark_price, timestamp) if trade: print(f"Trade: {trade}") Usage (live): arb = FundingArb(apr_threshold=0.30) signal = arb.signal(check_rates(time.time())) if signal["action"] != "HOLD": execute(signal) """ from __future__ import annotations import time from collections import deque from typing import Optional class FundingArb: """Delta-neutral funding rate carry strategy. Logic: - Entry: |annualized_funding| > apr_threshold AND no position - Exit: |annualized_funding| < apr_exit OR hold_time > max_hold_hours OR funding direction flips (paying instead of collecting) OR basis stop-loss triggered """ def __init__( self, apr_threshold: float = 0.30, apr_exit: float = 0.10, size: float = 0.001, max_hold_hours: float = 48.0, basis_stop_loss_pct: float = 0.03, taker_fee_pct: float = 0.00045, maker_fee_pct: float = 0.00015, ): self._apr_threshold = apr_threshold self._apr_exit = apr_exit self._size = size self._max_hold_seconds = max_hold_hours * 3600 self._basis_stop_loss = basis_stop_loss_pct self._taker_fee = taker_fee_pct self._maker_fee = maker_fee_pct self._position: int = 0 self._entry_price: float = 0.0 self._entry_time: float = 0.0 self._entry_apr: float = 0.0 self._funding_collected: float = 0.0 self._funding_paid: float = 0.0 self._trades: list[dict] = [] self._funding_history: deque[float] = deque(maxlen=200) self._mark_history: deque[float] = deque(maxlen=200) self._signals: deque[dict] = deque(maxlen=50) @property def position(self) -> int: return self._position @property def trades(self) -> list[dict]: return self._trades def tick( self, funding_rate_annual: float, mark_price: float, timestamp: Optional[float] = None, ) -> dict | None: """Process one funding rate observation. Returns trade dict if entry/exit occurred.""" if timestamp is None: timestamp = time.time() if mark_price <= 0: return None self._funding_history.append(funding_rate_annual) self._mark_history.append(mark_price) abs_apr = abs(funding_rate_annual) action = "HOLD" trade = None if self._position == 0: if abs_apr > self._apr_threshold: action = "SELL" if funding_rate_annual > 0 else "BUY" self._position = -1 if funding_rate_annual > 0 else 1 self._entry_price = mark_price self._entry_time = timestamp self._entry_apr = funding_rate_annual notional = self._size * mark_price fee = notional * self._taker_fee self._funding_paid += fee trade = { "action": action, "side": action, "size": self._size, "entry_price": mark_price, "apr": round(funding_rate_annual, 4), "apr_pct": round(funding_rate_annual * 100, 2), "fee": round(fee, 4), } self._signals.append({ "timestamp": timestamp, "action": action, "apr": funding_rate_annual, "price": mark_price, }) else: hold_seconds = timestamp - self._entry_time direction = "short" if self._position == -1 else "long" exit_reason = "" if abs_apr < self._apr_exit: exit_reason = f"apr_faded_to_{abs_apr*100:.1f}%" elif hold_seconds >= self._max_hold_seconds: exit_reason = f"max_hold_{hold_seconds/3600:.1f}h" elif (self._position == -1 and funding_rate_annual < 0) or \ (self._position == 1 and funding_rate_annual > 0): exit_reason = f"funding_flipped_to_{funding_rate_annual*100:.2f}%" else: price_move = (mark_price - self._entry_price) / self._entry_price position_pnl_pct = price_move * self._position if abs(position_pnl_pct) > self._basis_stop_loss: exit_reason = f"basis_stop_loss_{position_pnl_pct*100:.2f}%" if exit_reason: notional = self._size * mark_price fee = notional * self._taker_fee price_pnl = self._size * (mark_price - self._entry_price) * self._position funding_earned = 0.0 if isinstance(self._entry_apr, float) and self._entry_apr != 0: funding_rate_8h = self._entry_apr / 1095 funding_intervals = hold_seconds / (8 * 3600) funding_earned = notional * abs(funding_rate_8h) * funding_intervals * 0.95 net_pnl = price_pnl + funding_earned - fee - self._funding_paid action = "BUY" if self._position == -1 else "SELL" trade = { "action": f"EXIT_{exit_reason}", "side": action, "entry_price": round(self._entry_price, 2), "exit_price": round(mark_price, 2), "size": self._size, "direction": direction, "hold_hours": round(hold_seconds / 3600, 2), "entry_apr": round(self._entry_apr * 100, 2), "exit_apr": round(funding_rate_annual * 100, 2), "price_pnl": round(price_pnl, 4), "funding_earned": round(funding_earned, 4), "fees": round(fee + self._funding_paid, 4), "net_pnl": round(net_pnl, 4), "reason": exit_reason, } self._trades.append(trade) self._signals.append({ "timestamp": timestamp, "action": "EXIT", "apr": funding_rate_annual, "price": mark_price, "reason": exit_reason, "pnl": net_pnl, }) self._position = 0 self._entry_price = 0.0 self._entry_time = 0.0 self._entry_apr = 0.0 self._funding_paid = 0.0 return trade def signal( self, funding_rate_annual: float, mark_price: float = 0.0, timestamp: Optional[float] = None, ) -> dict: """Generate trading signal without executing.""" if timestamp is None: timestamp = time.time() abs_apr = abs(funding_rate_annual) if self._position == 0 and abs_apr > self._apr_threshold: return { "action": "SELL" if funding_rate_annual > 0 else "BUY", "size": self._size, "apr": round(funding_rate_annual * 100, 2), "reason": f"apr_{abs_apr*100:.1f}%_above_{self._apr_threshold*100:.0f}%", } elif self._position != 0 and abs_apr < self._apr_exit: return { "action": "EXIT", "reason": f"apr_faded_to_{abs_apr*100:.1f}%", } return {"action": "HOLD"} def summary(self) -> dict: if not self._trades: return { "total_trades": 0, "win_rate": 0.0, "total_net_pnl": 0.0, "avg_net_pnl": 0.0, "avg_hold_hours": 0.0, "total_funding_earned": 0.0, "position": self._position, } wins = sum(1 for t in self._trades if t["net_pnl"] > 0) total_net = sum(t["net_pnl"] for t in self._trades) total_funding = sum(t["funding_earned"] for t in self._trades) hold_hours = [t["hold_hours"] for t in self._trades] return { "total_trades": len(self._trades), "win_rate": round(wins / len(self._trades), 3), "total_net_pnl": round(total_net, 4), "avg_net_pnl": round(total_net / len(self._trades), 4), "avg_hold_hours": round(sum(hold_hours) / len(hold_hours), 2), "total_funding_earned": round(total_funding, 4), "best_trade": round(max(t["net_pnl"] for t in self._trades), 4), "worst_trade": round(min(t["net_pnl"] for t in self._trades), 4), "position": self._position, } def reset(self): self._position = 0 self._entry_price = 0.0 self._entry_time = 0.0 self._entry_apr = 0.0 self._funding_collected = 0.0 self._funding_paid = 0.0 self._trades.clear() self._signals.clear() self._funding_history.clear() self._mark_history.clear() def backtest_funding_arb( funding_rates: list[float], mark_prices: list[float], apr_threshold: float = 0.30, size: float = 0.001, taker_fee_pct: float = 0.00045, ) -> dict: """Run funding arb backtest on a series of funding rate observations. Args: funding_rates: list of annualized funding rates (e.g., from HL API) mark_prices: list of corresponding mark prices apr_threshold: minimum annual APR to enter size: trade size taker_fee_pct: taker fee per trade Returns dict with trades and summary. """ arb = FundingArb( apr_threshold=apr_threshold, size=size, taker_fee_pct=taker_fee_pct, ) min_len = min(len(funding_rates), len(mark_prices)) for i in range(min_len): arb.tick(funding_rates[i], mark_prices[i], float(i)) return arb.summary() def run_funding_discovery(data_dir: str = "data/raw", coin: str = "BTC", start_date: str = "2026-01-01", end_date: str = "2030-01-01") -> dict: """Run funding rate discovery — analyze historical funding rates to find optimal entry/exit thresholds. Returns dict with rate distribution percentiles and backtest results at different thresholds. """ import numpy as np from data.store import read_range msgs = read_range(data_dir, channel="funding", coin=coin, start_date=start_date, end_date=end_date) if not msgs: return {"error": "No funding data available"} rates = [] marks = [] for msg in msgs: payload = msg.get("payload", {}) rate = float(payload.get("funding", 0)) mark = float(payload.get("mark_px", 0)) if mark > 0: annual = rate * 1095 rates.append(annual) marks.append(mark) if not rates: return {"error": "No valid funding observations"} a = np.array(rates) abs_a = np.abs(a) percentiles = [10, 25, 50, 75, 90, 95, 99] result = { "n_observations": len(rates), "rate_distribution": { "mean_apr_pct": round(float(np.mean(a)) * 100, 2), "std_apr_pct": round(float(np.std(a)) * 100, 2), "max_apr_pct": round(float(np.max(a)) * 100, 2), "min_apr_pct": round(float(np.min(a)) * 100, 2), "abs_percentiles": { f"p{p}": round(float(np.percentile(abs_a, p)) * 100, 2) for p in percentiles }, }, "backtests": {}, } for threshold in [0.05, 0.10, 0.20, 0.30, 0.50]: summary = backtest_funding_arb(rates, marks, apr_threshold=threshold) if summary["total_trades"] > 0: result["backtests"][f"apr_{int(threshold*100)}pct"] = summary return result