""" Term structure monitor — perp vs quarterly vs bi-quarterly futures basis. Hyperliquid offers perpetual (funding-based), quarterly, and bi-quarterly futures contracts. The basis curve (perp→quarterly→bi-quarterly) contains information about market expectations and can be traded. Anomalies: - Perp funding deeply negative but quarterly basis remains steep → go long perp (collect funding), short quarterly (lock basis) - Quarterly futures converging to perp at expiration → calendar spread mean-reversion - Bi-quarterly premium over quarterly deviating from fair value → curve steepener/flattener trades """ from __future__ import annotations from collections import deque from typing import Optional class TermStructureMonitor: """Monitor perp/futures term structure for arbitrage opportunities. Tracks: - Perp funding rate and mark price - Quarterly futures price - Bi-quarterly futures price (if available) - Basis spreads: quarterly-perp, biq-quarterly, biq-perp - Calendar spread mean-reversion """ def __init__( self, funding_window: int = 1440, # 24h of 1-min samples basis_window: int = 100, ): self._funding_window = funding_window self._basis_window = basis_window self._perp_prices: dict[str, deque[float]] = {} self._quarterly_prices: dict[str, deque[float]] = {} self._biq_prices: dict[str, deque[float]] = {} self._funding_rates: dict[str, deque[float]] = {} self._funding_epoch_seconds: int = 8 * 3600 self._quarterly_expiry_days: int = 90 self._biq_expiry_days: int = 180 # ── Data feed ──────────────────────────────────────────── def update_perp(self, coin: str, price: float, funding_rate: float): c = coin.upper() self._perp_prices.setdefault(c, deque(maxlen=self._basis_window)).append(price) self._funding_rates.setdefault(c, deque(maxlen=self._funding_window)).append(funding_rate) def update_quarterly(self, coin: str, price: float): self._quarterly_prices.setdefault(coin.upper(), deque(maxlen=self._basis_window)).append(price) def update_biq(self, coin: str, price: float): self._biq_prices.setdefault(coin.upper(), deque(maxlen=self._basis_window)).append(price) # ── Basis computation ──────────────────────────────────── def quarterly_perp_basis(self, coin: str) -> dict | None: """Basis between quarterly future and perpetual.""" q = list(self._quarterly_prices.get(coin.upper(), [])) p = list(self._perp_prices.get(coin.upper(), [])) min_len = min(len(q), len(p)) if min_len < 2: return None qq = q[-min_len:] pp = p[-min_len:] basis_bps = [(qq[i] - pp[i]) / pp[i] * 10000 for i in range(min_len) if pp[i] > 0] if not basis_bps: return None return { "current_bps": round(basis_bps[-1], 2), "mean_bps": round(sum(basis_bps) / len(basis_bps), 2), "std_bps": round(_std(basis_bps), 2), "z_score": round((basis_bps[-1] - sum(basis_bps) / len(basis_bps)) / max(_std(basis_bps), 0.01), 2), "n_samples": len(basis_bps), } def biq_quarterly_basis(self, coin: str) -> dict | None: """Basis between bi-quarterly and quarterly futures (curve steepness).""" bq = list(self._biq_prices.get(coin.upper(), [])) q = list(self._quarterly_prices.get(coin.upper(), [])) min_len = min(len(bq), len(q)) if min_len < 2: return None bb = bq[-min_len:] qq = q[-min_len:] basis_bps = [(bb[i] - qq[i]) / qq[i] * 10000 for i in range(min_len) if qq[i] > 0] if not basis_bps: return None return { "current_bps": round(basis_bps[-1], 2), "mean_bps": round(sum(basis_bps) / len(basis_bps), 2), "std_bps": round(_std(basis_bps), 2), "z_score": round((basis_bps[-1] - sum(basis_bps) / len(basis_bps)) / max(_std(basis_bps), 0.01), 2), "n_samples": len(basis_bps), } def fair_quarterly_price(self, coin: str, risk_free_annual: float = 0.05) -> dict | None: """Compute fair quarterly price from perp via interest rate parity.""" p = list(self._perp_prices.get(coin.upper(), [])) if not p: return None perp_px = p[-1] days = self._quarterly_expiry_days funding = list(self._funding_rates.get(coin.upper(), [])) avg_funding = sum(funding[-100:]) / max(len(funding[-100:]), 1) if funding else 0 # Fair quarterly = perp * (1 + (r + avg_funding) * days/365) carry_rate = risk_free_annual + avg_funding * 3 * 365 # annualize 8h funding fair_px = perp_px * (1 + carry_rate * days / 365) return { "perp_price": perp_px, "fair_quarterly": round(fair_px, 2), "carry_rate_annual_pct": round(carry_rate * 100, 2), "days_to_expiry": days, } def signal(self, coin: str) -> dict: """Generate term-structure trading signal. Returns: signal: "buy_basis", "sell_basis", "curve_steepener", "curve_flattener", "none" """ c = coin.upper() qp = self.quarterly_perp_basis(c) bq = self.biq_quarterly_basis(c) fair = self.fair_quarterly_price(c) signals = [] # Check quarterly-perp basis anomalies if qp and abs(qp["z_score"]) > 2.0: if qp["z_score"] > 0: signals.append({ "signal": "sell_basis", "reason": f"Quarterly {qp['z_score']:.1f}σ rich vs perp", "confidence": min(1.0, abs(qp["z_score"]) / 4.0), }) else: signals.append({ "signal": "buy_basis", "reason": f"Quarterly {qp['z_score']:.1f}σ cheap vs perp", "confidence": min(1.0, abs(qp["z_score"]) / 4.0), }) # Check curve steepness if bq and abs(bq["z_score"]) > 2.0: if bq["z_score"] > 0: signals.append({ "signal": "curve_flattener", "reason": f"BiQ {bq['z_score']:.1f}σ rich vs quarterly", "confidence": min(1.0, abs(bq["z_score"]) / 4.0), }) else: signals.append({ "signal": "curve_steepener", "reason": f"BiQ {bq['z_score']:.1f}σ cheap vs quarterly", "confidence": min(1.0, abs(bq["z_score"]) / 4.0), }) result = { "coin": c, "signals": signals, "primary_signal": signals[0]["signal"] if signals else "none", "quarterly_perp_basis": qp, "biq_quarterly_basis": bq, "fair_quarterly": fair, } return result def summary(self) -> dict: return {coin: self.signal(coin) for coin in self._perp_prices} def _std(vals: list[float]) -> float: """Population standard deviation.""" if len(vals) < 2: return 0.0 mean = sum(vals) / len(vals) return (sum((v - mean) ** 2 for v in vals) / len(vals)) ** 0.5