"""r0702_ell_channel — "tecnica del canale" Elliott (Ftaonline): falsificazione onesta. FILONE (2026-07-02): l'unica parte pienamente meccanica/falsificabile del metodo Elliott dell'analista Ftaonline: - Swing meccanici (zigzag causale k*ATR): pivot 0 (origine), massimo "onda 1", minimo "onda 2" (vincolo: sopra l'origine, altrimenti conteggio NULLO). - Canale: retta 0 -> minimo onda 2; parallela dal massimo di onda 1. - SEGNALE 1: close FUORI dal lato alto del canale = "onda 3" -> long al close della barra di rottura; target = min(onda2) + 1.618 * ampiezza(onda 1); stop = min(onda 2). - SEGNALE 2 (variante): rottura del massimo di onda 3 dopo un pivot di onda 4 che NON sovrappone il territorio di onda 1 -> target = min(onda4) + 1.0 * ampiezza(onda 1). - REGOLA DISCRIMINANTE: movimento che NON esce mai dal canale = correttivo (nessun trade; segnale opposto alla violazione della base). Testata separatamente con null permutato. - Speculare per lo short. COVERAGE (scripts/research/alt/runs, sweep 104 famiglie 2026-06-20): BRK01 (Donchian LS/LF), BRK02 (Donchian+chandelier), BRK03 (Keltner), BRK04 (Bollinger), BRK05 (ATR-range), BRK08 (NR7), BRK09 (inside-bar), BRK10 (squeeze) + SKH01 coprono la famiglia breakout-canale, ma NESSUNO costruisce canali da pivot zigzag con vincoli d'onda e target 1.618 -> non identico, pero' stessa famiglia: per giudizio si confronta ANCHE contro un Donchian a pari geometria (stop = base canale, target = base + 1.618*larghezza) e pari frequenza di trade. ONESTA': - Pivot noti solo alla CONFERMA (reversal k*ATR dal running extreme); il canale usa solo pivot confermati al tempo t. Guard: al.causality_ok (prefix-recompute). - Entry a close[i] della barra che CHIUDE fuori dal canale (mai fill sull'estremo). - Exit a target/stop/timeout ESEGUITE AL CLOSE della barra che li tocca (gap-through-stop reale, lezione SKH01). Il fill-al-livello e' riportato SOLO come lens ottimista dichiarata. - Fee 0.10% RT + sweep 0.00-0.20% (al.study_weights). - Selezione cella IN-SAMPLE-ONLY + deflated Sharpe su TUTTA la griglia (al.study_family_honest); marginale vs TP01 (al.study_marginal). - 4h: banda d'ancora su offset 0/1/2/3h (regola anchor-luck 2026-07-02). Epoca ms ESPLICITA nel resample (MAI DatetimeIndex.view("int64")). - Hold-out 2025+ mai usato per selezionare. Timeout 150 barre FISSO (non cercato). Output temporanei: scratchpad ell_c_*. Diario: da scrivere a valle (a cura del chiamante). """ from __future__ import annotations import json import sys import time import numpy as np import pandas as pd sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al # noqa: E402 SCRATCH = ("/tmp/claude-1001/-opt-docker-PythagorasGoal/" "e00896d3-d4bb-4f2a-b471-55a1d88a12ba/scratchpad") TFS = ("1d", "4h", "1h") KS = (2.0, 3.0, 4.0) # zigzag: reversal = k * ATR(14) VARIANTS = ("s1_long", "s1_ls", "s12_ls") TIMEOUT = 150 # barre, fisso e dichiarato (non cercato) ATR_WIN = 14 TGT1, TGT2 = 1.618, 1.0 SEED = 20260702 _ZZ_CACHE: dict = {} _SIM_CACHE: dict = {} def _dfkey(df: pd.DataFrame, asset: str): return (asset, int(df["timestamp"].iloc[0]), int(df["timestamp"].iloc[-1]), len(df)) # =========================================================================== # 1) ZIGZAG CAUSALE — pivot confermato SOLO quando il close ritraccia k*ATR # dal running extreme. Ogni valore usa dati <= i (prefix-stable). # =========================================================================== def zigzag(df: pd.DataFrame, k: float, asset: str = "?"): """Ritorna lista di pivot confermati (conf_i, piv_i, piv_price, kind); kind +1=high, -1=low. Un pivot e' utilizzabile solo da conf_i in poi.""" key = _dfkey(df, asset) + (k,) if key in _ZZ_CACHE: return _ZZ_CACHE[key] h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) a = al.atr(df, ATR_WIN) n = len(c) piv = [] dir_ = 0 # 0=unknown, +1=gamba su (caccio un high), -1=gamba giu hi_v, hi_i = h[0], 0 lo_v, lo_i = l[0], 0 for i in range(1, n): thr = k * a[i] if dir_ >= 0 and h[i] > hi_v: hi_v, hi_i = h[i], i if dir_ <= 0 and l[i] < lo_v: lo_v, lo_i = l[i], i if dir_ >= 0 and c[i] < hi_v - thr: piv.append((i, hi_i, float(hi_v), +1)) dir_ = -1 j0 = hi_i + 1 if j0 <= i: seg = l[j0:i + 1] off = int(np.argmin(seg)) lo_v, lo_i = float(seg[off]), j0 + off else: lo_v, lo_i = l[i], i elif dir_ <= 0 and c[i] > lo_v + thr: piv.append((i, lo_i, float(lo_v), -1)) dir_ = +1 j0 = lo_i + 1 if j0 <= i: seg = h[j0:i + 1] off = int(np.argmax(seg)) hi_v, hi_i = float(seg[off]), j0 + off else: hi_v, hi_i = h[i], i _ZZ_CACHE[key] = piv return piv # =========================================================================== # 2) STATE MACHINE canale Elliott — forward-only, un trade alla volta. # =========================================================================== def simulate(df: pd.DataFrame, k: float, variant: str, asset: str = "?") -> dict: key = _dfkey(df, asset) + (k, variant) if key in _SIM_CACHE: return _SIM_CACHE[key] piv = zigzag(df, k, asset) h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) n = len(c) pos = np.zeros(n) trades: list = [] events: list = [] cnt = dict(setups_long=0, setups_short=0, null_count=0, superseded=0) allow_short = variant in ("s1_ls", "s12_ls") allow_s2 = variant == "s12_ls" LS = SS = None # setup long / short attivo S2L = S2S = None # stato onda-5 (dopo S1 chiuso a target) tr = None # trade aperto pi = 0 for i in range(1, n): # --- (a) gestione trade aperto: exit AL CLOSE della barra che tocca ------------ if tr is not None and i > tr["e"]: reason = None if tr["dir"] == +1: if l[i] <= tr["stp"]: reason = "stop" # stop prioritario se tocca entrambi elif h[i] >= tr["tgt"]: reason = "target" else: if h[i] >= tr["stp"]: reason = "stop" elif l[i] <= tr["tgt"]: reason = "target" if reason is None and i - tr["e"] >= TIMEOUT: reason = "timeout" if reason: tr["x"], tr["exit_px"], tr["reason"] = i, float(c[i]), reason trades.append(tr) if allow_s2 and tr["sig"] == "S1" and reason == "target": if tr["dir"] == +1: S2L = dict(stage="w3h", P1p=tr["P1p"], amp=tr["amp"], after=tr["e"]) else: S2S = dict(stage="w3l", P1p=tr["P1p"], amp=tr["amp"], after=tr["e"]) tr = None pos[i] = 0.0 else: pos[i] = tr["dir"] # --- (b) pivot confermati a questa barra ---------------------------------------- while pi < len(piv) and piv[pi][0] == i: _, p_i, p_px, kind = piv[pi] pi += 1 if kind == -1: # nuovo pivot LOW if S2L is not None and S2L.get("stage") == "w4l" and p_i > S2L["w3h_i"]: if p_px > S2L["P1p"]: # onda 4 NON sovrappone onda 1 S2L["w4l_px"], S2L["stage"] = p_px, "brk" else: S2L = None if S2S is not None and S2S.get("stage") == "w3l" and p_i > S2S["after"]: S2S["w3l_px"], S2S["w3l_i"], S2S["stage"] = p_px, p_i, "w4h" if pi >= 3: t0, t1, t2 = piv[pi - 3], piv[pi - 2], piv[pi - 1] if t0[3] == -1 and t1[3] == +1 and t2[3] == -1: if t2[2] > t0[2]: # vincolo: onda 2 sopra l'origine if LS is not None and not LS["done"]: cnt["superseded"] += 1 m = (t2[2] - t0[2]) / (t2[1] - t0[1]) LS = dict(P0i=t0[1], P0p=t0[2], P1i=t1[1], P1p=t1[2], P2i=t2[1], P2p=t2[2], m=m, conf=i, done=False) cnt["setups_long"] += 1 else: cnt["null_count"] += 1 LS = None else: # nuovo pivot HIGH if S2L is not None and S2L.get("stage") == "w3h" and p_i > S2L["after"]: S2L["w3h_px"], S2L["w3h_i"], S2L["stage"] = p_px, p_i, "w4l" if S2S is not None and S2S.get("stage") == "w4h" and p_i > S2S["w3l_i"]: if p_px < S2S["P1p"]: S2S["w4h_px"], S2S["stage"] = p_px, "brk" else: S2S = None if pi >= 3: t0, t1, t2 = piv[pi - 3], piv[pi - 2], piv[pi - 1] if t0[3] == +1 and t1[3] == -1 and t2[3] == +1: if t2[2] < t0[2]: # speculare: onda 2 sotto l'origine if SS is not None and not SS["done"]: cnt["superseded"] += 1 m = (t2[2] - t0[2]) / (t2[1] - t0[1]) SS = dict(P0i=t0[1], P0p=t0[2], P1i=t1[1], P1p=t1[2], P2i=t2[1], P2p=t2[2], m=m, conf=i, done=False) cnt["setups_short"] += 1 else: cnt["null_count"] += 1 SS = None # --- (c) monitoraggio setup + entry a close[i] ---------------------------------- if LS is not None and not LS["done"]: up = LS["P1p"] + LS["m"] * (i - LS["P1i"]) base = LS["P0p"] + LS["m"] * (i - LS["P0i"]) if c[i] > up: LS["done"] = True events.append(dict(kind="impulse", side=+1, bar=i)) if tr is None: amp = LS["P1p"] - LS["P0p"] tr = dict(dir=+1, sig="S1", e=i, entry_px=float(c[i]), tgt=LS["P2p"] + TGT1 * amp, stp=LS["P2p"], P1p=LS["P1p"], amp=amp) pos[i] = 1.0 elif c[i] < base: LS["done"] = True events.append(dict(kind="corrective", side=+1, bar=i)) if SS is not None and not SS["done"]: dn = SS["P1p"] + SS["m"] * (i - SS["P1i"]) base = SS["P0p"] + SS["m"] * (i - SS["P0i"]) if c[i] < dn: SS["done"] = True events.append(dict(kind="impulse", side=-1, bar=i)) if tr is None and allow_short: amp = SS["P0p"] - SS["P1p"] tr = dict(dir=-1, sig="S1", e=i, entry_px=float(c[i]), tgt=SS["P2p"] - TGT1 * amp, stp=SS["P2p"], P1p=SS["P1p"], amp=amp) pos[i] = -1.0 elif c[i] > base: SS["done"] = True events.append(dict(kind="corrective", side=-1, bar=i)) # --- (d) SEGNALE 2 (onda 5) ----------------------------------------------------- if allow_s2 and S2L is not None and S2L.get("stage") == "brk": if c[i] < S2L["w4l_px"]: S2L = None elif c[i] > S2L["w3h_px"] and tr is None: tr = dict(dir=+1, sig="S2", e=i, entry_px=float(c[i]), tgt=S2L["w4l_px"] + TGT2 * S2L["amp"], stp=S2L["w4l_px"], P1p=S2L["P1p"], amp=S2L["amp"]) pos[i] = 1.0 S2L = None if allow_s2 and S2S is not None and S2S.get("stage") == "brk": if c[i] > S2S["w4h_px"]: S2S = None elif c[i] < S2S["w3l_px"] and tr is None: tr = dict(dir=-1, sig="S2", e=i, entry_px=float(c[i]), tgt=S2S["w4h_px"] - TGT2 * S2S["amp"], stp=S2S["w4h_px"], P1p=S2S["P1p"], amp=S2S["amp"]) pos[i] = -1.0 S2S = None out = dict(pos=pos, trades=trades, events=events, counters=cnt) _SIM_CACHE[key] = out return out def make_target(k: float, variant: str): def target_fn(df, asset): return simulate(df, k, variant, asset)["pos"] return target_fn def factory(tf=None, k=3.0, variant="s1_ls"): # tf consumata da study_family_honest/candidate_daily (carica il df giusto) return make_target(k, variant) # =========================================================================== # 3) COMPARATORE DONCHIAN "banale" a pari geometria (stop=base canale, # target = base + 1.618*larghezza) — stesso engine close-exec. # =========================================================================== def donch_sim(df: pd.DataFrame, N: int, allow_short: bool, asset: str = "?") -> dict: key = _dfkey(df, asset) + ("donch", N, allow_short) if key in _SIM_CACHE: return _SIM_CACHE[key] hi, lo = al.donchian(df, N) # shiftati -> causali h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) n = len(c) pos = np.zeros(n) trades = [] tr = None for i in range(1, n): if tr is not None and i > tr["e"]: reason = None if tr["dir"] == +1: if l[i] <= tr["stp"]: reason = "stop" elif h[i] >= tr["tgt"]: reason = "target" else: if h[i] >= tr["stp"]: reason = "stop" elif l[i] <= tr["tgt"]: reason = "target" if reason is None and i - tr["e"] >= TIMEOUT: reason = "timeout" if reason: tr["x"], tr["exit_px"], tr["reason"] = i, float(c[i]), reason trades.append(tr) tr = None pos[i] = 0.0 else: pos[i] = tr["dir"] if tr is None and np.isfinite(hi[i]) and np.isfinite(lo[i]): W = hi[i] - lo[i] if c[i] > hi[i] and W > 0: tr = dict(dir=+1, sig="D", e=i, entry_px=float(c[i]), tgt=lo[i] + TGT1 * W, stp=lo[i]) pos[i] = 1.0 elif allow_short and c[i] < lo[i] and W > 0: tr = dict(dir=-1, sig="D", e=i, entry_px=float(c[i]), tgt=hi[i] - TGT1 * W, stp=hi[i]) pos[i] = -1.0 out = dict(pos=pos, trades=trades) _SIM_CACHE[key] = out return out # =========================================================================== # 4) 4h con ancora spostata (offset 0/1/2/3h) — epoca ms ESPLICITA. # =========================================================================== def get_4h_anchor(asset: str, off: int) -> pd.DataFrame: g = al.get(asset, "1h").copy() idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) idx.name = "dt" g.index = idx out = g.resample("4h", label="left", closed="left", offset=pd.Timedelta(hours=off)).agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) out = out.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"]] # =========================================================================== # 5) STATISTICHE: claim discriminante + target 1.618 vs null vol-matched. # =========================================================================== def discriminant_test(tf: str, k: float, H: int = 10, B: int = 2000) -> dict: """Claim: un movimento che NON esce dal canale (violazione base senza uscita alta) e' correttivo -> follow-through nella direzione del conteggio NEGATIVO (reversal). Statistica: media del forward-return H-barre ALLINEATO al conteggio, normalizzato ATR, vs null di barre casuali (stessi segni). p_low piccolo => claim supportata.""" rng = np.random.default_rng(SEED) out = {} for a in al.CERTIFIED: df = al.get(a, tf) sim = simulate(df, k, "s1_ls", a) c = df["close"].values.astype(float) atr_ = al.atr(df, ATR_WIN) n = len(c) f = np.full(n, np.nan) f[:n - H] = (c[H:] / c[:n - H] - 1.0) / np.maximum(atr_[:n - H] / c[:n - H], 1e-9) / np.sqrt(H) res = {} for kind in ("corrective", "impulse"): ev = [(e["bar"], e["side"]) for e in sim["events"] if e["kind"] == kind and e["bar"] + H < n and e["bar"] >= 50] if len(ev) < 5: res[kind] = dict(n=len(ev), mean=None, p_low=None, p_high=None) continue bars = np.array([b for b, _ in ev]) sides = np.array([s for _, s in ev], float) obs = float(np.mean(f[bars] * sides)) valid = np.arange(50, n - H - 1) draws = rng.choice(valid, size=(B, len(ev))) null = (f[draws] * sides[None, :]).mean(axis=1) res[kind] = dict(n=len(ev), mean=round(obs, 4), null_mean=round(float(np.mean(null)), 4), p_low=round(float(np.mean(null <= obs)), 4), p_high=round(float(np.mean(null >= obs)), 4)) out[a] = res return out def _first_touch(h, l, s, dirn, tgt, stp): wh = h[s + 1:s + 1 + TIMEOUT] wl = l[s + 1:s + 1 + TIMEOUT] if dirn == +1: mT, mS = wh >= tgt, wl <= stp else: mT, mS = wl <= tgt, wh >= stp iT = int(np.argmax(mT)) if mT.any() else 10 ** 9 iS = int(np.argmax(mS)) if mS.any() else 10 ** 9 return 1 if iT < iS else 0 # pari barra -> stop prioritario (come l'engine) def target_hit_vs_null(tf: str, k: float, variant: str, B: int = 300) -> dict: """Freq. con cui il target 1.618 viene toccato prima dello stop nei trade REALI vs null: stessa geometria (distanze % target/stop, direzione, timeout) da barre CASUALI. p_high piccolo => la struttura d'onda tempa meglio del caso a parita' di geometria.""" rng = np.random.default_rng(SEED + 1) out = {} for a in al.CERTIFIED: df = al.get(a, tf) sim = simulate(df, k, variant, a) h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) n = len(c) trs = [t for t in sim["trades"] if t["sig"] == "S1"] if len(trs) < 5: out[a] = dict(n=len(trs), real=None) continue real = float(np.mean([1 if t["reason"] == "target" else 0 for t in trs])) geo = [(t["dir"], abs(t["tgt"] / t["entry_px"] - 1.0), abs(1.0 - t["stp"] / t["entry_px"])) for t in trs] if len(geo) > 300: # cap dichiarato per il bootstrap (compute) sel = rng.choice(len(geo), size=300, replace=False) geo = [geo[int(s)] for s in sel] nulls = np.empty(B) lo_s, hi_s = 50, n - TIMEOUT - 2 for b in range(B): starts = rng.integers(lo_s, hi_s, size=len(geo)) hits = 0 for (dirn, dT, dS), s in zip(geo, starts): e = c[s] if dirn == +1: hits += _first_touch(h, l, s, +1, e * (1 + dT), e * (1 - dS)) else: hits += _first_touch(h, l, s, -1, e * (1 - dT), e * (1 + dS)) nulls[b] = hits / len(geo) out[a] = dict(n=len(trs), real=round(real, 3), null_mean=round(float(np.mean(nulls)), 3), p_high=round(float(np.mean(nulls >= real)), 4), p_low=round(float(np.mean(nulls <= real)), 4)) return out # =========================================================================== # 6) LENS OTTIMISTA (fill al livello) — dichiarata, solo per confronto. # =========================================================================== def lens_compare(tf: str, k: float, variant: str, fee_rt: float = 0.001) -> dict: out = {} for a in al.CERTIFIED: df = al.get(a, tf) trs = simulate(df, k, variant, a)["trades"] if not trs: out[a] = dict(n=0) continue close_r, level_r = [], [] for t in trs: rc = t["dir"] * (t["exit_px"] / t["entry_px"] - 1.0) - fee_rt if t["reason"] == "target": px = t["tgt"] elif t["reason"] == "stop": px = t["stp"] else: px = t["exit_px"] rl = t["dir"] * (px / t["entry_px"] - 1.0) - fee_rt close_r.append(rc) level_r.append(rl) out[a] = dict(n=len(trs), exp_close=round(float(np.mean(close_r)), 5), exp_level=round(float(np.mean(level_r)), 5), tot_close=round(float(np.prod(1 + np.array(close_r)) - 1), 4), tot_level=round(float(np.prod(1 + np.array(level_r)) - 1), 4)) return out # =========================================================================== # MAIN # =========================================================================== def main(): t0 = time.time() report = [] def say(s=""): print(s, flush=True) report.append(s) say("=" * 88) say("r0702_ell_channel — canale Elliott (Ftaonline), falsificazione onesta") say("=" * 88) # ---------- TABELLA CELLE ---------------------------------------------------------- rows = [] say("\n[1] TABELLA CELLE (TF x k x variante) — net 0.10% RT, exit al close (onesto)") hdr = (f"{'tf':>4} {'k':>3} {'variante':>8} | " f"{'BTC f/h':>12} {'nT':>4} {'DD%':>5} | {'ETH f/h':>12} {'nT':>4} {'DD%':>5} | " f"{'COMB full':>9} {'hold':>6} {'inS':>6} {'DD%':>5}") say(hdr) say("-" * len(hdr)) for tf in TFS: for k in KS: for v in VARIANTS: per, parts = {}, {} for a in al.CERTIFIED: df = al.get(a, tf) sim = simulate(df, k, v, a) ev = al.eval_weights(df, sim["pos"]) per[a] = dict(full=ev["full"], hold=ev["holdout"], ntr=len(sim["trades"]), cnt=sim["counters"]) parts[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(parts, axis=1, join="inner").fillna(0.0) comb = al._to_daily(0.5 * J["BTC"] + 0.5 * J["ETH"]) ins = comb[comb.index < al.HOLDOUT] hold = comb[comb.index >= al.HOLDOUT] r = dict(tf=tf, k=k, var=v, btc_full=per["BTC"]["full"]["sharpe"], btc_hold=per["BTC"]["hold"].get("sharpe", 0.0), btc_dd=per["BTC"]["full"]["maxdd"], btc_n=per["BTC"]["ntr"], eth_full=per["ETH"]["full"]["sharpe"], eth_hold=per["ETH"]["hold"].get("sharpe", 0.0), eth_dd=per["ETH"]["full"]["maxdd"], eth_n=per["ETH"]["ntr"], comb_full=round(al._sh(comb), 3), comb_hold=round(al._sh(hold), 3), comb_ins=round(al._sh(ins), 3), comb_dd=round(al._dd_ret(comb), 4), counters=dict(BTC=per["BTC"]["cnt"], ETH=per["ETH"]["cnt"])) rows.append(r) flag = " (<30 trade!)" if min(r["btc_n"], r["eth_n"]) < 30 else "" say(f"{tf:>4} {k:>3.0f} {v:>8} | " f"{r['btc_full']:>+5.2f}/{r['btc_hold']:>+5.2f} {r['btc_n']:>4d} " f"{r['btc_dd']*100:>5.1f} | " f"{r['eth_full']:>+5.2f}/{r['eth_hold']:>+5.2f} {r['eth_n']:>4d} " f"{r['eth_dd']*100:>5.1f} | " f"{r['comb_full']:>+9.2f} {r['comb_hold']:>+6.2f} {r['comb_ins']:>+6.2f} " f"{r['comb_dd']*100:>5.1f}{flag}") say(f"\n (tempo tabella: {time.time()-t0:.0f}s)") # ---------- FAMILY HONEST (selezione in-sample + deflated Sharpe) ------------------ say("\n[2] study_family_honest — cella scelta IN-SAMPLE-ONLY + DSR su tutta la griglia") grid = [dict(k=k, variant=v) for k in KS for v in VARIANTS] fam = al.study_family_honest("ELLCH", factory, grid, TFS) ch = fam["chosen"] say(f" celle valutate: {fam['n_cells']} cella in-sample: tf={ch['tf']} " f"params={ch['params']} (inS Sharpe {ch['insample_sharpe']}, full {ch['full_sharpe']})") say(f" deflated Sharpe = {fam['deflated_sharpe']} (null max atteso " f"{fam['expected_null_max']}) dsr_pass={fam['dsr_pass']}") say(f" earns_slot_marginal={fam['earns_slot_marginal']} " f"EARNS_SLOT_HONEST={fam['earns_slot_honest']}") say(al.fmt_marginal(fam["marginal"])) ck, cv, ctf = ch["params"]["k"], ch["params"]["variant"], ch["tf"] # ---------- CAUSALITA' ------------------------------------------------------------- say("\n[3] causality_ok (prefix-recompute) sulla cella scelta") for tf_chk in {ctf, "1h"}: cz = al.causality_ok(make_target(ck, cv), tf=tf_chk) say(f" tf={tf_chk}: ok={cz['ok']} max_tail_diff={cz['max_tail_diff']} " f"checked={cz['checked']}") # ---------- FEE SWEEP + SMALLCAP sulla cella scelta --------------------------------- say("\n[4] fee sweep 0.00-0.20% RT + haircut small-cap ($600, min order $5) — cella scelta") sw = al.study_weights(f"ELLCH k={ck} {cv}", make_target(ck, cv), tfs=(ctf,)) cell = sw["cells"][0] for a in al.CERTIFIED: say(f" {a}: fee_sweep={cell['per_asset'][a]['fee_sweep']}") df = al.get(a, ctf) sc = al.eval_weights_smallcap(df, simulate(df, ck, cv, a)["pos"]) say(f" smallcap: modeled Sh {sc['modeled']['sharpe']} -> real " f"{sc['realistic']['sharpe']} (haircut {sc['sharpe_haircut']}, " f"{sc['n_executed_trades']} ordini)") # ---------- COMPARATORE DONCHIAN ---------------------------------------------------- say("\n[5] Donchian 'banale' a pari geometria (stop=base, target=base+1.618*W), " "close-exec, stesso timeout") donch_rows = [] allow_short = cv in ("s1_ls", "s12_ls") for tf in TFS: ell_r = next(r for r in rows if r["tf"] == tf and r["k"] == ck and r["var"] == cv) ell_n = (ell_r["btc_n"] + ell_r["eth_n"]) / 2 for N in (20, 55, 100): per, parts = {}, {} for a in al.CERTIFIED: df = al.get(a, tf) sim = donch_sim(df, N, allow_short, a) ev = al.eval_weights(df, sim["pos"]) per[a] = dict(full=ev["full"]["sharpe"], hold=ev["holdout"].get("sharpe", 0.0), ntr=len(sim["trades"])) parts[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(parts, axis=1, join="inner").fillna(0.0) comb = al._to_daily(0.5 * J["BTC"] + 0.5 * J["ETH"]) dn = (per["BTC"]["ntr"] + per["ETH"]["ntr"]) / 2 donch_rows.append(dict(tf=tf, N=N, per=per, ntr=dn, comb_full=round(al._sh(comb), 3), comb_hold=round(al._sh(comb[comb.index >= al.HOLDOUT]), 3), comb=comb, freq_gap=abs(dn - ell_n))) best = min([d for d in donch_rows if d["tf"] == tf], key=lambda d: d["freq_gap"]) # corr Elliott(cella scelta k,cv su questo tf) vs Donchian matched parts_e = {} for a in al.CERTIFIED: df = al.get(a, tf) ev = al.eval_weights(df, simulate(df, ck, cv, a)["pos"]) parts_e[a] = pd.Series(ev["net"], index=ev["idx"]) JE = pd.concat(parts_e, axis=1, join="inner").fillna(0.0) comb_e = al._to_daily(0.5 * JE["BTC"] + 0.5 * JE["ETH"]) JJ = pd.concat({"E": comb_e, "D": best["comb"]}, axis=1, join="inner").dropna() corr = round(float(JJ["E"].corr(JJ["D"])), 3) if len(JJ) > 30 else None for d in [x for x in donch_rows if x["tf"] == tf]: mark = " <-- freq-matched" if d is best else "" say(f" {tf:>4} N={d['N']:>3d}: comb full {d['comb_full']:>+5.2f} hold " f"{d['comb_hold']:>+5.2f} nT/asset~{d['ntr']:.0f}{mark}") say(f" {tf:>4} ELLIOTT (k={ck:.0f},{cv}): comb full {ell_r['comb_full']:>+5.2f} " f"hold {ell_r['comb_hold']:>+5.2f} nT/asset~{ell_n:.0f} " f"corr(Elliott,Donch-matched)={corr}") # ---------- BANDA D'ANCORA 4h ------------------------------------------------------- say("\n[6] Banda d'ancora 4h (offset 0/1/2/3h) — cella 4h migliore IN-SAMPLE") r4 = [r for r in rows if r["tf"] == "4h"] best4 = max(r4, key=lambda r: r["comb_ins"]) say(f" cella 4h in-sample: k={best4['k']:.0f} var={best4['var']} " f"(inS {best4['comb_ins']}, full {best4['comb_full']}, hold {best4['comb_hold']})") anchor = {} for off in (0, 1, 2, 3): parts = {} for a in al.CERTIFIED: df = get_4h_anchor(a, off) sim = simulate(df, best4["k"], best4["var"], f"{a}@+{off}h") ev = al.eval_weights(df, sim["pos"]) parts[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(parts, axis=1, join="inner").fillna(0.0) comb = al._to_daily(0.5 * J["BTC"] + 0.5 * J["ETH"]) anchor[off] = dict(full=round(al._sh(comb), 3), hold=round(al._sh(comb[comb.index >= al.HOLDOUT]), 3)) say(f" offset +{off}h: comb full {anchor[off]['full']:>+5.2f} " f"hold {anchor[off]['hold']:>+5.2f}") fulls = [v["full"] for v in anchor.values()] holds = [v["hold"] for v in anchor.values()] say(f" banda full [{min(fulls):+.2f},{max(fulls):+.2f}] " f"hold [{min(holds):+.2f},{max(holds):+.2f}]") # ---------- CLAIM DISCRIMINANTE ----------------------------------------------------- say("\n[7] Claim discriminante: 'mai fuori dal canale = correttivo' (fwd 10 barre " "allineato al conteggio, ATR-norm, null permutato B=2000)") disc = {} for tf in TFS: disc[tf] = discriminant_test(tf, ck) for a in al.CERTIFIED: d = disc[tf][a] co, im = d["corrective"], d["impulse"] say(f" {tf:>4} {a}: corrective n={co['n']} mean={co.get('mean')} " f"(null {co.get('null_mean')}) p_low={co.get('p_low')} | " f"impulse n={im['n']} mean={im.get('mean')} " f"(null {im.get('null_mean')}) p_high={im.get('p_high')}") # ---------- TARGET 1.618 vs NULL ---------------------------------------------------- say("\n[8] Target 1.618 toccato prima dello stop: freq reale vs null vol/geometry-" "matched (B=300 bootstrap, stessa distanza %/direzione/timeout da barre casuali)") thit = {} for tf in TFS: thit[tf] = target_hit_vs_null(tf, ck, cv) for a in al.CERTIFIED: t = thit[tf][a] if t.get("real") is None: say(f" {tf:>4} {a}: n={t['n']} (troppo pochi trade)") else: say(f" {tf:>4} {a}: n={t['n']} real={t['real']} null={t['null_mean']} " f"p_high={t['p_high']} p_low={t['p_low']}") # ---------- LENS OTTIMISTA ---------------------------------------------------------- say("\n[9] Lens fill-al-livello (OTTIMISTA, dichiarata) vs close-exec — cella scelta, " f"tf={ctf}") lc = lens_compare(ctf, ck, cv) for a in al.CERTIFIED: d = lc[a] if d.get("n", 0) == 0: say(f" {a}: 0 trade") else: say(f" {a}: n={d['n']} expectancy/trade close={d['exp_close']:+.4f} " f"level={d['exp_level']:+.4f} tot close={d['tot_close']:+.2%} " f"level={d['tot_level']:+.2%}") say(f"\n(tempo totale {time.time()-t0:.0f}s)") # ---------- SALVATAGGI --------------------------------------------------------------- donch_save = [{kk: vv for kk, vv in d.items() if kk not in ("comb", "freq_gap")} for d in donch_rows] payload = dict(rows=rows, family=al._clean(fam), donchian=al._clean(donch_save), anchor_4h=anchor, discriminant=disc, target_hit=thit, lens=lc, chosen=dict(tf=ctf, k=ck, variant=cv)) with open(f"{SCRATCH}/ell_c_results.json", "w") as f: json.dump(al._clean(payload), f, default=str, indent=1) with open(f"{SCRATCH}/ell_c_report.txt", "w") as f: f.write("\n".join(report)) say(f"salvato: {SCRATCH}/ell_c_results.json + ell_c_report.txt") if __name__ == "__main__": main()