#!/usr/bin/env python """r0726_skh_partial_entry — MISURA DEDICATA: quanto costa che il live valuti gli INGRESSI di SKH01 su una barra 230m PARZIALE. IL FATTO (scoperto il 2026-07-26, diario `2026-07-26-t1-esecuzione-skh-live.md`). `resample_5m` NON scarta il bin 230m in corso e `_skyhook_positions` ci itera dentro. Sulle USCITE e' un bene (il tocco di SL/TP e' visto entro ~1h invece che a fine barra). Sugli INGRESSI e' un'altra cosa: `ent[n-1]` puo' venire da un breakout **non ancora confermato a fine barra**. Se il segnale evapora, il book apre e richiude -> churn. E se non evapora, l'ingresso avviene comunque a un PREZZO diverso da quello che il backtest assume (la chiusura del bin). Il primo probe (112 bin campionati) trovo' 1 disaccordo ma con sole 3 entry nel campione: sapevo che il fenomeno esiste, non quanto pesa. Questa e' la misura con potenza. MECCANISMO ESATTO (serve per capire perche' la misura e' fattibile senza simulare 60.000 cron). `merge_htf_to_ltf` unisce su `close_ts = ltf.timestamp + 230min`, che per una barra PARZIALE e' INVARIATO (l'etichetta resta l'apertura del bin). Quindi: * i 3 bin LTF dentro un gruppo HTF hanno close_ts = htf_start + 230 / 460 / 690; * la feature HTF parziale ha close_ts = htf_start + 690; * merge_asof BACKWARD => solo il TERZO bin LTF (close_ts == 690) aggancia la feature HTF PARZIALE. Nei primi due la feature viene da una barra HTF gia' CHIUSA, identica al backtest. => la divergenza di SEGNALE puo' nascere solo in 1 bin su 3. La divergenza di PREZZO d'ingresso, invece, riguarda TUTTI i bin (si entra a meta' barra, non alla sua chiusura). DISEGNO * per ogni bin 230m e ogni osservazione oraria al suo interno (il cron gira a ore piene), si ricostruiscono le barre LTF/HTF PARZIALI dai 5m disponibili fino a quel momento; * il segnale si calcola con le funzioni VERE (`htf_features`, `merge_htf_to_ltf`), su una FINESTRA di storia completa + la barra parziale in coda -> nessuna reimplementazione della strategia, e la finestra e' lecita perche' le feature hanno memoria finita (Donchian 45, chande 13, ATR EWM 14); * SELF-CHECK OBBLIGATORIO: a osservazione = fine bin, la ricostruzione deve riprodurre ESATTAMENTE `skyhook_entries` sul pannello completo. Se non lo fa, la misura non vale niente. COSA SI MISURA A. tasso di FALSI INGRESSI: il segnale spara a meta' bin e a fine bin NON c'e' (o ha direzione opposta) -> il live apre un trade che il backtest non apre. B. ANTICIPO: bin in cui il segnale c'e' anche a fine barra ma il live entra prima -> stesso trade, prezzo diverso (quanto diverso). C. EFFETTO SULL'EQUITY: si simula il path "ingresso alla prima osservazione che spara" e lo si confronta col backtest, sulla BANDA dei 23 offset (mediana delle DIFFERENZE appaiate, come impone la lezione del 26/07 mattina). Uso: `uv run python scripts/research/r0726_skh_partial_entry.py` """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) import r0702_anchor_skh01 as r02 # noqa: E402 from src.portfolio.portfolio import metrics # noqa: E402 from src.strategies.skyhook import (HTF_MIN, LTF_MIN, SKH01_V2_DD, atr, # noqa: E402 htf_features, merge_htf_to_ltf, skyhook_entries) MS5 = 5 * 60_000 MS_LTF = LTF_MIN * 60_000 MS_HTF = HTF_MIN * 60_000 MSH = 3_600_000 WARM_HTF = 200 # barre HTF di storia nella finestra (>> Donchian 45 / chande 13 / ATR 14) FEE_RT = 0.001 HOLDOUT = "2025-01-01" P = SKH01_V2_DD # --------------------------------------------------------------------- aggregazione parziale def _agg(ts5, o5, h5, l5, c5, v5, lo_ms: int, hi_ms: int): """OHLCV delle barre 5m in [lo_ms, hi_ms). None se il bin e' vuoto.""" a = int(np.searchsorted(ts5, lo_ms, side="left")) b = int(np.searchsorted(ts5, hi_ms, side="left")) if b <= a: return None return (float(o5[a]), float(h5[a:b].max()), float(l5[a:b].min()), float(c5[b - 1]), float(v5[a:b].sum())) def signal_at(obs_ms: int, ltf: pd.DataFrame, htf: pd.DataFrame, ts5, o5, h5, l5, c5, v5) -> dict | None: """Segnale che il LIVE vedrebbe osservando a `obs_ms`: storia completa + barra parziale. Ritorna dict(dir, close, comp_long, comp_short) o None se non c'e' segnale/dati. Usa le funzioni VERE della strategia su una finestra di storia (memoria finita). """ # ⚠️ -1: un'osservazione ESATTAMENTE alla chiusura del bin appartiene al bin che chiude, non # al successivo. Senza questo, `obs_ms // MS_LTF` salta al bin dopo e l'aggregato esce VUOTO # -> signal_at ritorna None -> il self-check confronta zeri con zeri e PASSA a vuoto. # E' il bug che ha reso finto il primo self-check (BTC 80/80 con want=0 ovunque). ltf_start = ((obs_ms - 1) // MS_LTF) * MS_LTF htf_start = ((obs_ms - 1) // MS_HTF) * MS_HTF p_ltf = _agg(ts5, o5, h5, l5, c5, v5, ltf_start, obs_ms) p_htf = _agg(ts5, o5, h5, l5, c5, v5, htf_start, obs_ms) if p_ltf is None or p_htf is None: return None kl = int(np.searchsorted(ltf["timestamp"].values, ltf_start, side="left")) kh = int(np.searchsorted(htf["timestamp"].values, htf_start, side="left")) if kl < 5 or kh < WARM_HTF: return None # --- pannello HTF: storia CHIUSA + barra parziale in coda hs = htf.iloc[max(0, kh - WARM_HTF):kh][["timestamp", "open", "high", "low", "close", "volume"]] hrow = pd.DataFrame([[htf_start, *p_htf]], columns=hs.columns) hp = pd.concat([hs, hrow], ignore_index=True) feat = htf_features(hp, P) # --- pannello LTF: storia CHIUSA + barra parziale in coda (serve solo l'ATR LTF + close) ls = ltf.iloc[max(0, kl - WARM_HTF):kl][["timestamp", "open", "high", "low", "close", "volume"]] lrow = pd.DataFrame([[ltf_start, *p_ltf]], columns=ls.columns) lp = pd.concat([ls, lrow], ignore_index=True) lp["datetime"] = pd.to_datetime(lp["timestamp"], unit="ms", utc=True) m = merge_htf_to_ltf(lp, feat) a = atr(m, P.ltf_atr_win) if not np.isfinite(a[-1]) or a[-1] <= 0: return None cl = bool(np.nan_to_num(m["comp_long"].values)[-1]) cs = bool(np.nan_to_num(m["comp_short"].values)[-1]) if not (cl or cs): return None return dict(dir=1 if cl else -1, close=float(m["close"].values[-1]), atr=float(a[-1]), bin_start=ltf_start) def levels(direction: int, close: float, a: float) -> tuple[float, float, int]: """SL/TP/max_bars col meccanismo congelato (pct asimmetrico di SKH01_V2_DD).""" if direction == 1: mode = P.exit_mode sl_p, tp_p, sl_a, tp_a, mb = P.sl_pct, P.tp_pct, P.sl_atr, P.tp_atr, P.uscitalong else: mode = P.exit_mode_short if P.exit_mode_short is not None else P.exit_mode 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 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 mb = P.uscitashort off_sl, off_tp = (sl_a * a, tp_a * a) if mode == "atr" else (sl_p * close, tp_p * close) return ((close - off_sl, close + off_tp, mb) if direction == 1 else (close + off_sl, close - off_tp, mb)) # --------------------------------------------------------------------- misura def observations(bin_start: int) -> list[int]: """Istanti in cui il cron ORARIO osserva DENTRO il bin (esclusa la sua chiusura).""" end = bin_start + MS_LTF first = ((bin_start + MSH - 1) // MSH) * MSH return [t for t in range(first, end, MSH) if t > bin_start] def signal_table(asset: str, off: int = 0) -> pd.DataFrame: """Segnale del composer a OGNI osservazione oraria (intra-bin) + a ogni chiusura di bin. ⚠️ E' una funzione PURA del prezzo: non dipende dallo stato del book. Per questo si puo' precalcolare una volta e poi far girare sopra la macchina a stati (cap giornaliero + non-overlap), che e' quello che il primo tentativo di misura NON faceva — e senza il quale i "falsi ingressi" sono sovracontati di ~4x (il composer spara spesso mentre il book e' gia' in posizione o ha gia' usato la quota del giorno). """ _, _, ltf, ent = r02.run_asset(asset, off) df5 = r02.get5m(asset) ts5 = df5["timestamp"].values.astype(np.int64) o5, h5 = df5["open"].values.astype(float), df5["high"].values.astype(float) l5, c5 = df5["low"].values.astype(float), df5["close"].values.astype(float) v5 = df5["volume"].values.astype(float) htf = r02.resample_off(df5, HTF_MIN, off) tsl = ltf["timestamp"].values.astype(np.int64) rows = [] for k in range(WARM_HTF * 3, len(ltf) - 1): b0 = int(tsl[k]) for t in observations(b0) + [b0 + MS_LTF]: s = signal_at(t, ltf, htf, ts5, o5, h5, l5, c5, v5) rows.append(dict(k=k, bin_start=b0, obs_ms=t, is_close=int(t == b0 + MS_LTF), dir=(s["dir"] if s else 0), close=(s["close"] if s else np.nan), atr=(s["atr"] if s else np.nan))) return pd.DataFrame(rows) def scan(asset: str, off: int = 0, limit_bins: int | None = None) -> pd.DataFrame: """Per ogni bin 230m: primo istante orario in cui il segnale spara, e segnale a fine bin.""" _, _, ltf, ent = r02.run_asset(asset, off) df5 = r02.get5m(asset) ts5 = df5["timestamp"].values.astype(np.int64) o5, h5 = df5["open"].values.astype(float), df5["high"].values.astype(float) l5, c5 = df5["low"].values.astype(float), df5["close"].values.astype(float) v5 = df5["volume"].values.astype(float) htf = r02.resample_off(df5, HTF_MIN, off) tsl = ltf["timestamp"].values.astype(np.int64) def px_at(t_ms: int) -> float: """Prezzo 5m alla prima barra che chiude >= t_ms (dove il cron eseguirebbe).""" j = int(np.searchsorted(ts5 + MS5, t_ms, side="left")) return float(c5[min(j, len(c5) - 1)]) rows = [] idx = range(WARM_HTF * 3, len(ltf) - 1) if limit_bins: idx = list(idx)[-limit_bins:] for k in idx: b0 = int(tsl[k]) full = ent[k] obs = observations(b0) sigs = [(t, signal_at(t, ltf, htf, ts5, o5, h5, l5, c5, v5)) for t in obs] fired = [(t, s) for t, s in sigs if s is not None] if not fired and not full: continue # bin muto: niente da confrontare first_t, first_s = (fired[0] if fired else (0, None)) # ultimo istante consecutivo in cui il segnale (stessa direzione) e' ancora vivo last_alive = first_t if first_s is not None: for t, s in sigs: if t < first_t: continue if s is not None and s["dir"] == first_s["dir"]: last_alive = t else: break rows.append(dict( k=k, bin_start=b0, full_dir=(full["dir"] if full else 0), full_close=float(ltf["close"].values[k]), part_dir=(first_s["dir"] if first_s else 0), part_close=(first_s["close"] if first_s else np.nan), part_atr=(first_s["atr"] if first_s else np.nan), obs_ms=first_t, last_alive_ms=last_alive, n_obs=len(obs), n_fired=len(fired), # prezzo al quale il cron chiuderebbe se il segnale muore prima della fine del bin px_death=px_at(min(last_alive + MSH, b0 + MS_LTF)), )) return pd.DataFrame(rows) def selfcheck(asset: str, off: int = 0, n_signal: int = 120, n_flat: int = 400) -> dict: """A osservazione = FINE bin la ricostruzione DEVE riprodurre `skyhook_entries`. ⚠️ Un self-check che campiona bin a caso e' VACUO: gli ingressi sono rari (~2% dei bin), quindi confronta zeri con zeri e passa anche se la ricostruzione e' rotta (successo alla prima stesura: BTC 80/80 con `want=0` ovunque). Qui si testano ESPLICITAMENTE i bin che HANNO un ingresso nel backtest, e separatamente un campione di bin flat. NB: `skyhook_entries` applica `max_per_day`; la ricostruzione no. Un segnale trovato dove il backtest ha 0 puo' quindi essere soppressione da cap, non divergenza -> si conta a parte. """ _, _, ltf, ent = r02.run_asset(asset, off) df5 = r02.get5m(asset) ts5 = df5["timestamp"].values.astype(np.int64) o5, h5 = df5["open"].values.astype(float), df5["high"].values.astype(float) l5, c5 = df5["low"].values.astype(float), df5["close"].values.astype(float) v5 = df5["volume"].values.astype(float) htf = r02.resample_off(df5, HTF_MIN, off) tsl = ltf["timestamp"].values.astype(np.int64) lo = WARM_HTF * 3 sig_idx = [k for k in range(lo, len(ltf) - 1) if ent[k]][-n_signal:] flat_idx = [k for k in range(lo, len(ltf) - 1) if not ent[k]] rng = np.random.default_rng(726) flat_idx = list(rng.choice(flat_idx, size=min(n_flat, len(flat_idx)), replace=False)) res = dict(sig_ok=0, sig_tot=0, flat_ok=0, flat_tot=0, flat_capped=0) for k in sig_idx: s = signal_at(int(tsl[k]) + MS_LTF, ltf, htf, ts5, o5, h5, l5, c5, v5) res["sig_tot"] += 1 res["sig_ok"] += int((s["dir"] if s else 0) == ent[k]["dir"]) for k in flat_idx: s = signal_at(int(tsl[k]) + MS_LTF, ltf, htf, ts5, o5, h5, l5, c5, v5) res["flat_tot"] += 1 if s is None: res["flat_ok"] += 1 else: res["flat_capped"] += 1 # segnale presente ma soppresso dal cap nel backtest return res def simulate(tab: pd.DataFrame, ltf: pd.DataFrame, ts5, h5, l5, c5, intra_entry: bool) -> dict: """Macchina a stati FEDELE al live: cap giornaliero + non-overlap + uscite intra-barra. intra_entry=True -> il segnale si valuta a OGNI osservazione oraria (= il live di oggi, che legge anche il bin 230m parziale); intra_entry=False -> solo alle CHIUSURE dei bin (= quello che il backtest assume). Le USCITE sono intra-barra in entrambi i casi (il live gia' le fa cosi', misura del 26/07): cosi' l'unica differenza fra i due path e' l'INGRESSO, che e' la domanda. """ T = tab if intra_entry else tab[tab.is_close == 1] T = T.sort_values("obs_ms") obs = T["obs_ms"].values.astype(np.int64) dirs = T["dir"].values.astype(int) cls = T["close"].values.astype(float) atrs = T["atr"].values.astype(float) tsl = ltf["timestamp"].values.astype(np.int64) def px_at(t_ms: int) -> float: j = int(np.searchsorted(ts5 + MS5, t_ms, side="left")) return float(c5[min(j, len(c5) - 1)]) def touched(t0: int, t1: int, lvl: float, up: bool) -> bool: a = int(np.searchsorted(ts5, t0, side="left")) b = int(np.searchsorted(ts5, t1, side="left")) if b <= a: return False return bool((h5[a:b] >= lvl).any() if up else (l5[a:b] <= lvl).any()) cap = 1.0 pos = None day_used = None n_tr = n_false = 0 rets = [] for i in range(len(obs)): t = int(obs[i]) day = t // 86_400_000 if pos is not None: # uscita: SL prioritario, poi TP, poi scadenza max_bars — valutata sui 5m dal # controllo precedente a questo (e' cosi' che il live la vede: intra-barra) hit_sl = touched(pos["t_prev"], t, pos["sl"], up=(pos["dir"] == -1)) hit_tp = touched(pos["t_prev"], t, pos["tp"], up=(pos["dir"] == 1)) expired = t >= pos["t_expire"] if hit_sl or hit_tp or expired: px = pos["sl"] if hit_sl else (pos["tp"] if hit_tp else px_at(t)) r = (px - pos["entry"]) / pos["entry"] * pos["dir"] - FEE_RT cap *= (1.0 + max(r, -0.99)) rets.append((t, r)) pos = None else: pos["t_prev"] = t # il segnale e' evaporato prima della chiusura del bin -> il live CHIUDE if pos["provisional"] and dirs[i] != pos["dir"] and t <= pos["bin_close"]: px = px_at(t) r = (px - pos["entry"]) / pos["entry"] * pos["dir"] - FEE_RT cap *= (1.0 + max(r, -0.99)) rets.append((t, r)) n_false += 1 pos = None elif pos["provisional"] and t > pos["bin_close"]: pos["provisional"] = False # confermato a fine bin: e' un trade vero if pos is None and dirs[i] != 0 and day != day_used and np.isfinite(atrs[i]) and atrs[i] > 0: d = int(dirs[i]) sl, tp, mb = levels(d, float(cls[i]), float(atrs[i])) k = int(np.searchsorted(tsl, t - 1, side="right")) - 1 bin_close = int(tsl[k]) + MS_LTF pos = dict(dir=d, entry=float(cls[i]), sl=sl, tp=tp, t_prev=t, t_expire=bin_close + mb * MS_LTF, bin_close=bin_close, provisional=bool(t < bin_close)) day_used = day n_tr += 1 idx = pd.to_datetime([t for t, _ in rets], unit="ms", utc=True) s = pd.Series([r for _, r in rets], index=idx).resample("1D").sum() return dict(equity=cap, n_trades=n_tr, n_false=n_false, daily=s) def main() -> None: print("=" * 100) print(" MISURA — ingressi di SKH01 valutati su barra 230m PARZIALE: quanto costa?") print("=" * 100) print(f"\n finestra di storia per la ricostruzione: {WARM_HTF} barre HTF " f"(Donchian {P.ptn_n}, chande {P.n_vola}/{P.n_volume}, ATR {P.atr_win})") print("\n" + "-" * 100) print(" 0. SELF-CHECK — a fine bin la ricostruzione deve dare lo stesso segnale del backtest") print("-" * 100) print(" ⚠️ testa ESPLICITAMENTE i bin CON ingresso: campionarli a caso rende il check vacuo") print(" (gli ingressi sono ~2% dei bin -> si confrontano zeri con zeri e passa comunque).") for a in ("BTC", "ETH"): r = selfcheck(a) verdict = "OK" if r["sig_ok"] == r["sig_tot"] else "*** DIVERGE ***" print(f" {a}: bin CON ingresso {r['sig_ok']}/{r['sig_tot']} riprodotti {verdict} | " f"bin flat {r['flat_ok']}/{r['flat_tot']} " f"(+{r['flat_capped']} con segnale soppresso da max_per_day, atteso)") if r["sig_ok"] != r["sig_tot"]: print(" ricostruzione non fedele -> i numeri sotto NON sono validi.") sys.exit(1) frames = {} for a in ("BTC", "ETH"): print(f"\n scansione {a} (ogni bin 230m x osservazioni orarie interne)...") frames[a] = scan(a) print("\n" + "-" * 100) print(" A. FALSI INGRESSI — il segnale spara a meta' bin ma a fine bin NON c'e'") print("-" * 100) print(f" {'asset':>6} {'bin':>7} {'segnale a fine bin':>19} {'segnale intra-bin':>18} " f"{'FALSI':>7} {'% dei bin':>10} {'% delle entry vere':>19}") tot_false = tot_true = 0 for a, D in frames.items(): n = len(D) n_full = int((D.full_dir != 0).sum()) n_part = int((D.part_dir != 0).sum()) false_e = int(((D.part_dir != 0) & (D.part_dir != D.full_dir)).sum()) tot_false += false_e tot_true += n_full print(f" {a:>6} {n:>7} {n_full:>19} {n_part:>18} {false_e:>7} " f"{false_e / n:>9.2%} {(false_e / max(n_full, 1)):>18.0%}") print(f"\n TOTALE: {tot_false} falsi ingressi contro {tot_true} ingressi veri " f"= {tot_false / max(tot_true, 1):.0%} in piu' di trade") print("\n" + "-" * 100) print(" B. ANTICIPO — quando il segnale c'e' ANCHE a fine bin, quanto cambia il prezzo?") print("-" * 100) for a, D in frames.items(): m = D[(D.full_dir != 0) & (D.part_dir == D.full_dir)] if not len(m): print(f" {a}: nessun caso") continue slip = (m.part_close - m.full_close) / m.full_close * m.full_dir lead = (m.bin_start + MS_LTF - m.obs_ms) / 60000.0 print(f" {a}: {len(m)} ingressi anticipati | anticipo mediano {lead.median():.0f} min | " f"prezzo d'ingresso {slip.mean()*100:+.3f}% medio ({slip.median()*100:+.3f}% mediano) " f"vs la chiusura del bin") print(f" (segno positivo = il live entra a un prezzo PEGGIORE del backtest)") print("\n" + "-" * 100) print(" C. COSTO dei falsi ingressi — apri a P(t) e richiudi quando il segnale evapora") print("-" * 100) print(" Ogni falso ingresso e' un round-trip completo: fee 0.10% + il movimento di prezzo") print(" nella finestra in cui la posizione e' stata tenuta per errore.\n") years = None tot_cost = 0.0 for a, D in frames.items(): f = D[(D.part_dir != 0) & (D.part_dir != D.full_dir)].copy() span = (D.bin_start.max() - D.bin_start.min()) / (365.25 * 86_400_000) years = span if not len(f): print(f" {a}: nessun falso ingresso") continue drift = (f.px_death - f.part_close) / f.part_close * f.part_dir pnl = drift - FEE_RT # netto del round-trip held = (f.last_alive_ms + MSH - f.obs_ms) / 60000.0 tot_cost += float(pnl.sum()) print(f" {a}: {len(f)} falsi ingressi in {span:.1f} anni " f"({len(f)/span:.1f}/anno) | tenuti mediana {held.median():.0f} min") print(f" drift di prezzo medio {drift.mean()*100:+.3f}% | " f"netto fee {pnl.mean()*100:+.3f}%/trade | " f"totale {pnl.sum()*100:+.2f}% su {span:.1f} anni " f"= {pnl.sum()/span*100:+.2f}%/anno sullo SLEEVE") if years: print(f"\n COSTO AGGREGATO (BTC+ETH, 50/50 dentro lo sleeve): " f"{tot_cost/2/years*100:+.2f}%/anno di sleeve " f"-> sul BOOK al peso 25%: {tot_cost/2/years*0.25*100:+.2f}%/anno") print("\n" + "-" * 100) print(" D. CANNIBALIZZAZIONE DEL CAP — un falso ingresso brucia la quota max_per_day=1") print("-" * 100) for a, D in frames.items(): f = D[(D.part_dir != 0) & (D.part_dir != D.full_dir)].copy() if not len(f): continue day_false = set(pd.to_datetime(f.bin_start, unit="ms", utc=True).dt.floor("D")) real = D[D.full_dir != 0] day_real = pd.to_datetime(real.bin_start, unit="ms", utc=True).dt.floor("D") # ingresso VERO successivo, nello stesso giorno UTC di un falso -> sarebbe stato bloccato clash = 0 for d, g in real.groupby(day_real): if d in day_false: fb = f[pd.to_datetime(f.bin_start, unit="ms", utc=True).dt.floor("D") == d] if len(fb) and (g.bin_start.min() > fb.bin_start.min()): clash += 1 print(f" {a}: {len(day_false)} giorni con un falso ingresso | " f"ingressi VERI che sarebbero stati bloccati dal cap: {clash} " f"({clash/max(len(real),1):.1%} di tutti gli ingressi veri)") print("\n" + "=" * 100) if __name__ == "__main__": main()