"""SKYHOOK (SKH01) — dual-timeframe regime+breakout system, ported to BTC/ETH (2026-06-23). NON e' un trend-follower: entra SOLO quando coincidono (a) un REGIME di volatilita'/volume e (b) un PATTERN di breakout/momentum. Porting onesto su BTC/ETH certificati (Deribit mainnet) di un sistema ES (E-mini S&P) genetico a doppio timeframe. Architettura (dal brief): * data2 = HTF 690 min (genera il SEGNALE: regime + pattern) * data1 = LTF 230 min (ESEGUE: ingressi/uscite) NB 690 = 3 x 230 (HTF = 3x LTF) Entrambi resampled dal feed 5m certificato con origin='epoch' -> i confini 690 sono un SOTTOINSIEME dei confini 230, quindi una barra HTF chiude esattamente su una chiusura LTF. Pipeline per barra (evaluate_bar): barre -> indicatori -> fasce regime -> pattern -> composer -> ingresso/uscita -> SkyhookDecision 1. INDICATORI (sul HTF, tipo-Chande, normalizzati 0-100): BuzVola = chande01(ATR) -> dove sei nel CICLO di volatilita' (flat -> 50) BuzVolume= chande01(volume) -> dove sei nel CICLO di volume (rampa -> 100) Ancore della demo del brief (trend lineare): ATR costante -> BuzVola=50 (neutro); volume in rampa -> BuzVolume=100. Entrambe RICOSTRUITE esattamente da chande01. 2. FASCE REGIME (Vola, Volume): trade ammesso solo se BuzVola in [vola_lo,vola_hi] E BuzVolume in [vol_lo,vol_hi]. (Le "fasce 4/3/2 - 4/2/2" del sistema originale sono ricostruite come bande-soglia tunabili: i magici interi non sono nel brief.) 3. PATTERN (breakout su data2/HTF): Donchian leak-free a `ptn_n` barre (default 13, da 13/13/1). ptn_long = close_htf rompe il massimo delle ptn_n barre PRECEDENTI ptn_short = close_htf rompe il minimo delle ptn_n barre PRECEDENTI 4. COMPOSER: contenitore_long = regime_ok AND ptn_long ; contenitore_short = regime_ok AND ptn_short 5. INGRESSO (max 1 al giorno): se il composer e' attivo -> OPEN_LONG / OPEN_SHORT alla chiusura LTF. (stop-and-reverse: non-overlap nell'engine -> il rovescio entra alla prima barra utile dopo l'uscita se il segnale persiste.) 6. USCITE: time-based ASIMMETRICO (uscitalong=24, uscitashort=18 barre LTF) + hard stop/profit. Lo "stop 2000 / profit 5000" in $ del sistema ES e' tradotto in CRYPTO come multipli di ATR LTF (scale-free): sl = k_sl*ATR, tp = k_tp*ATR (default 2.0/5.0 ~ il rapporto 40:100 pt ES), con modalita' 'pct' alternativa (stop/profit in percentuale). CAUSALITA': ogni feature usa dati <= close della barra (HTF: donchian con shift(1), chande01 rolling causale). Il merge HTF->LTF e' merge_asof BACKWARD sulla CHIUSURA HTF (<= chiusura LTF): una barra HTF e' usata solo quando e' realmente chiusa. backtest_signals apre a close[i]. API: from src.strategies.skyhook import SkyhookParams, build_frames, skyhook_entries ltf, htf = build_frames(load_data("BTC","5m")) # resample 5m -> 230m + 690m entries = skyhook_entries(ltf, htf, SkyhookParams()) # list[dict|None] len(ltf), per backtest_signals from src.backtest.harness import backtest_signals m = backtest_signals(ltf, entries, fee_rt=0.001); m.print_summary("SKH01 BTC") """ from __future__ import annotations from dataclasses import dataclass, field import numpy as np import pandas as pd # 690 = 3 x 230 ; entrambi multipli esatti di 5m (138 e 46 barre da 5m) HTF_MIN = 690 # data2 — segnale LTF_MIN = 230 # data1 — esecuzione # --------------------------------------------------------------------------- # Resample dal feed 5m certificato (origin='epoch' -> confini deterministici e allineati) # --------------------------------------------------------------------------- def resample_5m(df5: pd.DataFrame, minutes: int) -> pd.DataFrame: """5m -> `minutes` barre (origin epoch). Schema con 'datetime' + 'timestamp' (open-labeled).""" g = df5[["timestamp", "open", "high", "low", "close", "volume"]].copy() g.index = pd.to_datetime(g["timestamp"], unit="ms", utc=True) out = (g.resample(f"{minutes}min", label="left", closed="left", origin="epoch") .agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) .dropna(subset=["open"])) out["datetime"] = out.index epoch = pd.Timestamp("1970-01-01", tz="UTC") out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64") return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]] def build_frames(df5: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: """Da un feed 5m certificato -> (ltf 230m exec, htf 690m signal).""" return resample_5m(df5, LTF_MIN), resample_5m(df5, HTF_MIN) # --------------------------------------------------------------------------- # Indicatori causali # --------------------------------------------------------------------------- def atr(df: pd.DataFrame, win: 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).ewm(alpha=1.0 / win, adjust=False).mean().values def chande01(x: np.ndarray, n: int) -> np.ndarray: """Chande Momentum Oscillator su `x`, normalizzato 0-100 (tipo-Chande). CMO = (Su - Sd)/(Su + Sd) in [-1,1] sulle n variazioni; mappato (1+CMO)*50 -> [0,100]. Serie piatta (variazioni nulle) -> 50 (neutro). Causale (rolling fino a i).""" x = np.asarray(x, float) d = np.diff(x, prepend=x[0]) up = np.where(d > 0, d, 0.0) dn = np.where(d < 0, -d, 0.0) su = pd.Series(up).rolling(n, min_periods=n).sum().values sd = pd.Series(dn).rolling(n, min_periods=n).sum().values denom = su + sd cmo = np.divide(su - sd, denom, out=np.zeros_like(denom), where=denom > 0) out = 50.0 * (1.0 + cmo) out[~np.isfinite(out)] = 50.0 return out def donchian_breakout(df: pd.DataFrame, n: int) -> tuple[np.ndarray, np.ndarray]: """Breakout leak-free: close[i] rompe il max/min delle n barre STRETTAMENTE precedenti.""" hi = pd.Series(df["high"].values).rolling(n, min_periods=n).max().shift(1).values lo = pd.Series(df["low"].values).rolling(n, min_periods=n).min().shift(1).values c = df["close"].values.astype(float) return (c > hi), (c < lo) # --------------------------------------------------------------------------- # Parametri # --------------------------------------------------------------------------- @dataclass class SkyhookParams: # indicatori (HTF) atr_win: int = 14 n_vola: int = 13 # finestra Chande su ATR (da PtnL 13) n_volume: int = 13 # finestra Chande su volume (da PtnL 13) # fasce regime (bande-soglia su 0-100). Default = "regime di breakout": # volume vivo (BuzVolume alto) + volatilita' presente ma non da blow-off. vola_lo: float = 35.0 vola_hi: float = 95.0 vol_lo: float = 50.0 vol_hi: float = 100.0 # pattern (HTF) — Donchian breakout ptn_n: int = 13 # da PtnL 13/13/1 # composer / direzione long_only: bool = False # Skyhook e' L/S di natura; True = solo long (stile crypto difensivo) # ingresso max_per_day: int = 1 # uscite — time-based asimmetrico (barre LTF) uscitalong: int = 24 uscitashort: int = 18 # uscite — hard stop/profit (LONG, e SHORT se gli override sotto sono None) exit_mode: str = "atr" # 'atr' = multipli di ATR LTF ; 'pct' = percentuale fissa sl_atr: float = 2.0 tp_atr: float = 5.0 sl_pct: float = 0.03 tp_pct: float = 0.075 ltf_atr_win: int = 14 # uscite — OVERRIDE asimmetrico SHORT (None = usa i valori simmetrici sopra). # In crypto lo short si fa steamrollare da uno spike vola: stop short piu' stretti # tagliano il draw-down standalone senza toccare il segnale (vedi SKH01-V2-DD, diario). exit_mode_short: str | None = None sl_atr_short: float | None = None tp_atr_short: float | None = None sl_pct_short: float | None = None tp_pct_short: float | None = None # --------------------------------------------------------------------------- # Feature HTF -> merge causale su LTF # --------------------------------------------------------------------------- def htf_features(htf: pd.DataFrame, p: SkyhookParams) -> pd.DataFrame: """Calcola regime+pattern sull'HTF e li restituisce indicizzati per CHIUSURA HTF (timestamp di chiusura = open + 690min). Cosi' il merge backward su LTF e' strettamente causale.""" buz_vola = chande01(atr(htf, p.atr_win), p.n_vola) buz_volume = chande01(htf["volume"].values, p.n_volume) ptn_long, ptn_short = donchian_breakout(htf, p.ptn_n) regime_ok = ((buz_vola >= p.vola_lo) & (buz_vola <= p.vola_hi) & (buz_volume >= p.vol_lo) & (buz_volume <= p.vol_hi)) comp_long = regime_ok & ptn_long comp_short = regime_ok & ptn_short if p.long_only: comp_short = np.zeros_like(comp_short, dtype=bool) close_ts = htf["timestamp"].astype("int64").values + HTF_MIN * 60 * 1000 return pd.DataFrame({ "close_ts": close_ts, "buz_vola": buz_vola, "buz_volume": buz_volume, "comp_long": comp_long.astype(bool), "comp_short": comp_short.astype(bool), }) def merge_htf_to_ltf(ltf: pd.DataFrame, feat: pd.DataFrame) -> pd.DataFrame: """Attacca a ogni barra LTF l'ultima feature HTF la cui CHIUSURA <= chiusura LTF (causale).""" left = ltf.copy() left["close_ts"] = left["timestamp"].astype("int64").values + LTF_MIN * 60 * 1000 m = pd.merge_asof(left.sort_values("close_ts"), feat.sort_values("close_ts"), on="close_ts", direction="backward") return m.sort_index().reset_index(drop=True) # --------------------------------------------------------------------------- # Generatore di ingressi per backtest_signals ({'dir','tp','sl','max_bars'}) # --------------------------------------------------------------------------- def skyhook_entries(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams | None = None) -> list: """Lista di entry-dict (uno per barra LTF, None = niente segnale), pronta per backtest_signals. Max `max_per_day` ingressi/giorno (prima barra qualificante del giorno). sl/tp e max_bars asimmetrici per direzione. Tutto causale (decide a close[i]).""" p = p or SkyhookParams() feat = htf_features(htf, p) m = merge_htf_to_ltf(ltf, feat) c = m["close"].values.astype(float) a = atr(m, p.ltf_atr_win) comp_long = np.nan_to_num(m["comp_long"].values).astype(bool) comp_short = np.nan_to_num(m["comp_short"].values).astype(bool) days = pd.to_datetime(m["datetime"]).dt.floor("D").values entries: list = [None] * len(m) count_today: dict = {} for i in range(len(m)): if not np.isfinite(a[i]) or a[i] <= 0: continue day = days[i] if count_today.get(day, 0) >= p.max_per_day: continue if comp_long[i]: direction, mb = 1, p.uscitalong mode, sl_a, tp_a, sl_p, tp_p = p.exit_mode, p.sl_atr, p.tp_atr, p.sl_pct, p.tp_pct elif comp_short[i]: direction, mb = -1, p.uscitashort # SHORT: usa l'override asimmetrico dove presente, altrimenti i valori simmetrici. mode = p.exit_mode_short if p.exit_mode_short is not None else p.exit_mode sl_a = p.sl_atr_short if p.sl_atr_short is not None else p.sl_atr tp_a = p.tp_atr_short if p.tp_atr_short is not None else p.tp_atr sl_p = p.sl_pct_short if p.sl_pct_short is not None else p.sl_pct tp_p = p.tp_pct_short if p.tp_pct_short is not None else p.tp_pct else: continue if mode == "atr": sl_off, tp_off = sl_a * a[i], tp_a * a[i] else: sl_off, tp_off = sl_p * c[i], tp_p * c[i] if direction == 1: sl, tp = c[i] - sl_off, c[i] + tp_off else: sl, tp = c[i] + sl_off, c[i] - tp_off entries[i] = {"dir": direction, "tp": float(tp), "sl": float(sl), "max_bars": int(mb)} count_today[day] = count_today.get(day, 0) + 1 return entries # --------------------------------------------------------------------------- # Config canoniche (vedi docs/diary/2026-06-23-skyhook.md) # --------------------------------------------------------------------------- # SKH01-V1: vincente del primo lever-scout/grid (regime gate + breakout lento + stop larghi). SKH01_V1 = SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) # SKH01-V2-DD: vincente dell'onda DD-reduction (famiglia ASYM_LS). Stesso SEGNALE del winner # intermedio (ptn_n=45, banda vola larga) ma EXIT a percentuale fissa ASIMMETRICA: short con SL # piu' stretto (2% vs 4% long) -> taglia il draw-down standalone (maxDD BTC 21% / ETH 27% <30%) # alzando hold-out e uplift di portafoglio. Verificato leak-free + 2 scettici avversariali. SKH01_V2_DD = SkyhookParams( ptn_n=45, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0, uscitalong=24, uscitashort=16, exit_mode="pct", sl_pct=0.04, tp_pct=0.10, # LONG exit_mode_short="pct", sl_pct_short=0.02, tp_pct_short=0.08, # SHORT (SL piu' stretto) ) def signal_counts(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams | None = None) -> dict: """Diagnostica: quante barre passano regime/pattern/composer (prima del cap giornaliero).""" p = p or SkyhookParams() feat = htf_features(htf, p) m = merge_htf_to_ltf(ltf, feat) cl = np.nan_to_num(m["comp_long"].values).astype(bool) cs = np.nan_to_num(m["comp_short"].values).astype(bool) ent = skyhook_entries(ltf, htf, p) return dict(ltf_bars=len(m), comp_long=int(cl.sum()), comp_short=int(cs.sum()), entries=int(sum(e is not None for e in ent)))