"""r0717_wk_cond — WK-COND: esposizione weekend (ven->lun) CONDIZIONATA da gate causali. DOMANDA: esiste un gate causale (noto al venerdi' H_in) che rende la finestra weekend un edge, dove il drift nudo non lo e' (day-of-week nudo gia' morto, SEA ~0)? METODO (onesto, vettoriale su 1h, harness altlib): * Finestra: ven H_in -> lun H_out, H_in in {12,16,20}, H_out in {0,8,16} (9 timing). * Convenzione eval_weights: target[i] deciso con dati <= close[i], tenuto nella barra i+1. Le barre 1h sono open-labeled -> il decision time della barra i e' datetime[i]+1h. Barra d'ingresso = quella il cui CLOSE cade a ven H_in:00 UTC (decisione con dati fino a ven H_in:00 esatte, nessun leak). Ultima barra target = close lun H_out-1h -> l' esposizione reale copre esattamente [ven H_in, lun H_out). * Gate (tutti causali, calcolati al close della barra d'ingresso): A TSMOM multi-orizzonte 30/90/180g (LO long-se-up / SO short-se-down / LS entrambi) B FOLLOW del venerdi' (open ven 00:00 -> close H_in), + FADE come controllo C ritorno settimana (lun 00:00 -> ven H_in), follow + fade D regime vol: percentile ESPANDENTE causale della RV30g (LOW: pct<=0.30 / HIGH: >=0.70) E prevday level: close H_in vs high/low del GIOVEDI' -> breakout-follow L/S F NAKED long (riferimento: il drift nudo NON e' un edge) * Ogni cella = (gate-variante x timing); valutata su BTC ed ETH e come 50/50 daily (convenzione candidate_daily). TUTTI i trial contano per il deflated_sharpe. * Selezione cella SOLO in-sample (Sharpe 50/50 daily pre-2025), poi hold-out + banda dei 9 timing (lezione anchor timing-luck 2026-07-02). * Per ogni gate: study_marginal vs TP01 (corr/earns_slot/is_hedge), fee sweep, per-anno, day_boundary_robust (il segnale E' calendario-dipendente), causality_ok. TRIAL: 3+2+2+2+1+1 = 11 varianti x 9 timing = 99 combo (x2 asset = 198 stream valutati). VINCOLI: nessun file esistente modificato; niente rete (dati locali certificati). """ 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 H_INS = (12, 16, 20) # venerdi' UTC, ora di ingresso H_OUTS = (0, 8, 16) # lunedi' UTC, ora di uscita (0 = mezzanotte dom->lun) ASSETS = tuple(al.CERTIFIED) HOLDOUT = al.HOLDOUT # =========================================================================== # SEGNALI CAUSALI — memo per-df (chiave su timestamp: regge shift di calendario # di day_boundary_robust e troncamenti di causality_ok senza contaminazioni). # Tutti i valori all'indice i usano SOLO dati <= close[i]. # =========================================================================== _SIG_MEMO: dict = {} def get_signals(df: pd.DataFrame) -> dict: key = (int(df["timestamp"].iloc[0]), int(df["timestamp"].iloc[-1]), len(df)) if key in _SIG_MEMO: return _SIG_MEMO[key] c = df["close"].values.astype(float) o = df["open"].values.astype(float) h = df["high"].values.astype(float) lo_ = df["low"].values.astype(float) dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) close_dt = dt + pd.Timedelta(hours=1) # decision time della barra i # (A) TSMOM multi-orizzonte 30/90/180g su close 1h (voto di segni, come TP01) votes = np.zeros(len(c)) for hd in (30, 90, 180): hb = hd * 24 s = np.full(len(c), np.nan) if len(c) > hb: s[hb:] = np.sign(c[hb:] / c[:-hb] - 1.0) votes += np.nan_to_num(s) tsmom = np.sign(votes) # -1 / 0 / +1 # (B) direzione intraday del giorno corrente: close[i] vs OPEN del giorno UTC day = np.asarray(dt.normalize()) day_open = pd.Series(o).groupby(day).transform("first").values intraday = np.sign(c / day_open - 1.0) # (C) direzione della settimana: close[i] vs OPEN del lunedi' 00:00 della settimana week = np.asarray((dt - pd.to_timedelta(dt.dayofweek, unit="D")).normalize()) week_open = pd.Series(o).groupby(week).transform("first").values week_dir = np.sign(c / week_open - 1.0) # (D) percentile espandente CAUSALE della RV 30g (rank del valore corrente vs storia) r = al.simple_returns(c) rv = al.realized_vol(r, 30 * 24, 24 * 365.25) volpct = pd.Series(rv).expanding(min_periods=30 * 24).rank(pct=True).values # (E) breakout vs high/low del giorno PRECEDENTE completo (per ven = giovedi') dhi = pd.Series(h).groupby(day).max() dlo = pd.Series(lo_).groupby(day).min() prev_hi = dhi.shift(1).reindex(day).values prev_lo = dlo.shift(1).reindex(day).values brk = np.where(c > prev_hi, 1.0, np.where(c < prev_lo, -1.0, 0.0)) brk[~np.isfinite(prev_hi)] = 0.0 S = dict(tsmom=tsmom, intraday=intraday, week=week_dir, volpct=volpct, brk=brk, cdw=close_dt.dayofweek.values, chr=close_dt.hour.values) if len(_SIG_MEMO) > 48: _SIG_MEMO.clear() _SIG_MEMO[key] = S return S # =========================================================================== # GATE (variante -> direzione al momento dell'ingresso, da segnali causali) # =========================================================================== GATE_FAMILIES: dict[str, dict] = { "A-TSMOM": { "LO": lambda S: np.clip(S["tsmom"], 0.0, 1.0), # long solo se trend up "SO": lambda S: np.clip(S["tsmom"], -1.0, 0.0), # short solo se trend down "LS": lambda S: S["tsmom"], # entrambi }, "B-FRIDIR": { "FOLLOW": lambda S: S["intraday"], "FADE": lambda S: -S["intraday"], }, "C-WEEKDIR": { "FOLLOW": lambda S: S["week"], "FADE": lambda S: -S["week"], }, "D-VOLREG": { "LOW30": lambda S: np.where(S["volpct"] <= 0.30, 1.0, 0.0), "HIGH70": lambda S: np.where(S["volpct"] >= 0.70, 1.0, 0.0), }, "E-PREVDAY": { "BRKFOLLOW": lambda S: S["brk"], }, "F-NAKED": { "LONG": lambda S: np.ones(len(S["tsmom"])), }, } def make_target(fam: str, var: str, hin: int, hout: int): """target_fn(df) -> posizione per barra. Direzione congelata alla barra d'ingresso (close = ven H_in) e tenuta su tutta la finestra [ven H_in, lun H_out).""" gate_fn = GATE_FAMILIES[fam][var] def target_fn(df: pd.DataFrame) -> np.ndarray: S = get_signals(df) dirs = np.nan_to_num(np.asarray(gate_fn(S), float)) dw, hr = S["cdw"], S["chr"] # barre target = quelle il cui CLOSE cade in [ven H_in, lun H_out) in_win = ((dw == 4) & (hr >= hin)) | (dw == 5) | (dw == 6) if hout > 0: in_win |= (dw == 0) & (hr < hout) entry = (dw == 4) & (hr == hin) sig = pd.Series(np.where(entry, dirs, np.nan)).ffill().values return np.where(in_win, np.nan_to_num(sig), 0.0) target_fn.__name__ = f"wk_{fam}_{var}_F{hin}_M{hout}" return target_fn # =========================================================================== # VALUTAZIONE DI UNA CELLA (per-asset 1h + 50/50 daily, split in/hold) # =========================================================================== def combined_daily(target_fn, fee_side: float = al.FEE_SIDE) -> pd.Series: series = {} for a in ASSETS: df = al.get(a, "1h") ev = al.eval_weights(df, target_fn(df), fee_side=fee_side) series[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(series, axis=1, join="inner").fillna(0.0) return al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]]) def run_cell(fam: str, var: str, hin: int, hout: int) -> dict: tfn = make_target(fam, var, hin, hout) per_asset = {} for a in ASSETS: df = al.get(a, "1h") ev = al.eval_weights(df, tfn(df)) per_asset[a] = dict(full_sh=ev["full"]["sharpe"], hold_sh=ev["holdout"].get("sharpe", 0.0), dd=ev["full"]["maxdd"], tim=ev["time_in_market"], turnover=ev["turnover_per_year"], yearly=ev["yearly"]) D = combined_daily(tfn) ins, hold = D[D.index < HOLDOUT], D[D.index >= HOLDOUT] eqf = np.cumprod(1 + D.values) pk = np.maximum.accumulate(eqf) return dict(fam=fam, var=var, hin=hin, hout=hout, target_fn=tfn, daily=D, per_asset=per_asset, is_sh=round(al._sh(ins), 3), full_sh=round(al._sh(D), 3), hold_sh=round(al._sh(hold), 3), dd_comb=round(float(np.max((pk - eqf) / pk)), 4), ret_hold=round(float(np.prod(1 + hold.values) - 1), 4)) def fee_sweep_cell(cell: dict) -> dict: out = {} for f in al.FEE_SWEEP: D = combined_daily(cell["target_fn"], fee_side=f) ins, hold = D[D.index < HOLDOUT], D[D.index >= HOLDOUT] out[f"{2 * f * 100:.2f}%RT"] = dict(is_sh=round(al._sh(ins), 2), full=round(al._sh(D), 2), hold=round(al._sh(hold), 2)) return out def yearly_comb(cell: dict) -> dict: D = cell["daily"] out = {} for y, g in D.groupby(D.index.year): out[int(y)] = dict(ret=round(float(np.prod(1 + g.values) - 1), 4), sh=round(al._sh(g), 2)) return out def gate_selectivity(cell: dict) -> dict: """Quota di venerdi' in cui il gate apre una posizione (in-sample / hold-out), BTC.""" df = al.get("BTC", "1h") S = get_signals(df) dirs = np.nan_to_num(np.asarray(GATE_FAMILIES[cell["fam"]][cell["var"]](S), float)) entry = (S["cdw"] == 4) & (S["chr"] == cell["hin"]) dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) m_ins = entry & np.asarray(dt < HOLDOUT) m_hold = entry & np.asarray(dt >= HOLDOUT) def frac(m): return round(float(np.mean(dirs[m] != 0)), 3) if m.sum() else None def longshare(m): on = dirs[m][dirs[m] != 0] return round(float(np.mean(on > 0)), 3) if len(on) else None return dict(n_fridays=int(entry.sum()), on_ins=frac(m_ins), on_hold=frac(m_hold), long_share_ins=longshare(m_ins)) # =========================================================================== # MAIN # =========================================================================== def main() -> None: print("=" * 100) print("WK-COND — weekend ven->lun condizionato da gate causali | 1h BTC/ETH certificati") print(f"timing: H_in {H_INS} x H_out {H_OUTS} | fee {2 * al.FEE_SIDE * 100:.2f}%RT | HOLDOUT {HOLDOUT.date()}") print("=" * 100) # ---- sweep completo ----------------------------------------------------------- rows = [] for fam, variants in GATE_FAMILIES.items(): for var in variants: for hin in H_INS: for hout in H_OUTS: rows.append(run_cell(fam, var, hin, hout)) all_full = [r["full_sh"] for r in rows] n_combo = len(rows) print(f"\nTRIAL: {n_combo} combo (gate-variante x timing) x {len(ASSETS)} asset = " f"{n_combo * len(ASSETS)} stream valutati. DSR calcolato vs tutte le {n_combo} combo 50/50.") # ---- tabella compatta di famiglia (mediane per variante) ---------------------- print("\n--- PANORAMICA per variante (mediana e banda sui 9 timing, 50/50 daily) ---") hdr = f"{'variante':<22}{'IS med [min,max]':>24}{'FULL med':>10}{'HOLD med [min,max]':>26}" print(hdr) for fam, variants in GATE_FAMILIES.items(): for var in variants: sub = [r for r in rows if r["fam"] == fam and r["var"] == var] iss = [r["is_sh"] for r in sub]; hs = [r["hold_sh"] for r in sub] fs = [r["full_sh"] for r in sub] print(f"{fam + '/' + var:<22}" f"{np.median(iss):>8.2f} [{min(iss):+.2f},{max(iss):+.2f}]" f"{np.median(fs):>10.2f}" f"{np.median(hs):>12.2f} [{min(hs):+.2f},{max(hs):+.2f}]") # ---- selezione IN-SAMPLE-ONLY per gate + verifica cella ----------------------- print("\n" + "=" * 100) print("SELEZIONE IN-SAMPLE-ONLY (max Sharpe 50/50 daily pre-2025) + verifica per gate") print("=" * 100) summary = {} for fam in GATE_FAMILIES: sub = [r for r in rows if r["fam"] == fam] best = max(sub, key=lambda r: r["is_sh"]) # banda dei 9 timing per la variante scelta (lezione anchor timing-luck) band = sorted(r["hold_sh"] for r in sub if r["var"] == best["var"]) band_is = sorted(r["is_sh"] for r in sub if r["var"] == best["var"]) pctl = float(np.mean([b <= best["hold_sh"] for b in band])) dsr, sr0 = al.deflated_sharpe(al._sh(best["daily"]), all_full, best["daily"].values) name = f"{fam}/{best['var']} F{best['hin']}->M{best['hout']}" print(f"\n### {name} (cella scelta in-sample su {len(sub)} trial del gate)") print(f" 50/50: IS {best['is_sh']:+.2f} | FULL {best['full_sh']:+.2f} | " f"HOLD {best['hold_sh']:+.2f} (ret hold {best['ret_hold'] * 100:+.1f}%) | DD {best['dd_comb'] * 100:.1f}%") pa = best["per_asset"] for a in ASSETS: print(f" {a}: full {pa[a]['full_sh']:+.2f} hold {pa[a]['hold_sh']:+.2f} " f"DD {pa[a]['dd']*100:.0f}% tim {pa[a]['tim']*100:.0f}% turn/y {pa[a]['turnover']:.0f}") print(f" banda 9 timing (var {best['var']}): HOLD [{band[0]:+.2f} .. med {np.median(band):+.2f} .. {band[-1]:+.2f}] " f"(cella scelta al {pctl * 100:.0f}° pctl) | IS [{band_is[0]:+.2f}..{band_is[-1]:+.2f}]") print(f" deflated Sharpe vs {n_combo} trial: DSR={dsr:.3f} (null max atteso {sr0:.2f}) " f"{'PASS' if dsr >= 0.95 else 'FAIL'}") sel = gate_selectivity(best) print(f" selettivita' (BTC, ven {best['hin']}h): on IS {sel['on_ins']} / HOLD {sel['on_hold']} " f"(quota long IS {sel['long_share_ins']}) su {sel['n_fridays']} venerdi'") cok = al.causality_ok(best["target_fn"], tf="1h", tail=400) print(f" causality_ok: {cok['ok']} (max tail diff {cok['max_tail_diff']}, checked {cok['checked']})") print(f" fee sweep 50/50: " + " ".join( f"{k}: IS {v['is_sh']:+.2f}/FULL {v['full']:+.2f}/HOLD {v['hold']:+.2f}" for k, v in fee_sweep_cell(best).items())) yr = yearly_comb(best) print(" per-anno 50/50: " + " ".join(f"{y}:{d['ret']*100:+.1f}%({d['sh']:+.1f})" for y, d in yr.items())) summary[fam] = dict(best=best, dsr=dsr, band=band, name=name, survivor=(best["is_sh"] >= 0.5)) # ---- marginale vs TP01 + day-boundary per ogni gate a-e ----------------------- print("\n" + "=" * 100) print("MARGINALE vs TP01 (study_marginal, 1h) + day_boundary_robust — gate A..E") print("=" * 100) for fam, info in summary.items(): if fam == "F-NAKED": continue best = info["best"] sm = al.study_marginal(f"WK-{info['name']}", best["target_fn"], tf="1h") print("\n" + al.fmt_marginal(sm)) dbr = al.day_boundary_robust(best["target_fn"], tf="1h") print(f" day_boundary: {dbr['verdict']} (per_offset {dbr['per_offset']}, spread {dbr.get('spread')})") info["sm"] = sm info["dbr"] = dbr # ---- sintesi finale ------------------------------------------------------------ print("\n" + "=" * 100) print("SINTESI") print("=" * 100) for fam, info in summary.items(): b = info["best"] line = (f"{info['name']:<38} IS {b['is_sh']:+.2f} FULL {b['full_sh']:+.2f} " f"HOLD {b['hold_sh']:+.2f} DSR {info['dsr']:.2f} survivor_IS={info['survivor']}") if "sm" in info: m = info["sm"] line += (f" | marg={m['marginal_verdict']} earns_slot={m['earns_slot']} " f"corr {m['marginal'].get('corr_full')} hedge={m['marginal'].get('is_hedge')}" f" | boundary={info['dbr']['verdict']}") print(line) if __name__ == "__main__": main()