#!/usr/bin/env python """r0823_tp01_ls.py — §42 TP01-LS: la gamba SHORT del trend a 1d paga il proprio costo? DOMANDA. TP01 e' long-flat PER SCELTA: nei ribassi va a 0x invece di girarsi short. Il progetto lo descrive come ASSICURAZIONE (taglia il DD ~6x contro buy&hold, contributo hold-out negativo nel 99,1% delle ancore, LOO 26/07). L'obiettivo dichiarato e' il DRIFT, non lo Sharpe. La direzione mai presa al TF di deploy e' il TSMOM LONG-SHORT a 1d: il progetto ha UNA riga ("trend 1h L/S ~1.0" fra gli edge deboli), misurata a 1h e senza l'harness onesto di oggi. Decidibile in quattro pezzi: (a) la short aggiunge DRIFT? (b) sopravvive al GATE ISO-VOLATILITA'? (se aggiunge drift solo alzando l'esposizione e' leva travestita — null del de-levering al contrario, 5 occorrenze nel progetto) (c) NON e' semplicemente il beta del 2022? (si toglie il 2022 e si rifa') (d) resta ADDS quando la si giudica contro TP01 stesso (`marginal_vs_tp01`)? ATTESA A PRIORI DEL COORDINATORE, dichiarata PRIMA di misurare: il crypto ha drift positivo strutturale e la short di trend perde quasi sempre fuori dal 2022 -> il guadagno sara' UN ANNO SOLO (2022), lo stesso schema di SOL (22/08: 24/24 ancore positive, un anno solo). Se e' cosi' la conclusione non e' "la short funziona" ma "il 2022 e' un evento nel campione". MECCANISMO CONGELATO — UN SOLO GRADO DI LIBERTA'. Si parte dal TP01 di PRODUZIONE (src/strategies/trend_portfolio.TrendPortfolio + CANONICAL) e si cambia SOLO il pavimento della direzione TSMOM, da 0 (clip long-flat) a `floor`. Tutto il resto e' importato, non riscritto: orizzonti 30/90/180g, vol_win 30g, target_vol 20%, leva cap 2x, fee 0.05%/lato, 50/50 BTC+ETH, lo shift causale e la contabilita' fee stanno in `TrendPortfolio.net_returns` EREDITATA. La direzione del blend a 3 orizzonti vive in {-1,-1/3,+1/3,+1}, quindi il pavimento e' un vero knob discreto: floor= 0 -> {0,0,1/3,1} = CANONICO (long-flat) floor=-1/3 -> {-1/3,-1/3,1/3,1} = short "timida" (solo 1/3 di size al ribasso pieno) floor=-2/3 -> {-2/3,-1/3,1/3,1} floor=-1 -> {-1,-1/3,1/3,1} = LONG-SHORT pieno (== long_only=False di produzione) GRIGLIA DICHIARATA PRIMA DI GUARDARE: 4 celle (i 4 floor) su UN solo TF (1d, quello di deploy). Nessun altro parametro riaperto -> 4 trial, contati nel deflated-Sharpe. La griglia di 24 ANCORE non e' una griglia di parametri (il DSR non la conta): si giudica con `anchor_luck_band` / `anchor_luck_delta`, come prescritto dal gate del 26/07. REPLICA BIT-EXACT PRIMA DI OGNI DELTA (tre vie, richiesta max|diff| == 0): (1) floor=0 -> target_series == TrendPortfolio(**CANONICAL).target_series (2) floor=-1 -> target_series == TrendPortfolio(long_only=False).target_series (cioe' la mia unica riga nuova coincide con la produzione a ENTRAMBI gli estremi: il codice davvero nuovo sono solo i due floor intermedi) (3) serie 50/50 a floor=0 vs src.portfolio.sleeves._tp01_returns() (lo sleeve del libro) Onesta': altlib (`marginal_vs_tp01`, `deflated_sharpe`, `causality_ok`, `implausible_sharpe`, `eval_weights_smallcap`, `anchor_luck_*`) importato, non riscritto. Controlli positivi OBBLIGATORI su ogni gate (un gate che non segnala puo' essere semplicemente rotto). Vincoli: NESSUN file toccato fuori da questo script, nessun ordine inviato, sola lettura su data/raw. Runtime ~1-2 min su 2 core. """ from __future__ import annotations import sys import time from dataclasses import dataclass from functools import lru_cache from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) sys.path.insert(0, str(ROOT)) import altlib as al # noqa: E402 from src.strategies.trend_portfolio import ( # noqa: E402 CANONICAL, TrendPortfolio, realized_vol, simple_returns, tsmom_blend, ) ASSETS = ("BTC", "ETH") HOLDOUT = al.HOLDOUT FLOORS = (0.0, -1.0 / 3.0, -2.0 / 3.0, -1.0) # LA GRIGLIA, dichiarata prima FLOOR_LBL = {0.0: "0 (canonico LF)", -1.0 / 3.0: "-1/3", -2.0 / 3.0: "-2/3", -1.0: "-1 (LS pieno)"} OFFSETS = tuple(range(24)) # banda d'ancora giornaliera W_TP01_BOOK, W_SKH_BOOK = 0.75, 0.25 # src/live/book.py CAPITAL = 635.0 # conto reale dichiarato nel briefing # =========================================================================== # IL MECCANISMO — TP01 di produzione con UN grado di liberta' in piu' # =========================================================================== @dataclass class TPFloor(TrendPortfolio): """TP01 con pavimento della direzione parametrico. UNICA riga diversa dalla produzione: `np.clip(direction, self.floor, None)` al posto di `np.clip(direction, 0, None)`. Tutto il resto (vol realizzata, blend TSMOM, scaling, cap di leva, net_returns con lo shift causale e le fee sul turnover) e' ereditato/importato dal modulo di produzione.""" floor: float = 0.0 def target_series(self, df: pd.DataFrame) -> np.ndarray: c = df["close"].values.astype(float) bpd = self._bpd(df) bpy = bpd * 365.25 r = simple_returns(c) vol = realized_vol(r, self.vol_win_days * bpd, bpy) horizons = tuple(d * bpd for d in self.horizons_days) direction = np.clip(tsmom_blend(c, horizons), self.floor, None) # <-- IL GRADO DI LIBERTA' scal = np.where((vol > 0) & np.isfinite(vol), self.target_vol / vol, 0.0) tgt = np.clip(direction * scal, -self.leverage, self.leverage) tgt[~np.isfinite(tgt)] = 0.0 return tgt _CFG = {k: v for k, v in CANONICAL.items() if k != "long_only"} def tp_floor(floor: float) -> TPFloor: return TPFloor(long_only=False, floor=floor, **_CFG) # =========================================================================== # DATI — resample 1h -> 24h con ANCORA h (stessa convenzione di resample_tf) # =========================================================================== @lru_cache(maxsize=8) def _h1(asset: str) -> pd.DataFrame: return al.get(asset, "1h") @lru_cache(maxsize=128) def bars(asset: str, off: int) -> pd.DataFrame: g = _h1(asset).copy() idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) idx.name = "dt" g.index = idx out = g.resample("24h", offset=pd.Timedelta(hours=off), label="left", closed="left").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"]] @lru_cache(maxsize=512) def sleeve(floor: float, off: int = 0, fee_side: float | None = None) -> pd.Series: """Serie 50/50 BTC+ETH dei rendimenti NETTI per barra giornaliera, ancora `off`. Usa `TrendPortfolio.net_returns` EREDITATA -> shift causale e fee di produzione.""" tp = tp_floor(floor) if fee_side is not None: tp.fee_side = fee_side ser = {} for a in ASSETS: df = bars(a, off) net, ts = tp.net_returns(df) ser[a] = pd.Series(net, index=pd.DatetimeIndex(pd.to_datetime(ts.values, utc=True))) J = pd.concat(ser, axis=1, join="inner").fillna(0.0) return pd.Series(0.5 * J[ASSETS[0]].values + 0.5 * J[ASSETS[1]].values, index=J.index) # =========================================================================== # METRICHE # =========================================================================== def sh(s: pd.Series) -> float: return al._sh(s) def dd(s: pd.Series) -> float: return al._dd_ret(s) def drift(s: pd.Series) -> float: """Drift annualizzato in LOG (la grandezza su cui vivono i muri di capitale; il carry si misura sul log, mai sulla media aritmetica delle differenze — regola 23/08).""" r = np.asarray(s.dropna().values, float) return float(np.mean(np.log1p(np.clip(r, -0.99, None))) * 365.25) if len(r) else 0.0 def cagr(s: pd.Series) -> float: r = np.asarray(s.dropna().values, float) if len(r) < 2: return 0.0 return float(np.exp(np.mean(np.log1p(np.clip(r, -0.99, None))) * 365.25) - 1.0) def vol(s: pd.Series) -> float: r = np.asarray(s.dropna().values, float) return float(np.std(r) * np.sqrt(365.25)) if len(r) else 0.0 def hold(s: pd.Series) -> pd.Series: return s[s.index >= HOLDOUT] def no2022(s: pd.Series) -> pd.Series: return s[s.index.year != 2022] def row(s: pd.Series) -> dict: return dict(sh=sh(s), sh_hold=sh(hold(s)), cagr=cagr(s), drift=drift(s), vol=vol(s), dd=dd(s), n=len(s)) def turnover_year(floor: float, off: int = 0) -> float: tot = 0.0 for a in ASSETS: df = bars(a, off) tgt = tp_floor(floor).target_series(df) held = np.zeros(len(tgt)); held[1:] = tgt[:-1] turn = np.abs(np.diff(held, prepend=0.0)).sum() yrs = (df["datetime"].iloc[-1] - df["datetime"].iloc[0]).total_seconds() / 86400 / 365.25 tot += turn / yrs return tot / len(ASSETS) def time_in_market(floor: float, off: int = 0) -> tuple[float, float]: lo, sho = [], [] for a in ASSETS: tgt = tp_floor(floor).target_series(bars(a, off)) lo.append(float(np.mean(tgt > 0))); sho.append(float(np.mean(tgt < 0))) return float(np.mean(lo)), float(np.mean(sho)) def hr(t=""): print("\n" + "=" * 78) if t: print(t) print("=" * 78) # =========================================================================== def main() -> None: t_start = time.time() print(__doc__.split("\n\n")[0]) print(f"\nAvvio {pd.Timestamp.now('UTC')} | griglia dichiarata: {len(FLOORS)} floor x 1 TF (1d) = " f"{len(FLOORS)} trial | ancore: {len(OFFSETS)} (NON trial: gate d'ancora)") # ------------------------------------------------------------------ 1. REPLICA BIT-EXACT hr("1. REPLICA BIT-EXACT DEL CANONICO (prima di qualunque delta)") ok_all = True canon = TrendPortfolio(**CANONICAL) ls_prod = TrendPortfolio(**{**CANONICAL, "long_only": False}) for a in ASSETS: df = bars(a, 0) d0 = float(np.max(np.abs(tp_floor(0.0).target_series(df) - canon.target_series(df)))) d1 = float(np.max(np.abs(tp_floor(-1.0).target_series(df) - ls_prod.target_series(df)))) print(f" {a}: max|floor=0 - CANONICAL(long_only=True) | = {d0:.1e} " f"max|floor=-1 - CANONICAL(long_only=False)| = {d1:.1e}") ok_all &= (d0 == 0.0) and (d1 == 0.0) from src.portfolio.sleeves import _tp01_returns prod = _tp01_returns() mine = sleeve(0.0, 0) JJ = pd.concat({"p": prod, "m": mine}, axis=1, join="inner").dropna() d2 = float(np.max(np.abs(JJ["p"] - JJ["m"]))) print(f" sleeve 50/50 vs src.portfolio.sleeves._tp01_returns(): n={len(JJ)} max|diff| = {d2:.1e}") ok_all &= (d2 == 0.0) print(f" ==> REPLICA {'OK (3/3 a 0.0)' if ok_all else 'FALLITA — NON procedere'}") if not ok_all: sys.exit(1) # ------------------------------------------------------------------ 2. CONTROLLI POSITIVI hr("2. CONTROLLI POSITIVI DEI GATE (un gate che non segnala puo' essere rotto)") def tgt_fn(floor): return lambda df: tp_floor(floor).target_series(df) def leaky(df): # guarda il futuro: DEVE fallire causality_ok t = tp_floor(-1.0).target_series(df) out = np.zeros_like(t); out[:-1] = t[1:] return out c_ok = al.causality_ok(tgt_fn(-1.0), tf="1d") c_bad = al.causality_ok(leaky, tf="1d") print(f" causality_ok(floor=-1) -> ok={c_ok['ok']} max_tail_diff={c_ok['max_tail_diff']}") print(f" causality_ok(variante LEAKY) -> ok={c_bad['ok']} max_tail_diff={c_bad['max_tail_diff']}" f" [controllo positivo: deve essere False]") ctrl1 = c_ok["ok"] and not c_bad["ok"] rng = np.random.default_rng(20260823) idx = pd.date_range("2020-01-01", periods=900, freq="1D", tz="UTC") fake_good = pd.Series(np.abs(rng.normal(0.002, 0.0005, 900)), index=idx) # 0 perdite fake_noise = pd.Series(rng.normal(0.0004, 0.012, 900), index=idx) ip_g = al.implausible_sharpe(fake_good) ip_n = al.implausible_sharpe(fake_noise) print(f" implausible_sharpe(serie senza perdite) -> implausible={ip_g['implausible']} " f"[deve essere True] (rumore Sh {ip_n['sharpe']:.2f} -> {ip_n['implausible']} [deve essere False])") ctrl2 = ip_g["implausible"] and not ip_n["implausible"] base0 = sleeve(0.0, 0) lev13 = base0 * 1.3 # LEVA PURA: il gate iso-vol deve bocciarla k_lev = vol(base0) / vol(lev13) print(f" GATE ISO-VOL, controllo positivo su LEVA PURA (canonico x1.3):") print(f" grezzo dCAGR = {cagr(lev13) - cagr(base0):+.4f} (positivo: sembra 'piu' drift')") print(f" iso-vol dCAGR = {cagr(lev13 * k_lev) - cagr(base0):+.6f} [deve essere ~0 -> e' leva]") ctrl3 = abs(cagr(lev13 * k_lev) - cagr(base0)) < 1e-6 ad_self = al.anchor_luck_delta(lambda o: sleeve(0.0, o), lambda o: sleeve(0.0, o), OFFSETS, metric=sh) print(f" anchor_luck_delta(A vs A) -> mediana {ad_self['median_paired']:+.6f}, " f"positivi {ad_self['n_positive']}/{ad_self['n_anchors']} [deve essere 0 e 0/24]") ctrl4 = (ad_self["median_paired"] == 0.0) and (ad_self["n_positive"] == 0) print(f" ==> CONTROLLI POSITIVI: {sum([ctrl1, ctrl2, ctrl3, ctrl4])}/4 superati") # ------------------------------------------------------------------ 3. TABELLA PRINCIPALE hr("3. TABELLA PRINCIPALE (ancora canonica h=0, fee 0.10% RT, 50/50 BTC+ETH)") print(f" {'floor':<16} {'ShFULL':>7} {'ShHOLD':>7} {'CAGR':>8} {'drift':>8} {'vol':>7} " f"{'maxDD':>7} {'turn/a':>7} {'%long':>6} {'%short':>7}") rows = {} for f in FLOORS: s = sleeve(f, 0) r = row(s); rows[f] = r tl, ts = time_in_market(f) print(f" {FLOOR_LBL[f]:<16} {r['sh']:>7.3f} {r['sh_hold']:>7.3f} {r['cagr']:>7.2%} " f"{r['drift']:>7.2%} {r['vol']:>6.2%} {r['dd']:>6.2%} {turnover_year(f):>7.2f} " f"{tl:>5.1%} {ts:>6.1%}") print(f" (n barre = {rows[0.0]['n']}, dal {base0.index[0].date()} al {base0.index[-1].date()})") # ------------------------------------------------------------------ 4. GATE ISO-VOL hr("4. GATE ISO-VOLATILITA' (null del de-levering AL CONTRARIO: la variante ALZA la vol)") print(" Convenzione: si DE-LEVA la variante fino alla vol del canonico (k<1, gratis: nessuna") print(" leva richiesta). Se a pari vol il canonico regge, la short era leva travestita.") print(f" {'floor':<16} {'k de-lev':>9} {'CAGR iso':>9} {'dCAGR iso':>10} {'dSharpe':>9} " f"{'maxDD iso':>10} {'esito':>10}") isovol = {} for f in FLOORS: s = sleeve(f, 0) k = vol(base0) / vol(s) if vol(s) > 0 else 1.0 si = s * k dc = cagr(si) - cagr(base0) isovol[f] = dict(k=k, dcagr=dc, dsh=sh(s) - sh(base0), dd=dd(si)) v = "—" if f == 0.0 else ("PASS" if dc > 0 else "FAIL") print(f" {FLOOR_LBL[f]:<16} {k:>9.4f} {cagr(si):>8.2%} {dc:>+10.2%} " f"{sh(s) - sh(base0):>+9.3f} {dd(si):>9.2%} {v:>10}") # ------------------------------------------------------------------ 5. BANDA D'ANCORA hr("5. BANDA D'ANCORA — 24 ore, MEDIANA DELLE DIFFERENZE APPAIATE (variante - canonico)") print(" (un Delta misurato a una sola ancora eredita la fortuna di quell'ancora: regola 26/07)") band = {} for f in FLOORS[1:]: d_sh = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=sh) d_ho = al.anchor_luck_delta(lambda o, f=f: hold(sleeve(f, o)), lambda o: hold(sleeve(0.0, o)), OFFSETS, metric=sh) d_dr = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=drift) d_dd = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=dd) band[f] = dict(sh=d_sh, ho=d_ho, dr=d_dr, dd=d_dd) print(f" floor {FLOOR_LBL[f]:<14} dShFULL {d_sh['median_paired']:+.3f} " f"[{d_sh['lo']:+.3f},{d_sh['hi']:+.3f}] {d_sh['n_positive']:>2}/24 | " f"dShHOLD {d_ho['median_paired']:+.3f} {d_ho['n_positive']:>2}/24 | " f"dDRIFT {d_dr['median_paired']:+.2%} {d_dr['n_positive']:>2}/24 | " f"dMAXDD {d_dd['median_paired']:+.2%} {d_dd['n_positive']:>2}/24") print("\n ⚠️ L'UNICA metrica che va nel verso della short e' l'HOLD-OUT, e all'ancora") print(" CANONICA non si vede: e' il caso speculare della lezione 26/07 (li' l'ancora") print(" canonica INVENTAVA un danno, qui NASCONDE un vantaggio).") print(f" {'floor':<16} {'dShHOLD @h=0':>13} {'mediana 24':>11} {'p10':>8} {'p90':>8}") for f in FLOORS[1:]: v = np.array(list(band[f]['ho']['per_offset'].values()), float) print(f" {FLOOR_LBL[f]:<16} {band[f]['ho']['per_offset'][0]:>+13.3f} " f"{np.median(v):>+11.3f} {np.percentile(v, 10):>+8.3f} {np.percentile(v, 90):>+8.3f}") b_can = al.anchor_luck_band(lambda o: sleeve(0.0, o), OFFSETS, metric=sh) b_ls = al.anchor_luck_band(lambda o: sleeve(-1.0, o), OFFSETS, metric=sh) print(f" livelli: canonico ShFULL canonica {b_can['canonical']:.3f} (pctl {b_can['canonical_pctl']:.2f}) " f"mediana {b_can['median']:.3f} | LS pieno canonica {b_ls['canonical']:.3f} " f"(pctl {b_ls['canonical_pctl']:.2f}) mediana {b_ls['median']:.3f}") # ------------------------------------------------------------------ 6. PER ANNO / SENZA 2022 hr("6. SCOMPOSIZIONE PER ANNO — e la domanda che decide: E SENZA IL 2022?") years = sorted(set(base0.index.year)) print(f" Sharpe per anno (ancora h=0)") print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS)) for y in years: line = f" {y:<6}" for f in FLOORS: s = sleeve(f, 0); sy = s[s.index.year == y] line += f"{sh(sy):>18.2f}" print(line) print(f"\n Rendimento per anno") print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS)) for y in years: line = f" {y:<6}" for f in FLOORS: s = sleeve(f, 0); sy = s[s.index.year == y] line += f"{float(np.prod(1 + sy.values) - 1):>17.1%} " print(line) print(f"\n dSharpe ANNUALE della variante vs canonico (positivo = la short paga QUELL'anno)") print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS[1:])) npos = {f: 0 for f in FLOORS[1:]} for y in years: line = f" {y:<6}" s0 = sleeve(0.0, 0) for f in FLOORS[1:]: s = sleeve(f, 0) d = sh(s[s.index.year == y]) - sh(s0[s0.index.year == y]) npos[f] += int(d > 0) line += f"{d:>+18.2f}" print(line) print(" anni positivi: " + " ".join(f"{FLOOR_LBL[f]} {npos[f]}/{len(years)}" for f in FLOORS[1:])) print(f"\n --- SENZA IL 2022 (ancora h=0) ---") print(f" {'floor':<16} {'ShFULL':>7} {'dSh':>7} {'CAGR':>8} {'dCAGR':>8} {'iso-vol dCAGR':>14}") b0n = no2022(base0) for f in FLOORS: sn = no2022(sleeve(f, 0)) k = vol(b0n) / vol(sn) if vol(sn) > 0 else 1.0 print(f" {FLOOR_LBL[f]:<16} {sh(sn):>7.3f} {sh(sn) - sh(b0n):>+7.3f} {cagr(sn):>7.2%} " f"{cagr(sn) - cagr(b0n):>+7.2%} {cagr(sn * k) - cagr(b0n):>+14.2%}") print(f"\n --- BANDA D'ANCORA SENZA IL 2022 (mediana differenze appaiate) ---") for f in FLOORS[1:]: d_sh = al.anchor_luck_delta(lambda o, f=f: no2022(sleeve(f, o)), lambda o: no2022(sleeve(0.0, o)), OFFSETS, metric=sh) d_dr = al.anchor_luck_delta(lambda o, f=f: no2022(sleeve(f, o)), lambda o: no2022(sleeve(0.0, o)), OFFSETS, metric=drift) print(f" floor {FLOOR_LBL[f]:<14} dShFULL {d_sh['median_paired']:+.3f} " f"[{d_sh['lo']:+.3f},{d_sh['hi']:+.3f}] {d_sh['n_positive']:>2}/24 | " f"dDRIFT {d_dr['median_paired']:+.2%} {d_dr['n_positive']:>2}/24") # ------------------------------------------------------------------ 7. PROTEZIONE DD hr("7. LA PROTEZIONE — il valore DICHIARATO di TP01 e' il taglio del DD: la short lo rompe?") bh = {} for a in ASSETS: df = bars(a, 0) bh[a] = pd.Series(simple_returns(df["close"].values.astype(float)), index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))) JB = pd.concat(bh, axis=1, join="inner").fillna(0.0) bh50 = pd.Series(0.5 * JB[ASSETS[0]].values + 0.5 * JB[ASSETS[1]].values, index=JB.index) bh50 = bh50.reindex(base0.index).fillna(0.0) print(f" buy&hold 50/50: Sharpe {sh(bh50):.2f} CAGR {cagr(bh50):.1%} maxDD {dd(bh50):.1%}") print(f" Criterio (B) di edge_watch: in un anno con DD del buy&hold > 10%, il DD dello sleeve") print(f" deve restare sotto il 75% di quello. Rapporto DD_sleeve/DD_B&H per anno di sinistro:") print(f" {'anno':<6} {'DD B&H':>8}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS)) fails = {f: 0 for f in FLOORS}; sinistri = 0 for y in years: by = bh50[bh50.index.year == y] if dd(by) <= 0.10: continue sinistri += 1 line = f" {y:<6} {dd(by):>7.1%}" for f in FLOORS: s = sleeve(f, 0); r_ = dd(s[s.index.year == y]) / dd(by) fails[f] += int(r_ >= 0.75) line += f"{r_:>17.2f}x" if False else f"{r_:>18.2f}" print(line) print(f" anni di sinistro: {sinistri}. Violazioni del criterio (rapporto >= 0.75): " + " ".join(f"{FLOOR_LBL[f]} {fails[f]}" for f in FLOORS)) print(f" maxDD full: " + " ".join(f"{FLOOR_LBL[f]} {rows[f]['dd']:.2%}" for f in FLOORS)) # ------------------------------------------------------------------ 8. COSTI hr("8. COSTI — la short raddoppia il turnover? curva fee 0-20 bps/lato") print(f" {'fee/lato':>9} {'%RT':>7}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS)) for bps in (0, 2.5, 5, 10, 15, 20): line = f" {bps:>8.1f}b {bps * 2 / 100:>6.2f}%" for f in FLOORS: line += f"{sh(sleeve(f, 0, bps / 10000.0)):>18.3f}" print(line) print(" turnover/anno (media asset): " + " ".join(f"{FLOOR_LBL[f]} {turnover_year(f):.2f}" for f in FLOORS)) # ------------------------------------------------------------------ 9. LIBRO 75/25 + SKH01 hr("9. IMPATTO SUL LIBRO 75/25 CON SKH01 (che e' GIA' long/short: ridondanza?)") from src.portfolio.sleeves import _skyhook_returns skh = _skyhook_returns() skh.index = pd.DatetimeIndex(skh.index).tz_convert("UTC") if skh.index.tz else \ pd.DatetimeIndex(skh.index).tz_localize("UTC") def book(floor: float, off: int = 0) -> pd.Series: s = sleeve(floor, off).copy() s.index = s.index.floor("1D") # ancora h -> data di calendario (1 barra per data) J = pd.concat({"t": s, "k": skh}, axis=1, join="inner").dropna() return pd.Series(W_TP01_BOOK * J["t"].values + W_SKH_BOOK * J["k"].values, index=J.index) print(f" {'floor':<16} {'ShFULL':>7} {'ShHOLD':>7} {'CAGR':>8} {'maxDD':>7}") for f in FLOORS: b = book(f) print(f" {FLOOR_LBL[f]:<16} {sh(b):>7.3f} {sh(hold(b)):>7.3f} {cagr(b):>7.2%} {dd(b):>6.2%}") print(f" banda d'ancora del LIBRO (24 ancore su TP01, SKH01 all'ancora canonica):") for f in FLOORS[1:]: d_b = al.anchor_luck_delta(lambda o, f=f: book(f, o), lambda o: book(0.0, o), OFFSETS, metric=sh) d_c = al.anchor_luck_delta(lambda o, f=f: book(f, o), lambda o: book(0.0, o), OFFSETS, metric=drift) print(f" floor {FLOOR_LBL[f]:<14} dSh libro {d_b['median_paired']:+.3f} " f"[{d_b['lo']:+.3f},{d_b['hi']:+.3f}] {d_b['n_positive']:>2}/24 | " f"dDRIFT {d_c['median_paired']:+.2%} {d_c['n_positive']:>2}/24") print(f"\n RIDONDANZA con SKH01 — la 'gamba short' isolata = variante meno canonico:") for f in FLOORS[1:]: leg = (sleeve(f, 0) - sleeve(0.0, 0)) leg.index = leg.index.floor("1D") JL = pd.concat({"l": leg, "k": skh}, axis=1, join="inner").dropna() print(f" floor {FLOOR_LBL[f]:<14} corr(gamba short, SKH01) = {JL['l'].corr(JL['k']):+.3f} " f"Sharpe della gamba isolata {sh(JL['l']):+.3f} drift {drift(JL['l']):+.2%}") # ------------------------------------------------------------------ 10. MARGINALE + DSR hr("10. MARGINALE vs TP01 STESSO + DEFLATED SHARPE (4 trial dichiarati)") cd0 = al.candidate_daily(tgt_fn(0.0), tf="1d") bl = al.tp01_baseline_daily() JC = pd.concat({"a": cd0, "b": bl}, axis=1, join="inner").dropna() print(f" 4a replica: candidate_daily(floor=0) vs tp01_baseline_daily() max|diff| " f"{float(np.max(np.abs(JC['a'] - JC['b']))):.1e} (n={len(JC)})") # trial set del DSR: si prende dal percorso CODIFICATO (select_cell_insample), non a mano. # ⚠️ errore mio catturato: la prima stesura costruiva la lista dei trial da solo. Due percorsi # per lo stesso numero e' esattamente cio' che il gate codificato serve a evitare -> qui li # confronto invece di sostituirne uno in silenzio. sel = al.select_cell_insample(lambda tf, floor: tgt_fn(floor), [dict(floor=f) for f in FLOORS], ("1d",)) all_sh = sel["all_full_sharpe"] mine_sh = [round(sh(sleeve(f, 0)), 3) for f in FLOORS] print(f" trial set DSR da select_cell_insample: {all_sh}") print(f" stesso conto per via indipendente (sleeve): {mine_sh} -> " f"{'COINCIDONO' if all_sh == mine_sh else 'DIVERGONO (fermarsi)'}") for f in FLOORS[1:]: cd = al.candidate_daily(tgt_fn(f), tf="1d") m = al.marginal_vs_tp01(cd) d, sr0 = al.deflated_sharpe(sh(sleeve(f, 0)), all_sh, sleeve(f, 0)) ip = al.implausible_sharpe(sleeve(f, 0)) print(f"\n floor {FLOOR_LBL[f]}: marginal_verdict = {m['marginal_verdict']}") print(f" corr->TP01 {m['corr_full']} beta {m['beta_to_tp01']} alpha/a {m['alpha_ann']:+.4f} " f"resid Sharpe {m['resid_sharpe_full']}") print(f" blend w25 uplift full {m['blends']['w25']['uplift_full']:+.3f} / hold " f"{m['blends']['w25']['uplift_hold']:+.3f} w50 full " f"{m['blends']['w50']['uplift_full']:+.3f} / hold {m['blends']['w50']['uplift_hold']:+.3f}") print(f" robust_oos={m['robust_oos']} multicut={m['multicut_persistent']} " f"insample_edge={m['has_insample_edge']} (Sh IS {m['cand_insample_sharpe']}) " f"is_hedge={m['is_hedge']}") print(f" HEDGE vs ALPHA: corr(Sharpe-TP01 annuo, uplift annuo) = {m['hedge_yearly_corr']} " f"uplift quando TP01 SALE {m['uplift_tp01_up']} / quando TP01 SCENDE {m['uplift_tp01_down']}") print(f" multicut per anno: {m['multicut_uplift']}") print(f" deflated-Sharpe {d:.3f} (sr0 nullo {sr0:.3f}, N={len(all_sh)}) " f"implausible={ip['implausible']}") print(f"\n --- GATE SELECTION-ON-HOLDOUT (codificato 29/06): quale floor si sceglie ") print(f" guardando SOLO il pre-2025? Se e' il canonico, il guadagno hold-out NON e'") print(f" selezionabile e sceglierlo sarebbe selezione-sull'hold-out. ---") print(f" {'floor':<16} {'Sh IN-SAMPLE (<2025)':>21} {'Sh HOLD-OUT':>12} {'Sh FULL':>9}") for f in FLOORS: s_ = sleeve(f, 0) print(f" {FLOOR_LBL[f]:<16} {sh(s_[s_.index < HOLDOUT]):>21.3f} {sh(hold(s_)):>12.3f} {sh(s_):>9.3f}") fb = sel["chosen"]["params"]["floor"] fh = max(FLOORS, key=lambda f: sh(hold(sleeve(f, 0)))) print(f" cella scelta AL BUIO (max Sharpe in-sample) = floor {fb:+.4f} -> " f"{'IL CANONICO' if abs(fb) < 1e-9 else 'NON il canonico'}") print(f" cella scelta SULL'HOLD-OUT (max Sharpe hold-out) = floor {fh:+.4f} -> " f"{'diversa: firma di selezione-sull-hold-out' if abs(fh - fb) > 1e-9 else 'la stessa'}") print(f"\n ⚠️ NOTA ONESTA sul DSR: con 4 celle quasi identiche sr0 e' minuscolo -> il DSR e'") print(f" alto per COSTRUZIONE (dipende dalla VARIANZA della griglia, lezione §10 del 22/08).") print(f" Qui non e' il gate che decide: e' un contorno. Il gate che decide e' l'iso-vol +") print(f" la banda d'ancora + il per-anno.") # ------------------------------------------------------------------ 11. ESEGUIBILITA' hr(f"11. ESEGUIBILITA' A ${CAPITAL:.0f} (min_order $5) e vincolo di NETTING del libro") for f in (0.0, -1.0): for a in ASSETS: df = bars(a, 0) sc = al.eval_weights_smallcap(df, tp_floor(f).target_series(df), capital=CAPITAL, min_order=5.0) print(f" floor {FLOOR_LBL[f]:<14} {a}: Sharpe modellato {sc['modeled']['sharpe']:.3f} -> " f"reale {sc['realistic']['sharpe']:.3f} haircut {sc['sharpe_haircut']:+.3f} " f"trade eseguiti {sc['n_executed_trades']}") src_book = (ROOT / "src" / "live" / "book.py").read_text() clipped = "max(tp_frac, 0.0)" in src_book print(f"\n VERIFICA SUL CODICE CHE GIRA: src/live/book.py contiene 'max(tp_frac, 0.0)' -> {clipped}") print(f" Cioe' l'esecutore del libro CLIPPA a zero la frazione di TP01: il long-flat non e' solo") print(f" in CANONICAL, e' cablato anche in `book_net_target`. Attivare la short richiederebbe") print(f" di toccare src/live/book.py, cioe' CODICE su un percorso con soldi veri; e TP01 e SKH01") print(f" si nettano su UNA sola posizione per strumento, quindi la short di TP01 andrebbe a") print(f" sommarsi/cancellarsi con quella di SKH01 invece di essere una gamba a se'.") hr("12. VERDETTO E COSA NON E' STATO MISURATO") dsh_ls = band[-1.0]["sh"]["median_paired"]; ddr_ls = band[-1.0]["dr"]["median_paired"] print(f" floor=-1 (LS pieno): dShFULL appaiato {dsh_ls:+.3f} ({band[-1.0]['sh']['n_positive']}/24), " f"dDRIFT {ddr_ls:+.2%} ({band[-1.0]['dr']['n_positive']}/24), " f"dMAXDD {band[-1.0]['dd']['median_paired']:+.2%} ({band[-1.0]['dd']['n_positive']}/24)") print(f" iso-vol dCAGR {isovol[-1.0]['dcagr']:+.2%} | anni positivi {npos[-1.0]}/{len(years)} | " f"cella scelta al buio = il canonico") print(f"\n >>> VERDETTO: SCARTATO. La gamba short NON paga il proprio costo: toglie drift") print(f" ({ddr_ls:+.2%}/anno a 0/24 ancore), FALLISCE il gate iso-vol, peggiora il maxDD") print(f" ({band[-1.0]['dd']['median_paired']:+.2%} a 24/24) e non e' selezionabile in-sample.") print(f" L'attesa a priori del coordinatore (il guadagno e' il 2022) e' CONFERMATA e") print(f" superata: il 2022 c'e' ({npos[-1.0]}/8 anni positivi, dSharpe 2022 +1.84) ma NON") print(f" compensa gli altri sei nemmeno con se stesso dentro il campione.") print(f" E NON e' morte-per-fee: a fee ZERO il canonico fa {sh(sleeve(0.0, 0, 0.0)):.3f} contro " f"{sh(sleeve(-1.0, 0, 0.0)):.3f} -> l'edge lordo non c'e'.") print(f"\n COSA NON HO MISURATO (dichiarato, non nascosto):") print(f" (a) FUNDING. Nessun backtest del progetto lo modella (§35, 22/08) e qui nemmeno.") print(f" E' l'unica voce che potrebbe muovere il numero nel verso della short: una") print(f" posizione corta INCASSA il funding quando il tasso e' positivo. MA il") print(f" meccanismo misurato il 22/08 e' che esposizione e funding sono POSITIVAMENTE") print(f" correlati, e un trend-short entra quando il mercato scende, cioe' quando il") print(f" funding e' tipicamente NEGATIVO (e li' e' lo short a pagare). Argomento, non") print(f" misura: si chiude con la macchineria di r0822d_funding.py, non a parole.") print(f" Ordine di grandezza in gioco (~1-2%/anno) contro un drift perso di " f"{-ddr_ls:.2%}/anno.") print(f" (b) TF diversi dal 1d. Riaprire il TF riapre la FAMIGLIA (griglia 4 -> 4xN celle)") print(f" e imporrebbe study_family_honest; qui il TF e' congelato a quello di deploy.") print(f" (c) La banda d'ancora del LIBRO tiene SKH01 alla sua ancora canonica (le 23 fasi") print(f" di SKH01 non sono state incrociate): il delta di libro e' appaiato solo sulle") print(f" 24 ancore di TP01. Non cambia il segno (0/24 su tutte e tre le celle).") print(f" (d) La lente WICK accoppiata (25/07): tutto qui e' close-only. Irrilevante per il") print(f" verdetto (non c'e' barriera di rovina in gioco), rilevante se qualcuno") print(f" riproponesse la variante sul canale funded.") print(f"\n Tempo totale {time.time() - t_start:.1f}s") if __name__ == "__main__": main()