research: rebuild certified BTC/ETH feed + honest backtest harness
- rebuilt BTC/ETH from Deribit mainnet (certified 1.7-1.9bps vs Coinbase) - archived contaminated alt data to Old/data/raw - add src/backtest/harness.py: leakage-free, fee-aware signal engine (entry at close[i], intrabar TP/SL, CAGR/Sharpe/DD/per-year/OOS)
This commit is contained in:
@@ -0,0 +1,249 @@
|
||||
"""HONEST BACKTEST HARNESS — universo certificato BTC/ETH (Deribit mainnet).
|
||||
|
||||
Foundation per la ricerca post-reset (2026-06-19). Tutte le strategie nuove devono
|
||||
usare QUESTO harness per garantire:
|
||||
1. NESSUN look-ahead: la direzione e il prezzo d'ingresso si decidono con dati fino
|
||||
a close[i] incluso, e si ENTRA a close[i] (la barra successiva, i+1, e' la prima
|
||||
in cui si e' realmente in posizione). L'exit intrabar guarda high/low di i+1..
|
||||
2. Fee realistiche Deribit: 0.10% round-trip (taker) di default.
|
||||
3. Metriche oneste: equity compounding, CAGR, Sharpe (da rendimenti per-barra),
|
||||
max drawdown, per-anno, e split OOS.
|
||||
|
||||
Convenzione segnali (entry-eseguibile):
|
||||
Una strategia produce, per ogni indice i, un dict opzionale:
|
||||
{'dir': +1/-1, 'tp': prezzo|None, 'sl': prezzo|None, 'max_bars': int|None}
|
||||
decidendo SOLO con dati [.. i] (close[i] incluso). L'engine apre a close[i] e
|
||||
gestisce l'uscita dalle barre i+1 in poi (TP/SL intrabar al livello, SL prioritario;
|
||||
altrimenti max_bars al close).
|
||||
|
||||
Uso tipico:
|
||||
from src.backtest.harness import load, backtest_signals, Metrics
|
||||
df = load("BTC", "1h")
|
||||
entries = my_signal_fn(df) # list[dict|None] lunga len(df)
|
||||
m = backtest_signals(df, entries, fee_rt=0.001, leverage=1.0)
|
||||
m.print_summary("MYSTRAT BTC 1h")
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.data.downloader import load_data
|
||||
|
||||
CERTIFIED = {"BTC", "ETH"}
|
||||
|
||||
|
||||
def load(asset: str, tf: str) -> pd.DataFrame:
|
||||
"""Carica un feed certificato. Solleva su asset non certificato (guardrail fisico)."""
|
||||
if asset.upper() not in CERTIFIED:
|
||||
raise ValueError(f"Asset non certificato: {asset}. Universo = {CERTIFIED}.")
|
||||
df = load_data(asset, tf).reset_index(drop=True)
|
||||
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
return df
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Metriche
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass
|
||||
class Metrics:
|
||||
asset: str = ""
|
||||
tf: str = ""
|
||||
n_trades: int = 0
|
||||
wins: int = 0
|
||||
net_return: float = 0.0 # ritorno totale frazionale (final/initial - 1)
|
||||
cagr: float = 0.0
|
||||
sharpe: float = 0.0 # annualizzato dai rendimenti per-barra dell'equity
|
||||
max_dd: float = 0.0 # frazione (0.10 = 10%)
|
||||
time_in_market: float = 0.0 # frazione barre in posizione
|
||||
avg_bars: float = 0.0
|
||||
final_capital: float = 0.0
|
||||
initial_capital: float = 0.0
|
||||
bars_per_year: float = 0.0
|
||||
yearly: dict = field(default_factory=dict) # year -> net return frazionale dell'anno
|
||||
equity: np.ndarray = field(default_factory=lambda: np.array([]))
|
||||
eq_index: pd.DatetimeIndex | None = None
|
||||
|
||||
@property
|
||||
def win_rate(self) -> float:
|
||||
return self.wins / self.n_trades * 100 if self.n_trades else 0.0
|
||||
|
||||
@property
|
||||
def profit_per_day_on(self, capital: float = 2000.0) -> float: # placeholder
|
||||
return 0.0
|
||||
|
||||
def daily_profit(self, capital: float = 2000.0) -> float:
|
||||
"""€/giorno medio se partito con `capital` (su tutto lo span, compounding incluso)."""
|
||||
if self.eq_index is None or len(self.equity) < 2:
|
||||
return 0.0
|
||||
idx = self.eq_index
|
||||
days = (idx.iloc[-1] - idx.iloc[0]).total_seconds() / 86400 if hasattr(idx, "iloc") \
|
||||
else (idx[-1] - idx[0]).total_seconds() / 86400
|
||||
if days <= 0:
|
||||
return 0.0
|
||||
final = capital * (self.final_capital / self.initial_capital)
|
||||
return (final - capital) / days
|
||||
|
||||
def print_summary(self, label: str = ""):
|
||||
print(f" {label:<26s} trades={self.n_trades:>5d} wr={self.win_rate:>4.1f}% "
|
||||
f"ret={self.net_return*100:>+8.0f}% CAGR={self.cagr*100:>+6.1f}% "
|
||||
f"Sharpe={self.sharpe:>5.2f} DD={self.max_dd*100:>4.1f}% "
|
||||
f"mkt={self.time_in_market*100:>4.0f}% €/d(2k)={self.daily_profit(2000):>+6.2f}")
|
||||
|
||||
def print_yearly(self):
|
||||
for y in sorted(self.yearly):
|
||||
print(f" {y}: {self.yearly[y]*100:>+7.1f}%")
|
||||
|
||||
|
||||
def _sharpe(equity: np.ndarray, bars_per_year: float) -> float:
|
||||
if len(equity) < 3:
|
||||
return 0.0
|
||||
r = np.diff(equity) / equity[:-1]
|
||||
r = r[np.isfinite(r)]
|
||||
if len(r) == 0 or np.std(r) == 0:
|
||||
return 0.0
|
||||
return float(np.mean(r) / np.std(r) * np.sqrt(bars_per_year))
|
||||
|
||||
|
||||
def _max_dd(equity: np.ndarray) -> float:
|
||||
peak = np.maximum.accumulate(equity)
|
||||
dd = (peak - equity) / peak
|
||||
return float(np.max(dd)) if len(dd) else 0.0
|
||||
|
||||
|
||||
def backtest_signals(
|
||||
df: pd.DataFrame,
|
||||
entries: list,
|
||||
fee_rt: float = 0.001,
|
||||
leverage: float = 1.0,
|
||||
position_size: float = 1.0,
|
||||
initial_capital: float = 1000.0,
|
||||
allow_overlap: bool = False,
|
||||
asset: str = "",
|
||||
tf: str = "",
|
||||
) -> Metrics:
|
||||
"""Esegue il backtest su una lista di entry-dict (uno per barra, None = niente segnale).
|
||||
|
||||
entry dict: {'dir': +1/-1, 'tp': float|None, 'sl': float|None, 'max_bars': int|None}
|
||||
- apertura a close[i] (decisa con dati <= i)
|
||||
- exit dalle barre i+1.. : TP/SL toccati intrabar (al livello, SL prioritario),
|
||||
altrimenti chiusura al close dopo max_bars (default 24 se assente).
|
||||
- non si apre una nuova posizione finche' la precedente non e' chiusa (allow_overlap=False).
|
||||
- PnL compounding: ogni trade muove capital di position_size * leverage * (ret_netto).
|
||||
"""
|
||||
c = df["close"].values.astype(float)
|
||||
h = df["high"].values.astype(float)
|
||||
l = df["low"].values.astype(float)
|
||||
n = len(c)
|
||||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
|
||||
capital = float(initial_capital)
|
||||
equity = np.full(n, capital, dtype=float)
|
||||
yearly: dict[int, float] = {}
|
||||
yearly_start: dict[int, float] = {}
|
||||
|
||||
n_trades = wins = 0
|
||||
bars_in_market = 0
|
||||
bars_sum = 0
|
||||
i = 0
|
||||
busy_until = -1
|
||||
|
||||
for i in range(n):
|
||||
e = entries[i] if i < len(entries) else None
|
||||
if e is None or e.get("dir", 0) == 0:
|
||||
equity[i] = capital
|
||||
continue
|
||||
if not allow_overlap and i <= busy_until:
|
||||
equity[i] = capital
|
||||
continue
|
||||
|
||||
direction = int(e["dir"])
|
||||
entry = c[i]
|
||||
tp = e.get("tp")
|
||||
sl = e.get("sl")
|
||||
max_bars = int(e.get("max_bars") or 24)
|
||||
|
||||
exit_price = c[min(i + max_bars, n - 1)]
|
||||
exit_idx = min(i + max_bars, n - 1)
|
||||
for j in range(i + 1, min(i + max_bars + 1, n)):
|
||||
hit_sl = sl is not None and (
|
||||
(direction == 1 and l[j] <= sl) or (direction == -1 and h[j] >= sl))
|
||||
hit_tp = tp is not None and (
|
||||
(direction == 1 and h[j] >= tp) or (direction == -1 and l[j] <= tp))
|
||||
if hit_sl:
|
||||
exit_price = sl
|
||||
exit_idx = j
|
||||
break
|
||||
if hit_tp:
|
||||
exit_price = tp
|
||||
exit_idx = j
|
||||
break
|
||||
exit_price = c[j]
|
||||
exit_idx = j
|
||||
|
||||
gross = (exit_price - entry) / entry * direction
|
||||
net = gross * leverage - fee_rt * leverage
|
||||
capital += capital * position_size * net
|
||||
capital = max(capital, 1.0)
|
||||
|
||||
year = ts.iloc[i].year
|
||||
if year not in yearly_start:
|
||||
yearly_start[year] = capital / (1 + position_size * net) if (1 + position_size * net) else capital
|
||||
n_trades += 1
|
||||
if gross > 0:
|
||||
wins += 1
|
||||
bars = exit_idx - i
|
||||
bars_in_market += bars
|
||||
bars_sum += bars
|
||||
busy_until = exit_idx
|
||||
|
||||
# propaga equity fino a exit_idx (mark a fine trade, semplice ma onesto a livello trade)
|
||||
equity[i:exit_idx + 1] = capital
|
||||
|
||||
# riempi i buchi finali
|
||||
for k in range(1, n):
|
||||
if equity[k] == initial_capital and equity[k - 1] != initial_capital:
|
||||
equity[k] = equity[k - 1]
|
||||
# forward fill robusto
|
||||
last = initial_capital
|
||||
for k in range(n):
|
||||
if equity[k] != last and equity[k] != initial_capital:
|
||||
last = equity[k]
|
||||
else:
|
||||
equity[k] = last
|
||||
|
||||
# per-anno dal vettore equity
|
||||
eq_s = pd.Series(equity, index=ts)
|
||||
yearly_ret = {}
|
||||
for y, grp in eq_s.groupby(eq_s.index.year):
|
||||
if len(grp) > 1 and grp.iloc[0] > 0:
|
||||
yearly_ret[int(y)] = float(grp.iloc[-1] / grp.iloc[0] - 1)
|
||||
|
||||
span_days = (ts.iloc[-1] - ts.iloc[0]).total_seconds() / 86400
|
||||
years = span_days / 365.25 if span_days > 0 else 1.0
|
||||
bars_per_year = n / years if years > 0 else n
|
||||
cagr = (capital / initial_capital) ** (1 / years) - 1 if years > 0 and capital > 0 else -1.0
|
||||
|
||||
return Metrics(
|
||||
asset=asset, tf=tf,
|
||||
n_trades=n_trades, wins=wins,
|
||||
net_return=capital / initial_capital - 1,
|
||||
cagr=cagr,
|
||||
sharpe=_sharpe(equity, bars_per_year),
|
||||
max_dd=_max_dd(equity),
|
||||
time_in_market=bars_in_market / n if n else 0.0,
|
||||
avg_bars=bars_sum / n_trades if n_trades else 0.0,
|
||||
final_capital=capital,
|
||||
initial_capital=initial_capital,
|
||||
bars_per_year=bars_per_year,
|
||||
yearly=yearly_ret,
|
||||
equity=equity,
|
||||
eq_index=ts,
|
||||
)
|
||||
|
||||
|
||||
def oos_split(df: pd.DataFrame, frac: float = 0.65):
|
||||
"""Indice di taglio IS/OOS (default 65% in-sample)."""
|
||||
return int(len(df) * frac)
|
||||
Reference in New Issue
Block a user