Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9f55652d9903a35d44b3f6da2de83a7fc2cfd9a9
PythagorasGoal/scripts/research/r0822b_tp01_sinistro.py
T

955 lines
53 KiB
Python
Raw Blame History

"""r0822b_tp01_sinistro — quanto costa l'ASSICURAZIONE TP01 su un conto a barriera (2026-08-22).
IL BUCO CHE CHIUDE
------------------
`r0822_prop_alloc.py` (stessa ondata) ha ottimizzato l'allocazione di un conto **funded** sotto
barriera di drawdown e ha trovato l'ottimo a **TP01 25 / SKH01 25 / XS01 50**, cioe' TP01 pesato
**tre volte meno** del libro live (75%). Ha anche dichiarato il proprio limite:
"la sotto-pesatura di TP01, lo sleeve DIFENSIVO, e' misurata su un campione SENZA SINISTRO
(2024-2026 non contiene un crash): su una barriera assorbente quella scommessa si paga una
volta sola."
E questo si scontra con due misure del progetto:
(a) il **leave-one-out de-luckato** del 26/07: TP01 e' il maggior contributore al Sharpe FULL
(+0.390, positivo nel 100% delle estrazioni) e il suo hold-out negativo (-0.200, negativo
nel 99.1%) e' *la firma dell'assicurazione, non della morte* — "uno sleeve difensivo si
giudica sul SINISTRO, non sul premio";
(b) `edge_watch` criterio B: negli **8 anni su 8** con drawdown buy&hold > 10% TP01 ha protetto
da **1.8x a 34.4x**.
DOMANDA: qual e' il peso giusto di TP01 se l'obiettivo include sopravvivere a un crash che nel
campione di ottimizzazione NON c'e'?
IPOTESI REGISTRATA PRIMA DI MISURARE
------------------------------------
Mi aspettavo che (1) la sotto-pesatura di TP01 costasse molto **nel sinistro** e poco fuori, e che
(2) esistesse un peso di compromesso (~50%) dentro la regione robusta del funded e non
catastrofico al crash. In altre parole mi aspettavo di **confermare** il caveat di PROP-ALLOC.
Se la misura dice il contrario, e' un risultato e va scritto come tale.
CIO' CHE E' UN'ASSUNZIONE E CIO' CHE E' UN DATO
-----------------------------------------------
* Lo **scenario di stress e' uno SCENARIO, non una previsione.** Due costruzioni, entrambe
dichiarate: (S1) bootstrap MISTO — ogni blocco ha probabilita' `p` di essere estratto dal pool
dei sinistri 2019-2023 invece che dalla finestra normale; (S2) **replay** — un episodio di
crash VERO innestato per intero, in ordine, a una data casuale dentro l'anno. In piu' il
replay deterministico del **2022 intero** (aprile-dicembre), che non e' un modello: e' successo.
* Il vincolo binding di TUTTO il filone e' che i sinistri osservati sono **QUATTRO**. Nessun
bootstrap crea informazione: l'ampiezza effettiva della coda resta 4 episodi, e ogni numero qui
eredita quell'errore. Detto una volta, vale per ogni tabella.
* Lente **ACCOPPIATA** (`r0725_prop_coupled`): il minimo intraday si compone ESATTO sul path
orario condiviso TP01/SKH01. Non e' un dettaglio: su una regola a UN giorno (daily-loss) la
lente close-only e' misurata **esattamente cieca** (0.00 breach contro 0.40-1.21 veri), mentre
su una statistica multi-giorno l'errore vale il 3.5%. Il vincolo funded E' una regola a un giorno.
* de-luck sul drift **x0.89** (fattore MISURATO il 26/07). ⚠️ L'offset di de-luck e' calcolato UNA
VOLTA sulla finestra 2019+ e sottratto ovunque in valore ASSOLUTO: de-luckare un pool di crash
sulla sua PROPRIA media (negativa) *aggiungerebbe* drift proprio dove non deve.
* macchineria RIUSATA, non riscritta: `r0725_prop_coupled` (tuple accoppiate) e `r0822_prop_alloc`
(`eval_sim`, `funded_sim`, `boot_idx`, `FIRMS`). Qui si aggiungono solo i pool e gli innesti.
Uso: `nice -n 19 timeout 900 uv run python scripts/research/r0822b_tp01_sinistro.py`
"""
from __future__ import annotations
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"))
from src.data.downloader import load_data # noqa: E402
from src.portfolio.sleeves import _tp01_returns, _skyhook_returns # noqa: E402
import r0725_prop_coupled as pc # noqa: E402
import r0822_prop_alloc as PA # noqa: E402
# ------------------------------------------------------------------ costanti DICHIARATE
DELUCK = PA.DELUCK # 0.89, misurato il 26/07
FULL_START = "2019-03-14" # inizio del recon crypto (TP01+SKH01)
RECENT_START = "2024-01-01" # finestra di ottimizzazione di PROP-ALLOC (nasce XS01)
BLOCK = 20 # blocco del bootstrap, in giorni
N_PATHS = 2500
EPI_LEN = 30 # lunghezza di un episodio di sinistro, in giorni
HORIZON = 365
LEVS = (0.25, 0.375, 0.50, 0.625, 0.75, 1.00, 1.25, 1.50)
W_GRID = tuple(i / 8 for i in range(9)) # peso di TP01 nella coppia crypto, 0 -> 1
SURV_TARGET = 0.90 # livello di sopravvivenza della frontiera ISO-SOPRAVVIVENZA
N_EVAL = 0 # contatore onesto delle valutazioni (griglia dichiarata)
def hr(t: str = "") -> None:
print("\n" + "-" * 106)
if t:
print(f" {t}")
print("-" * 106)
# ================================================================== §A. PANNELLO
_TUP: dict = {}
def crypto(w_tp: float) -> tuple[np.ndarray, np.ndarray, pd.DatetimeIndex]:
"""(R, gap) giornalieri della coppia crypto a peso TP01 = w_tp, SKH01 = 1-w_tp.
gap = minimo_intraday - chiusura (<=0). Storia piena dal 2019-03."""
k = round(float(w_tp), 6)
if k not in _TUP:
c = pc.crypto_daily_tuples(k, 1.0 - k)
R = c["R"].values.astype(float)
G = np.minimum(c["m"].values, R) - R
_TUP[k] = (R, G, c.index)
R, G, I = _TUP[k]
return R.copy(), G.copy(), I
_OFF: dict = {}
def deluck_offset(w_tp: float, factor: float = DELUCK) -> float:
"""Offset ASSOLUTO di de-luck, calcolato UNA VOLTA sulla finestra 2019+ e riusato ovunque.
Calcolarlo sulla media di un sotto-campione di crash (negativa) *aggiungerebbe* drift."""
key = (round(float(w_tp), 6), round(float(factor), 4))
if key not in _OFF:
R, _, I = crypto(w_tp)
_OFF[key] = float((1.0 - factor) * R[I >= pd.Timestamp(FULL_START, tz="UTC")].mean())
return _OFF[key]
def buyhold_daily(idx: pd.DatetimeIndex) -> pd.Series:
"""Buy&hold 50/50 BTC+ETH sulla stessa griglia giornaliera del libro."""
bh = None
for a in ("BTC", "ETH"):
d = load_data(a, "1h")
i = pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))
s = pd.Series(d["close"].astype(float).values, index=i).resample("1D").last()
r = s.pct_change().fillna(0.0) * 0.5
bh = r if bh is None else bh.add(r, fill_value=0.0)
return bh.reindex(idx).fillna(0.0)
def worst_windows(lbh: np.ndarray, idx: pd.DatetimeIndex, K: int, L: int,
hi: int | None = None) -> list[int]:
"""Le K finestre di L giorni NON sovrapposte col peggior rendimento buy&hold, greedy sul
minimo. `hi` limita i possibili inizi (per restare dentro 2019-2023)."""
cs = np.concatenate(([0.0], np.cumsum(lbh)))
win = np.expm1(cs[L:] - cs[:-L])
cand = np.arange(len(win)) if hi is None else np.arange(min(hi, len(win)))
order = cand[np.argsort(win[cand])]
out: list[int] = []
for t in order:
if all(abs(int(t) - q) >= L for q in out):
out.append(int(t))
if len(out) == K:
break
return sorted(out)
def binom_two_sided(k: int, n: int) -> float:
"""p-value esatto a due code di un test dei segni con p0=0.5 (n piccolo: si conta)."""
from math import comb
tot = 2 ** n
kk = max(k, n - k)
tail = sum(comb(n, j) for j in range(kk, n + 1))
return min(1.0, 2.0 * tail / tot)
def maxdd(r: np.ndarray) -> float:
eq = np.cumprod(1.0 + r)
return float((eq / np.maximum.accumulate(eq) - 1.0).min())
# ================================================================== §B. SIMULAZIONE
def J_of(R: np.ndarray, G: np.ndarray, ev_idx: np.ndarray, fu_idx: np.ndarray,
lev: float, firm: str = "HYRO") -> dict:
"""J = P(passare l'eval) x P(essere vivo a 12 mesi da funded), lente ACCOPPIATA."""
global N_EVAL
N_EVAL += 1
e = PA.eval_sim(R, G, ev_idx, lev, firm, "coupled")
f = PA.funded_sim(R, G, fu_idx, lev, firm, 100_000.0, "coupled")
return dict(J=e["p_pass"] * f["p_alive"], p_pass=e["p_pass"], p_alive=f["p_alive"],
e_payout=f["e_payout"], jvec=(e["passed"] & f["alive"]).astype(float),
alive=f["alive"])
def best_lev(R: np.ndarray, G: np.ndarray, ev_idx, fu_idx, firm: str = "HYRO") -> tuple:
rows = [(lv, J_of(R, G, ev_idx, fu_idx, lv, firm)) for lv in LEVS]
return max(rows, key=lambda x: x[1]["J"])
def pool_paths(seg_bounds: list[tuple[int, int]], n_days: int, n_paths: int, seed: int,
block: int = BLOCK) -> np.ndarray:
"""Indici di bootstrap a blocchi ristretti a segmenti CONTIGUI (ogni blocco vive dentro un
solo segmento -> non si incollano giorni di episodi diversi)."""
rng = np.random.default_rng(seed)
starts = []
for a, b in seg_bounds:
if b - a >= block:
starts.extend(range(a, b - block + 1))
starts = np.array(starts)
nb = int(np.ceil(n_days / block))
st = starts[rng.integers(0, len(starts), size=(n_paths, nb))]
return (st[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1)[:, :n_days]
def mix_paths(base_idx: np.ndarray, crash_starts: np.ndarray, p: float, seed: int,
block: int = BLOCK) -> np.ndarray:
"""S1 — bootstrap MISTO: ogni blocco della path normale ha probabilita' `p` di essere
sostituito da un blocco estratto dal pool dei SINISTRI."""
rng = np.random.default_rng(seed)
out = base_idx.copy()
P, D = out.shape
nb = int(np.ceil(D / block))
hit = rng.random((P, nb)) < p
for b in range(nb):
sl = slice(b * block, min((b + 1) * block, D))
L = sl.stop - sl.start
rows = np.where(hit[:, b])[0]
if len(rows) == 0:
continue
st = crash_starts[rng.integers(0, len(crash_starts), size=len(rows))]
out[rows, sl] = st[:, None] + np.arange(L)[None, :]
return out
def splice_paths(base_idx: np.ndarray, epi_idx: np.ndarray, seed: int) -> np.ndarray:
"""S2 — REPLAY: l'episodio VERO (contiguo, in ordine) innestato una volta sola a una data
casuale dentro l'orizzonte. E' la domanda operativa: *se il sinistro arriva*."""
rng = np.random.default_rng(seed)
out = base_idx.copy()
P, D = out.shape
L = len(epi_idx)
st = rng.integers(0, max(D - L, 1), size=P)
for i in range(P):
a = int(st[i])
out[i, a:a + L] = epi_idx[:min(L, D - a)]
return out
# ================================================================== main
def main() -> None: # noqa: C901
try:
sys.stdout.reconfigure(line_buffering=True)
except Exception: # pragma: no cover
pass
t0 = time.time()
def tick(lab: str) -> None:
print(f" [{time.time()-t0:6.0f}s] {lab} (valutazioni finora: {N_EVAL})")
print("=" * 106)
print(" TP01-SINISTRO — quanto costa l'assicurazione su un conto a barriera, e quanto rende")
print(" quando il sinistro arriva. Lente ACCOPPIATA, firm HYRO.")
print("=" * 106)
R75, G75, IDX = crypto(0.75)
n_full = len(IDX)
i_rec = int(np.searchsorted(IDX, pd.Timestamp(RECENT_START, tz="UTC")))
print(f"\n finestra crypto piena: {n_full} giorni {IDX[0].date()} -> {IDX[-1].date()}")
print(f" finestra di ottimizzazione di PROP-ALLOC: {n_full - i_rec} giorni dal "
f"{IDX[i_rec].date()} (e' anche la finestra di SCOPERTA di XS01)")
# --------------------------------------------------------- sanity: recon vs sleeve ufficiale
print("\n SANITY — il recon orario contro gli sleeve ufficiali. TP01/SKH01 NON coincidono")
print(" barra per barra per costruzione (il recon e' un mark-to-market ORARIO, lo sleeve")
print(" marca a fine-trade): il controllo giusto e' il log-totale.")
for nm, off_s, w in (("TP01", _tp01_returns(), 1.0), ("SKH01", _skyhook_returns(), 0.0)):
o = off_s.dropna().sort_index()
if o.index.tz is None:
o.index = o.index.tz_localize("UTC")
Rn, _, In = crypto(w)
com = o.index.intersection(In)
a = o.reindex(com).values
b = pd.Series(Rn, index=In).reindex(com).values
print(f" {nm:>6} log-tot sleeve {float(np.log1p(a).sum()):+8.4f} "
f"recon {float(np.log1p(b).sum()):+8.4f} scarto "
f"{float(np.log1p(a).sum()-np.log1p(b).sum()):+7.4f}")
# ============================================================ §1. CHE COS'E' UN SINISTRO
hr("§1. CHE COS'E' UN SINISTRO — e la frase '2024-2026 non contiene un crash' e' FALSA")
bh = buyhold_daily(IDX)
lbh = np.log1p(bh.values)
print(" Peggior rendimento buy&hold 50/50 su finestra mobile, per periodo:")
print(f" {'W':>4} {'2019-2023 min':>15} {'2024+ min':>12} {'rapporto':>10} "
f"{'2019-23 p1':>11} {'2024+ p1':>10} {'data del min 2024+':>20}")
ratios = []
for W in (1, 5, 10, 20, 60):
cs = np.concatenate(([0.0], np.cumsum(lbh)))
roll = np.expm1(cs[W:] - cs[:-W]) # finestra che FINISCE in t=W-1+k
tt = IDX[W - 1:]
pre = roll[tt < pd.Timestamp(RECENT_START, tz="UTC")]
post = roll[tt >= pd.Timestamp(RECENT_START, tz="UTC")]
tpost = tt[tt >= pd.Timestamp(RECENT_START, tz="UTC")]
rr = pre.min() / post.min()
ratios.append(rr)
print(f" {W:>4} {pre.min():>15.1%} {post.min():>12.1%} {rr:>10.2f}x "
f"{np.percentile(pre, 1):>11.1%} {np.percentile(post, 1):>10.1%} "
f"{str(tpost[int(np.argmin(post))].date()):>20}")
print(f"\n -> il 2024+ **contiene** un crash: peggior giorno -14.5%, peggiori 20g -39.6%,")
print(f" drawdown buy&hold fino a -60%. Cio' che NON contiene e' un crash della TAGLIA")
print(f" dei precedenti: il rapporto e' {np.mean(ratios[:3]):.1f}x sulla coda veloce (1-10g)")
print(f" e {ratios[-1]:.1f}x a 60 giorni. **La frase giusta non e' 'senza sinistro' ma")
print(f" 'con un sinistro di taglia dimezzata'** — e cambia la domanda: non 'cosa succede")
print(f" se arriva un crash' ma 'cosa succede se ne arriva uno DUE VOLTE piu' grande'.")
# ----- replica del criterio B di edge_watch (protezione per anno)
print("\n Criterio B di `edge_watch` replicato qui (anni con DD buy&hold > 10%):")
tp_only, _, _ = crypto(1.0)
ser_bh = pd.Series(bh.values, index=IDX)
ser_tp = pd.Series(tp_only, index=IDX)
print(f" {'anno':>6} {'DD b&h':>9} {'DD TP01':>9} {'protezione':>12} {'DD SKH01':>10} "
f"{'DD 75/25':>10}")
skh_only, _, _ = crypto(0.0)
ser_sk = pd.Series(skh_only, index=IDX)
ser_75 = pd.Series(R75, index=IDX)
n_sin = n_ok = 0
for y, g in ser_bh.groupby(ser_bh.index.year):
d_bh = maxdd(g.values)
if d_bh > -0.10:
continue
n_sin += 1
d_tp = maxdd(ser_tp.loc[g.index].values)
d_sk = maxdd(ser_sk.loc[g.index].values)
d_75 = maxdd(ser_75.loc[g.index].values)
ok = (d_tp / d_bh) <= 0.75
n_ok += ok
print(f" {y:>6} {d_bh:>9.1%} {d_tp:>9.1%} {d_bh/d_tp:>11.1f}x {d_sk:>10.1%} "
f"{d_75:>10.1%} {'ok' if ok else 'FAIL'}")
print(f" -> anni di sinistro {n_sin}, superati {n_ok}. **Ogni anno e' un anno di sinistro")
print(f" per il buy&hold**: la 'frequenza del sinistro' non e' un evento raro, e' lo")
print(f" stato normale dell'asset. Cio' che e' raro e' la TAGLIA (riga sopra).")
# ----- definizione degli episodi
hr("§1-bis. GLI EPISODI DI SINISTRO — regola di selezione DICHIARATA PRIMA")
print(f" Regola: le {3} finestre di {EPI_LEN} giorni NON sovrapposte con il peggior rendimento")
print(f" buy&hold nel 2019-2023 (greedy sul minimo), PIU' l'FTX 2022-11-05 aggiunto **per")
print(f" NOME** (evento noto, non selezionato dal minimo: e' un controllo, non una scelta).")
cs = np.concatenate(([0.0], np.cumsum(lbh)))
win = np.expm1(cs[EPI_LEN:] - cs[:-EPI_LEN]) # finestra che INIZIA in t
pre_mask = IDX[:len(win)] < pd.Timestamp(RECENT_START, tz="UTC")
cand = np.where(pre_mask)[0]
order = cand[np.argsort(win[cand])]
picked = []
for t in order:
if all(abs(int(t) - p) >= EPI_LEN for p in picked):
picked.append(int(t))
if len(picked) == 3:
break
t_ftx = int(np.searchsorted(IDX, pd.Timestamp("2022-11-05", tz="UTC")))
epis = sorted(picked + [t_ftx])
NAMES = {}
print(f"\n {'episodio':>12} {'dal':>12} {'al':>12} {'b&h 30g':>10} {'peggior 1g b&h':>16}")
for t in epis:
lab = ("COVID" if IDX[t].year == 2020 else
"MAG-2021" if (IDX[t].year, IDX[t].month) == (2021, 5) else
"LUNA" if (IDX[t].year, IDX[t].month) in ((2022, 5), (2022, 6)) else
"FTX" if (IDX[t].year, IDX[t].month) == (2022, 11) else f"EPI-{IDX[t].date()}")
NAMES[t] = lab
seg = bh.values[t:t + EPI_LEN]
print(f" {lab:>12} {str(IDX[t].date()):>12} {str(IDX[t+EPI_LEN-1].date()):>12} "
f"{win[t]:>10.1%} {seg.min():>16.1%}")
crash_days = np.concatenate([np.arange(t, t + EPI_LEN) for t in epis])
freq_day = len(crash_days) / n_full
print(f"\n giorni di sinistro {len(crash_days)} su {n_full} = **{freq_day:.1%} del tempo**")
print(f" ⚠️ AMPIEZZA EFFETTIVA DELLA CODA = **4 EPISODI**. Nessun bootstrap la aumenta.")
print(f" Ogni numero di questo file eredita l'errore di un campione di quattro eventi.")
tick("§1 sinistri")
# ============================================================ §2. IL CONTO DELL'ASSICURAZIONE
hr("§2. IL CONTO DELL'ASSICURAZIONE — premio fuori dal sinistro, indennizzo dentro")
print(" Confronto DUE ripartizioni della coppia crypto, appaiato giorno per giorno:")
print(" LIVE = TP01 75 / SKH01 25 (il libro che gira)")
print(" OPT = TP01 38 / SKH01 62 (l'ottimo di PROP-ALLOC sul campione 2019+)")
in_sin = np.zeros(n_full, bool)
in_sin[crash_days] = True
print(f"\n {'config':>10} {'drift fuori':>12} {'drift dentro':>13} {'Sharpe full':>12} "
f"{'maxDD':>8} {'peggior 1g':>11} {'peggior min 1g':>15}")
prof = {}
for lab, w in (("LIVE 75/25", 0.75), ("OPT 38/62", 0.375), ("TP01 puro", 1.0),
("SKH01 puro", 0.0)):
R, G, _ = crypto(w)
R = R - deluck_offset(w)
out_d = R[~in_sin].mean() * 365.0
in_d = R[in_sin].mean() * 365.0
vol = R.std() * np.sqrt(365.0)
prof[lab] = dict(out=out_d, ind=in_d, sh=R.mean() * 365.0 / vol, dd=maxdd(R),
w1=R.min(), wm=(R + G).min())
p = prof[lab]
print(f" {lab:>10} {p['out']:>12.1%} {p['ind']:>13.1%} {p['sh']:>12.2f} {p['dd']:>8.1%} "
f"{p['w1']:>11.2%} {p['wm']:>15.2%}")
prem = prof["LIVE 75/25"]["out"] - prof["OPT 38/62"]["out"]
ind = prof["LIVE 75/25"]["ind"] - prof["OPT 38/62"]["ind"]
R_live = crypto(0.75)[0] - deluck_offset(0.75)
R_opt = crypto(0.375)[0] - deluck_offset(0.375)
cum_live = float(np.expm1(np.log1p(R_live[in_sin]).sum()))
cum_opt = float(np.expm1(np.log1p(R_opt[in_sin]).sum()))
print(f"\n PREMIO (LIVE - OPT) fuori dal sinistro : {prem:+.2%}/anno = "
f"{prem*100_000:+,.0f} $/anno su un funded da $100k")
print(f" INDENNIZZO (LIVE - OPT) sui 120 giorni di sinistro, CUMULATO (non annualizzato,")
print(f" che su 120 giorni sarebbe una finta precisione): LIVE {cum_live:+.1%} contro "
f"OPT {cum_opt:+.1%} = {cum_live-cum_opt:+.1%}")
print(f" su un funded da $100k: {(cum_live-cum_opt)*100_000:+,.0f} $ per l'INTERO "
f"monte-sinistri di {n_full/365.0:.1f} anni")
print(f" frequenza: {freq_day:.1%} del tempo dentro un sinistro (4 episodi in "
f"{n_full/365.0:.1f} anni = uno ogni {n_full/365.0/4:.1f} anni)")
if ind < 0:
print("\n 🚨 **IL SEGNO E' ROVESCIATO RISPETTO ALLA MIA IPOTESI.** Dentro il sinistro il")
print(" libro con MENO TP01 fa MEGLIO, non peggio. Il meccanismo si legge in una riga:")
print(" TP01 nel crash va FLAT (non perde, non guadagna: e' assicurazione), SKH01 va")
print(" SHORT (guadagna). Su un conto a barriera *guadagnare batte non-perdere*.")
print("\n Rendimento dei 4 episodi, per peso di TP01 (30 giorni ciascuno, chiusura e MINIMO):")
print(f" {'w TP01':>8} " + " ".join(f"{NAMES[t]:>18}" for t in epis) + f" {'peggior min':>13}")
for w in (1.0, 0.75, 0.5, 0.375, 0.25, 0.0):
R, G, _ = crypto(w)
cells = []
wm = 0.0
for t in epis:
seg = R[t:t + EPI_LEN]
gp = G[t:t + EPI_LEN]
eq = np.cumprod(1.0 + seg)
mn = float(np.minimum(eq - 1.0, np.concatenate(([1.0], eq[:-1])) * (1 + seg + gp) - 1.0).min())
wm = min(wm, mn)
cells.append(f"{float(eq[-1]-1.0):+7.1%}/{mn:+7.1%}")
print(f" {w:>8.3f} " + " ".join(f"{c:>18}" for c in cells) + f" {wm:>13.1%}")
print(" (formato: chiusura a 30g / minimo intraday cumulato dell'episodio)")
print("\n ⚠️ QUANTA POTENZA HA QUESTO SEGNO? Con 4 episodi, '4 su 4 a favore' ha un test dei")
print(" segni con p = 0.125 a due code: **non e' significativo al 5%**. Il modo onesto di")
print(" guadagnare potenza non e' ripetere il bootstrap (non crea eventi) ma ESTENDERE la")
print(" stessa regola di selezione a piu' finestre. Le 12 peggiori finestre di 30g NON")
print(" sovrapposte su TUTTO il 2019+ (stessa regola greedy, K da 3 a 12):")
w12 = worst_windows(lbh, IDX, 12, EPI_LEN)
print(f"\n {'finestra':>12} {'b&h 30g':>9} {'LIVE 75/25':>11} {'OPT 38/62':>11} "
f"{'TP01 puro':>10} {'SKH01 puro':>11} {'OPT>LIVE':>9}")
wins = 0
wins_sk = 0
lr_live, lr_opt = np.log1p(R_live), np.log1p(R_opt)
R_tp = crypto(1.0)[0] - deluck_offset(1.0)
R_sk = crypto(0.0)[0] - deluck_offset(0.0)
for t in w12:
a = float(np.expm1(lr_live[t:t + EPI_LEN].sum()))
b = float(np.expm1(lr_opt[t:t + EPI_LEN].sum()))
c = float(np.expm1(np.log1p(R_tp[t:t + EPI_LEN]).sum()))
d = float(np.expm1(np.log1p(R_sk[t:t + EPI_LEN]).sum()))
wins += b > a
wins_sk += d > c
print(f" {str(IDX[t].date()):>12} {win[t]:>9.1%} {a:>11.1%} {b:>11.1%} "
f"{c:>10.1%} {d:>11.1%} {('si' if b > a else 'no'):>9}")
pre = [t for t in w12 if IDX[t] < pd.Timestamp(RECENT_START, tz="UTC")]
wins_pre = sum(1 for t in pre
if float(np.expm1(lr_opt[t:t + EPI_LEN].sum()))
> float(np.expm1(lr_live[t:t + EPI_LEN].sum())))
print(f"\n OPT batte LIVE in **{wins}/12** finestre -> test dei segni "
f"p = {binom_two_sided(wins, 12):.4f}")
print(f" SKH01 puro batte TP01 puro in **{wins_sk}/12** -> p = "
f"{binom_two_sided(wins_sk, 12):.4f}")
print(f" ⚠️ {12-len(pre)} di queste 12 finestre stanno nel 2024+, cioe' DENTRO la finestra")
print(" su cui SKH01 e' stato tarato e su cui PROP-ALLOC ha ottimizzato: la potenza sale")
print(" ma l'indipendenza scende. **Sul solo sotto-campione pre-2024**, che e' quello")
print(f" indipendente: OPT batte LIVE in **{wins_pre}/{len(pre)}** -> p = "
f"{binom_two_sided(wins_pre, len(pre)):.4f}. E' questo il numero da citare.")
tick("§2 conto assicurazione")
# ============================================================ §3. SOTTO BARRIERA
hr("§3. SOTTO BARRIERA — griglia peso TP01 x leva su TRE lenti di campione")
print(f" {len(W_GRID)} pesi x {len(LEVS)} leve x 3 finestre. J = P(pass) x P(vivo 12m), HYRO")
print(f" (eval +10%/-6% statico/-4% giornaliero; funded -6% statico/-4% giornaliero).")
ev_f = PA.boot_idx(n_full, HORIZON, N_PATHS, seed=8230)
fu_f = PA.boot_idx(n_full, HORIZON, N_PATHS, seed=8231)
n_rec = n_full - i_rec
ev_r = PA.boot_idx(n_rec, HORIZON, N_PATHS, seed=8240)
fu_r = PA.boot_idx(n_rec, HORIZON, N_PATHS, seed=8241)
seg_sin = [(t, t + EPI_LEN) for t in epis]
ev_s = pool_paths(seg_sin, HORIZON, N_PATHS, seed=8250)
fu_s = pool_paths(seg_sin, HORIZON, N_PATHS, seed=8251)
print(f"\n {'w TP01':>7} | {'2019+ (col 2022)':>26} | {'2024+ (PROP-ALLOC)':>26} | "
f"{'SOLO-SINISTRO':>26}")
print(f" {'':>7} | {'leva':>6} {'J':>7} {'P(viv)':>7} | {'leva':>6} {'J':>7} {'P(viv)':>7} | "
f"{'leva':>6} {'J':>7} {'P(viv)':>7}")
tab = {}
for w in W_GRID:
R, G, _ = crypto(w)
off = deluck_offset(w)
Rf, Gf = R - off, G
Rr, Gr = Rf[i_rec:], Gf[i_rec:]
lv1, o1 = best_lev(Rf, Gf, ev_f, fu_f)
lv2, o2 = best_lev(Rr, Gr, ev_r, fu_r)
lv3, o3 = best_lev(Rf, Gf, ev_s, fu_s)
tab[w] = dict(full=(lv1, o1), rec=(lv2, o2), sin=(lv3, o3))
print(f" {w:>7.3f} | {lv1:>6.3f} {o1['J']:>7.3f} {o1['p_alive']:>7.1%} | "
f"{lv2:>6.3f} {o2['J']:>7.3f} {o2['p_alive']:>7.1%} | "
f"{lv3:>6.3f} {o3['J']:>7.3f} {o3['p_alive']:>7.1%}")
a_full = max(tab, key=lambda w: tab[w]["full"][1]["J"])
a_rec = max(tab, key=lambda w: tab[w]["rec"][1]["J"])
a_sin = max(tab, key=lambda w: tab[w]["sin"][1]["J"])
print(f"\n argmax: 2019+ -> TP01 {a_full:.3f} | 2024+ -> TP01 {a_rec:.3f} | "
f"SOLO-SINISTRO -> TP01 {a_sin:.3f}")
print(f" ⚠️ RISOLUZIONE: un LIVELLO di J ha errore MC ~{np.sqrt(0.25/N_PATHS):.3f} "
f"(N={N_PATHS}) -> celle entro ~0.02 NON sono distinguibili, e infatti 0.375 e 0.500")
print(" si scambiano il posto fra semi diversi. La lettura e' 'la regione 0.25-0.50',")
print(" non 'il punto 0.375' (stessa lezione del null dei tilt casuali di PROP-ALLOC).")
print(" ⚠️ La colonna SOLO-SINISTRO e' una lente ESTREMA e va letta come tale: l'anno intero")
print(" e' fatto di soli blocchi di crash (4 episodi ricampionati). Non e' uno scenario")
print(" plausibile, e' il **limite superiore del danno**: se il verso della risposta e' lo")
print(" stesso qui e sul 2019+, non dipende dalla frazione di sinistro assunta.")
# replica di controllo del numero pubblicato da PROP-ALLOC (E-ter)
print("\n REPLICA DI CONTROLLO — PROP-ALLOC E-ter pubblica, sul 2019+: argmax della coppia")
print(" crypto **38/62 @0.50x con J 0.521**, e **75/25 J 0.443**. Stesse seed, stessa lente:")
j38 = tab[0.375]["full"]
j75 = tab[0.75]["full"]
print(f" qui: 38/62 @{j38[0]:g}x J {j38[1]['J']:.3f} | 75/25 @{j75[0]:g}x "
f"J {j75[1]['J']:.3f} -> "
f"{'REPLICA OK' if abs(j38[1]['J']-0.521) < 0.02 and abs(j75[1]['J']-0.443) < 0.02 else 'NON REPLICA — leggere con cautela'}")
tick("§3 griglia 3 lenti")
# ============================================================ §4. LO STRESS ESPLICITO
hr("§4. LO STRESS ESPLICITO — iniettare nel 2024+ i crash che il 2024+ non ha")
print(" Costruzione DICHIARATA. Base = blocchi del 2024+ (la finestra di ottimizzazione).")
print(" S1 MISTO: ogni blocco ha probabilita' p di venire dal pool dei 4 sinistri 2019-2023.")
print(f" p = {freq_day:.3f} e' la frequenza STORICA; si girano anche 2x e 4x.")
print(" S2 REPLAY: un episodio VERO innestato per intero, in ordine, a data casuale dentro")
print(" l'anno. E' la domanda operativa: *se il sinistro arriva, chi resta in piedi?*")
print(" ⚠️ Sono SCENARI, non previsioni. E poggiano su 4 episodi.")
# pool = 2024+ (blocchi normali) + gli episodi in coda, su un indice unico
def stress_panel(w: float) -> tuple[np.ndarray, np.ndarray, np.ndarray, list]:
R, G, _ = crypto(w)
off = deluck_offset(w)
Rn, Gn = R[i_rec:] - off, G[i_rec:]
segs = []
Rc, Gc = [], []
cur = len(Rn)
for t in epis:
Rc.append(R[t:t + EPI_LEN] - off)
Gc.append(G[t:t + EPI_LEN])
segs.append((cur, cur + EPI_LEN))
cur += EPI_LEN
Rp = np.concatenate([Rn] + Rc)
Gp = np.concatenate([Gn] + Gc)
return Rp, Gp, np.array([s for s, _ in segs]), segs
base_ev = PA.boot_idx(n_rec, HORIZON, N_PATHS, seed=8260)
base_fu = PA.boot_idx(n_rec, HORIZON, N_PATHS, seed=8261)
print(f"\n S1 — bootstrap MISTO (base 2024+, blocchi di sinistro con probabilita' p)")
print(f" {'w TP01':>7} | " + " | ".join(f"p={p:.3f}: {'J':>6} {'P(viv)':>7}"
for p in (0.0, freq_day, 2 * freq_day, 4 * freq_day)))
s1 = {}
for w in W_GRID:
Rp, Gp, cst, segs = stress_panel(w)
starts = np.concatenate([np.arange(a, b - BLOCK + 1) for a, b in segs])
cells = []
for j, p in enumerate((0.0, freq_day, 2 * freq_day, 4 * freq_day)):
ev = base_ev if p == 0 else mix_paths(base_ev, starts, p, seed=8270 + j)
fu = base_fu if p == 0 else mix_paths(base_fu, starts, p, seed=8280 + j)
lv, o = best_lev(Rp, Gp, ev, fu)
cells.append((lv, o))
s1[w] = cells
print(f" {w:>7.3f} | " + " | ".join(f" @{c[0]:.3g}x {c[1]['J']:>6.3f} "
f"{c[1]['p_alive']:>7.1%}" for c in cells))
for j, p in enumerate((0.0, freq_day, 2 * freq_day, 4 * freq_day)):
am = max(s1, key=lambda w: s1[w][j][1]["J"])
print(f" p={p:.3f} -> argmax TP01 {am:.3f} (J {s1[am][j][1]['J']:.3f}; "
f"75/25 fa {s1[0.75][j][1]['J']:.3f})")
tick("§4 S1 misto")
print(f"\n S2 — REPLAY: un episodio innestato una volta nell'anno funded (365g)")
print(f" Metrica: **P(vivo a 12 mesi) con il crash dentro**, alla leva scelta sul 2024+")
print(f" (cioe' la leva che uno sceglierebbe NON sapendo del crash: e' il punto).")
hdr = " ".join(f"{NAMES[t]:>10}" for t in epis)
print(" ⚠️ DUE colonne di leva, e servono entrambe: la leva 'da 2024+' e' quella che uno")
print(" sceglierebbe non sapendo del crash (realistica, ma CONFONDE peso e leva perche' non")
print(" e' la stessa per tutte le righe); la leva COMUNE 0.50x isola l'effetto del peso.")
print(f"\n {'w TP01':>7} {'leva (da 2024+)':>16} {hdr:>44} {'peggiore':>10} "
f"{'peggiore @0.50x':>16} {'P(viv) 2024+':>13}")
s2 = {}
for w in W_GRID:
Rp, Gp, _, segs = stress_panel(w)
lv = tab[w]["rec"][0]
cells, cells_c = [], []
for k, (a, b) in enumerate(segs):
fu = splice_paths(base_fu, np.arange(a, b), seed=8290 + k)
cells.append(J_of(Rp, Gp, base_ev, fu, lv)["p_alive"])
cells_c.append(J_of(Rp, Gp, base_ev, fu, 0.50)["p_alive"])
s2[w] = (lv, cells, min(cells_c), tab[w]["rec"][1]["p_alive"])
print(f" {w:>7.3f} {lv:>16.3f} " + " ".join(f"{c:>10.1%}" for c in cells)
+ f" {min(cells):>10.1%} {min(cells_c):>16.1%} "
f"{tab[w]['rec'][1]['p_alive']:>13.1%}")
bw_a = max(s2, key=lambda w: min(s2[w][1]))
bw_c = max(s2, key=lambda w: s2[w][2])
print(f"\n argmax della colonna 'peggiore': leva-da-2024+ -> TP01 {bw_a:.3f} | "
f"leva comune 0.50x -> TP01 {bw_c:.3f}")
print(" ⚠️ LE DUE COLONNE DICONO COSE DIVERSE, E LA SECONDA E' A CAMPANA: a leva COMUNE la")
print(f" sopravvivenza al crash SALE col peso di TP01 fino a {bw_c:.3f} ("
f"{s2[0.0][2]:.1%} a TP01=0 -> {s2[bw_c][2]:.1%}) e poi ricade.")
print(" **Quindi TP01 protegge davvero: a pari leva, il libro con piu' difensivo sopravvive")
print(" di piu' al crash.** Il punto e' che quella protezione si compra anche abbassando la")
print(" leva — ed e' esattamente cio' che misura §5. Alla leva che ciascun libro sceglierebbe")
print(" (prima colonna) il vantaggio sparisce, perche' il libro senza TP01 la sceglie piu'")
print(" bassa. E' la firma del **null del de-levering**, qui in veste di 'protezione dal")
print(" crash' invece che di 'meno drawdown'.")
tick("§4 S2 replay")
# ----- replay deterministico del 2022
print("\n S2-bis — REPLAY DETERMINISTICO DEL 2022 (2022-04-01 -> 2022-12-31, 275 giorni).")
print(" Non e' un modello: e' successo. Un conto funded HYRO da $100k che parte il 1° aprile")
print(" 2022 con max-loss STATICO -6% e daily-loss -4%, alla leva scelta sul 2024+:")
a22 = int(np.searchsorted(IDX, pd.Timestamp("2022-04-01", tz="UTC")))
b22 = int(np.searchsorted(IDX, pd.Timestamp("2023-01-01", tz="UTC")))
def replay22(w: float, lev: float) -> tuple:
R, G, _ = crypto(w)
off = deluck_offset(w)
r = (R[a22:b22] - off) * lev
g = G[a22:b22] * lev
eq, mn, bust, why = 1.0, 1.0, None, ""
for i in range(len(r)):
low = eq * (1.0 + r[i] + g[i])
eq = eq * (1.0 + r[i])
mn = min(mn, low)
if bust is None and (low < 0.94 or (r[i] + g[i]) < -0.04):
bust = IDX[a22 + i].date()
why = "daily-loss -4%" if (r[i] + g[i]) < -0.04 else "max-loss -6%"
return eq - 1.0, mn - 1.0, bust, why, float((r + g).min())
print(f"\n {'w TP01':>7} {'leva':>6} {'rend. 9 mesi':>13} {'peggior giorno':>15} "
f"{'min equity':>11} {'esito':>22}")
for w in W_GRID:
lv = tab[w]["rec"][0]
ret, mn, bust, why, wd = replay22(w, lv)
esito = "VIVO" if bust is None else f"BUSTATO {bust}"
print(f" {w:>7.3f} {lv:>6.3f} {ret:>13.1%} {wd:>15.2%} {mn:>11.2%} {esito:>22}")
print("\n Tutti VIVI, e non e' una buona notizia mal letta: **alle leve funded il 2022 non")
print(" era una minaccia per questo libro** (min equity -1.4/-2.4% contro una barriera al")
print(" -6%). Cio' che uccide un conto a barriera non e' il crash del sottostante: e' la")
print(" regola a UN giorno su una giornata qualunque. Ecco fin dove si puo' spingere la leva:")
print(f"\n {'w TP01':>7} {'leva max viva sul 2022':>23} {'rend. a quella leva':>20} "
f"{'primo vincolo che morde':>25}")
for w in W_GRID:
lo, hi = 0.1, 8.0
for _ in range(18):
mid = 0.5 * (lo + hi)
if replay22(w, mid)[2] is None:
lo = mid
else:
hi = mid
ret, _, _, _, _ = replay22(w, lo)
_, _, _, why, _ = replay22(w, hi)
print(f" {w:>7.3f} {lo:>23.2f}x {ret:>20.1%} {why:>25}")
print("\n 📌 La colonna 'leva max viva' CRESCE col peso di TP01 (fino a TP01=0.875; a TP01=1")
print(" ricade perche' li' a mordere non e' piu' il giorno singolo ma la perdita CUMULATA).")
print(" **L'assicurazione e' reale e si vede: piu' TP01 = piu' leva sopportabile nel 2022.**")
print(" E accanto c'e' il prezzo: alla PROPRIA leva massima il libro senza TP01 porta a casa")
print(" il +23.3% e quello a TP01 pieno il -5.9%. TP01 compra capacita' di leva, e la")
print(" capacita' di leva non si converte in rendimento perche' cio' che TP01 toglie e'")
print(" proprio il rendimento. E' il null del de-levering scritto in una riga sola.")
print(" ⚠️ Cambia anche il VINCOLO che morde: sotto TP01 0.50 e' la regola a UN GIORNO,")
print(" sopra e' il max-loss cumulato. Due libri diversi muoiono di due morti diverse, e una")
print(" sola delle due lenti (l'accoppiata) vede la prima.")
tick("§4 replay 2022")
# ============================================================ §5. NULL DEL DE-LEVERING
hr("§5. IL NULL DEL DE-LEVERING — la protezione di TP01 e' altro rispetto a meno leva?")
print(" Il null classico ('esiste k<1 che da' lo stesso rischio con risultato migliore?') qui")
print(" prende la forma giusta per una barriera: **frontiera ISO-SOPRAVVIVENZA**. Per ogni")
print(f" peso si cerca la leva k(w) tale che P(vivo 12m | replay del sinistro PEGGIORE) = "
f"{SURV_TARGET:.0%},")
print(" poi si confronta il PAYOUT atteso sulla finestra normale a quella leva. Se un libro")
print(" con meno TP01 e meno leva paga di piu' a pari sopravvivenza, la protezione di TP01")
print(" e' de-levering; se paga di meno, TP01 compra qualcosa che la leva non compra.")
_SP: dict = {}
def p_alive_stress(w: float, lev: float, seg: tuple) -> float:
"""P(vivo 12m) col replay del sinistro dentro. Solo `funded_sim`: l'eval non entra
nella definizione della frontiera, e calcolarlo raddoppierebbe il costo per niente."""
global N_EVAL
N_EVAL += 1
k = (round(w, 6), seg)
if k not in _SP:
Rp, Gp, _, _ = stress_panel(w)
_SP[k] = (Rp, Gp, splice_paths(base_fu, np.arange(seg[0], seg[1]), seed=8291))
Rp, Gp, fu = _SP[k]
return PA.funded_sim(Rp, Gp, fu, lev, "HYRO", 100_000.0, "coupled")["p_alive"]
print(f"\n {'w TP01':>7} {'k iso-surv':>11} {'P(viv|crash)':>13} {'E[pay/a] normale':>17} "
f"{'P(pass) normale':>16} {'J normale':>11}")
front = {}
for w in W_GRID:
Rp, Gp, _, segs = stress_panel(w)
# episodio peggiore per QUESTO peso (il piu' severo, misurato)
worst_seg = min(segs, key=lambda sg: np.prod(1.0 + Rp[sg[0]:sg[1]]))
lo, hi = 0.05, 3.0
for _ in range(14):
mid = 0.5 * (lo + hi)
if p_alive_stress(w, mid, worst_seg) >= SURV_TARGET:
lo = mid
else:
hi = mid
k = lo
R, G, _ = crypto(w)
off = deluck_offset(w)
o = J_of(R[i_rec:] - off, G[i_rec:], ev_r, fu_r, k)
front[w] = (k, o)
print(f" {w:>7.3f} {k:>11.3f} {p_alive_stress(w, k, worst_seg):>13.1%} "
f"{o['e_payout']:>16,.0f}$ {o['p_pass']:>16.1%} {o['J']:>11.3f}")
kbest = max(front, key=lambda w: front[w][1]["e_payout"])
jbest = max(front, key=lambda w: front[w][1]["J"])
print(f"\n argmax a ISO-SOPRAVVIVENZA: payout -> TP01 {kbest:.3f} | J -> TP01 {jbest:.3f}")
print(" ⚠️ La leva k(w) e' tarata SU un episodio che nel campione di ottimizzazione non c'e':")
print(" e' un esercizio con l'informazione del futuro. Serve a rispondere alla domanda del")
print(" filone ('se sapessi che arriva, come mi allocherei?'), NON e' una politica.")
tick("§5 frontiera iso-sopravvivenza")
# ============================================================ §6. IL TERZO SLEEVE
hr("§6. XS01 — la parte della domanda che NON e' rispondibile, e perche'")
print(" L'ottimo di PROP-ALLOC mette il 50% su XS01, che **nasce nel 2024**: per lui non")
print(" esiste NESSUN sinistro nel campione. Qui non lo si simula col bootstrap della sua")
print(" finestra (sarebbe assumere che nel crash si comporti come in un giorno normale, che")
print(" e' esattamente l'assunzione sotto esame). Si dichiarano TRE comportamenti e si")
print(" guarda se la conclusione dipende da quale:")
print(" A) FLAT nel crash (il gate di dispersione lo spegne) -> generosa")
print(" B) la sua PEGGIORE finestra di 30g del 2024+, contigua e in ordine -> severa")
print(" C) uno shock una-tantum di taglia X sul giorno peggiore del crash -> si cerca")
print(" la X di ROTTURA, cioe' quanto grande deve essere per portare P(vivo) sotto 50%")
xs = pc.xsec_daily_tuples()
xsR = xs["R"].values
xsG = np.minimum(xs["m"].values, xsR) - xsR
lxs = np.log1p(xsR)
cxs = np.concatenate(([0.0], np.cumsum(lxs)))
wxs30 = np.expm1(cxs[EPI_LEN:] - cxs[:-EPI_LEN])
t_xw = int(np.argmin(wxs30))
print(f"\n XS01: {len(xs)} giorni {xs.index[0].date()} -> {xs.index[-1].date()}; peggior "
f"giorno {xsR.min():.2%}, peggior minimo {xs['m'].min():.2%}")
print(f" peggior finestra di {EPI_LEN}g di XS01: {xs.index[t_xw].date()} -> "
f"{xs.index[t_xw+EPI_LEN-1].date()}, cumulato {wxs30[t_xw]:.1%} "
f"(ipotesi B). Per confronto la coppia crypto nel COVID fa -5.4% a TP01 pieno.")
print(" 📌 Il numero che conta e' questo: **la peggior finestra di 30g mai vista da XS01 vale")
print(f" {wxs30[t_xw]:.1%}**. Se nel crash XS01 facesse cosi', al 50% di peso porterebbe")
print(f" {0.5*wxs30[t_xw]:.1%} al libro. La domanda non e' se XS01 sopravvive a cio' che ha")
print(" gia' visto, e' se il suo comportamento nel crash somiglia a cio' che ha gia' visto —")
print(" e su questo il campione non ha NIENTE da dire.")
# ⚠ le ultime tre righe hanno XS01 FISSO al 50%: senza quel controllo si confonderebbe
# "piu' TP01" con "meno XS01", che e' un'altra domanda.
W3 = [(0.75, 0.25, 0.00), (0.375, 0.625, 0.00), (0.50, 0.25, 0.25),
(0.125, 0.375, 0.50), (0.25, 0.25, 0.50), (0.375, 0.125, 0.50)]
com = xs.index.intersection(IDX[i_rec:])
xs_al = xs.reindex(com)
LEV_COM = 0.625 # leva comune per il blocco a XS01=50% (la minima fra le tre righe)
print(f"\n {'TP/SKH/XS':>16} {'ipotesi XS':>12} {'leva':>6} {'J normale':>10} "
f"{'P(viv|replay pegg.)':>21} {'idem @0.625x':>14} {'shock di rottura':>18}")
for cfg in W3:
wtp, wsk, wxs = cfg
base = wtp + wsk
wtp_n = wtp / base if base > 0 else 0.0
Rc, Gc, Ic = crypto(wtp_n)
offc = deluck_offset(wtp_n)
m1 = Ic.isin(com)
Rn = base * (Rc[m1] - offc) + wxs * xs_al["R"].values
Gn = base * Gc[m1] + wxs * (np.minimum(xs_al["m"].values, xs_al["R"].values)
- xs_al["R"].values)
nn = len(Rn)
evn = PA.boot_idx(nn, HORIZON, N_PATHS, seed=8300)
fun = PA.boot_idx(nn, HORIZON, N_PATHS, seed=8301)
lv, on = best_lev(Rn, Gn, evn, fun)
bfu = PA.boot_idx(nn, HORIZON, N_PATHS, seed=8302)
lab = f"{int(round(wtp*100))}/{int(round(wsk*100))}/{int(round(wxs*100))}"
def build(hyp: str, shock: float = 0.0, lev_: float | None = None):
segsR, segsG = [], []
for t in epis:
rc = base * (crypto(wtp_n)[0][t:t + EPI_LEN] - offc)
gc = base * crypto(wtp_n)[1][t:t + EPI_LEN]
if wxs > 0:
if hyp == "A-flat":
rx = np.zeros(EPI_LEN); gx = np.zeros(EPI_LEN)
elif hyp == "B-peggiore":
rx = xsR[t_xw:t_xw + EPI_LEN].copy()
gx = xsG[t_xw:t_xw + EPI_LEN].copy()
else:
rx = np.zeros(EPI_LEN); gx = np.zeros(EPI_LEN)
rx[int(np.argmin(rc))] = -shock
segsR.append(rc + wxs * rx); segsG.append(gc + wxs * gx)
else:
segsR.append(rc); segsG.append(gc)
Rp = np.concatenate([Rn] + segsR)
Gp = np.concatenate([Gn] + segsG)
sg = [(nn + k * EPI_LEN, nn + (k + 1) * EPI_LEN) for k in range(len(epis))]
worst = 1.0
for k, g2 in enumerate(sg):
fu = splice_paths(bfu, np.arange(g2[0], g2[1]), seed=8320 + k)
worst = min(worst, J_of(Rp, Gp, evn, fu, lv if lev_ is None else lev_)["p_alive"])
return worst
if wxs == 0:
print(f" {lab:>16} {'n/a (no XS)':>12} {lv:>6.3f} {on['J']:>10.3f} "
f"{build('A-flat'):>21.1%} {build('A-flat', lev_=LEV_COM):>14.1%} {'-':>18}")
continue
for hyp in ("A-flat", "B-peggiore"):
print(f" {lab:>16} {hyp:>12} {lv:>6.3f} {on['J']:>10.3f} "
f"{build(hyp):>21.1%} {build(hyp, lev_=LEV_COM):>14.1%} {'-':>18}")
lo, hi = 0.0, 0.40 # shock di rottura: bisezione su P(vivo) = 50%
for _ in range(9):
mid = 0.5 * (lo + hi)
if build("C-shock", mid) >= 0.50:
lo = mid
else:
hi = mid
lo2, hi2 = 0.0, 0.40
for _ in range(9):
mid = 0.5 * (lo2 + hi2)
if build("C-shock", mid, lev_=LEV_COM) >= 0.50:
lo2 = mid
else:
hi2 = mid
print(f" {lab:>16} {'C-shock':>12} {lv:>6.3f} {on['J']:>10.3f} {'(50% per def.)':>21} "
f"{'(50% per def.)':>14} {-lo:>17.1%} / {-lo2:.1%} @0.625x")
print("\n Lettura: 'shock di rottura' = il crollo di UN giorno su XS01 che porta P(vivo) al")
print(" 50% quando il crash arriva. Piu' e' piccolo, piu' l'allocazione XS01-pesante e' a un")
print(" giorno di coda dalla morte. Non e' una previsione: e' la taglia della scommessa.")
print(" ⚠️ Le ultime tre righe hanno XS01 FISSO al 50% e differiscono SOLO per come si")
print(" divide il resto fra TP01 e SKH01: e' li' che si legge se TP01 fa da zavorra contro")
print(" l'ignoto di XS01. **La colonna @0.625x e' quella da leggere**, perche' a leva libera")
print(" la riga con meno TP01 sceglie una leva piu' bassa e vince per quello.")
tick("§6 XS01")
# ============================================================ §7. BANDA + VERDETTO
hr("§7. SENSIBILITA' AL DE-LUCK — l'ORDINE fra i pesi sopravvive alla banda?")
print(" La banda d'ancora non e' ri-misurabile sotto questa lente (costo, dichiarato anche da")
print(" PROP-ALLOC). Si applica il fattore misurato x0.89 e si controlla l'unica cosa che")
print(" conta: se l'argmax sui pesi si muove fra 1.00 (nessuna correzione) e 0.60 (la stima")
print(" a occhio del 25/07, poi ri-misurata troppo severa).")
print(f"\n {'fattore':>9} {'argmax 2019+':>14} {'argmax 2024+':>14} {'argmax SOLO-SIN':>16} "
f"{'J(75/25) 2019+':>16} {'J(38/62) 2019+':>16}")
for fct in (1.00, 0.89, 0.60):
best_f = best_r = best_s = None
vals = {}
for w in W_GRID:
R, G, _ = crypto(w)
o = deluck_offset(w, fct)
Rf = R - o
j1 = best_lev(Rf, G, ev_f, fu_f)[1]["J"]
j2 = best_lev(Rf[i_rec:], G[i_rec:], ev_r, fu_r)[1]["J"]
j3 = best_lev(Rf, G, ev_s, fu_s)[1]["J"]
vals[w] = (j1, j2, j3)
best_f = max(vals, key=lambda w: vals[w][0])
best_r = max(vals, key=lambda w: vals[w][1])
best_s = max(vals, key=lambda w: vals[w][2])
print(f" {fct:>9.2f} {best_f:>14.3f} {best_r:>14.3f} {best_s:>16.3f} "
f"{vals[0.75][0]:>16.3f} {vals[0.375][0]:>16.3f}")
hr("§8. I GATE — cosa e' girato, cosa NON e' girato e perche'")
print(" Questo filone non propone una strategia nuova: propone (o rifiuta) un'ALLOCAZIONE per")
print(" un conto che non esiste. I gate del progetto vanno dichiarati uno per uno.")
try:
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
import altlib as A # noqa: E402
for lab, w in (("LIVE 75/25", 0.75), ("OPT 38/62", 0.375)):
R = crypto(w)[0] - deluck_offset(w)
d = pd.Series(R, index=IDX)
n_act = int((np.abs(R) > 1e-12).sum())
n_los = int((R[np.abs(R) > 1e-12] < 0).sum())
f = A.implausible_sharpe(d, n_trades=n_act, n_losing_trades=n_los)
print(f" implausible_sharpe {lab:>11}: implausible={f['implausible']} "
f"(Sharpe {f['sharpe']:.2f}, maxDD {f['maxdd']:.1%}, quota barre in perdita "
f"{f['loss_frac']:.1%}, Calmar {f['calmar']:.2f}) {f['reasons']}")
except Exception as exc: # pragma: no cover
print(f" implausible_sharpe: NON GIRATO ({exc})")
print(" marginal_vs_tp01 / study_family_honest / deflated_sharpe : **NON GIRATI, e non")
print(" sono applicabili** — non c'e' uno stream candidato da confrontare con TP01: gli")
print(" stream sono i due sleeve gia' ammessi, e l'oggetto e' il loro PESO. Il gate dei")
print(" pesi del progetto e' `weights_tilt_null`, che pero' giudica lo SHARPE del libro")
print(" di RICERCA: qui l'obiettivo e' J su una barriera, e il conto e' separato.")
print(" (PROP-ALLOC lo ha girato: gate_pass False. Non cambia nulla qui: nessuna")
print(" proposta di questo file tocca il libro live.)")
print(" causality_ok : **non girato, non applicabile** — nessun segnale nuovo; le serie")
print(" sono quelle di produzione, gia' verificate leak-free a monte.")
print(" anchor_luck_band : **non ri-misurata** (costo, come in PROP-ALLOC). Si e' girata")
print(" la sensibilita' al FATTORE di de-luck (§7), che e' meno: dice se l'ORDINE regge,")
print(" non qual e' la banda. Distorsione dichiarata: la lente e' l'ancora CANONICA e")
print(" SKH01 e' lo sleeve con la fortuna d'ancora piu' grande da restituire (26/07)")
print(" -> **la distorsione e' A FAVORE della conclusione di questo file**, cioe' del")
print(" peso basso su TP01. E' il caveat piu' importante del filone.")
print(" null del de-levering : **GIRATO** (§5, frontiera iso-sopravvivenza) — e superato")
print(" NEL VERSO OPPOSTO a quello atteso: e' la protezione di TP01 a essere replicabile")
print(" con meno leva, e meglio.")
hr("§9. LA RISPOSTA OPERATIVA")
print(f" valutazioni MC totali di questo file: **{N_EVAL}** "
f"(ognuna = {N_PATHS} percorsi di eval + {N_PATHS} di funded)")
print(f"\n {'lente':>26} {'argmax w TP01':>14} {'J argmax':>10} {'J a 0.75':>10} "
f"{'J a 0.375':>11}")
for lab, key in (("2019+ (contiene il 2022)", "full"), ("2024+ (PROP-ALLOC)", "rec"),
("SOLO-SINISTRO", "sin")):
am = max(tab, key=lambda w: tab[w][key][1]["J"])
print(f" {lab:>26} {am:>14.3f} {tab[am][key][1]['J']:>10.3f} "
f"{tab[0.75][key][1]['J']:>10.3f} {tab[0.375][key][1]['J']:>11.3f}")
print("\n ESISTE UN PESO DI COMPROMESSO? Il criterio dichiarato: (a) dentro il 5% del")
print(" massimo di J sulla finestra 2024+ (la 'regione robusta' del funded) E (b) dentro il")
print(" 5% del massimo di J sulla finestra 2019+, che il sinistro ce l'ha dentro.")
jr_max = max(tab[w]["rec"][1]["J"] for w in W_GRID)
jf_max = max(tab[w]["full"][1]["J"] for w in W_GRID)
js_max = max(tab[w]["sin"][1]["J"] for w in W_GRID)
print(f"\n {'w TP01':>7} {'J 2024+':>9} {'entro 5%':>9} {'J 2019+':>9} {'entro 5%':>9} "
f"{'J solo-sin':>11} {'entro 5%':>9} {'compromesso':>13}")
ok_both = []
for w in W_GRID:
a, b, c = (tab[w]["rec"][1]["J"], tab[w]["full"][1]["J"], tab[w]["sin"][1]["J"])
oa, ob, oc = a >= 0.95 * jr_max, b >= 0.95 * jf_max, c >= 0.95 * js_max
if oa and ob:
ok_both.append(w)
print(f" {w:>7.3f} {a:>9.3f} {('si' if oa else 'no'):>9} {b:>9.3f} "
f"{('si' if ob else 'no'):>9} {c:>11.3f} {('si' if oc else 'no'):>9} "
f"{('SI' if (oa and ob) else '-'):>13}")
print(f"\n -> pesi che soddisfano (a) E (b): {ok_both if ok_both else 'NESSUNO'}")
print(f" il libro live 75/25 e' {'dentro' if 0.75 in ok_both else 'FUORI'} da questa regione.")
print("\n LA RISPOSTA, IN QUATTRO RIGHE")
print(" 1) La premessa del caveat e' sbagliata nella FORMA: il 2024+ un crash ce l'ha (b&h")
print(" -60% di drawdown, -39.6% in 20 giorni). Manca la TAGLIA, ~2x sulla coda veloce.")
print(" 2) Il compromesso ESISTE e non e' un compromesso: lo stesso peso (TP01 ~0.375) e'")
print(" l'argmax sulla finestra con il 2022 dentro E su quella senza. Il caveat di")
print(" PROP-ALLOC e' onesto come cautela e **falsificato come previsione**.")
print(" 3) TP01 protegge davvero (a pari leva sopravvive di piu' al crash, e alza la leva")
print(" sopportabile nel 2022 da 0.93x a 2.20x) ma la stessa protezione si compra piu' a")
print(" buon mercato ABBASSANDO LA LEVA: a iso-sopravvivenza il payout e' massimo a")
print(" TP01 0.125-0.250 e cala monotonamente da li'. **7ª occorrenza del null del")
print(" de-levering**, in una veste nuova: 'protezione dal crash' invece di 'meno DD'.")
print(" 4) Il rischio vero dell'ottimo di PROP-ALLOC NON e' TP01 sotto-pesato: e' XS01 al")
print(" 50% con un comportamento nel crash che il campione non contiene. Un solo giorno")
print(" a ~-10% su XS01 dimezza la sopravvivenza quando il crash arriva.")
print(" ⚠️ IPOTESI MIA, NATA E REFUTATA IN SESSIONE: avevo scritto che TP01 fa da")
print(" ZAVORRA contro quell'ignoto (a XS01 25% serviva uno shock -12.3% contro -8.0% a")
print(" XS01 50%). Il controllo a **XS01 FISSO al 50% e leva comune** lo smonta: lo shock")
print(" di rottura e' -10.1 / -10.2 / -10.2% muovendo TP01 da 12% a 38%, cioe' INVARIANTE,")
print(" e la sopravvivenza sotto l'ipotesi severa CALA (74.1 -> 72.6 -> 64.7%). Cio' che")
print(" protegge dall'ignoto di XS01 e' **meno XS01**, non piu' TP01: il primo numero")
print(" confrontava due pesi di XS01 diversi e attribuiva a TP01 il merito dell'altro.")
print("\n ⚠️ COSA MI SMENTIREBBE: un crash in cui SKH01 NON si gira short in tempo (gap")
print(" notturno oltre lo stop, o venue fermo). Il campione ne ha quattro e in tutti e")
print(" quattro SKH01 ha girato. Basterebbe un episodio in cui SKH01 perde quanto il")
print(" sottostante per rovesciare §2 e §5 — e con n=4 quell'episodio non e' improbabile.")
print("\n" + "=" * 106)
print(f" fatto in {time.time()-t0:.0f}s | valutazioni: {N_EVAL}")
print("=" * 106)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink