"""signal_inout_1leg — 1-GAMBA a SEGNALI classici (entrata+uscita) su BTC/ETH (2026-06-29). TESI ---- Un filone con un VANTAGGIO STRUTTURALE diverso dai book multi-gamba (XS01, CC01, basis, short-vol) che restano STAT-MODE perche' a $600 non si eseguono: qui ogni strategia e' a **1 SOLA GAMBA** (un singolo asset, BTC o ETH) con entrata E uscita gestite da SEGNALI CLASSICI (MACD, RSI, Supertrend/ATR-trail, Donchian, Bollinger, EMA-cross, MACD+ADX). Turnover basso => realmente ESEGUIBILE a $600 (cap $300/asset, min $5). Il vantaggio del filone NON e' lo Sharpe assoluto — e' l'eseguibilita'. MA la lezione del progetto e' brutale: un 1-gamba direzionale su BTC/ETH o e' (a) TREND-FOLLOWING -> corr ~0.7-0.9 a TP01, marginal REDUNDANT (TP01 travestito), oppure (b) MEAN-REVERSION -> morto sul feed reale (fade negativo anche a fee zero, v2.0.0), o muore di fee a sub-daily. Il soffitto direzionale BTC/ETH e' ~1.3 (= TP01). Quindi NON si giudica sullo Sharpe assoluto ma sul MARGINALE vs TP01 (earns_slot), e la cella si sceglie IN-SAMPLE-ONLY (no peeking del hold-out — il punto cieco del filone B / intraday-ERM). GATE (tutti dall'harness condiviso altlib, leak-free by construction) --------------------------------------------------------------------- 1. study_family_honest(name, factory, grid, tfs): - sceglie la cella per Sharpe IN-SAMPLE (pre-2025), MAI per max hold-out; - study_marginal sulla cella scelta (corr vs TP01, blend uplift full/hold, is_hedge, has_insample_edge, robust_oos multi-cut) -> earns_slot; - deflated-Sharpe (Bailey & Lopez de Prado) su TUTTE le celle del grid (multiple-testing); - earns_slot_honest = earns_slot AND DSR>=0.95. 2. causality_ok: la costruzione del segnale e' causale (max_tail_diff ~0, no peeking). 3. eval_weights_smallcap(capital=600, min_order=5): HAIRCUT $600 + n. trade eseguiti reali (il punto FORTE del filone — turnover basso = eseguibile davvero). 4. fee-sweep 0.00-0.20% RT (dentro study_weights) a frequenza reale. 5. day_boundary_robust: per completezza (segnali di prezzo => INVARIANT atteso). Decisione finale per ogni segnale: SLEEVE-CANDIDATE-eseguibile -> earns_slot_honest=True AND haircut ~0 (executable). RARO. LEAD-forward-monitor -> marginal ADDS ma DSR/hold corto, o edge esile: monitor. REDUNDANT-vs-TP01 -> trend travestito (corr alta, uplift ~0). SCARTATO -> abs FAIL (negativo), DILUTES, o morte per fee. USO: cd /opt/docker/PythagorasGoal && uv run python scripts/research/signal_inout_1leg.py Idempotente, solo stdout. NON committa (lo fa il coordinatore). """ from __future__ import annotations import sys import numpy as np import pandas as pd sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al # noqa: E402 CERTIFIED = al.CERTIFIED TFS = ("1d", "12h", "8h") # =========================================================================== # INDICATORI extra (causali) non gia' in altlib # =========================================================================== def macd_lines(c: np.ndarray, fast: int, slow: int, sig: int): """MACD = EMA(fast) - EMA(slow); signal = EMA(MACD, sig). EMA adjust=False => causale.""" macd = al.ema(c, fast) - al.ema(c, slow) signal = al.ema(macd, sig) return macd, signal def supertrend_dir(df: pd.DataFrame, atr_win: int, mult: float) -> np.ndarray: """Supertrend classico -> direzione {+1 up, -1 down}. CAUSALE: dir[i] usa close[i] e le bande finali calcolate fino a i-1 (mai high/low di i come prezzo d'ingresso).""" h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) a = al.atr(df, atr_win) hl2 = (h + l) / 2.0 upper = hl2 + mult * a lower = hl2 - mult * a n = len(c) fu = np.zeros(n) fl = np.zeros(n) d = np.ones(n) for i in range(n): if i == 0 or not np.isfinite(a[i]): fu[i], fl[i], d[i] = upper[i], lower[i], 1.0 continue fu[i] = upper[i] if (upper[i] < fu[i - 1] or c[i - 1] > fu[i - 1]) else fu[i - 1] fl[i] = lower[i] if (lower[i] > fl[i - 1] or c[i - 1] < fl[i - 1]) else fl[i - 1] if c[i] > fu[i - 1]: d[i] = 1.0 elif c[i] < fl[i - 1]: d[i] = -1.0 else: d[i] = d[i - 1] return d def adx(df: pd.DataFrame, win: int = 14) -> np.ndarray: """ADX di Wilder (forza di trend), causale via EWM. Usato come gate anti-whipsaw.""" h = df["high"].values.astype(float) l = df["low"].values.astype(float) up = np.zeros(len(h)) dn = np.zeros(len(h)) up[1:] = h[1:] - h[:-1] dn[1:] = l[:-1] - l[1:] plus_dm = np.where((up > dn) & (up > 0), up, 0.0) minus_dm = np.where((dn > up) & (dn > 0), dn, 0.0) atr_ = al.atr(df, win) den = np.where(atr_ > 0, atr_, np.nan) plus_di = 100 * pd.Series(plus_dm).ewm(alpha=1 / win, adjust=False).mean().values / den minus_di = 100 * pd.Series(minus_dm).ewm(alpha=1 / win, adjust=False).mean().values / den s = plus_di + minus_di dx = 100 * np.abs(plus_di - minus_di) / np.where(s > 0, s, np.nan) return pd.Series(dx).ewm(alpha=1 / win, adjust=False).mean().values def _hold(pos_raw: np.ndarray) -> np.ndarray: """ffill dello stato fra i segnali (NaN = mantieni posizione precedente), start flat.""" return pd.Series(pos_raw).ffill().fillna(0.0).values # =========================================================================== # FACTORY dei segnali — ognuna: factory(tf, **params) -> target_fn(df)->posizione causale # (tf passa solo per firma; la granularita' la sceglie candidate_daily caricando get(asset,tf)) # =========================================================================== def make_macd(mode: str): def factory(tf, fast, slow, sig): def target(df): c = df["close"].values.astype(float) m, s = macd_lines(c, fast, slow, sig) if mode == "LF": return np.where(m > s, 1.0, 0.0) return np.where(m > s, 1.0, -1.0) return target return factory def make_rsi(): """Mean-reversion long-flat: entra quando RSIoverbought, HOLD in mezzo.""" def factory(tf, win, oversold, overbought): def target(df): c = df["close"].values.astype(float) ind = al.rsi(c, win) raw = np.full(len(c), np.nan) raw[ind < oversold] = 1.0 raw[ind > overbought] = 0.0 return _hold(raw) return target return factory def make_supertrend(mode: str): def factory(tf, atr_win, mult): def target(df): d = supertrend_dir(df, atr_win, mult) return np.clip(d, 0, None) if mode == "LF" else d return target return factory def make_donchian(mode: str): """Turtle: long su breakout del canale alto, esce/short sul canale basso, HOLD in mezzo.""" def factory(tf, win): def target(df): hi, lo = al.donchian(df, win) c = df["close"].values.astype(float) raw = np.full(len(c), np.nan) raw[c > hi] = 1.0 raw[c < lo] = 0.0 if mode == "LF" else -1.0 return _hold(raw) return target return factory def make_bbands(mode: str): """MR: long sotto banda bassa, esce al ritorno sopra la media. BO: long sopra banda alta, flat/short sotto banda bassa.""" def factory(tf, win, k): def target(df): c = df["close"].values.astype(float) up, mid, lo = al.bbands(c, win, k) raw = np.full(len(c), np.nan) if mode == "MR": raw[c < lo] = 1.0 raw[c > mid] = 0.0 else: # breakout raw[c > up] = 1.0 raw[c < lo] = 0.0 return _hold(raw) return target return factory def make_ema(mode: str): def factory(tf, fast, slow): def target(df): c = df["close"].values.astype(float) ef, es = al.ema(c, fast), al.ema(c, slow) if mode == "LF": return np.where(ef > es, 1.0, 0.0) return np.where(ef > es, 1.0, -1.0) return target return factory def make_macd_adx(): """MACD long-flat gated da ADX (entra long solo se MACD>signal E trend abbastanza forte).""" def factory(tf, fast, slow, sig, adx_win, adx_thr): def target(df): c = df["close"].values.astype(float) m, s = macd_lines(c, fast, slow, sig) a = adx(df, adx_win) return np.where((m > s) & (a > adx_thr), 1.0, 0.0) return target return factory # =========================================================================== # GRID # =========================================================================== def macd_grid(): g = [] for fast in (8, 12, 16): for slow in (21, 26, 34): for sig in (9,): if fast < slow: g.append(dict(fast=fast, slow=slow, sig=sig)) return g RSI_GRID = [dict(win=14, oversold=os, overbought=ob) for os in (25, 30, 35) for ob in (60, 65, 70)] SUPER_GRID = [dict(atr_win=w, mult=m) for w in (10, 14, 20) for m in (2.0, 2.5, 3.0)] DONCH_GRID = [dict(win=w) for w in (20, 30, 40, 55)] BB_GRID = [dict(win=20, k=k) for k in (2.0, 2.5)] EMA_GRID = [dict(fast=f, slow=s) for f in (10, 20, 30) for s in (50, 100, 200) if f < s] MACD_ADX_GRID = [dict(fast=12, slow=26, sig=9, adx_win=14, adx_thr=t) for t in (15, 20, 25)] # =========================================================================== # DRIVER # =========================================================================== def _yr_line(absolute: dict) -> str: """Per-anno minimo-fra-asset dalla parte assoluta (cella scelta), TF best.""" cells = absolute.get("cells", []) if not cells: return "" c = cells[0] out = [] for a, pa in c["per_asset"].items(): yr = " ".join(f"{y}:{(d['ret'] if isinstance(d, dict) else d) * 100:+.0f}%" for y, d in pa["yearly"].items()) out.append(f" {a}: {yr}") return "\n".join(out) def classify(rep: dict, haircut_ok: bool) -> tuple[str, str]: """Verdetto a 4 etichette + motivo di una riga.""" if rep.get("chosen") is None: return "SCARTATO", "nessuna cella in-sample valida" m = rep["marginal"]["marginal"] mv = rep["marginal"]["marginal_verdict"] abs_grade = rep["marginal"]["abs_grade"] corr = m.get("corr_full") uph = (m.get("blends", {}).get("w25", {}) or {}).get("uplift_hold") hsh = m.get("cand_hold_sharpe") trend_like = corr is not None and corr >= 0.5 if rep.get("earns_slot_honest"): if haircut_ok: return "SLEEVE-CANDIDATE-eseguibile", "earns_slot_honest=True + haircut $600 ~0 (extra-scettico: possibile selection/fee artifact)" return "LEAD-forward-monitor", "earns_slot_honest=True ma haircut $600 non trascurabile" if mv == "HEDGE": return "LEAD-forward-monitor", "HEDGE: low-corr ma paga solo quando TP01 e' debole (non alpha standing); DSR/abs sotto soglia" if mv == "DILUTES": return "SCARTATO", "DILUTES: trascina giu' il blend TP01 (no edge marginale)" if abs_grade == "FAIL": if trend_like: return "REDUNDANT-vs-TP01", f"trend = TP01 travestito (corr {corr}); la cella in-sample-best (sub-daily) overfitta -> hold-Sh {hsh} OOS" return "SCARTATO", f"abs FAIL: la cella in-sample-best non generalizza OOS (hold-Sh {hsh})" if mv == "REDUNDANT" or trend_like: return "REDUNDANT-vs-TP01", f"trend travestito: corr {corr} a TP01, marginal {mv}, uplift-hold non persistente" if mv == "ADDS": return "LEAD-forward-monitor", "marginal ADDS ma deflated-Sharpe non passa (multiple-testing)" return "REDUNDANT-vs-TP01", f"{mv}: nessun uplift marginale robusto (corr {corr})" def run_family(name: str, factory, grid, tfs=TFS): print("\n" + "=" * 100) print(f"### {name} (grid {len(grid)} celle x {len(tfs)} TF = {len(grid) * len(tfs)} prove)") print("=" * 100) rep = al.study_family_honest(name, factory, grid, tfs) ch = rep.get("chosen") if ch is None: print(f" -> {rep.get('reason', 'no chosen')}") v, why = classify(rep, False) print(f" VERDETTO: {v} — {why}") return rep, (name, v, why, None) fn = factory(tf=ch["tf"], **ch["params"]) m = rep["marginal"]["marginal"] mv = rep["marginal"]["marginal_verdict"] absr = rep["marginal"]["absolute"] bl = m.get("blends", {}) w25 = bl.get("w25", {}) w50 = bl.get("w50", {}) print(f" cella scelta IN-SAMPLE: tf={ch['tf']} {ch['params']} " f"insample-Sh={ch['insample_sharpe']} full-Sh={ch['full_sharpe']}") print(f" ABS grade={rep['marginal']['abs_grade']} | cand full-Sh={m.get('cand_full_sharpe')} " f"hold-Sh={m.get('cand_hold_sharpe')} (TP01 full {m.get('tp01_full_sharpe')}/hold {m.get('tp01_hold_sharpe')})") print(f" MARGINAL={mv} corr->TP01 full={m.get('corr_full')} hold={m.get('corr_hold')} " f"is_hedge={m.get('is_hedge')} has_insample_edge={m.get('has_insample_edge')} robust_oos={m.get('robust_oos')}") print(f" blend w25: full {w25.get('full')} (uplift {w25.get('uplift_full')}) " f"hold {w25.get('hold')} (uplift {w25.get('uplift_hold')}) DD {w25.get('dd')}") print(f" blend w50: full {w50.get('full')} (uplift {w50.get('uplift_full')}) " f"hold {w50.get('hold')} (uplift {w50.get('uplift_hold')})") print(f" multicut uplift {m.get('multicut_uplift')} persistent={m.get('multicut_persistent')}") print(f" DEFLATED-SHARPE={rep.get('deflated_sharpe')} (null-max {rep.get('expected_null_max')}, " f"n_cells {rep.get('n_cells')}) dsr_pass={rep.get('dsr_pass')}") print(f" >>> earns_slot(marginal)={rep['earns_slot_marginal']} EARNS_SLOT_HONEST={rep['earns_slot_honest']}") # causalita' caus = al.causality_ok(fn, tf=ch["tf"]) print(f" causality_ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}") # HAIRCUT $600 + n. trade (il punto forte del filone) print(" ESEGUIBILITA' $600 (haircut + n.trade eseguiti reali):") haircut_ok = True for a in CERTIFIED: df = al.get(a, ch["tf"]) tgt = al._call_target(fn, df, a) sc = al.eval_weights_smallcap(df, tgt, capital=600, min_order=5) print(f" {a}: modeled Sh={sc['modeled']['sharpe']} real$600 Sh={sc['realistic']['sharpe']} " f"haircut={sc['sharpe_haircut']} n_trade_eseguiti={sc['n_executed_trades']} " f"turnover/yr={sc['executed_turnover_per_year']}") if abs(sc["sharpe_haircut"]) > 0.25: haircut_ok = False # day-boundary (segnale di prezzo => INVARIANT atteso) try: db = al.day_boundary_robust(fn, tf=ch["tf"]) print(f" day_boundary: {db['verdict']} (spread {db.get('spread')})") except Exception as e: # noqa print(f" day_boundary: skip ({type(e).__name__})") # per-anno yr = _yr_line(absr) if yr: print(" per-anno (cella scelta):") print(yr) v, why = classify(rep, haircut_ok) print(f" VERDETTO: {v} — {why}") return rep, (name, v, why, rep.get("earns_slot_honest")) def main(): print("#" * 100) print("# SIGNAL-IN/OUT 1-GAMBA — MACD/RSI/Supertrend/Donchian/Bollinger/EMA/MACD+ADX su BTC,ETH") print(f"# TP01 baseline daily full Sharpe = {round(al._sh(al.tp01_baseline_daily()), 3)} (il soffitto da battere al MARGINE)") print(f"# TF testati: {TFS} | fee 0.10% RT + sweep 0..0.20% | capitale reale $600") print("#" * 100) summary = [] families = [ ("MACD-LF (long-flat)", make_macd("LF"), macd_grid()), ("MACD-LS (long-short)", make_macd("LS"), macd_grid()), ("RSI-MR (mean-rev long-flat)", make_rsi(), RSI_GRID), ("SUPERTREND-LF", make_supertrend("LF"), SUPER_GRID), ("SUPERTREND-LS", make_supertrend("LS"), SUPER_GRID), ("DONCHIAN-LF (turtle)", make_donchian("LF"), DONCH_GRID), ("DONCHIAN-LS (turtle)", make_donchian("LS"), DONCH_GRID), ("BBANDS-MR (mean-rev)", make_bbands("MR"), BB_GRID), ("BBANDS-BO (breakout)", make_bbands("BO"), BB_GRID), ("EMA-CROSS-LF", make_ema("LF"), EMA_GRID), ("EMA-CROSS-LS", make_ema("LS"), EMA_GRID), ("MACD+ADX-LF (gated)", make_macd_adx(), MACD_ADX_GRID), ] for nm, fac, grid in families: try: _, row = run_family(nm, fac, grid) summary.append(row) except Exception as e: # noqa import traceback print(f"\n[ERRORE in {nm}] {type(e).__name__}: {e}") traceback.print_exc() summary.append((nm, "ERRORE", str(e), None)) print("\n" + "#" * 100) print("# SOMMARIO FINALE") print("#" * 100) for nm, v, why, esh in summary: flag = " <<< ESEGUIBILE+SCORRELATO" if esh else "" print(f" {nm:32s} -> {v}{flag}") print(f" {why}") any_slot = any(esh for *_, esh in summary) print("\nCONCLUSIONE: c'e' un 1-gamba a segnale che AGGIUNGE oltre TP01 ED e' eseguibile a $600?") print(f" -> {'SI (verificare extra-scetticismo: selection/fee artifact)' if any_slot else 'NO — tutto REDUNDANT (trend=TP01) o SCARTATO (MR morta / fee). Risultato valido: base-rate confermata.'}") if __name__ == "__main__": main()