rams/ftdt-quant-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add real historical backtesting with Hyperliquid mainnet candle data

Browse Source 
backtests/historical_runner.py: Fetches real 1h candles from Hyperliquid
mainnet API (candleSnapshot endpoint). Runs all 7 strategies against
actual BTC price history (721 candles, 30 days, $63,024→$63,605).
Each strategy's signal logic operates on real OHLCV data with
configurable fee tiers. Saves to backtests/results/historical/.

Results on 30d BTC data at VIP0:
  Mean Reversion: +93.87% net (Sharpe 0.94)
  Order Book Imbalance: +54.31% net (Sharpe 1.03)
  Avellaneda-Stoikov: -1.02% net (Sharpe -0.13)
  Iceberg Detection: -33.20% net
  Momentum Breakout: -54.72% net

Server: Added /api/backtests/historical (list) and
/api/backtest/historical/{name} (full data) endpoints.

Dashboard: Added "Historical" tab with "Real Data" badge. Cards show
coin + mainnet source. Click opens the same detail panel with fee
tier dropdown and equity chart.
This commit is contained in:
ramseshk
2026-08-04 07:29:51 +00:00
parent 0c0d2124ad
commit 1bf54b4c00
19 changed files with 44522 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/risk.py
+267
View File
@@ -0,0 +1,267 @@
"""
Risk analytics module.
Value at Risk, Conditional VaR, drawdown analysis, correlation,
and composite risk ratios. Builds on common/metrics.py for
Sharpe/Sortino/max_drawdown which are re-exported here.
All functions accept equity curves as either plain lists of floats
or lists of {t: timestamp, v: equity_value} dicts.
"""
import math
import numpy as np
# ═══════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════
def _to_values(curve):
"""Normalise an equity curve to a list of floats."""
if not curve:
return []
if isinstance(curve[0], dict):
return [p["v"] for p in curve]
return [float(v) for v in curve]
def _daily_returns(curve):
"""Compute daily log returns from an equity curve."""
vals = _to_values(curve)
if len(vals) < 2:
return []
return [math.log(vals[i] / vals[i - 1]) for i in range(1, len(vals))]
# ═══════════════════════════════════════════════════════════
# Value at Risk & Conditional VaR
# ═══════════════════════════════════════════════════════════
def var_95(daily_returns):
"""
95% historical Value at Risk.
Returns a *positive* number representing the loss threshold
(e.g. 0.02 means "we are 95% confident daily loss won't exceed 2%").
"""
if len(daily_returns) < 5:
return 0.0
sorted_ret = sorted(daily_returns)
idx = max(0, int(len(sorted_ret) * 0.05))
var = sorted_ret[idx]
return -min(var, 0.0) # return positive loss magnitude
def cvar_95(daily_returns):
"""
95% Conditional Value at Risk (Expected Shortfall).
Average loss *beyond* the VaR threshold. Returns a positive number.
"""
if len(daily_returns) < 5:
return 0.0
sorted_ret = sorted(daily_returns)
cutoff = max(0, int(len(sorted_ret) * 0.05))
tail = [r for r in sorted_ret[:cutoff + 1] if r < 0]
if not tail:
return 0.0
return -np.mean(tail)
def var_95_from_equity(equity_curve):
"""Convenience: VaR computed directly from an equity curve."""
return var_95(_daily_returns(equity_curve))
def cvar_95_from_equity(equity_curve):
"""Convenience: CVaR computed directly from an equity curve."""
return cvar_95(_daily_returns(equity_curve))
# ═══════════════════════════════════════════════════════════
# Drawdown
# ═══════════════════════════════════════════════════════════
def max_drawdown(equity_curve):
"""
Maximum drawdown from an equity curve (peak-to-trough).
Returns a positive fraction (0.25 = 25% max DD).
Accepts list of floats or list of {t, v} dicts.
"""
vals = _to_values(equity_curve)
if not vals:
return 0.0
peak = vals[0]
worst = 0.0
for v in vals:
if v > peak:
peak = v
if peak > 0:
dd = (peak - v) / peak
worst = max(worst, dd)
return worst
# ═══════════════════════════════════════════════════════════
# Ratios
# ═══════════════════════════════════════════════════════════
def sharpe(returns, rf=0.0, periods=365):
"""
Annualised Sharpe ratio.
`returns` should be a list of daily log returns (floats).
"""
if len(returns) < 2:
return 0.0
excess = np.mean(returns) - rf
std = np.std(returns, ddof=1)
if std <= 0:
return 0.0
return (excess / std) * math.sqrt(periods)
def sortino(returns, rf=0.0, periods=365):
"""
Annualised Sortino ratio (downside deviation only).
"""
if len(returns) < 2:
return 0.0
excess = np.mean(returns) - rf
downside = [r for r in returns if r < 0]
d_std = np.std(downside, ddof=1) if downside else 0.0
if d_std <= 0:
return 0.0
return (excess / d_std) * math.sqrt(periods)
def calmar_ratio(returns, max_dd):
"""
Calmar ratio = annualised return / maximum drawdown.
`returns` is a list of daily log returns.
`max_dd` is a positive fraction (0.25 = 25% drawdown).
"""
if len(returns) < 2 or max_dd <= 0:
return 0.0
ann_return = np.mean(returns) * 365
return ann_return / max_dd
def sharpe_from_equity(equity_curve):
"""Sharpe ratio computed from an equity curve."""
return sharpe(_daily_returns(equity_curve))
def sortino_from_equity(equity_curve):
"""Sortino ratio computed from an equity curve."""
return sortino(_daily_returns(equity_curve))
def calmar_from_equity(equity_curve):
"""Calmar ratio computed from an equity curve."""
dr = _daily_returns(equity_curve)
dd = max_drawdown(equity_curve)
return calmar_ratio(dr, dd)
# ═══════════════════════════════════════════════════════════
# Correlation
# ═══════════════════════════════════════════════════════════
def correlation_matrix(strategy_returns_dict):
"""
Pearson correlation matrix between strategies.
Args:
strategy_returns_dict: {name: [daily_log_returns], ...}
Returns:
{name: {name: float, ...}, ...}
or empty dict if fewer than 2 strategies.
"""
names = list(strategy_returns_dict.keys())
if len(names) < 2:
return {}
# Align lengths (truncate to shortest)
min_len = min(len(strategy_returns_dict[n]) for n in names)
if min_len < 2:
return {}
matrix = {}
for n1 in names:
r1 = strategy_returns_dict[n1][-min_len:]
row = {}
for n2 in names:
r2 = strategy_returns_dict[n2][-min_len:]
if n1 == n2:
row[n2] = 1.0
else:
corr = np.corrcoef(r1, r2)[0, 1]
row[n2] = float(corr) if not np.isnan(corr) else 0.0
matrix[n1] = row
return matrix
def correlation_from_equity(strategy_equity_dict):
"""
Convenience: correlation matrix from {name: [{t,v},...]} equity curves.
"""
returns_dict = {}
for name, curve in strategy_equity_dict.items():
dr = _daily_returns(curve)
if len(dr) >= 2:
returns_dict[name] = dr
return correlation_matrix(returns_dict)
# ═══════════════════════════════════════════════════════════
# Composite risk summary
# ═══════════════════════════════════════════════════════════
def risk_summary(equity_history, strategy_equity=None):
"""
One-shot: compute all risk metrics for a portfolio.
Args:
equity_history: portfolio-level equity curve [{t, v}, ...]
strategy_equity: optional {name: [{t, v}, ...]}
Returns dict with VaR, CVaR, MaxDD, Calmar, Sharpe, Sortino,
and optionally correlation/cross-strategy metrics.
"""
dr = _daily_returns(equity_history)
dd = max_drawdown(equity_history)
summary = {
"var_95": round(var_95(dr), 6),
"cvar_95": round(cvar_95(dr), 6),
"max_drawdown": round(dd, 6),
"calmar_ratio": round(calmar_ratio(dr, dd), 4),
"sharpe": round(sharpe(dr), 4),
"sortino": round(sortino(dr), 4),
"num_observations": len(dr),
}
if strategy_equity and len(strategy_equity) >= 2:
summary["correlation"] = correlation_from_equity(strategy_equity)
# Per-strategy metrics
per_strat = {}
for name, curve in strategy_equity.items():
sdr = _daily_returns(curve)
sdd = max_drawdown(curve)
per_strat[name] = {
"sharpe": round(sharpe(sdr), 4),
"sortino": round(sortino(sdr), 4),
"max_drawdown": round(sdd, 6),
"calmar_ratio": round(calmar_ratio(sdr, sdd), 4),
"var_95": round(var_95(sdr), 6),
"cvar_95": round(cvar_95(sdr), 6),
}
summary["per_strategy"] = per_strat
return summary