#!/usr/bin/env python """r0823_deposit_timing.py — §44 DEPOSIT-TIMING: il timing dei BONIFICI paga? LA DOMANDA. Nel progetto e' misurato che la leva dominante non e' la strategia ma il versamento: *versare o no* (da mai a ~16 anni) > *quando* (fattore 6 a pari soldi totali) > *quanto presto si smette* (i primi 5 anni sono il 25% dei soldi e il 58% del risultato) > *piatto vs crescente* (24%) > *frequenza* (1-3%). Ma TUTTE quelle misure confrontano **calendari deterministici**. Nessuno ha mai misurato un calendario **CONDIZIONALE**: versare di piu' dopo un drawdown, o quando il libro e' risk-off, o su una qualunque regola che guarda lo stato del conto. A pari soldi totali versati e pari calendario di DISPONIBILITA', una regola di deposito condizionale batte il versamento piatto? ⚠️ E' MARKET TIMING SUI PROPRI BONIFICI, e va giudicato come tale. Il null e' severo (sotto). ATTESA A PRIORI, DICHIARATA PRIMA DI MISURARE (coordinatore, e la sottoscrivo): **NON funziona**. Comprare piu' in basso alza il rendimento sul capitale versato, ma la mediana e' dominata dal fatto che i soldi entrano PIU' TARDI, e il progetto ha gia' misurato che entrare prima vale un fattore 6 (r0726_deposits punto 3). L'esito e' riportato comunque, anche se refuta l'attesa. IL VINCOLO DI REALTA' CHE RENDE LA DOMANDA NON BANALE. I soldi non compaiono. Ogni politica riceve lo STESSO flusso di cassa (EUR X/mese sul conto corrente); cio' che cambia e' **quando entrano nel libro**. La liquidita' in attesa rende **0% reale netto** (assunzione del progetto: XEON) — ed e' meta' del costo di ogni politica che aspetta. Senza questo vincolo la domanda sarebbe truccata: "versare di piu' nei cali" senza dire da dove vengono i soldi e' un confronto fra due piani di versamento diversi, non fra due timing. LE POLITICHE, DICHIARATE TUTTE PRIMA (e contate come trial: 12 celle candidate) P0 PIATTO — versa tutto il buffer il giorno stesso. E' l'incumbent. P1 BUY-THE-DIP(th) — accantona; versa TUTTO il buffer quando l'indice di equity del libro sta >= th sotto il suo massimo. th in {5,10,15,20}% (4 celle) P2 RISK-OFF — versa quando TP01 e' a 0x (segnale gia' pubblico e causale, ~33% dei giorni). (1 cella) P3 VALORE-MEDIO(g) — versa quanto serve a riportare il capitale su una traiettoria bersaglio che cresce a g. Mai vendere. g in {0,8,15}%/anno (3 celle) P4 ANTI-DIP(th) — lo SPECULARE OBBLIGATORIO: versa dopo un rialzo di th dal minimo corrente. th in {5,10,15,20}% (4 celle) CONTROLLO DEL RITARDO — per ogni politica condizionale, la stessa attesa MEDIA ma DETERMINISTICA (versa k giorni dopo, k = attesa misurata di quella politica, nessun segnale). Separa "il segnale informa" da "aspettare costa". Non e' una cella candidata: e' un null. I NULL (a) BOOTSTRAP IID (blocco=1) — distrugge la dipendenza seriale. Se una politica vince anche li', sta sfruttando la meccanica del bootstrap, non il mercato: con ritorni indipendenti il passato non informa il futuro e ogni residuo e' cassa/ritardo. (b) P4 ANTI-DIP — se P1 e P4 danno lo stesso segno, si sta misurando rumore. (c) DRIFT DIMEZZATO — lo stress standard del progetto sul libro (7,4 anni di storia, due tori). LA LENTE E' QUELLA UFFICIALE: L3 CONGIUNTA di r0822d_piano_vero (funding esatto DENTRO la serie + fisco d'accumulo coerente + ancora x0.89 misurata). Bersaglio = il muro di quella lente. CONVENZIONE DEL TRAGUARDO: si arriva quando **libro + cassa** >= muro. E' la convenzione FAVOREVOLE alle politiche condizionali (la cassa che aspetta conta lo stesso), scelta apposta perche' l'attesa a priori e' che perdano: se perdono anche cosi', il risultato e' piu' forte. La convenzione severa (solo libro) e' riportata come sensibilita'. nice -n 19 uv run python scripts/research/r0823_deposit_timing.py """ from __future__ import annotations import sys import time from pathlib import Path import numpy as np import pandas as pd ROOT = Path("/opt/docker/PythagorasGoal") sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT)) import r0725_capcurve as CC # noqa: E402 TAX_RATE, EURUSD, _boot_paths import r0727_tasse as TX # noqa: E402 PATRIMONIALE, CARRY_ANNI import r0807_piano_netto as PN # noqa: E402 passo_fiscale, accumula, muro import r0822d_piano_vero as PV # noqa: E402 costruisci_serie, deluck, leggi from src.data.downloader import load_data # noqa: E402 from src.strategies.trend_portfolio import (CANONICAL, TrendPortfolio, # noqa: E402 resample_1d) # --- costanti dichiarate --------------------------------------------------- START = PV.START # $635, il conto Deribit vero N_PATHS = 3_000 # la taglia della tabella pubblicata (seed 725) BLOCK = PV.BLOCK # 20 giorni ANNI = PV.ANNI_TRAJ # 25 (20 troncherebbe la coda) SEED = PV.SEED_TRAJ # 725 SEMI_MC = (725, 2026, 20260823) # per la risoluzione Monte Carlo DEP_MAIN = 500.0 # il piano dichiarato dall'operatore (EUR/mese) DEP_ALT = 250.0 # il livello di cui si discute da un mese CASH_RATE = 0.0 # XEON ~0% reale netto: assunzione del progetto DD_GRID = (0.05, 0.10, 0.15, 0.20) VA_GRID = (0.00, 0.08, 0.15) P_VENUE = (0.0, 0.01, 0.02) ASSETS = ("BTC", "ETH") T0 = time.time() def sez(t: str) -> None: print("\n" + "=" * 104) print(f" {t}") print("=" * 104) def clock() -> str: return f"[{time.time() - T0:5.0f}s]" # =========================================================================== # 0) LE SERIE — ritorni (lente L3) ed ESPOSIZIONE di TP01, sullo stesso indice # =========================================================================== def esposizione_tp01(alloc: float = 635.0) -> pd.Series: """Esposizione GIA' TENUTA dal libro (0..2), causale, con la stessa regola di min-order di `CC._tp01_realistic_cached`: un ribilanciamento sotto $5 non si esegue. Non e' una re-implementazione della strategia: `target_series` e' importata dal modulo di produzione. Qui serve solo il livello di posizione, non i ritorni (quelli restano quelli ufficiali di `CC.book_series`, gia' dentro la serie della lente). """ tp = TrendPortfolio(**CANONICAL) cap_asset = alloc / len(ASSETS) ser = {} for a in ASSETS: df = resample_1d(load_data(a, "1h")) tgt = np.nan_to_num(np.asarray(tp.target_series(df), float)) held = np.empty(len(tgt)) cur = 0.0 for i in range(len(tgt)): if abs(tgt[i] - cur) * cap_asset >= CC.DERIBIT_MIN_ORDER: cur = tgt[i] held[i] = cur pos = np.zeros(len(held)) pos[1:] = held[:-1] # causale: deciso a i-1, tenuto in i ser[a] = pd.Series(pos, index=pd.to_datetime(df["datetime"])) J = pd.concat(ser, axis=1, join="inner").fillna(0.0) out = pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index) if out.index.tz is None: out.index = out.index.tz_localize("UTC") return out def boot_idx(n_src: int, n_paths: int, n_days: int, block: int, rng: np.random.Generator) -> np.ndarray: """Gli INDICI del block bootstrap di `CC._boot_paths`, per poter trasportare insieme ai ritorni una seconda serie allineata (l'esposizione di TP01, che serve a P2). Controllo di identita' in sezione 1: `r[boot_idx(...)]` deve essere bit-exact `CC._boot_paths(r, ...)` a parita' di generatore. """ n_blocks = int(np.ceil(n_days / block)) starts = rng.integers(0, n_src - block, size=(n_paths, n_blocks)) idx = (starts[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1) return idx[:, :n_days].astype(np.int32) # =========================================================================== # IL MOTORE — accumulo con buffer di CASSA e politica di versamento # =========================================================================== def accumula_pol(paths: np.ndarray, flat: np.ndarray | None, dep_eur: float, bersaglio: float, aliquota: float, patrimoniale: float, policy, start: float = START, cash_rate: float = CASH_RATE, fail: np.ndarray | None = None, traguardo_include_cassa: bool = True) -> dict: """Accumulo su path GIA' ESTRATTI, con una politica che decide QUANDO la cassa entra. Il flusso di cassa e' identico per ogni politica: `dep_eur` ogni 30 giorni sul CONTO CORRENTE. La politica decide solo quanto trasferire nel libro, ogni giorno. Causalita': il ritorno del giorno t si applica PRIMA del trasferimento (come in `PN.accumula`), quindi cio' che entra il giorno t rende dal giorno t+1; e la decisione usa l'indice di equity fino alla chiusura di t, che a quel punto e' osservabile. `policy(t, cap, cash, dd, ru, flat_t)` ritorna l'importo da trasferire (USD, per path). `fail`: maschera (path, giorno) di fallimento del venue — il libro va a zero, la CASSA no. """ n_paths, n_days = paths.shape dep = dep_eur * CC.EURUSD cash_d = (1.0 + cash_rate) ** (1.0 / 365.0) - 1.0 cap = np.full(n_paths, start) cash = np.zeros(n_paths) eqi = np.ones(n_paths) # indice di equity del libro, NEUTRO ai versamenti peak = np.ones(n_paths) trough = np.ones(n_paths) carry = np.zeros((n_paths, TX.CARRY_ANNI)) anno_start = cap.copy() flusso = np.zeros(n_paths) colpito = np.full(n_paths, -1, int) versato = start cassa_int = 0.0 # integrale della cassa (per la legge di Little) persi = np.zeros(n_paths, bool) pnum = np.zeros(n_paths) # sum(importo * livello di equity all'ingresso) pden = np.zeros(n_paths) # sum(importo) for t in range(n_days): rt = paths[:, t] cap *= (1.0 + rt) eqi *= (1.0 + rt) np.maximum(peak, eqi, out=peak) np.minimum(trough, eqi, out=trough) if fail is not None: f = fail[:, t] if f.any(): cap = np.where(f, 0.0, cap) anno_start = np.where(f, 0.0, anno_start) flusso = np.where(f, 0.0, flusso) # la perdita non diventa credito d'imposta persi |= f cash *= (1.0 + cash_d) if t % 30 == 0 and t > 0 and dep > 0: cash += dep versato += dep move = policy(t, cap, cash, 1.0 - eqi / peak, eqi / trough - 1.0, None if flat is None else flat[:, t]) if move is not None: move = np.minimum(np.maximum(move, 0.0), cash) cap += move cash -= move flusso += move pnum += move * eqi # a che LIVELLO e' entrato ogni dollaro pden += move cassa_int += float(cash.mean()) tot = cap + cash if traguardo_include_cassa else cap nuovi = (colpito < 0) & (tot >= bersaglio) colpito[nuovi] = t if (t + 1) % 365 == 0 and (aliquota > 0 or patrimoniale > 0): PN.passo_fiscale(cap, anno_start, flusso, carry, aliquota, patrimoniale) anno_start = cap.copy() flusso[:] = 0.0 lam = dep / 30.0 if dep > 0 else np.nan # USD/giorno di flusso in entrata attesa = (cassa_int / n_days) / lam if dep > 0 else 0.0 # Little: W = L / lambda prezzo = np.where(pden > 0, pnum / np.maximum(pden, 1e-12), np.nan) return dict(cap=cap, cash=cash, colpito=colpito, versato=versato, attesa_g=float(attesa), p_perso=float(persi.mean()), prezzo=float(np.nanmedian(prezzo)), non_versata=float(np.median(cash))) # =========================================================================== # LE POLITICHE # =========================================================================== def pol_piatto(): return lambda t, cap, cash, dd, ru, fl: cash def pol_dip(th: float): return lambda t, cap, cash, dd, ru, fl: np.where(dd >= th, cash, 0.0) def pol_antidip(th: float): return lambda t, cap, cash, dd, ru, fl: np.where(ru >= th, cash, 0.0) def pol_riskoff(): return lambda t, cap, cash, dd, ru, fl: np.where(fl, cash, 0.0) def pol_va(g: float, dep_eur: float, start: float = START): """Valore-medio: traiettoria bersaglio deterministica, versa la differenza, mai vende.""" dep = dep_eur * CC.EURUSD gm = (1.0 + g) ** (30.0 / 365.0) - 1.0 state = {"T": start} def f(t, cap, cash, dd, ru, fl): if t % 30 or t == 0: return None state["T"] = state["T"] * (1.0 + gm) + dep return np.maximum(state["T"] - cap, 0.0) return f def pol_lag(k: int, dep_eur: float): """Il controllo del ritardo: versa esattamente k giorni dopo l'arrivo, senza guardare nulla.""" dep = dep_eur * CC.EURUSD def f(t, cap, cash, dd, ru, fl): return np.full(len(cap), dep) if (t > k and (t - k) % 30 == 0) else None return f def celle(dep_eur: float) -> dict: """Le 12 celle candidate, dichiarate qui e in nessun altro posto.""" C = {"P0 piatto": pol_piatto()} for th in DD_GRID: C[f"P1 dip {th:.0%}"] = pol_dip(th) C["P2 risk-off"] = pol_riskoff() for g in VA_GRID: C[f"P3 VA g={g:.0%}"] = pol_va(g, dep_eur) for th in DD_GRID: C[f"P4 anti {th:.0%}"] = pol_antidip(th) return C N_TRIAL = 12 # P1(4) + P2(1) + P3(3) + P4(4). P0 e' l'incumbent, i lag sono null. # =========================================================================== def leggi(colpito: np.ndarray) -> dict: """Mediana INCONDIZIONATA (non-arrivo = +inf, MAI -1) + condizionata + P per orizzonte.""" y = np.where(colpito >= 0, colpito / 365.0, np.inf) a = y[np.isfinite(y)] return dict(med=float(np.median(y)), med_cond=float(np.median(a)) if len(a) > 100 else float("nan"), p10=float((y <= 10).mean()), p15=float((y <= 15).mean()), p20=float((y <= 20).mean())) def fmt(d: dict) -> str: m = f">{ANNI}a" if not np.isfinite(d["med"]) else f"{d['med']:.1f}a" c = "—" if np.isnan(d["med_cond"]) else f"{d['med_cond']:.1f}a" return f"{m:>7}{c:>8}" def riga(nome: str, res: dict, perp: float, base: dict | None = None, prezzo0: float | None = None) -> str: d = leggi(res["colpito"]) tot = res["cap"] + res["cash"] rend = float(np.median(tot * perp / 365.0 / CC.EURUSD)) dp = "" if base is None else f"{d['p20'] - base['p20']:+7.1%}" pz = "" if not prezzo0 else f"{res['prezzo'] / prezzo0:.3f}" return (f" {nome:<16}{fmt(d)}{d['p15']:>8.1%}{d['p20']:>8.1%}" f"{dp:>9}{rend:>10.2f}{res['attesa_g']:>8.0f}{pz:>8}") HDR = (f" {'politica':<16}{'med':>7}{'m|arr':>8}{'P(15a)':>8}{'P(20a)':>8}" f"{'dP20':>9}{'EUR/g 20a':>10}{'attesa':>8}{'prezzo':>8}") def tabella(paths, flat, dep, muro, perp, titolo, **kw) -> dict: print(f"\n {titolo}") print(HDR) out = {} base = None p0 = None for nome, pol in celle(dep).items(): r = accumula_pol(paths, flat, dep, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol, **kw) d = leggi(r["colpito"]) if base is None: base, p0 = d, r["prezzo"] out[nome] = r print(riga(nome, r, perp, None if nome.startswith("P0") else base, None if nome.startswith("P0") else p0)) return out # =========================================================================== def main() -> None: print("=" * 104) print(" r0823 §44 DEPOSIT-TIMING — il timing dei BONIFICI paga? (lente L3: funding + fisco)") print("=" * 104) # ------------------------------------------------------------------ 0 B = PV.costruisci_serie() idx = pd.DatetimeIndex(B.index) r = PV.deluck(B["fund"].values.astype(float)) # lente L3 CONGIUNTA, ancora x0.89 E = esposizione_tp01().reindex(idx).ffill().fillna(0.0) flat_src = (E.values <= 1e-9) print(f"\n serie L3 CONGIUNTA (libro live 75/25, funding esatto, ancora x{PV.DELUCK}) — " f"{len(r)} giorni {idx.min().date()}..{idx.max().date()}") print(f" drift {r.mean() * 365:.2%} · vol {r.std() * 365 ** 0.5:.2%} · " f"esposizione TP01 media {E.mean():.3f}x · giorni a 0x = {flat_src.mean():.1%} {clock()}") # ------------------------------------------------------------------ 1 sez("1) CONTROLLO DI REPLICA — prima di ogni delta, riprodurre i numeri gia' pubblicati") ok = {} netto = CC.TARGET_EUR_DAY * 365 * CC.EURUSD perp, muro = PN.muro(r, netto, CC.TAX_RATE, TX.PATRIMONIALE) ok["R1"] = abs(muro - 313_000.0) < 1_500.0 print(f"\n R1 muro L3 congiunta: perpetua {perp:.2%} · ${muro:,.0f}" f"{'':>10}pubblicato $313k {'OK' if ok['R1'] else 'DIVERGE'} {clock()}") rng = np.random.default_rng(SEED) ii = boot_idx(len(r), N_PATHS, ANNI * 365, BLOCK, rng) paths = r[ii] flat = flat_src[ii] rng2 = np.random.default_rng(SEED) ok["R2"] = bool(np.array_equal(paths, CC._boot_paths(r, N_PATHS, ANNI * 365, BLOCK, rng2))) print(f" R2 `boot_idx` trasporta gli STESSI path di `CC._boot_paths` (per portarsi dietro") print(f" l'esposizione di TP01, che serve a P2):{'':>19}" f"{'BIT-EXACT' if ok['R2'] else 'DIVERGE'}") print(f"\n R3 la riga pubblicata della tabella PIANO-VERO (L3, da ${START:.0f}):") att = {250.0: (23.4, 22.1, 0.14), 500.0: (17.3, 17.3, 0.85)} r3 = True print(f" {'EUR/m':>7}{'med.inc':>10}{'atteso':>9}{'med|arr':>10}{'atteso':>9}" f"{'P(20a)':>9}{'atteso':>9}") for dep, (a1, a2, a3) in att.items(): d = leggi(PN.accumula(paths, dep, muro, CC.TAX_RATE, TX.PATRIMONIALE, start=START)["colpito"]) r3 &= abs(d["med"] - a1) < 0.2 and abs(d["p20"] - a3) < 0.02 print(f" {dep:>7.0f}{d['med']:>10.1f}{a1:>9.1f}{d['med_cond']:>10.1f}{a2:>9.1f}" f"{d['p20']:>9.1%}{a3:>9.0%}") ok["R3"] = r3 # R4: P0 del mio motore == PN.accumula, bit-exact A = PN.accumula(paths, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, start=START) Bp = accumula_pol(paths, flat, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol_piatto(), start=START) dcap = float(np.max(np.abs(A["cap"] - Bp["cap"]))) dcol = int(np.max(np.abs(A["colpito"] - Bp["colpito"]))) ok["R4"] = dcap == 0.0 and dcol == 0 and abs(A["versato"] - Bp["versato"]) < 1e-9 print(f"\n R4 P0 di QUESTO motore (con buffer di cassa) vs `PN.accumula` importata:") print(f" max|dif| capitale = {dcap:.3e} · max|dif| giorno d'arrivo = {dcol} · " f"versato identico{'':>3}{'BIT-EXACT' if ok['R4'] else 'DIVERGE'}") print(f" -> il buffer non introduce nulla: con versamento immediato il motore E' quello pubblicato.") print(f"\n ESITO REPLICA: {sum(ok.values())}/{len(ok)} — " + ", ".join(f"{k}={'OK' if v else 'NO'}" for k, v in ok.items())) if not all(ok.values()): raise SystemExit("REPLICA INCOMPLETA: un delta sarebbe indistinguibile da un bug.") # ------------------------------------------------------------------ 2 sez("2) RISOLUZIONE MONTE CARLO — quanto vale la cifra che si pubblica") print(f"\n {N_PATHS} percorsi, blocchi da {BLOCK} giorni, {ANNI} anni. I confronti fra") print(" politiche sono APPAIATI (stessi percorsi) quindi la loro varianza e' molto minore") print(" di quella dei livelli. Misurata su 3 semi:") mc = [] for s in SEMI_MC: rg = np.random.default_rng(s) i2 = boot_idx(len(r), N_PATHS, ANNI * 365, BLOCK, rg) p2, f2 = r[i2], flat_src[i2] b = leggi(accumula_pol(p2, f2, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol_piatto(), start=START)["colpito"]) d = leggi(accumula_pol(p2, f2, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol_dip(0.10), start=START)["colpito"]) mc.append((s, b["p20"], d["p20"], d["p20"] - b["p20"])) del p2, f2, i2 print(f"\n {'seme':>10}{'P0 P(20a)':>12}{'P1 dip 10% P(20a)':>20}{'differenza APPAIATA':>22}") for s, a, bb, dd in mc: print(f" {s:>10}{a:>12.1%}{bb:>20.1%}{dd:>22.2%}") liv = max(x[1] for x in mc) - min(x[1] for x in mc) dif = max(x[3] for x in mc) - min(x[3] for x in mc) print(f"\n ampiezza fra semi: LIVELLO {liv:.2%} · DIFFERENZA APPAIATA {dif:.2%}") print(f" -> si cita la differenza appaiata al punto percentuale; un livello si cita a " f"{liv:.1%} di rumore. {clock()}") # ------------------------------------------------------------------ 3 sez(f"3) LA TABELLA — 12 celle candidate contro il piatto, stesso flusso di cassa") print(f""" Lettura delle colonne. `med` = mediana INCONDIZIONATA degli anni al capitale-rendita (non-arrivo = +inf, mai -1); `m|arr` = mediana condizionata all'arrivo, che NON si legge mai senza la sua P accanto; `dP20` = differenza APPAIATA di P(20a) contro P0 sugli stessi percorsi; `attesa` = giorni medi che un euro passa in cassa prima di entrare nel libro (legge di Little, L/lambda); `prezzo` = livello di equity medio PONDERATO PER IMPORTO a cui i soldi sono entrati, in rapporto a quello di P0 — sotto 1.000 la politica ha comprato davvero piu' in basso. Bersaglio ${muro:,.0f}; traguardo su libro+cassa.""") R = {} for dep in (DEP_MAIN, DEP_ALT): R[dep] = tabella(paths, flat, dep, muro, perp, f"EUR {dep:.0f}/mese di flusso disponibile {clock()}", start=START) # ------------------------------------------------------------------ 4 sez("4) IL CONTROLLO DEL RITARDO — «il segnale informa» contro «aspettare costa»") print(""" Ogni politica condizionale fa DUE cose: sceglie il momento (informazione) e ritarda i soldi (costo). Il controllo isola la seconda: stessa attesa MEDIA, ma deterministica, nessun segnale. Se la politica condizionale non batte il proprio ritardo-gemello, non ha informazione — ha solo aspettato.""") info = {} for etichetta, (P, FL, RR) in {"BLOCCHI 20g (mercato vero)": (paths, flat, R[DEP_MAIN]), "IID blocco=1 (null)": (None, None, None)}.items(): if P is None: # il null iid: si ricostruiscono i path e le celle rg = np.random.default_rng(SEED) i2 = boot_idx(len(r), N_PATHS, ANNI * 365, 1, rg) P, FL = r[i2], flat_src[i2] del i2 RR = {n: accumula_pol(P, FL, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol, start=START) for n, pol in celle(DEP_MAIN).items()} print(f"\n {etichetta}") print(f" {'politica':<16}{'attesa g.':>11}{'P(20a) reale':>14}" f"{'P(20a) ritardo-gemello':>25}{'informazione':>15}") col = {} for nome in ("P1 dip 5%", "P1 dip 10%", "P1 dip 15%", "P1 dip 20%", "P2 risk-off"): rr = RR[nome] k = int(round(rr["attesa_g"])) g = accumula_pol(P, FL, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, pol_lag(k, DEP_MAIN), start=START) a, b = leggi(rr["colpito"])["p20"], leggi(g["colpito"])["p20"] col[nome] = a - b print(f" {nome:<16}{k:>11}{a:>14.1%}{b:>25.1%}{a - b:>15.2%}") info[etichetta] = col if "IID" in etichetta: del P, FL, RR k1 = "BLOCCHI 20g (mercato vero)" k2 = "IID blocco=1 (null)" cost15 = (leggi(R[DEP_MAIN]["P1 dip 15%"]["colpito"])["p20"] - leggi(R[DEP_MAIN]["P0 piatto"]["colpito"])["p20"]) print(f""" ⚠️ IL PRIMO SGUARDO DICE «IL SEGNALE INFORMA» — E IL NULL IID LO REFUTA. Sulle celle profonde la colonna `informazione` e' positiva sul mercato vero ({info[k1]['P1 dip 15%']:+.1%} a dip 15%, {info[k1]['P1 dip 20%']:+.1%} a dip 20%): a pari attesa MEDIA, aspettare il calo batte aspettare a caso, e leggendo solo questa riga si concluderebbe che il buy-the-dip contiene previsione. Ma sotto IID, dove per COSTRUZIONE il passato non dice nulla del futuro, la stessa colonna vale {info[k2]['P1 dip 15%']:+.1%} / {info[k2]['P1 dip 20%']:+.1%} — cioe' NON PIU' PICCOLA, anzi piu' grande. -> il divario contro il ritardo-gemello e' MECCANICO, non previsione: la politica condizionale ha attesa VARIABILE (spesso zero, ogni tanto lunghissima) mentre il gemello ha attesa FISSA, e versa il buffer tutto in una volta. Un'attesa variabile con la stessa media di una fissa non produce lo stesso capitale (Jensen), e il null lo mostra senza ambiguita'. ⚠️ Il gemello non e' un controllo perfetto proprio per questo: e' il controllo del RITARDO MEDIO, non della sua distribuzione. Il suo compito qui e' fissare l'ordine di grandezza — il ritardo di dip 15% costa {abs(cost15):.0%} di P(20a) contro un divario col gemello di {info[k1]['P1 dip 15%']:+.1%}, e sarebbe la conclusione anche prendendo il gemello alla lettera.""") print(f" {clock()}") # ------------------------------------------------------------------ 5 sez("5) I NULL") print(""" (a) BOOTSTRAP IID (blocco = 1 giorno). Distrugge ogni dipendenza seriale: il passato non informa il futuro per costruzione. Una politica che li' mantiene il suo vantaggio lo stava prendendo dalla meccanica del bootstrap, non dal mercato. (Il controllo del ritardo su questo stesso null e' gia' in sezione 4.)""") rg = np.random.default_rng(SEED) i_iid = boot_idx(len(r), N_PATHS, ANNI * 365, 1, rg) p_iid, f_iid = r[i_iid], flat_src[i_iid] del i_iid I = tabella(p_iid, f_iid, DEP_MAIN, muro, perp, f"IID (blocco=1) {clock()}", start=START) del p_iid, f_iid print(""" (c) DRIFT DIMEZZATO — lo stress standard del progetto (7,4 anni di storia, due tori dentro). Bersaglio tenuto FISSO al muro della lente piena: si stressa l'accumulo, non l'obiettivo (col muro ricalcolato non arriverebbe nessuno e il confronto perderebbe potenza).""") r_half = r - 0.5 * r.mean() rg = np.random.default_rng(SEED) i_h = boot_idx(len(r_half), N_PATHS, ANNI * 365, BLOCK, rg) p_h, f_h = r_half[i_h], flat_src[i_h] del i_h H = tabella(p_h, f_h, DEP_MAIN, muro, perp, f"DRIFT DIMEZZATO ({r_half.mean() * 365:.2%}/anno) {clock()}", start=START) half20 = (leggi(H["P1 dip 20%"]["colpito"])["p20"] - leggi(H["P0 piatto"]["colpito"])["p20"]) del p_h, f_h # ------------------------------------------------------------------ 6 sez("6) RISCHIO DI VENUE — una politica che tiene i soldi FUORI e' anche una politica di rischio") print(""" Il fatto strutturale: la cassa che aspetta non e' sull'exchange. Se il vantaggio (o il minor danno) di una politica condizionale venisse da li', sarebbe un risultato sul RISCHIO, non sul timing. I due effetti si separano leggendo la stessa coppia a p=0 (solo timing) e a p>0. Convenzione: al fallimento il LIBRO va a zero, la CASSA sopravvive, e si riparte (conto nuovo, versamenti che continuano) — dichiarata, e diversa da quella di r0726_venue_risk, che nel caso a un venue solo scartava anche i versamenti futuri. `p` NON e' stimato: e' sensibilita'.""") print(f"\n {'p annua':>9} {'politica':<16}{'P(20a)':>9}{'dP20 vs P0':>13}" f"{'P(libro azzerato)':>20}{'EUR/g a 20a':>13}{'effetto rischio':>17}") d0 = {} for p in P_VENUE: h = 1.0 - (1.0 - p) ** (1.0 / 365.0) if p > 0 else 0.0 F = None if p > 0: rgv = np.random.default_rng(726) F = np.zeros((N_PATHS, ANNI * 365), bool) tt = rgv.geometric(h, size=N_PATHS) for _ in range(8): m = tt < ANNI * 365 if not m.any(): break F[np.where(m)[0], tt[m]] = True tt = tt + rgv.geometric(h, size=N_PATHS) base = None for nome in ("P0 piatto", "P1 dip 10%", "P1 dip 20%", "P4 anti 10%"): rr = accumula_pol(paths, flat, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, celle(DEP_MAIN)[nome], start=START, fail=F) d = leggi(rr["colpito"]) if base is None: base = d["p20"] rend = float(np.median((rr["cap"] + rr["cash"]) * perp / 365.0 / CC.EURUSD)) dd = d["p20"] - base if p == 0.0: d0[nome] = dd eff = "" if p == 0.0 else f"{dd - d0[nome]:+.2%}" print(f" {f'{p:.1%}' if nome.startswith('P0') else '':>9} {nome:<16}" f"{d['p20']:>9.1%}{dd:>13.2%}{rr['p_perso']:>20.1%}{rend:>13.2f}{eff:>17}") del F print(f""" `effetto rischio` = di quanto il rischio di venue RIDUCE la penalita' della politica condizionale rispetto a p=0. E' il pezzo di risultato che NON e' timing: la cassa in attesa non e' sull'exchange, quindi non brucia. Il meccanismo esiste ed e' misurabile, ma vale qualche punto percentuale contro una penalita' di timing di decine — e la colonna `P(libro azzerato)` e' IDENTICA per tutte le politiche (il conto salta comunque: cambia quanto c'e' sopra, non se salta). Chi volesse comprare quella protezione la compra con un SECONDO CONTO, non ritardando i bonifici (25/07 §4: «un rischio non-diversificabile si compra con un secondo CONTO, non con un secondo sleeve»).""") print(f" {clock()}") # ------------------------------------------------------------------ 7 sez("7) SENSIBILITA' — le due convenzioni che potrebbero decidere al posto della misura") print(f"\n (i) CASSA a 0% (assunzione XEON) contro cassa al 2%/anno nominale:") print(f" {'politica':<16}{'P(20a) cassa 0%':>18}{'P(20a) cassa 2%':>18}{'delta':>10}") for nome in ("P0 piatto", "P1 dip 10%", "P1 dip 20%", "P2 risk-off", "P3 VA g=0%"): a = leggi(R[DEP_MAIN][nome]["colpito"])["p20"] b = leggi(accumula_pol(paths, flat, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, celle(DEP_MAIN)[nome], start=START, cash_rate=0.02)["colpito"])["p20"] print(f" {nome:<16}{a:>18.1%}{b:>18.1%}{b - a:>10.2%}") print(f"\n (ii) TRAGUARDO su libro+cassa (favorevole alle condizionali, usato sopra) contro") print(f" traguardo su SOLO LIBRO (severo: la cassa a 0% non produce rendita):") print(f" {'politica':<16}{'P(20a) libro+cassa':>21}{'P(20a) solo libro':>20}{'delta':>10}") for nome in ("P0 piatto", "P1 dip 10%", "P1 dip 20%", "P2 risk-off", "P3 VA g=0%"): a = leggi(R[DEP_MAIN][nome]["colpito"])["p20"] b = leggi(accumula_pol(paths, flat, DEP_MAIN, muro, CC.TAX_RATE, TX.PATRIMONIALE, celle(DEP_MAIN)[nome], start=START, traguardo_include_cassa=False)["colpito"])["p20"] print(f" {nome:<16}{a:>21.1%}{b:>20.1%}{b - a:>10.2%}") print(f" {clock()}") # ------------------------------------------------------------------ 8 sez("VERDETTO") P0 = R[DEP_MAIN]["P0 piatto"] base = leggi(P0["colpito"]) tab = {n: leggi(v["colpito"]) for n, v in R[DEP_MAIN].items() if n != "P0 piatto"} best = max(tab, key=lambda k: tab[k]["p20"]) vere = {n: v for n, v in tab.items() if R[DEP_MAIN][n]["attesa_g"] >= 30.0} bvera = max(vere, key=lambda k: vere[k]["p20"]) sopra = sum(1 for n in tab if tab[n]["p20"] > base["p20"]) dips = [tab[n]["p20"] - base["p20"] for n in tab if n.startswith("P1")] anti = [tab[n]["p20"] - base["p20"] for n in tab if n.startswith("P4")] pz = {n: R[DEP_MAIN][n]["prezzo"] / P0["prezzo"] for n in tab} def conta(D: dict) -> str: b = leggi(D["P0 piatto"]["colpito"])["p20"] dd = {n: leggi(v["colpito"])["p20"] - b for n, v in D.items() if n != "P0 piatto"} k = max(dd, key=lambda x: dd[x]) return f"{sum(1 for x in dd.values() if x > 0)}/12 ({k} {dd[k]:+.2%})" conte = " · ".join(f"{lab} {conta(D)}" for lab, D in (("EUR500", R[DEP_MAIN]), ("EUR250", R[DEP_ALT]), ("IID", I), ("drift/2", H))) d250 = leggi(R[DEP_ALT]["P0 piatto"]["colpito"]) print(f""" 1. NESSUNA delle {N_TRIAL} celle candidate batte il versamento piatto: {sopra}/{N_TRIAL} sopra P0, migliore in assoluto {best} a {tab[best]['p20'] - base['p20']:+.2%} appaiato — dentro la risoluzione Monte Carlo della differenza appaiata ({dif:.2%}) e su una cella che aspetta {R[DEP_MAIN][best]['attesa_g']:.0f} giorno, cioe' e' P0 travestito. La migliore cella DAVVERO condizionale (attesa >= 30 giorni) e' {bvera}: {vere[bvera]['p20'] - base['p20']:+.1%}. Segno medio: P1 buy-the-dip {np.mean(dips):+.1%} · P4 anti-dip {np.mean(anti):+.1%}. Per lente (celle STRETTAMENTE sopra P0 su 12, e la migliore): {conte}. Nessuna lente produce una cella sopra il rumore MC ({dif:.1%}) -> non c'e' niente da deflazionare: il conteggio dei trial non e' il vincolo, il vincolo e' che il segno e' negativo e monotono nella profondita' della condizione. 2. IL SEGNALE FUNZIONA E PERDE LO STESSO, ed e' la riga che spiega tutto. Colonna `prezzo`: P1 dip 15% compra al {pz['P1 dip 15%']:.3f} del livello medio di P0 e P1 dip 20% al {pz['P1 dip 20%']:.3f} — cioe' {1 - pz['P1 dip 20%']:.0%} PIU' IN BASSO, esattamente cio' che la politica promette. E arriva al capitale-rendita nel {tab['P1 dip 20%']['p20']:.1%} dei casi contro {base['p20']:.1%}. Comprare meglio e comprare TARDI sono la stessa mossa, e la seconda meta' domina di un ordine di grandezza. ⚠️ E il dettaglio che ribalta l'intuizione sulle soglie BASSE: P1 dip 5% compra a {pz['P1 dip 5%']:.3f}, cioe' PIU' IN ALTO di P0. Su una serie che sale, una condizione poco profonda non compra il calo: ritarda dentro la salita. Solo le soglie che fanno male comprano davvero piu' in basso — ed e' esattamente il motivo per cui non conviene. 3. P4 ANTI-DIP, il controllo speculare, NON e' simmetrico — e la sua asimmetria e' la misura del meccanismo: su una serie che sale, un rialzo del 20% dal minimo arriva in {R[DEP_MAIN]['P4 anti 20%']['attesa_g']:.0f} giorni mentre un ribasso del 20% dal massimo ne chiede {R[DEP_MAIN]['P1 dip 20%']['attesa_g']:.0f}. Il costo non e' «scommettere sul lato sbagliato»: e' «mettere una condizione rara davanti a un bonifico». Qualunque condizione rara, in qualunque direzione, costa. P1 e P4 NON danno lo stesso segno di taglia: se lo dessero staremmo misurando rumore. 4. ORDINE DI GRANDEZZA, che e' il modo in cui questo risultato va usato: la migliore politica di timing vale {tab[bvera]['p20'] - base['p20']:+.1%} di P(20a) — indistinguibile da zero sulla risoluzione MC di {dif:.1%}; passare da EUR {DEP_ALT:.0f} a EUR {DEP_MAIN:.0f} al mese vale {base['p20'] - d250['p20']:+.0%} ({d250['p20']:.0%} -> {base['p20']:.0%}). Non c'e' rapporto da calcolare fra un effetto misurato e uno che non si distingue dal rumore: la prima colonna e' una decisione, la seconda no. 5. ATTESA A PRIORI (dichiarata prima di misurare): «non funziona, perche' entrare piu' tardi domina il comprare piu' in basso». CONFERMATA, e col meccanismo misurato invece che assunto. Cio' che l'attesa NON prevedeva e' il punto 2: il segnale compra davvero molto piu' in basso — {1 - pz['P1 dip 20%']:.0%} — e perde comunque. Sapere PERCHE' non serve vale piu' del verdetto: la domanda non tornera' sotto forma di «e se aspettassi un calo piu' profondo?», perche' e' proprio la profondita' a peggiorare il risultato, in modo monotono su tutte e 4 le soglie e su tutte e 4 le lenti. 6. DUE ERRORI CATTURATI SU ME STESSO, entrambi da un controllo e non a occhio. (i) La colonna `informazione` di sezione 4 sul solo mercato vero mi avrebbe fatto scrivere «il buy-the-dip contiene previsione»: il null IID la rende PIU' GRANDE su ritorni dove non c'e' niente da prevedere -> e' meccanica (attesa variabile contro fissa). Una colonna positiva non e' un'evidenza finche' non gira anche sul null. (ii) Senza la colonna `attesa` avrei letto P4 anti-dip come «stesso segno di P1 => rumore» (regola dichiarata nel mandato). Falso: P4 e' piatto perche' in un mercato che sale la sua condizione e' quasi sempre gia' vera ({R[DEP_MAIN]['P4 anti 20%']['attesa_g']:.0f} giorni di attesa contro {R[DEP_MAIN]['P1 dip 20%']['attesa_g']:.0f}), cioe' e' P0 travestito, non uno specchio informativo. Lo specchio va letto sulla colonna dell'ATTESA, non su quella del segno. COSA QUESTO NON MISURA: politiche che cambiano il FLUSSO disponibile (versare di piu' quando si guadagna di piu' e' un'altra domanda, ed e' fuori dal controllo del libro); orizzonti diversi da 25 anni; il caso in cui la cassa in attesa renda davvero qualcosa (sez. 7 (i): al 2% nominale la penalita' si riduce di ~4pp e non cambia segno); e il fatto che la finestra storica del libro sono 7,4 anni con due tori — su un campione con un orso lungo il costo del ritardo sarebbe minore (lo si vede nella colonna DRIFT DIMEZZATO: la penalita' scende da {tab['P1 dip 20%']['p20'] - base['p20']:+.0%} a {half20:+.0%}, ma resta negativa a tutte e 12 le celle). """) print(f" fine {clock()}") if __name__ == "__main__": main()