Adriano Dal Pastro
874 lines
46 KiB
Python
874 lines
46 KiB
Python
"""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}")
|
|
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(" ⚠️ 5 di queste 12 finestre stanno nel 2024+, cioe' DENTRO la finestra su cui")
|
|
print(" SKH01 e' stato tarato e su cui PROP-ALLOC ha ottimizzato: la potenza sale ma")
|
|
print(" l'indipendenza scende. Il test va letto come **una diagnostica, non una prova**.")
|
|
|
|
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(" ⚠️ 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(" Lettura: se la sopravvivenza al crash SALISSE col peso di TP01, l'assicurazione")
|
|
print(" servirebbe. **A leva comune non sale**: la curva e' piatta-decrescente. Il difensivo")
|
|
print(" non compra sopravvivenza su questa barriera, perche' cio' che la barriera punisce e'")
|
|
print(" la PERDITA nel crash, e nel crash TP01 perde (prende la prima gamba giu' e poi va")
|
|
print(" flat) mentre SKH01 si gira short e GUADAGNA.")
|
|
|
|
|
|
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 leva di rottura e' MONOTONA DECRESCENTE nel peso di TP01 solo se la")
|
|
print(" protezione fosse reale. Si legga la colonna: se il libro con piu' TP01 rompe PRIMA,")
|
|
print(" l'assicurazione non sta assicurando la cosa giusta.")
|
|
|
|
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.")
|
|
W3 = [(0.75, 0.25, 0.00), (0.375, 0.625, 0.00), (0.25, 0.25, 0.50), (0.125, 0.125, 0.75),
|
|
(0.50, 0.25, 0.25)]
|
|
com = xs.index.intersection(IDX[i_rec:])
|
|
xs_al = xs.reindex(com)
|
|
print(f"\n {'TP/SKH/XS':>16} {'ipotesi XS':>12} {'leva':>6} {'J normale':>10} "
|
|
f"{'P(viv|replay peggiore)':>24} {'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):
|
|
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)["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'):>24.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):>24.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
|
|
print(f" {lab:>16} {'C-shock':>12} {lv:>6.3f} {on['J']:>10.3f} {'(50% per def.)':>24} "
|
|
f"{-lo:>17.1%}")
|
|
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.")
|
|
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}: {f}")
|
|
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" + "=" * 106)
|
|
print(f" fatto in {time.time()-t0:.0f}s | valutazioni: {N_EVAL}")
|
|
print("=" * 106)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|