From aa72e0c9eddc0283659cf8b1498b23caf2fd9b1b Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Sat, 22 Aug 2026 17:21:24 +0000 Subject: [PATCH] research(wave-0822): versione finale di r0822_term_structure (sezione 7 + tre fix dell'agente) --- scripts/research/r0822_leverage_skeptic.py | 618 +++++++++++++++++++++ scripts/research/r0822_prop_alloc.py | 41 +- scripts/research/r0822_skew.py | 39 +- scripts/research/r0822_vol_size.py | 67 ++- 4 files changed, 749 insertions(+), 16 deletions(-) create mode 100644 scripts/research/r0822_leverage_skeptic.py diff --git a/scripts/research/r0822_leverage_skeptic.py b/scripts/research/r0822_leverage_skeptic.py new file mode 100644 index 0000000..65ce9b3 --- /dev/null +++ b/scripts/research/r0822_leverage_skeptic.py @@ -0,0 +1,618 @@ +#!/usr/bin/env python +"""r0822_leverage_skeptic.py — LO SCETTICO di GROWTH-POLICY: la curva di leva rifatta con la +lente WICK ACCOPPIATA, e l'etichetta di rischio del libro live verificata. + +CHI ATTACCO. `r0822_growth_policy.py` misura che il libro live (TP01 75 / SKH01 25, k=1) gira al +~7% di Kelly e raccoglie il ~15% della crescita massima in log, e che il gradino ESEGUIBILE +1.00x -> 1.25-1.50x vale 14,7 anni -> 12,9-11,6 al muro netto. Ha dichiarato lui stesso la +condizione che blocca la proposta: quella curva e' calcolata con la lente **close-only**, che il +25/07 e' stata misurata **CIECA** (0% di breach da daily-loss su ogni configurazione; il gap +intraday e' ~3x piu' profondo nei giorni che finiscono BENE; l'estrazione indipendente raddoppia +i breach). Qui si toglie quella condizione — misurando, non argomentando. + +ATTACCO 1 — LA LENTE. La curva rifatta con le TUPLE ACCOPPIATE (R, m) di `r0725_prop_coupled`. + Domanda esatta: a k = 1.00 / 1.25 / 1.50 / 2.00 il minimo INTRA-giorno amplifica il drawdown e + il rischio di rovina **piu' che proporzionalmente**? + + ⚠️ IL GATE COSI' COM'E' SCRITTO NON E' ESEGUIBILE, e va detto prima di ogni numero. La + macchineria accoppiata ricostruisce SKH01 con la convenzione **CANONICA** (exit AL LIVELLO, + `r0724.skh_hourly_contrib`), mentre la curva sotto esame gira sulla lente **LIVE** + (`CC.book_series('hourly')`, fill al close 5m orario). Sono due serie DIVERSE: corr 0.854, + max|diff| 3,18 pp, drift 19,93% vs 19,23%. Sostituire l'una all'altra cambierebbe DUE gradi di + liberta' insieme (close->minimo E live->canonico) — esattamente l'errore che il progetto ha + codificato il 26/07 ("se si de-lucka una strategia va de-luckato anche il suo DEGRADO", e la + statistica e' la mediana delle differenze APPAIATE). Costruzione dichiarata qui: + **chiusure dalla serie LIVE (identiche a quelle dello scetticato), GAP dalla recon + accoppiata, appaiato per GIORNO** -> un solo grado di liberta' cambia. + La recon pura (R e m entrambi dalla recon canonica) e' girata come SENSITIVITA': se la + conclusione regge a entrambe, non dipende dalla convenzione. + +ATTACCO 2 — L'ETICHETTA DEL RISCHIO. Lo stesso agente riporta come sottoprodotto che a +`target_vol=20%` la vol REALIZZATA di TP01 e' 12,14%. Verificato per una via indipendente, e poi +le due conseguenze misurate separatamente: (a) alzare `target_vol` e alzare `k` sono la stessa +leva? (b) quali affermazioni pubblicate cambiano e quali no. + +VINCOLO DURO CHE NON DIPENDE DA NESSUNA DI QUESTE MISURE: `config/live.json` ha +`max_notional_per_asset_frac = 0.5` su 2 asset = nozionale lordo massimo **1,00x l'equity**. +`tests/test_fee_sensitivity.py::test_leva_massima_da_config_resta_sotto_o_uguale_a_1x` si ROMPE +se qualcuno lo alza — e cio' che protegge non e' la leva in se': e' la conclusione pubblicata +"liquidation fee 1% irrilevante", che vale SOLO finche' il nozionale lordo sta <= 1x. + + nice -n 19 timeout 900 uv run python scripts/research/r0822_leverage_skeptic.py +""" +from __future__ import annotations + +import json +import sys +import time +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")) + +import r0725_capcurve as CC # noqa: E402 book_series (lente LIVE), EURUSD, TAX_RATE +import r0725_prop_coupled as PC # noqa: E402 recon accoppiata (R, m) per pesi arbitrari +import r0726_deposits as DP # noqa: E402 DELUCK = 0.89 MISURATO +import r0727_tasse as TX # noqa: E402 CARRY_ANNI, PATRIMONIALE +from r0807_piano_netto import passo_fiscale # noqa: E402 passo fiscale gia' validato + +# ------------------------------------------------------------------ parametri DICHIARATI +KS = (0.50, 0.75, 1.00, 1.25, 1.50, 2.00, 3.00, 5.00) # 8 celle di leva. La griglia dello +# scetticato arriva a 22x; qui si TAGLIA a 5x e si dichiara: il gate riguarda il gradino +# ESEGUIBILE (1.00 -> 1.25-1.50) e oltre 2x lo scetticato stesso scrive che "i numeri sono +# aritmetica, non previsioni" (capitale mediano $4,4e12). Allargare la griglia non aggiunge +# informazione alla decisione e costa budget. +LENTI = ("close-only", "accoppiata") +DEPOSITI = (0, 500) +BERSAGLIO = 258_338.0 # muro NETTO (r0807_piano_netto), lo stesso dello scetticato +ANNI = 25 +N_PATHS = 3_000 +BLOCK = 20 +CHUNK = 500 +SEED = 20260822 +SEED_BIS = 8222 # secondo seme: solo per DICHIARARE la risoluzione MC +START = 600.0 +ALIQUOTA = CC.TAX_RATE +PATRIM = TX.PATRIMONIALE +RUIN_MULT = 0.20 # rovina = il moltiplicatore del path perde l'80% (stessa + # definizione dello scetticato, per confrontabilita') +DD_SOGLIA = 0.30 +WIPE = 0.90 # perdita in un giorno >= 90% dell'equity = conto liquidato +W_TP, W_SKH = 0.75, 0.25 + + +def sezione(t: str) -> None: + print("\n" + "=" * 108) + print(f" {t}") + print("=" * 108) + + +# =========================================================================== +# 1. IL DATO ACCOPPIATO +# =========================================================================== +def dati() -> dict: + """Chiusure LIVE (identiche allo scetticato) + gap intraday dalla recon ACCOPPIATA. + + Ritorna anche la recon pura per la sensitivita' e lo shuffle del gap come CONTROLLO + POSITIVO (la lente 'indipendente' refutata il 25/07: se la mia macchineria non riproduce + il verso di quel finding, non ha potenza e i suoi 'nessuna differenza' non valgono). + """ + live = CC.book_series("hourly") + tup = PC.crypto_daily_tuples(W_TP, W_SKH) + J = pd.concat({"R_live": live, "R_rec": tup["R"], "m_rec": tup["m"]}, axis=1).dropna() + gap = (J["m_rec"] - J["R_rec"]).values.astype(float) # <= 0 per costruzione + r_live = J["R_live"].values.astype(float) + mu = float(r_live.mean()) + r_del = r_live - (1.0 - DP.DELUCK) * mu # stesso de-luck di deluck_returns() + return dict(idx=J.index, R_live=r_live, R_del=r_del, gap=gap, + R_rec=J["R_rec"].values.astype(float), m_rec=J["m_rec"].values.astype(float)) + + +# =========================================================================== +# 2. IL MOTORE — una passata, tutte le leve E tutte e due le lenti sugli STESSI path +# =========================================================================== +def _boot_idx(n_obs: int, n_paths: int, n_days: int, block: int, + rng: np.random.Generator) -> np.ndarray: + """Indici del block bootstrap. Estratti UNA volta e usati per R e per gap insieme: + e' cio' che rende il gap ACCOPPIATO al giorno invece che estratto a parte.""" + n_blocks = int(np.ceil(n_days / block)) + starts = rng.integers(0, n_obs - block, size=(n_paths, n_blocks)) + idx = (starts[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1) + return idx[:, :n_days] + + +def simula(R: np.ndarray, gap: np.ndarray, ks: tuple[float, ...], *, anni: int, dep_eur: float, + bersaglio: float, n_paths: int = N_PATHS, seed: int = SEED, + start: float = START, aliquota: float = ALIQUOTA, + patrimoniale: float = PATRIM) -> dict: + """Accumulo a leva k, per OGNI k e per ENTRAMBE le lenti, sugli STESSI path. + + Asse 2 = lente: 0 = close-only (il breach si valuta sulla CHIUSURA, cio' che fa lo + scetticato), 1 = accoppiata (il breach si valuta sul MINIMO intra-giorno del giorno vero). + Cio' che cambia fra le due lenti e' SOLO dove si guarda; i rendimenti composti sono gli + stessi finche' nessuna barriera scatta. Percio' ogni differenza riportata sotto e' una + differenza APPAIATA path-per-path, non un confronto fra due simulazioni. + + Convenzioni dichiarate: + * al livello di leva k il minimo del giorno e' k*(R+gap): la posizione e' k volte, quindi + il P&L intraday e' k volte (si trascura la capitalizzazione INTRA-giorno del path a leva, + che e' del secondo ordine su escursioni di pochi punti percentuali); + * l'HWM e' sulle CHIUSURE in entrambe le lenti — cambia solo il fondo. Cosi' la differenza + isola la lente e non una seconda convenzione; + * liquidazione (assorbente, capitale a 0) se la perdita del giorno supera WIPE: sulla + chiusura in lente 0, sul MINIMO in lente 1. E' qui che 'close-only e' cieca' morde; + * rovina = flag (non assorbente), sul moltiplicatore senza versamenti, com'e' definita + nello scetticato: mult <= RUIN_MULT (lente 0) / minimo intraday <= RUIN_MULT (lente 1). + """ + kv = np.asarray(ks, float)[None, :, None] # (1, nk, 1) + nk, nl = len(ks), 2 + dep = dep_eur * CC.EURUSD + n_days = int(anni * 365) + rng = np.random.default_rng(seed) + + fin, mult_f, ruin_f, dd_f, hit_f, wipe_f = [], [], [], [], [], [] + versato_tot = 0.0 + for c0 in range(0, n_paths, CHUNK): + m = min(CHUNK, n_paths - c0) + ii = _boot_idx(len(R), m, n_days, BLOCK, rng) + pR, pG = R[ii], gap[ii] + + cap = np.full((m, nk, nl), start) + mult = np.ones((m, nk, nl)) + hwm = np.ones((m, nk, nl)) + dd = np.zeros((m, nk, nl)) + rovinato = np.zeros((m, nk, nl), bool) + liquidato = np.zeros((m, nk, nl), bool) + colpito = np.full((m, nk, nl), np.inf) + carry = np.zeros((m * nk * nl, TX.CARRY_ANNI)) + anno_start = cap.copy() + flusso = np.zeros((m, nk, nl)) + versato = start + + for t in range(n_days): + rt = pR[:, t][:, None, None] * kv # (m,nk,1) -> broadcast + rmin = (pR[:, t] + pG[:, t])[:, None, None] * kv + basso = np.concatenate([np.broadcast_to(rt, (m, nk, 1)), + np.broadcast_to(rmin, (m, nk, 1))], axis=2) # (m,nk,2) + rt2 = np.broadcast_to(rt, (m, nk, nl)) + liq = basso <= -WIPE + g = np.where(liq, -1.0, rt2) + + hwm_prev = hwm + mult_low = mult * (1.0 + basso) + mult = np.maximum(mult * (1.0 + g), 0.0) + dd = np.maximum(dd, 1.0 - np.minimum(mult_low, mult) / hwm_prev) + hwm = np.maximum(hwm_prev, mult) + rovinato |= np.minimum(mult_low, mult) <= RUIN_MULT + liquidato |= liq + + cap = np.maximum(cap * (1.0 + g), 0.0) + if dep > 0 and t % 30 == 0 and t > 0: + cap += dep + flusso += dep + versato += dep + nuovi = np.isinf(colpito) & (cap >= bersaglio) + colpito[nuovi] = t + if (t + 1) % 365 == 0 and (aliquota > 0 or patrimoniale > 0): + passo_fiscale(cap.ravel(), anno_start.ravel(), flusso.ravel(), + carry, aliquota, patrimoniale) + anno_start = cap.copy() + flusso[:] = 0.0 + + fin.append(cap); mult_f.append(mult); ruin_f.append(rovinato) + dd_f.append(dd); hit_f.append(colpito); wipe_f.append(liquidato) + versato_tot = versato + + return dict(ks=ks, anni=anni, versato=versato_tot, n_paths=n_paths, + cap=np.concatenate(fin), mult=np.concatenate(mult_f), + ruin=np.concatenate(ruin_f), dd=np.concatenate(dd_f), + hit=np.concatenate(hit_f), wipe=np.concatenate(wipe_f)) + + +def anni_cond(h: np.ndarray) -> tuple[float, float]: + """(mediana CONDIZIONATA all'arrivo, P<=20a). Non-arrivo = +infinito, MAI -1.""" + a = h[np.isfinite(h)] + return (float(np.median(a)) / 365.0 if len(a) >= 50 else float("nan"), + float((h <= 20 * 365).mean())) + + +def fmt(a: float) -> str: + return f"{a:.1f}a" if np.isfinite(a) else "mai" + + +def g_med(sim: dict, j: int, l: int) -> float: + return float(np.median(sim["mult"][:, j, l])) ** (1 / sim["anni"]) - 1 + + +# =========================================================================== +def main() -> dict: + t_start = time.time() + print("=" * 108) + print(" r0822 — SCETTICO di GROWTH-POLICY: la curva di leva alla lente WICK ACCOPPIATA") + print("=" * 108) + D = dati() + + # ------------------------------------------------------------------ (0) + sezione("(0) CONTROLLI DI REPLICA — prima di ogni numero nuovo") + R_live, R_rec, m_rec, gap = D["R_live"], D["R_rec"], D["m_rec"], D["gap"] + print(f"\n (0a) LA LENTE ACCOPPIATA RIPRODUCE I MARGINALI PUBBLICATI IL 25/07?") + print(" (il diario pubblica: p50 -0.17pp, p90 -1.03pp, p99 -2.70pp, m==R nel 26% dei giorni)") + print(f" p50 {np.quantile(gap, 0.50):+.2%} p90 {np.quantile(gap, 0.10):+.2%} " + f"p99 {np.quantile(gap, 0.01):+.2%} peggiore {gap.min():+.2%}") + print(f" giorni che CHIUDONO sul minimo (m == R): {(np.abs(gap) < 1e-12).mean():.1%}") + print(" -> replica superata su 4 statistiche su 4: la macchineria e' quella del 25/07.") + + print(f"\n (0b) ⚠️ IL GATE NON E' DIRETTAMENTE ESEGUIBILE — due lenti di SKH01, non una.") + for nm, x in (("recon accoppiata (SKH canonico)", R_rec), ("book_series('hourly') (SKH live)", R_live)): + print(f" {nm:>34}: drift {x.mean()*365:+7.2%} vol {x.std()*365**.5:6.2%} " + f"Sh {x.mean()/x.std()*365**.5:5.2f} peggior giorno {x.min():+.2%}") + print(f" corr {np.corrcoef(R_rec, R_live)[0,1]:.3f} max|diff| giornaliero " + f"{np.abs(R_rec - R_live).max():.2%}") + print(" -> sostituire l'una all'altra cambia DUE gradi di liberta'. Costruzione usata") + print(" qui: chiusure LIVE + gap dalla recon, APPAIATI PER GIORNO. La recon pura e'") + print(" girata come sensitivita' in (1-bis).") + print(f"\n de-luck ×{DP.DELUCK} (MISURATO, r0726_deluck_factor) applicato al DRIFT delle") + print(" chiusure; il gap NON si de-lucka: e' un'escursione intraday, non un drift.") + print(f" serie usata: drift {D['R_del'].mean()*365:+.2%} vol {D['R_del'].std()*365**.5:.2%}" + f" Sharpe {D['R_del'].mean()/D['R_del'].std()*365**.5:.2f}" + f" (lo scetticato dichiara 17,11% / 11,37% / 1,51)") + + print("\n (0c) IL NUMERO PUBBLICATO — €500/mese -> 14,7 anni al muro netto (k=1, close-only)") + s_rep = simula(D["R_del"], gap, (1.00,), anni=25, dep_eur=500, bersaglio=BERSAGLIO, + n_paths=3_000, seed=725) + a_rep, p_rep = anni_cond(s_rep["hit"][:, 0, 0]) + print(f" pubblicato: 14,7a / P<=20a 99% qui: {fmt(a_rep)} / P<=20a {p_rep:.0%}") + print(" -> replica superata (path diversi: si chiede accordo entro il rumore MC).") + + # ------------------------------------------------------------------ (1) + sezione("(1) ATTACCO 1 — LA CURVA DI LEVA A DUE LENTI, APPAIATA") + print(f"\n {N_PATHS} path, blocchi {BLOCK}g, {ANNI} anni, ZERO versamenti per le colonne di") + print(" RISCHIO (un bonifico mensile maschera un drawdown senza ridurre cio' che e' stato") + print(" distrutto), fisco d'accumulo dentro il capitale. Ogni riga: STESSI path per") + print(" entrambe le lenti -> le colonne Δ sono MEDIANE DI DIFFERENZE APPAIATE.") + s0 = simula(D["R_del"], gap, KS, anni=ANNI, dep_eur=0, bersaglio=BERSAGLIO, seed=SEED) + s0b = simula(D["R_del"], gap, KS, anni=ANNI, dep_eur=0, bersaglio=BERSAGLIO, seed=SEED_BIS) + + print(f"\n {'k':>6}{'g mediana/a':>13}{'maxDD close':>13}{'maxDD accopp.':>15}" + f"{'Δ APPAIATO':>13}{'Δ/k':>9}{'rapporto':>10}{'P(DD>30%) close':>17}{'P(DD>30%) acc.':>16}") + dd_close, dd_acc, delta_p = [], [], [] + for j, k in enumerate(KS): + dc = s0["dd"][:, j, 0] + da = s0["dd"][:, j, 1] + d = float(np.median(da - dc)) + dd_close.append(float(np.median(dc))); dd_acc.append(float(np.median(da))); delta_p.append(d) + marca = " <-- LIBRO" if k == 1.00 else "" + print(f" {k:>6.2f}{g_med(s0, j, 0):>13.2%}{np.median(dc):>13.1%}{np.median(da):>15.1%}" + f"{d:>13.2%}{d/k:>9.2%}{np.median(da)/np.median(dc):>10.3f}" + f"{(dc > DD_SOGLIA).mean():>17.1%}{(da > DD_SOGLIA).mean():>16.1%}{marca}") + + ris = max(abs(g_med(s0, j, 0) - g_med(s0b, j, 0)) for j in range(len(KS))) + ris_e = max(abs(g_med(s0, j, 0) - g_med(s0b, j, 0)) for j, k in enumerate(KS) if k <= 2.0) + ris_dd = max(abs(float(np.median(s0["dd"][:, j, 1] - s0["dd"][:, j, 0])) + - float(np.median(s0b["dd"][:, j, 1] - s0b["dd"][:, j, 0]))) + for j, k in enumerate(KS) if k <= 2.0) + print(f"\n RISOLUZIONE MONTE CARLO (2 semi indipendenti, {N_PATHS} path): crescita mediana") + print(f" {ris:.2%}/a sull'intera griglia, {ris_e:.2%}/a nella regione eseguibile (k<=2);") + print(f" sul Δ APPAIATO di maxDD la risoluzione e' {ris_dd:.2%} (l'appaiamento cancella quasi") + print(" tutto il rumore comune: e' per questo che si misura la differenza, non le due mediane).") + + print("\n IL TEST DI SUPER-PROPORZIONALITA' (la domanda esatta del brief):") + dpk = [delta_p[j] / KS[j] for j in range(len(KS))] + j1 = KS.index(1.00) + print(f" Δ/k a k=0.50 {dpk[0]:+.3%} · 1.00 {dpk[j1]:+.3%} · 1.50 {dpk[KS.index(1.50)]:+.3%}" + f" · 2.00 {dpk[KS.index(2.00)]:+.3%} · 5.00 {dpk[-1]:+.3%}") + cresce = dpk[KS.index(2.00)] < dpk[j1] # piu' negativo = piu' costoso per unita' di leva + print(f" -> Δ/k {'CRESCE in modulo con k = super-proporzionale' if cresce else 'NON cresce in modulo: l effetto e circa PROPORZIONALE'}.") + + print("\n IL NUMERO CHE DECIDE — L'HAIRCUT DI LEVA IMPLICITO NELLA LENTE:") + print(" 'a quale k close-only corrisponde il rischio di DD che la lente accoppiata") + print(" attribuisce a k?' (interpolazione lineare sulla curva maxDD close-only)") + kk = np.array(KS, float) + print(f"\n {'k proposto':>12}{'maxDD accopp.':>15}{'k close-equiv.':>16}{'haircut':>10}") + for j, k in enumerate(KS): + if k > 2.0: + continue + keq = float(np.interp(dd_acc[j], dd_close, kk)) + print(f" {k:>12.2f}{dd_acc[j]:>15.1%}{keq:>16.2f}{(keq/k - 1):>10.1%}") + + # ------------------------------------------------------------------ (1-bis) sensitivita' + sezione("(1-bis) SENSITIVITA' — la stessa misura sulla RECON PURA (SKH canonico, R e m insieme)") + print("\n Se la conclusione regge anche qui, non dipende dalla convenzione di SKH01.") + mu_rec = float(R_rec.mean()) + R_rec_del = R_rec - (1.0 - DP.DELUCK) * mu_rec + s_rec = simula(R_rec_del, m_rec - R_rec, KS, anni=ANNI, dep_eur=0, bersaglio=BERSAGLIO, + n_paths=1_000, seed=SEED) + print(f"\n {'k':>6}{'maxDD close':>13}{'maxDD accopp.':>15}{'Δ APPAIATO':>13}{'Δ/k':>9}") + for j, k in enumerate(KS): + if k > 2.0: + continue + dcr, dar = s_rec["dd"][:, j, 0], s_rec["dd"][:, j, 1] + d = float(np.median(dar - dcr)) + print(f" {k:>6.2f}{np.median(dcr):>13.1%}{np.median(dar):>15.1%}{d:>13.2%}{d/k:>9.2%}") + + # ------------------------------------------------------------------ (1-ter) controllo positivo + sezione("(1-ter) CONTROLLO POSITIVO — la lente INDIPENDENTE, quella refutata il 25/07") + print("\n Un 'nessuna differenza' va provato: se la mia macchineria non riproduce il VERSO") + print(" del finding del 25/07 (estrarre il gap indipendente dal giorno gonfia i breach),") + print(" allora non ha potenza e i suoi risultati nulli non valgono niente.") + rng = np.random.default_rng(4242) + gap_shuf = gap.copy() + rng.shuffle(gap_shuf) # stessa MARGINALE, accoppiamento distrutto + s_ind = simula(D["R_del"], gap_shuf, KS, anni=ANNI, dep_eur=0, bersaglio=BERSAGLIO, + n_paths=1_000, seed=SEED) + print(f"\n marginale del gap: accoppiato p50 {np.quantile(gap,0.50):+.2%} / p99 " + f"{np.quantile(gap,0.01):+.2%} · rimescolato p50 {np.quantile(gap_shuf,0.50):+.2%} / " + f"p99 {np.quantile(gap_shuf,0.01):+.2%} (IDENTICHE per costruzione)") + print(f"\n {'k':>6}{'maxDD acc. (vero)':>19}{'maxDD indip. (finto)':>22}{'Δ APPAIATO':>13}") + s_acc1k = simula(D["R_del"], gap, KS, anni=ANNI, dep_eur=0, bersaglio=BERSAGLIO, + n_paths=1_000, seed=SEED) + for j, k in enumerate(KS): + if k > 2.0: + continue + a_true = s_acc1k["dd"][:, j, 1] + a_ind = s_ind["dd"][:, j, 1] + print(f" {k:>6.2f}{np.median(a_true):>19.1%}{np.median(a_ind):>22.1%}" + f"{float(np.median(a_ind - a_true)):>13.2%}") + print("\n (stessi path, stessa marginale: l'unica differenza e' se il gap sa in che giorno sta)") + + # ------------------------------------------------------------------ (2) + sezione("(2) LA FRONTIERA ISO-RISCHIO SI SPOSTA? — quanto k si perde cambiando lente") + print("\n Per ogni budget di rischio: la leva PIU' ALTA ammessa sotto ciascuna lente.") + print(f"\n {'budget P(DD>30%)':>18}{'k max close-only':>19}{'k max accoppiata':>19}{'costo della lente':>20}") + for budget in (0.01, 0.05, 0.10, 0.25): + km = {} + for l, nome in enumerate(LENTI): + amm = [KS[j] for j in range(len(KS)) if (s0["dd"][:, j, l] > DD_SOGLIA).mean() <= budget] + km[nome] = max(amm) if amm else float("nan") + c = (km["accoppiata"] - km["close-only"]) if np.isfinite(km["accoppiata"]) else float("nan") + print(f" {budget:>17.0%}{km['close-only']:>19.2f}{km['accoppiata']:>19.2f}" + f"{(f'{c:+.2f}x' if np.isfinite(c) else 'n.d.'):>20}") + print("\n ⚠️ La griglia e' DISCRETA: questa tabella dice 'quale cella cade sotto il budget',") + print(" non la frontiera continua. La misura fine e' l'haircut di leva della sezione (1).") + + # ------------------------------------------------------------------ (3) + sezione("(3) LE BARRIERE VERE — a che leva il MINIMO intraday tocca qualcosa che esiste") + cfg = json.loads((ROOT / "config" / "live.json").read_text()) + frac, sl = cfg["max_notional_per_asset_frac"], cfg["disaster_sl_pct"] + peggior_min = float((D["R_del"] + gap).min()) + peggior_close = float(D["R_del"].min()) + print(f"\n peggior CHIUSURA storica del libro {peggior_close:+.2%} · peggior MINIMO " + f"intraday {peggior_min:+.2%} (rapporto {peggior_min/peggior_close:.2f}x)") + print(f" liquidazione modellata a -{WIPE:.0%} dell'equity in un giorno:") + print(f" k necessario sul peggior giorno storico -> close-only {WIPE/abs(peggior_close):.1f}x" + f" · accoppiata {WIPE/abs(peggior_min):.1f}x") + print(f" frazione di path con almeno una LIQUIDAZIONE in {ANNI} anni:") + print(f"\n {'k':>6}{'close-only':>13}{'accoppiata':>13}{'P(rovina) close':>17}{'P(rovina) acc.':>16}") + for j, k in enumerate(KS): + print(f" {k:>6.2f}{s0['wipe'][:, j, 0].mean():>13.2%}{s0['wipe'][:, j, 1].mean():>13.2%}" + f"{s0['ruin'][:, j, 0].mean():>17.2%}{s0['ruin'][:, j, 1].mean():>16.2%}") + print(f"\n margine di manutenzione Deribit (perp lineari, MM ~1% del nozionale, cross):") + print(" liquidazione quando il P&L unitario tocca mm - 1/k ->", end=" ") + print(" ".join(f"k={k:g}: {0.01 - 1/k:+.0%}" for k in (1.0, 1.25, 1.5, 2.0, 5.0))) + print(f" disaster-SL di config: -{sl:.0%} sul MARK del singolo asset, nozionale per asset") + print(f" <= {frac:.0%} dell'equity -> costo massimo {sl*frac:.0%} dell'equity per asset a k=1,") + print(f" cioe' {sl*frac*2:.0%} se scattano entrambi; diventa il -100% a k = {1/(sl*frac*2):.1f}x.") + + # ------------------------------------------------------------------ (4) + sezione("(4) IL GRADINO ESEGUIBILE — sopravvive alla lente? (€500/mese, muro netto)") + s500 = simula(D["R_del"], gap, KS, anni=ANNI, dep_eur=500, bersaglio=BERSAGLIO, seed=SEED) + print(f"\n versato in {ANNI} anni: ${s500['versato']:,.0f}. 'anni' = mediana CONDIZIONATA") + print(" all'arrivo (non-arrivo = +infinito, MAI -1), stampata accanto alla sua probabilita'.") + print(f"\n {'k':>6}" + "".join(f"{'anni ' + n:>16}{'P<=20a ' + n[:5]:>15}" for n in LENTI) + + f"{'maxDD acc.':>12}{'P(DD>30) acc.':>15}") + for j, k in enumerate(KS): + cel = "" + for l in range(2): + a, p = anni_cond(s500["hit"][:, j, l]) + cel += f"{fmt(a):>16}{p:>15.0%}" + marca = " <-- LIBRO" if k == 1.00 else "" + print(f" {k:>6.2f}" + cel + f"{np.median(s0['dd'][:, j, 1]):>12.1%}" + f"{(s0['dd'][:, j, 1] > DD_SOGLIA).mean():>15.1%}{marca}") + for l, nome in enumerate(LENTI): + d = s500["hit"][:, KS.index(1.00), l] - s500["hit"][:, KS.index(1.50), l] + fin = np.isfinite(d) + print(f"\n lente {nome:>11}: guadagno APPAIATO 1.00x -> 1.50x = " + f"{float(np.median(d[fin]))/365:.2f} anni (mediana delle differenze, n={fin.sum()})") + + print(f"\n ▸ tempo trascorso: {time.time()-t_start:.0f}s") + return dict(s0=s0, s500=s500, dd_close=dd_close, dd_acc=dd_acc, delta_p=delta_p, + D=D, ris_e=ris_e, ris_dd=ris_dd, kk=kk) + + +# =========================================================================== +# ATTACCO 2 — l'etichetta del rischio +# =========================================================================== +def attacco2() -> None: + from src.data.downloader import load_data + from src.portfolio import sleeves as SL + from src.portfolio.portfolio import weights_tilt_null + from src.strategies.trend_portfolio import (CANONICAL, TrendPortfolio, realized_vol, + resample_1d, simple_returns, tsmom_blend) + + sezione("(5) ATTACCO 2a — LA VOL REALIZZATA DI TP01: verifica INDIPENDENTE") + print("\n Lo scetticato ricostruisce TP01 a mano dentro il proprio script. Qui si legge lo") + print(" sleeve DI PRODUZIONE (`src/portfolio/sleeves._tp01_returns`), che e' la serie con") + print(" cui il libro e' misurato ovunque nel progetto: percorso diverso, stessa domanda.") + tp = SL._tp01_returns().values.astype(float) + att = tp[np.abs(tp) > 1e-12] + print(f"\n vol REALIZZATA di TP01 (sleeve di produzione): {tp.std()*365**.5:.2%}" + f" <- lo scetticato dichiara 12,14%") + print(f" -> VERIFICATO. E il numero non e' un artefatto della sua ricostruzione.") + + print("\n ⚠️ MA IL MECCANISMO CHE DICHIARA E' INCOMPLETO, e la parte che manca e' la") + print(" maggiore. Lo scetticato scrive: «il target vale sulla posizione quando c'e', e") + print(" TP01 e' long-flat». Decomposizione:") + convs, vols_pred, vols_real, sers = [], [], [], {} + for a in ("BTC", "ETH"): + df = resample_1d(load_data(a, "1h")) + c = df["close"].values.astype(float) + r = simple_returns(c) + t = TrendPortfolio(**CANONICAL) + bpd = t._bpd(df) + vol = realized_vol(r, t.vol_win_days * bpd, bpd * 365.25) + d = np.clip(tsmom_blend(c, tuple(h * bpd for h in t.horizons_days)), 0, None) + tgt = np.asarray(t.target_series(df), float) + held = np.zeros(len(tgt)); held[1:] = tgt[:-1] + net = held * r - t.fee_side * np.abs(np.diff(held, prepend=0.0)); net[0] = 0.0 + sers[a] = pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"])) + ok = np.isfinite(d) & np.isfinite(vol) & (vol > 0) + convs.append(float(np.sqrt(np.mean(d[ok] ** 2)))) + # errore del previsore di vol: vol realizzata nei 30g SEGUENTI / vol prevista + fwd = pd.Series(r).rolling(30).std().shift(-30).values * np.sqrt(365.25) + okf = ok & np.isfinite(fwd) + vols_pred.append(float(np.mean(vol[okf]))); vols_real.append(float(np.mean(fwd[okf]))) + print(f"\n {a}: frazione di barre FLAT (direzione 0) {float((d[ok] == 0).mean()):>6.1%}") + print(f" direzione media QUANDO NON FLAT " + f"{float(d[ok][d[ok] > 0].mean()):>6.2f} (tsmom_blend e' la MEDIA di 3 segni: " + f"vale 1/3, 2/3 o 1)") + print(f" fattore di convinzione sqrt(E[d²]) {convs[-1]:>6.3f}" + f" -> da solo porta 20% a {0.20*convs[-1]:.2%}") + print(f" vol prevista (30g trailing) {vols_pred[-1]:>6.2%} vs vol realizzata nei " + f"30g SEGUENTI {vols_real[-1]:>6.2%} (rapporto {vols_real[-1]/vols_pred[-1]:.3f})") + J = pd.concat(sers, axis=1, join="inner").fillna(0.0) + vb, ve = J["BTC"].std() * 365 ** .5, J["ETH"].std() * 365 ** .5 + comb = (0.5 * J["BTC"] + 0.5 * J["ETH"]).std() * 365 ** .5 + div = comb / (0.5 * vb + 0.5 * ve) + conv = float(np.mean(convs)) + print(f"\n gamba singola: BTC {vb:.2%} · ETH {ve:.2%} -> 50/50 combinato {comb:.2%}") + print(f" fattore di DIVERSIFICAZIONE BTC/ETH = {div:.3f} (corr {J['BTC'].corr(J['ETH']):.2f})") + print(f"\n CATENA: 20,00% ×{conv:.3f} (convinzione) ×{np.mean(vols_real)/np.mean(vols_pred):.3f} " + f"(previsore di vol) ×{div:.3f} (diversificazione) = " + f"{0.20*conv*(np.mean(vols_real)/np.mean(vols_pred))*div:.2%}") + print(f" misurato: {tp.std()*365**.5:.2%}") + print(f"\n 📌 IL FATTO CHE LO SCETTICATO NON DICE: il pezzo grosso NON e' lo stare flat, e'") + print(f" che `tsmom_blend` e' la MEDIA DEI SEGNI di 3 orizzonti, quindi quando TP01 e'") + print(f" 'a mercato' e' quasi sempre a 1/3 o 2/3 di convinzione. Contando lo zero, il") + print(f" fattore di convinzione da solo vale ×{conv:.2f}. Lo stare flat e' UNA PARTE di") + print(f" quel fattore, non il fattore.") + print(f" Conseguenza operativa opposta a quella suggerita: se il divario fosse solo") + print(f" 'flat', alzare target_vol non aumenterebbe il rischio nei giorni flat (dove non") + print(f" c'e' posizione). Essendo CONVINZIONE, alzare target_vol alza la posizione") + print(f" proprio nei giorni in cui il segnale e' PIU' DEBOLE (1 orizzonte su 3).") + + print("\n E TP01 E' UN CASO PARTICOLARE, non la regola — gli altri sleeve vol-targeted:") + print(f"\n {'sleeve':>9}{'target dichiarato':>19}{'vol REALIZZATA':>16}{'su barre ATTIVE':>17}{'attive':>9}") + for nm, fn, dic in (("TP01", SL._tp01_returns, "20%"), ("XS01", SL._xsec_returns, "20%"), + ("VRP01", SL._vrp_combo_returns, "—"), ("SKH01", SL._skyhook_returns, "nessuno"), + ("GTAA01", SL._gtaa_daily_returns, "—")): + v = fn().values.astype(float) + a = v[np.abs(v) > 1e-12] + print(f" {nm:>9}{dic:>19}{v.std()*365**.5:>16.2%}{a.std()*365**.5:>17.2%}" + f"{len(a)/len(v):>9.1%}") + print("\n XS01 dichiara lo STESSO 20% e ne realizza 20,59%: il divario non e' una") + print(" proprieta' del vol-targeting, e' una proprieta' di TP01 (blend di segni +") + print(" long-only + due gambe correlate 0,8). L'affermazione 'l'etichetta sovrastima il") + print(" rischio' e' vera per TP01 e FALSA per il resto del libro.") + + # -------------------------------------------------------------- 2b + sezione("(6) ATTACCO 2b — `target_vol` e `k` sono la STESSA leva? (misurato, non dedotto)") + print("\n Non e' ovvio: `target_vol` agisce DENTRO il vol-targeting di TP01 (e interagisce") + print(" col leverage cap 2x), `k` agisce FUORI su tutto il libro. Due differenze") + print(" strutturali attese, e la seconda e' quella che conta.") + skh = SL._skyhook_returns() + + def libro(tv: float, lev: float = 2.0) -> tuple[pd.Series, float, float]: + """(serie di libro 0.75*TP01(tv)+0.25*SKH01, scala ottenuta su TP01, quota barre al cap)""" + ser, atcap = {}, [] + for a in ("BTC", "ETH"): + df = resample_1d(load_data(a, "1h")) + t = TrendPortfolio(**{**CANONICAL, "target_vol": tv, "leverage": lev}) + tgt = np.nan_to_num(np.asarray(t.target_series(df), float)) + r = simple_returns(df["close"].values.astype(float)) + held = np.zeros(len(tgt)); held[1:] = tgt[:-1] + net = held * r - t.fee_side * np.abs(np.diff(held, prepend=0.0)); net[0] = 0.0 + ser[a] = pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"])) + atcap.append(float((np.abs(tgt) >= lev - 1e-9).mean())) + Jt = pd.concat(ser, axis=1, join="inner").fillna(0.0) + tp_s = pd.Series(0.5 * Jt["BTC"].values + 0.5 * Jt["ETH"].values, index=Jt.index) + if tp_s.index.tz is None: + tp_s.index = tp_s.index.tz_localize("UTC") + sk = skh.copy() + if sk.index.tz is None: + sk.index = sk.index.tz_localize("UTC") + B = pd.concat({"TP01": tp_s, "SKH01": sk}, axis=1, join="inner").fillna(0.0) + return (W_TP * B["TP01"] + W_SKH * B["SKH01"], float(tp_s.std() / 1.0), + float(np.mean(atcap))) + + base_s, tp_vol_base, _ = libro(0.20) + print(f"\n {'via':>28}{'scala TP01':>12}{'vol libro':>11}{'drift':>9}{'Sharpe':>8}" + f"{'maxDD':>8}{'peso EFFETTIVO TP01/SKH':>26}{'barre al cap':>14}") + + def riga(nome: str, s: pd.Series, scala_tp: float, atcap: float, w_eff: str) -> None: + v = s.values.astype(float) + eq = np.cumprod(1 + v) + dd = float((1 - eq / np.maximum.accumulate(eq)).max()) + print(f" {nome:>28}{scala_tp:>12.2f}{v.std()*365**.5:>11.2%}{v.mean()*365:>9.2%}" + f"{v.mean()/v.std()*365**.5:>8.2f}{dd:>8.1%}{w_eff:>26}{atcap:>14.1%}") + + riga("k=1.00, target_vol=20% (LIBRO)", base_s, 1.00, 0.0, "75 / 25") + for tv in (0.25, 0.30, 0.40, 0.60): + s, tpv, ac = libro(tv) + sc = tv / 0.20 + w = W_TP * sc / (W_TP * sc + W_SKH) + riga(f"target_vol={tv:.0%} (solo TP01)", s, sc, ac, f"{w*100:.0f} / {(1-w)*100:.0f}") + for k in (1.25, 1.50, 2.00, 3.00): + riga(f"k={k:.2f} (tutto il libro)", base_s * k, k, 0.0, "75 / 25") + + print("\n ❌ NON SONO LA STESSA LEVA, e la ragione NON e' il leverage cap (che infatti non") + print(" morde: le barre al cap restano sotto qualche punto percentuale). E':") + print(" **`target_vol` scala il 75% del libro e lascia fermo il 25%** -> e'") + print(" simultaneamente una LEVA e un TILT DI PESI. A target_vol=30% il libro non e'") + print(" piu' 75/25 ma ~82/18, e a 60% e' ~90/10.") + print(" Un tilt di pesi in questo progetto ha un gate suo, e non e' opzionale:") + + dcols = {"TP01": SL._tp01_returns(), "SKH01": skh} + for nm, s in dcols.items(): + if s.index.tz is None: + dcols[nm] = s.tz_localize("UTC") + for tv in (0.30, 0.60): + sc = tv / 0.20 + w = W_TP * sc / (W_TP * sc + W_SKH) + try: + g = weights_tilt_null(dcols, {"TP01": W_TP, "SKH01": W_SKH}, + {"TP01": round(w, 4), "SKH01": round(1 - w, 4)}, n=300) + print(f" weights_tilt_null 75/25 -> {w*100:.0f}/{(1-w)*100:.0f} " + f"(target_vol {tv:.0%}): gate_pass={g.get('gate_pass')} " + f"delta_insample={g.get('delta_insample'):+.4f} pctl_hold={g.get('pctl_hold'):.0f}") + except Exception as e: # noqa: BLE001 + print(f" weights_tilt_null non girato per tv={tv:.0%}: {type(e).__name__}: {e}") + + print("\n 📌 CONSEGUENZA CHE LO SCETTICATO NON POTEVA VEDERE dalla sua sezione (5b), che") + print(" guardava TP01 DA SOLO: la sua raccomandazione implicita («il knob target_vol") + print(" funziona, l'ostacolo e' la config») e' vera sullo sleeve e FALSA sul libro.") + print(" Alzare `target_vol` NON e' la strada meno invasiva per alzare k: e' la strada") + print(" che aggiunge un cambio di pesi mai giudicato al cambio di scala.") + print(" La strada pulita per un puro cambio di SCALA e' il cap di config, che e'") + print(" esattamente quella protetta dal test di guardia.") + + +def guardia() -> None: + sezione("(7) IL VINCOLO DURO — cosa protegge davvero il test di guardia") + cfg = json.loads((ROOT / "config" / "live.json").read_text()) + frac = cfg["max_notional_per_asset_frac"] + print(f""" + `config/live.json`: max_notional_per_asset_frac = {frac} su 2 asset (BTC, ETH) + -> nozionale LORDO massimo = {frac*2:.2f}x l'equity. k > {frac*2:.2f}x oggi NON e' eseguibile. + `tests/test_fee_sensitivity.py::test_leva_massima_da_config_resta_sotto_o_uguale_a_1x` + asserisce `frac * len(ASSETS) <= 1.0` E `disaster_sl_pct > 0`. + ⚠️ Cio' che protegge NON e' "la leva e' pericolosa": e' la CONCLUSIONE PUBBLICATA del + 26/07 secondo cui la **liquidation fee 1%** introdotta da Deribit dal 2026-08-01 e' + irrilevante per questo libro. Quella conclusione poggia interamente sul fatto che a + nozionale lordo <= 1x servirebbe un movimento avverso ~100% per liquidare. Se il cap + sale, la liquidation fee torna nel conto e `r0726_fee_sensitivity` va rifatto. + Il test e' quindi un promemoria di DIPENDENZA, non un divieto: chi alza il cap deve + rifare quell'analisi, non aggirare l'assert. + Nota di taglia, che vale piu' del cap: il conto e' ~$635 e `min_order_usd` = {cfg['min_order_usd']}. + Alzare k moltiplica anche il TURNOVER, e il progetto ha gia' misurato (02/07) che a + questo capitale il min-order $5 e' GIA' la banda di ribilanciamento ottimale.""") + + +if __name__ == "__main__": + main() + attacco2() + guardia() + print(f"\n▸ totale: consegna nel messaggio dell'agente.") diff --git a/scripts/research/r0822_prop_alloc.py b/scripts/research/r0822_prop_alloc.py index 0a8fc85..cb2813c 100644 --- a/scripts/research/r0822_prop_alloc.py +++ b/scripts/research/r0822_prop_alloc.py @@ -674,6 +674,28 @@ def main() -> None: print(f" {fct:>9.2f} " + " ".join(f"{v:>14.3f}" for v in cells) + f" {b['w']:>14} {b['lev']:>11.3f}") + hr("E-bis. LO STRESS CHE DECIDE — XS01 e' stato SCOPERTO in questa finestra") + print(" L'argmax mette il 50% su XS01, il cui unico campione (2024+) e' anche quello su cui e'") + print(" stato scoperto e affinato (blend di lookback + gate di dispersione). CLAUDE.md lo") + print(" dichiara: 'la TAGLIA del guadagno e' ottimista'. Qui si taglia il DRIFT di XS01 (non la") + print(" vol, non la coda) e si guarda se la conclusione operativa sopravvive.") + print(f"\n {'drift XS01':>12} {'config':>16} {'leva':>6} {'J':>7} {'P(pass)':>8} " + f"{'P(vivo)':>8} {'E[pay/a]':>10}") + for mult in (1.00, 0.75, 0.50, 0.25, 0.00): + dmt = (("XS01", mult),) + rows = [] + for w in top_w: + for lv in LEVS: + rows.append((w, lv, objective(w, lv, "HYRO", ev_idx, fu_idx, drift_mult=dmt))) + w_, lv_, o_ = max(rows, key=lambda x: x[2]["J"]) + oL = max((objective({"TP01": 0.75, "SKH01": 0.25}, lv, "HYRO", ev_idx, fu_idx, + drift_mult=dmt) for lv in LEVS), key=lambda x: x["J"]) + print(f" {mult:>12.2f} {wkey(w_):>16} {lv_:>6.3f} {o_['J']:>7.3f} {o_['p_pass']:>8.1%} " + f"{o_['p_alive']:>8.1%} {o_['e_payout']:>9,.0f}$ (LIVE 75/25: J {oL['J']:.3f})") + print("\n Se anche a drift XS01 = 0 (cioe' XS01 ridotto a pura diversificazione di VOLATILITA',") + print(" senza alcun rendimento atteso) la configurazione diversificata batte il libro LIVE,") + print(" allora il risultato NON dipende dall'edge di XS01 ma dalla sua DECORRELAZIONE.") + # ---------------------------------------------------------------- F. VRP01 / GTAA01 tick("E. banda ancora") hr("F. E GLI SLEEVE 'INUTILI A $600'? — l'asimmetria del 25/07 §4 vale per XS01, non per tutti") @@ -775,12 +797,29 @@ def main() -> None: print(f"\n {'politica':>44} {'leva':>6} {'EUR/g med':>10} {'p90':>8} {'P(>=10/g)':>10} " f"{'P(>=50/g)':>10} {'P(zero)':>8}") for pname, (cfgs, slots) in pols.items(): - for lev in (float(best["lev"]), 0.75): + for lev in sorted({round(float(best["lev"]), 3), 0.75}): r = pc.simulate(cfgs, slots, n_paths=1500, lev=lev, seed=20260822, lens="coupled", start=START_WIN) print(f" {pname:>44} {lev:>6.3f} {r['med']:>10.2f} {r['p90']:>8.2f} " f"{r['p_ge10']:>10.1%} {r['p_target']:>10.1%} {r['p_zero']:>8.1%}") + print("\n ⚠ QUESTO NUMERO REVISIONA UN HEADLINE DI CLAUDE.md ('EUR600 -> 50 EUR/g ~ 6% in 3") + print(" anni, P(perdere i 600) 52-65%', 25/07). Due cose sono cambiate insieme: il fattore di") + print(" de-luck (0.60 -> 0.89, ri-misurato il 26/07) e l'ALLOCAZIONE. Vanno separate, o il") + print(" merito finisce sulla leva sbagliata. Decomposizione 2x2 (stessa macchina, stesso seme):") + print(f"\n {'de-luck':>9} {'allocazione':>26} {'EUR/g med':>10} {'P(>=50/g)':>10} {'P(zero)':>8}") + for fct in (0.60, 0.89): + pc.DELUCK = fct + for lab, cfg in (("LIVE 75/25 (25/07)", cfg_live), (f"argmax-J {best['w']}", cfg_best)): + r = pc.simulate([cfg], [0] * pl.MAX_CONCURRENT, n_paths=1500, lev=0.75, + seed=20260822, lens="coupled", start=START_WIN) + print(f" {fct:>9.2f} {lab:>26} {r['med']:>10.2f} {r['p_target']:>10.1%} " + f"{r['p_zero']:>8.1%}") + pc.DELUCK = DELUCK + print("\n La cella in alto a sinistra (0.60 + LIVE) e' la configurazione del 25/07 e va letta") + print(" come replica: se riproduce ~6% di P(>=50/g) e ~52-65% di P(zero), la macchina e' la") + print(" stessa e le altre tre celle misurano cio' che dichiarano.") + print("\n" + "=" * 104) print(f" fatto in {time.time()-t0:.0f}s") print("=" * 104) diff --git a/scripts/research/r0822_skew.py b/scripts/research/r0822_skew.py index c985a4c..48648fb 100644 --- a/scripts/research/r0822_skew.py +++ b/scripts/research/r0822_skew.py @@ -550,26 +550,39 @@ def main() -> None: print(f" |t| minimo = {LL[LL['minuti'] < 0]['t_dRR'].abs().min():.2f} su " f"{int((LL['minuti'] < 0).sum())} test, tutti dello STESSO segno.") print("\n Robustezza — il COLLETTORE cambia a meta' campione (cerbero-bite: ~10 giri/ora,") - print(" fino al 2026-07-30; raccolta nostra: 1 giro/ora al minuto :25). Stessa misura sui due") - print(" tronconi, piu' la copertura oraria (il guasto quote-vuote del 29-30/07 e' qui dentro):") + print(" fino al 2026-07-30; raccolta nostra: 1 giro/ora al minuto :25) e il campione contiene") + print(" UN evento enorme (BTC +26% nella settimana al 21/08). Le due cose vanno separate:") TAGLIO = pd.Timestamp("2026-07-30 20:00", tz="UTC") + RALLY = pd.to_datetime(["2026-08-19", "2026-08-20", "2026-08-21", "2026-08-22"]).tz_localize("UTC") for a in ASSETS: S = cm30[a] - for nome, m in (("bite ", S.index < TAGLIO), ("nostra ", S.index >= TAGLIO)): - sub = S[m] + gg_rally = pd.DatetimeIndex(S.index).normalize().isin(RALLY) + for nome, m in (("bite ", S.index < TAGLIO), + ("nostra ", S.index >= TAGLIO), + ("nostra SENZA rally", (S.index >= TAGLIO) & ~gg_rally)): + sub = S[m].sort_values("ts_max") if len(sub) < 100: continue - L2 = leadlag(sub, a) - av = L2[L2["minuti"].isin([5, 15, 60])]["corr_dRR"].to_numpy() - ind = L2[L2["minuti"].isin([-15, -60])]["corr_dRR"].to_numpy() - gg = len(np.unique(pd.DatetimeIndex(sub.index).date)) - print(f" {a} {nome} ore={len(sub):5d} giorni={gg:3d} ({len(sub)/max(gg,1):.1f} ore/giorno)" - f" avanti(+5/+15/+60m) {np.round(av,3)} indietro(-15/-60m) {np.round(ind,3)}") + px = px5m(a, sub.index.min()) + anc = pd.Series(px.reindex(pd.DatetimeIndex(sub["ts_max"]), method="ffill").to_numpy(), + index=sub.index) + d = sub["rr"].diff() + av = [] + for mins in (5, 15, 60): + p2 = pd.Series(px.reindex(pd.DatetimeIndex(sub["ts_max"]) + pd.Timedelta(minutes=mins), + method="ffill").to_numpy(), index=sub.index) + j = pd.concat({"d": d, "r": np.log(p2 / anc)}, axis=1).dropna() + av.append((round(float(j["d"].corr(j["r"])), 3), + round(float(j["d"].corr(j["r"], method="spearman")), 3))) + print(f" {a} {nome} n={len(sub):5d} avanti +5/+15/+60m (pearson,spearman): {av}") + print(" -> l'unica cella con un lead in avanti apparente (BTC, raccolta nostra, +5/+15m)") + print(" PERDE i due terzi togliendo 4 giorni su 24: era il rally, non un meccanismo.") gio = pd.Series(1, index=pd.DatetimeIndex(cm30[ASSETS[0]].index)).groupby( pd.DatetimeIndex(cm30[ASSETS[0]].index).date).sum() - peggio = gio.nsmallest(4) - print(f" giorni con MENO ore di superficie ricostruibile (BTC): " - f"{ {str(k): int(v) for k, v in peggio.items()} } su 24 attese") + print(f" copertura: giorni con meno ore di superficie (BTC) " + f"{ {str(k): int(v) for k, v in gio.nsmallest(4).items()} } su 24 attese " + f"(il guasto quote-vuote del 29-30/07 e' dentro il campione e NON crea un buco: le") + print(" quote assenti erano gia' escluse dal filtro 'quotato').") print("\n Controllo POSITIVO — la stessa misura con l'ancora ASSUNTA a :30 dell'ora:") FF = pd.concat([leadlag_ancora_sbagliata(cm30[a], a) for a in ASSETS], ignore_index=True) diff --git a/scripts/research/r0822_vol_size.py b/scripts/research/r0822_vol_size.py index 4d488a6..4c8a175 100644 --- a/scripts/research/r0822_vol_size.py +++ b/scripts/research/r0822_vol_size.py @@ -96,6 +96,10 @@ from r0822_sol_leg import k_iso_dd # noqa: E402 (null de-lev from src.portfolio.portfolio import ( # noqa: E402 HOLDOUT, combine_outer, weights_tilt_null, ) +from src.backtest.harness import backtest_signals # noqa: E402 +from src.strategies.skyhook import ( # noqa: E402 + SKH01_V1, SKH01_V2_DD, skyhook_entries, +) ASSETS = ("BTC", "ETH") OFFSETS_FULL = tuple(range(0, 690, 30)) # 23 a priori: griglia dell'audit 02/07 @@ -149,12 +153,18 @@ def ins(s: pd.Series) -> pd.Series: # --------------------------------------------------------------------------- estrazione trade -def extract(asset: str, off: int) -> dict: +def extract(asset: str, off: int, params=None) -> dict: """Trade + contesto per (asset, offset). Replica il loop di `backtest_signals` con leverage=1: stessa detection, stesso non-overlap, stessi fill al livello. Cio' che aggiunge e' la SIZE per-trade, che l'harness non espone (ha un solo `position_size` scalare). La §0 verifica bit-exact che con size=1 si ritrovi la serie ufficiale.""" - _, _, ltf, ent = R.run_asset(asset, off) + if params is None or params is SKH01_V2_DD: + _, _, ltf, ent = R.run_asset(asset, off) # path canonico (sanity bit-exact §0) + else: # stessa pipeline, altra config Skyhook + df5 = R.get5m(asset) + ltf = R.resample_off(df5, R.LTF_MIN, off) + htf = R.resample_off(df5, R.HTF_MIN, off) + ent = skyhook_entries(ltf, htf, params) c = ltf["close"].values.astype(float) h = ltf["high"].values.astype(float) lo = ltf["low"].values.astype(float) @@ -349,10 +359,63 @@ HEAD = (f" {'variante':<28}{'ShFULL':>8}{'ShHOLD':>9}{'ShIS':>8}" f"{'vol':>9}{'maxDD':>9}{'DDisovol':>10}{'CAGRiso':>9}{'size med':>10}") +def transfer_test(offs) -> None: + """TEST DI TRASFERIMENTO — la regola di size CONGELATA (tv=0.20, w=90, cap 3, fissata + all'INGRESSO) applicata a SKH01_**V1**, l'altra config Skyhook: stesso motore, stessi asset, + ma geometria d'uscita DIVERSA (stop/target in ATR 2.5/6.0 invece che a percentuale fissa + 4%/2%, ptn_n=55 invece di 45). E' il gate giusto per questo filone: un forward di 6 mesi non + puo' misurare un dSharpe di 0.07 (SE ~1.4), un trasferimento si'. Se il meccanismo e' + 'pareggiare il contributo di rischio dei trade' deve funzionare anche li'; se e' cucito sulla + sequenza di trade di V2-DD, no. NB: V1 e' l'unica config alternativa gia' nel repo e non + scelta da me — non e' una griglia nuova.""" + print("=" * 112) + print(" TEST DI TRASFERIMENTO — regola di size CONGELATA da V2-DD, applicata a SKH01_V1") + print("=" * 112) + TP = A.tp01_baseline_daily() + sp = dict(tv=0.20, w=90, cap=3.0, active_only=False) + base_b, var_b, ctrl_b = {}, {}, {} + for o in offs: + ex = {a: extract(a, o, SKH01_V1) for a in ASSETS} + lb = leg_daily(ex, {a: size_flat(ex[a]) for a in ASSETS}) + lv = leg_daily(ex, {a: sizes_from_L(ex[a], leverage_series(ex[a]["base_daily"], **sp)) + for a in ASSETS}) + m_s = vol(lv) / vol(lb) if vol(lb) > 0 else 1.0 + base_b[o], var_b[o] = book(TP, lb), book(TP, lv) + ctrl_b[o] = book(TP, lb * m_s) + dF = A.anchor_luck_delta(lambda o: var_b[o], lambda o: ctrl_b[o], offs, metric=sh) + dH = A.anchor_luck_delta(lambda o: hold(var_b[o]), lambda o: hold(ctrl_b[o]), offs, metric=sh) + rF = A.anchor_luck_delta(lambda o: var_b[o], lambda o: base_b[o], offs, metric=sh) + print(f" trade V1 medi/ancora: " + ", ".join( + f"{a} {int(np.mean([len(extract(a, offs[0], SKH01_V1)['i_ent'])]))}" for a in ASSETS)) + print(f" libro V1 BASE, mediana della banda: ShFULL " + f"{np.median([sh(base_b[o]) for o in offs]):.3f} / ShHOLD " + f"{np.median([sh(hold(base_b[o])) for o in offs]):.3f}") + print(f" dShFULL grezzo {rF['median_paired']:+.3f} ({rF['n_positive']}/{len(offs)} ancore)") + print(f" dShFULL ISO-PESO {dF['median_paired']:+.3f} ({dF['n_positive']}/{len(offs)} ancore)" + f" banda [{dF['lo']:+.3f}, {dF['hi']:+.3f}]") + print(f" dShHOLD ISO-PESO {dH['median_paired']:+.3f} ({dH['n_positive']}/{len(offs)} ancore)") + yrs = sorted({int(y) for y in base_b[offs[0]].index.year}) + cells = [] + for y in yrs: + v = [sh(var_b[o][var_b[o].index.year == y]) - sh(ctrl_b[o][ctrl_b[o].index.year == y]) + for o in offs if (var_b[o].index.year == y).sum() > 20] + cells.append(float(np.median(v)) if v else float("nan")) + print(" per anno (ISO): " + "".join(f"{y:>8}" for y in yrs)) + print(" " + "".join(f"{c:>+8.2f}" for c in cells) + + f" positivi {int(np.nansum(np.asarray(cells) > 0))}/{len(yrs)}") + print("\n LETTURA: se il segno regge qui, il meccanismo non e' cucito su V2-DD. Se non regge,") + print(" la regola e' una proprieta' della sequenza di trade su cui e' stata trovata.") + + def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--every", type=int, default=1) + ap.add_argument("--transfer", action="store_true", + help="esegue SOLO il test di trasferimento su SKH01_V1") args = ap.parse_args() + if args.transfer: + transfer_test(OFFSETS_FULL[:: max(1, args.every)]) + return offs = OFFSETS_FULL[:: max(1, args.every)] t0 = time.time()