""" Historical backtest runner — uses REAL Hyperliquid candle data. Fetches hourly candles from Hyperliquid mainnet info API, runs each strategy's logic against actual price history, simulates fills with configurable fee tiers. No more random walks — every backtest is reproducible from real data. Usage: python backtests/historical_runner.py --coin BTC --strategy all python backtests/historical_runner.py --coin ETH --fee-tier 3 --staking-tier gold """ import argparse import json import os import sys import time import math import random from datetime import datetime, timedelta from pathlib import Path from collections import deque sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import requests from config.fee_tiers import get_perp_fees, PERPS_TIERS, STAKING_TIERS, STRATEGY_FEE_MODELS from common.metrics import sharpe, sortino, max_drawdown, win_rate # ── Constants ── MAINNET_API = "https://api.hyperliquid.xyz/info" RESULTS_DIR = Path(__file__).resolve().parent / "results" / "historical" os.makedirs(RESULTS_DIR, exist_ok=True) # Strategy configs matching paper trader STRATEGIES = { "ofi": {"name": "Order Book Imbalance", "size": 0.002, "fee_model": "taker"}, "iceberg": {"name": "Iceberg Detection", "size": 0.001, "fee_model": "taker"}, "funding_arb":{"name": "Funding Rate Arb", "size": 0.005, "fee_model": "taker"}, "pairs": {"name": "Pairs Trading", "size": 0.006, "fee_model": "taker"}, "avellaneda":{"name": "Avellaneda-Stoikov", "size": 0.001, "fee_model": "maker"}, "momentum": {"name": "Momentum Breakout", "size": 0.002, "fee_model": "taker"}, "mean_rev": {"name": "Mean Reversion", "size": 0.002, "fee_model": "taker"}, "kalman_pairs": {"name": "Kalman Pairs", "size": 0.005, "fee_model": "taker"}, } def fetch_candles(coin: str, interval: str = "1h", limit: int = 720) -> list[dict]: """Fetch historical candles from Hyperliquid mainnet. Returns list of {t: timestamp_ms, o: open, h: high, l: low, c: close, v: volume} Most recent first. """ now_ms = int(time.time() * 1000) # 720 hours = 30 days start_ms = now_ms - (limit * 3600 * 1000) payload = { "type": "candleSnapshot", "req": { "coin": coin, "interval": interval, "startTime": start_ms, "endTime": now_ms, } } try: r = requests.post(MAINNET_API, json=payload, timeout=15) data = r.json() if isinstance(data, list): # Sort oldest first data.sort(key=lambda x: x["t"]) return data return [] except Exception as e: print(f" Error fetching candles for {coin}: {e}") return [] def simulate_strategy_on_candles( key: str, candles: list[dict], coin_name: str = "BTC", allocation: float = 100.0, fee_tier: int = 0, staking_tier: str = "none", ) -> dict: """Run a strategy against real candle data.""" cfg = STRATEGIES[key] name = cfg["name"] fee_model = cfg["fee_model"] fee_rate = get_perp_fees(fee_tier, staking_tier, fee_model) sz = cfg["size"] eq = allocation curve = [] trades = [] position = 0.0 entry_price = 0.0 total_fees = 0.0 wins = 0 prices_20 = deque(maxlen=20) eth_prices_20 = deque(maxlen=20) # for pairs for i, candle in enumerate(candles): close = float(candle["c"]) high = float(candle["h"]) low = float(candle["l"]) t = datetime.fromtimestamp(candle["t"] / 1000) prices_20.append(close) signal = None reason = "" signal_strength = 0.0 # ── Strategy-specific signal generation ── if key == "ofi" and len(prices_20) >= 20: # Order Book Imbalance proxy: price momentum + volume confirmation vol = float(candle.get("v", 0)) ret_5 = (close - prices_20[-5]) / prices_20[-5] if len(prices_20) >= 5 else 0 ret_20 = (close - prices_20[0]) / prices_20[0] avg_vol = sum(float(c.get("v", 0)) for c in candles[max(0,i-20):i+1]) / min(i+1, 20) if ret_5 > 0.001 and vol > avg_vol * 1.2: signal = "SELL" # momentum up, sell into strength reason = f"OFI: +{ret_5*100:.2f}% 5-period, vol {vol/avg_vol:.1f}x avg" signal_strength = abs(ret_5) * 100 elif ret_5 < -0.001 and vol > avg_vol * 1.2: signal = "BUY" reason = f"OFI: {ret_5*100:.2f}% 5-period, vol {vol/avg_vol:.1f}x avg" signal_strength = abs(ret_5) * 100 elif key == "iceberg" and len(prices_20) >= 10: # Iceberg: consecutive directional moves up_count = sum(1 for j in range(-9, 0) if prices_20[j+1] > prices_20[j]) if up_count >= 7: signal = "BUY" reason = f"Iceberg: {up_count}/10 upward ticks" signal_strength = up_count / 10 elif up_count <= 3: signal = "SELL" reason = f"Iceberg: {up_count}/10 upward ticks" signal_strength = 1 - up_count / 10 elif key == "funding_arb" and len(prices_20) >= 20: # Funding Rate Arb: hourly price trend as funding proxy long_return = (close - prices_20[0]) / prices_20[0] annual_rate = long_return * 365 * 24 # hourly to annual if abs(annual_rate) > 0.03: # >3% annualized signal = "SELL" if annual_rate > 0 else "BUY" reason = f"Fund: {annual_rate*100:.1f}% APR ({long_return*100:.2f}% 1h)" signal_strength = min(1.0, abs(annual_rate) * 5) elif key == "pairs" and len(prices_20) >= 20: # Pairs: BTC/ETH ratio Z-score (only works if we have both) # For single-coin backtest, use high/low range as proxy ranges = [float(c["h"]) - float(c["l"]) for c in candles[max(0,i-20):i+1]] avg_range = sum(ranges) / len(ranges) if ranges else 0 current_range = high - low if avg_range > 0: z = (current_range - avg_range) / (avg_range * 0.5) if avg_range > 0 else 0 if z > 1.5: signal = "SELL" reason = f"Pairs: range Z={z:.1f} (wide range → mean reversion sell)" signal_strength = z elif z < -1.5: signal = "BUY" reason = f"Pairs: range Z={z:.1f} (tight range → expansion buy)" signal_strength = abs(z) elif key == "avellaneda" and len(prices_20) >= 20: # A-S: volatility-based quoting — simulates spread capture returns_20 = [(prices_20[j] - prices_20[j-1]) / prices_20[j-1] for j in range(1, len(prices_20))] vol = (sum(r*r for r in returns_20) / len(returns_20)) ** 0.5 if returns_20 else 0 annual_vol = vol * (365 * 24) ** 0.5 # Fill probability based on volatility regime fill_prob = 0.25 if annual_vol < 0.15 else (0.08 if annual_vol > 0.60 else 0.15) if random.random() < fill_prob: spread = close * vol # proxy spread signal = "BUY" if position <= 0 else "SELL" reason = f"A-S: vol={annual_vol:.1%}, fill_prob={fill_prob:.0%}, regime={'LOW' if annual_vol<0.15 else 'HIGH' if annual_vol>0.6 else 'NORMAL'}" signal_strength = fill_prob elif key == "momentum" and len(prices_20) >= 20: # Bollinger breakout avg = sum(prices_20) / len(prices_20) var = sum((p - avg)**2 for p in prices_20) / len(prices_20) std = var ** 0.5 if std > 0 and close > avg + 2*std: signal = "BUY" reason = f"Bollinger: {close:.0f} > {avg+2*std:.0f} (2σ breakout)" signal_strength = (close - avg - 2*std) / std elif std > 0 and close < avg - 2*std: signal = "SELL" reason = f"Bollinger: {close:.0f} < {avg-2*std:.0f} (2σ breakdown)" signal_strength = (avg - 2*std - close) / std elif key == "mean_rev" and len(prices_20) >= 20: # VWAP mean reversion weights = [1 + j/len(prices_20) for j in range(len(prices_20))] vwap = sum(p*w for p, w in zip(prices_20, weights)) / sum(weights) std = (sum((p - vwap)**2 for p in prices_20) / len(prices_20)) ** 0.5 dev = (close - vwap) / std if std > 0 else 0 if dev > 1.5: signal = "SELL" reason = f"VWAP: dev={dev:.1f}σ above VWAP ${vwap:.0f}" signal_strength = dev elif dev < -1.5: signal = "BUY" reason = f"VWAP: dev={dev:.1f}σ below VWAP ${vwap:.0f}" signal_strength = abs(dev) elif key == "kalman_pairs" and len(prices_20) >= 20: if "_kalman_trader" not in dir(): import sys as _sys _sys.path.insert(0, ".") from strategies.kalman_pairs import KalmanPairsTrader globals()["_kalman_trader"] = KalmanPairsTrader( transition_covariance=1e-4, observation_covariance=1e-2, z_entry=2.0, z_exit=0.5, warmup_bars=20, ) result = globals()["_kalman_trader"].step(close, close * 0.05 + (high - low) * 10) if result["signal"] != 0: signal = "BUY" if result["signal"] > 0 else "SELL" reason = f"K-pairs z={result['z_score']:.2f}" signal_strength = abs(result["z_score"]) / 4.0 # ── Execute signal ── if signal and signal_strength > 0.15: # minimum strength filter notional = sz * close fee = notional * fee_rate * 2 # entry + exit total_fees += fee if signal.startswith("BUY"): if position < 0: # Close short close_pnl = abs(position) * (entry_price - close) eq += close_pnl if close_pnl > 0: wins += 1 trades.append({ "time": t.strftime("%Y-%m-%d %H:%M"), "side": "BUY (close short)", "size": abs(position), "price": close, "pnl_gross": round(close_pnl, 4), "pnl_net": round(close_pnl - fee, 4), "fee": round(fee, 6), "reason": reason, }) position = 0 if position == 0: entry_price = close position = sz eq -= fee else: entry_price = (entry_price * position + close * sz) / (position + sz) position += sz eq -= fee else: # SELL if position > 0: close_pnl = position * (close - entry_price) eq += close_pnl if close_pnl > 0: wins += 1 trades.append({ "time": t.strftime("%Y-%m-%d %H:%M"), "side": "SELL (close long)", "size": position, "price": close, "pnl_gross": round(close_pnl, 4), "pnl_net": round(close_pnl - fee, 4), "fee": round(fee, 6), "reason": reason, }) position = 0 if position == 0: entry_price = close position = -sz eq -= fee else: entry_price = (entry_price * abs(position) + close * sz) / (abs(position) + sz) position -= sz eq -= fee # Track equity curve unrealized = position * (close - entry_price) if position != 0 else 0 curve.append({"t": t.isoformat(), "v": round(eq + unrealized, 4)}) # Close any open position at last price if position != 0 and candles: last_close = float(candles[-1]["c"]) close_pnl = abs(position) * (last_close - entry_price) * (1 if position > 0 else -1) eq += close_pnl if close_pnl > 0: wins += 1 # ── Compute metrics ── pnl_net = eq - allocation pnl_gross = pnl_net + total_fees returns = [] for i in range(1, len(curve)): if curve[i-1]["v"] > 0: returns.append((curve[i]["v"] - curve[i-1]["v"]) / curve[i-1]["v"]) padded_equity = [allocation] * 10 + [c["v"] for c in curve] return { "strategy": name, "strategy_key": key, "coin": coin_name, # actual ticker (BTC, ETH, etc.) "allocation": allocation, "start_time": curve[0]["t"] if curve else "", "end_time": curve[-1]["t"] if curve else "", "start_equity": allocation, "end_equity": round(eq, 4), "pnl": round(pnl_net, 4), "pnl_pct": round(pnl_net / allocation * 100, 2), "pnl_gross": round(pnl_gross, 4), "pnl_gross_pct": round(pnl_gross / allocation * 100, 2), "fees_total": round(total_fees, 4), "fee_tier": fee_tier, "staking_tier": staking_tier, "fee_model": fee_model, "sharpe": round(sharpe(returns) if returns else 0, 4), "sortino": round(sortino(returns) if returns else 0, 4), "max_dd": round(max_drawdown(padded_equity), 4), "max_dd_pct": round(max_drawdown(padded_equity) * 100, 2), "win_rate": round(win_rate(trades), 4), "total_trades": len(trades), "equity_curve": curve, "trades": trades[-100:], "num_periods": len(candles), "data_source": "Hyperliquid Mainnet", "generated_at": datetime.now().isoformat(), } def main(): p = argparse.ArgumentParser(description="FTDT Historical Backtest Runner") p.add_argument("--coin", default="BTC", choices=["BTC", "ETH", "SOL", "HYPE", "VVV"], help="Coin to backtest") p.add_argument("--strategy", "-s", choices=list(STRATEGIES) + ["all"], default="all") p.add_argument("--fee-tier", type=int, default=0, choices=range(7)) p.add_argument("--staking-tier", default="none", choices=list(STAKING_TIERS.keys())) p.add_argument("--hours", type=int, default=720, help="Hours of history (default: 720 = 30 days)") a = p.parse_args() ft_info = PERPS_TIERS[a.fee_tier] st_info = STAKING_TIERS[a.staking_tier] print("=" * 60) print(f" FTDT Quant Lab — HISTORICAL Backtest Runner") print(f" Coin: {a.coin} | Period: {a.hours}h ({a.hours//24} days)") print(f" Fee Tier: {ft_info['name']} | Staking: {st_info['name']}") print("=" * 60) # Fetch real candles print(f"\n Fetching {a.coin} candles from Hyperliquid mainnet...") candles = fetch_candles(a.coin, interval="1h", limit=a.hours) if not candles: print(" ERROR: No candle data returned. Check API connectivity.") sys.exit(1) print(f" Got {len(candles)} candles: " f"{datetime.fromtimestamp(candles[0]['t']/1000).strftime('%Y-%m-%d')} → " f"{datetime.fromtimestamp(candles[-1]['t']/1000).strftime('%Y-%m-%d')}") print(f" Price range: ${float(candles[0]['c']):.0f} → ${float(candles[-1]['c']):.0f}") keys = list(STRATEGIES) if a.strategy == "all" else [a.strategy] for key in keys: cfg = STRATEGIES[key] print(f"\n Running: {cfg['name']} on {a.coin}...") result = simulate_strategy_on_candles( key, candles, a.coin, fee_tier=a.fee_tier, staking_tier=a.staking_tier, ) # Save ts = datetime.now().strftime("%Y%m%d-%H%M%S") filename = f"{key}_{a.coin}_{ts}.json" filepath = RESULTS_DIR / filename with open(filepath, "w") as f: json.dump(result, f, indent=2, default=str) print(f" Saved: {filepath}") print(f" Net PnL: {result['pnl_pct']:+.2f}% | " f"Gross: {result['pnl_gross_pct']:+.2f}% | " f"Fees: ${result['fees_total']:.2f} | " f"Sharpe: {result['sharpe']:.2f} | " f"Trades: {result['total_trades']} | " f"Win: {result['win_rate']:.0%}") print("\n" + "=" * 60) print(f" Results in backtests/results/historical/") print(f" View at: https://ftdt.io/cv") print("=" * 60) if __name__ == "__main__": main()