"""Harness ONESTO condiviso per esplorare nuove famiglie di strategie. Regole NON negoziabili (per non ripetere l'errore squeeze look-ahead): - direzione e prezzo decisi con dati FINO a close[i] incluso, mai con la barra i usata per scegliere la direzione e poi entrare a i-1; - ingresso ESEGUIBILE a close[i]; - exit: take-profit / stop-loss intrabar (high/low) e/o time-limit max_bars; tp/sl possono essere None -> exit solo a tempo (utile per stagionalita'); - una posizione per volta (non-overlap), capitale composto; - NETTO dopo fee round-trip (default 0.10% RT reale Deribit) e leva; - validazione OOS (held-out, ultimo 30%) + sweep fee 0.00-0.20% RT. Le strategie ad alta frequenza muoiono di fee: ogni entry costa fee_rt*lev sul notional. Tienine conto: meno operazioni e edge > costi. Asset disponibili: ADA BNB BTC DOGE ETH LTC SOL XRP (1h, 15m; BTC/ETH anche 5m). """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from src.data.downloader import load_data FEE_RT = 0.001 # Deribit perp realistico: taker ~0.05%/lato = 0.10% RT LEV = 3.0 POS = 0.15 OOS_FRAC = 0.30 ASSETS = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"] BARS_PER_YEAR = {"5m": 105120, "15m": 35040, "1h": 8760, "4h": 2190, "1d": 365} # --------------------------- dati --------------------------- def get_df(asset: str, tf: str) -> pd.DataFrame: """OHLCV con colonna dt (UTC). tf nativo (5m,15m,1h) o resample da 1h (4h,1d). timestamp resta ms-epoch reale anche dopo il resample (no placeholder).""" if tf in ("5m", "15m", "1h"): df = load_data(asset, tf).reset_index(drop=True) else: base = load_data(asset, "1h").copy() base["dt"] = pd.to_datetime(base["timestamp"], unit="ms", utc=True) base = base.set_index("dt") rule = {"4h": "4h", "1d": "1D"}[tf] agg = base.resample(rule).agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"} ).dropna() epoch = pd.Timestamp("1970-01-01", tz="UTC") # ms-epoch portabile (qualsiasi risoluzione) agg["timestamp"] = ((agg.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64") df = agg.reset_index(drop=True) df["dt"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) return df def _dt(df: pd.DataFrame) -> pd.DatetimeIndex: return pd.to_datetime(df["timestamp"], unit="ms", utc=True) # --------------------------- indicatori --------------------------- def atr(df: pd.DataFrame, n: int = 14) -> np.ndarray: h, l, c = df["high"].values, df["low"].values, df["close"].values pc = np.roll(c, 1); pc[0] = c[0] tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc))) return pd.Series(tr).rolling(n).mean().values def ema(x: np.ndarray, n: int) -> np.ndarray: return pd.Series(x).ewm(span=n, adjust=False).mean().values def rsi(close: np.ndarray, n: int = 14) -> np.ndarray: d = np.diff(close, prepend=close[0]) up = pd.Series(np.where(d > 0, d, 0.0)).ewm(alpha=1/n, adjust=False).mean() dn = pd.Series(np.where(d < 0, -d, 0.0)).ewm(alpha=1/n, adjust=False).mean() rs = up / dn.replace(0, np.nan) return (100 - 100 / (1 + rs)).values # --------------------------- engine --------------------------- def simulate(entries: list[dict], df: pd.DataFrame, fee_rt: float = FEE_RT, lev: float = LEV, pos: float = POS, split: int = -1) -> dict: """entries: dict con i(idx), d(+1/-1), max_bars; tp/sl opzionali (None=solo tempo). split: se >0, conta solo entries con i>=split (finestra OOS).""" h, l, c = df["high"].values, df["low"].values, df["close"].values n = len(c) ts = _dt(df) cap = peak = 1000.0 max_dd = 0.0 fee = fee_rt * lev trades = wins = 0 last_exit = -1 bars_in = 0 yearly: dict[int, float] = {} rets: list[float] = [] for e in entries: i, d = e["i"], e["d"] if i <= last_exit or i + 1 >= n or i < split: continue entry = c[i] tp, sl, mb = e.get("tp"), e.get("sl"), e["max_bars"] exit_p = c[min(i + mb, n - 1)] j = min(i + mb, n - 1) for k in range(1, mb + 1): j = i + k if j >= n: exit_p = c[n - 1]; break hit_sl = sl is not None and ((d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)) hit_tp = tp is not None and ((d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)) if hit_sl: exit_p = sl; break if hit_tp: exit_p = tp; break if k == mb: exit_p = c[j] ret = (exit_p - entry) / entry * d * lev - fee cb = cap cap = max(cb + cb * pos * ret, 10.0) peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak) trades += 1; wins += ret > 0; bars_in += (j - i) last_exit = j rets.append(ret * pos) yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100 sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0 return { "trades": trades, "win": wins / trades * 100 if trades else 0.0, "ret": (cap / 1000 - 1) * 100, "dd": max_dd * 100, "sharpe": sharpe, "yearly": yearly, "exposure": bars_in / n * 100 if n else 0.0, } def evaluate(name: str, entries: list[dict], df: pd.DataFrame, fees=(0.0, 0.0005, 0.001, 0.002)) -> dict: """Valuta una lista di entries: FULL, OOS e sweep fee. Stampa una riga sintetica.""" split = int(len(df) * (1 - OOS_FRAC)) full = simulate(entries, df) oos = simulate(entries, df, split=split) sweep = {f: simulate(entries, df, fee_rt=f)["ret"] for f in fees} sweep_oos = {f: simulate(entries, df, fee_rt=f, split=split)["ret"] for f in fees} yrs = full["yearly"]; pos_yrs = sum(1 for v in yrs.values() if v > 0) print(f" {name:<24s} trd={full['trades']:>5d} win={full['win']:>4.1f}% " f"FULL={full['ret']:>+7.0f}% OOS={oos['ret']:>+7.0f}% DD={full['dd']:>4.0f}% " f"oDD={oos['dd']:>4.0f}% Shrp={full['sharpe']:>4.2f} exp={full['exposure']:>4.1f}% " f"anniPos={pos_yrs}/{len(yrs)} | fee0.2%: FULL={sweep[0.002]:>+6.0f} OOS={sweep_oos[0.002]:>+6.0f}") return {"full": full, "oos": oos, "sweep": sweep, "sweep_oos": sweep_oos, "pos_yrs": pos_yrs, "n_yrs": len(yrs)} def robust(res: dict) -> bool: """Verdetto onesto: positivo FULL e OOS, regge a fee 0.20% RT, quasi tutti gli anni positivi.""" return (res["full"]["ret"] > 0 and res["oos"]["ret"] > 0 and res["sweep"][0.002] > 0 and res["sweep_oos"][0.002] > 0 and res["pos_yrs"] >= max(res["n_yrs"] - 1, 1)) if __name__ == "__main__": # smoke test: una stagionalita' banale (hour-of-day) su BTC 1h df = get_df("BTC", "1h"); ts = _dt(df) ents = [{"i": i, "d": 1, "max_bars": 6, "tp": None, "sl": None} for i in range(len(df) - 7) if ts.iloc[i].hour == 0] print("smoke test — BTC long ad ogni 00:00 UTC, hold 6h:") evaluate("seasonality_h0", ents, df)