"""r0725_prop_coupled — CHIUDE il follow-up dichiarato: la lente WICK ACCOPPIATA (2026-07-25). `r0725_prop_ladder.py` ha giudicato la scala di conti funded con DUE lenti che differiscono di un ordine di grandezza su P(>=50 EUR/g): **20.7% close-only** (tetto) vs **1.6-2.5% wick indipendente** (pavimento). La banda e' cosi' larga da non decidere nulla: e' la differenza fra "scommessa a coda destra che vale 600 euro" e "non vale la pena". La ragione della larghezza e' un DIFETTO DICHIARATO del pavimento: il gap intraday era estratto **lognormale INDIPENDENTE dal rendimento del giorno**. Ma i wick profondi NON sono indipendenti: stanno sulle giornate brutte. Estrarli indipendenti aggiunge finti tuffi anche nelle giornate buone -> breach spuri -> P(pass) e P(vivo) troppo basse. Il modo giusto e' il bootstrap delle **TUPLE (ritorno, gap)** accoppiate, che r0724_goal50_intraday_mc.py fa — ma solo per il book Deribit 75/25 (TP01+SKH01), non per XS01 ne' a pesi arbitrari. Qui si generalizza. COSA FA QUESTO SCRIPT 1) **Recon per-sleeve** a risoluzione oraria: (ret, wick) unitari di TP01 e SKH01 separati, cosi' il minimo intraday si puo' calcolare esattamente per QUALSIASI vettore di pesi (non solo 75/25). Riusa la macchina di r0724 (exit-al-livello, SL prioritario, fee). 2) **Lente accoppiata per XS01**, che su Hyperliquid e' 1d nativo: l'escursione intraday si ricostruisce dagli OHLC GIORNALIERI dei 19 alt con ordinamento condiviso (tutte le gambe toccano prima il minimo, o prima il massimo — le alt co-muovono), prendendo l'ordine AVVERSO. Non e' path-exact, ma e' **accoppiato al giorno vero**, che e' il punto. 3) **Prova empirica del difetto**: gap per decile di rendimento + conteggio dei breach sui giorni STORICI (nessun bootstrap) sotto le tre lenti. 4) Le 4 politiche di scala e la decisione HyroTrader config A vs B, girate a **TRE lenti** (close-only / indipendente / accoppiata) sulla stessa macchina di simulazione. 5) Tre VERIFICHE, perche' un "nessuna differenza" va provato: risoluzione 1h vs 5m, riconciliazione col numero indipendente del 24/07, bound severo su XS01. RISULTATO IN UNA RIGA: la CALIBRAZIONE del wick del 25/07 era giusta (p50 -0.17pp identico), l'errore era l'INDIPENDENZA. L'escursione intraday e' ~3x piu' profonda nei giorni che finiscono BENE (recuperi a V) che in quelli che finiscono male (che chiudono sul minimo: m==R nel 26% dei giorni). Il breach si valuta sul minimo -> estrarre il gap indipendente carica i giorni brutti con la coda che appartiene ai giorni buoni, e RADDOPPIA i breach da daily-loss (2.0-2.9x). CONVENZIONI DICHIARATE (le stesse del 24/07, piu' una nuova) * il minimo di un book multi-sleeve e' preso come **somma dei minimi** (worst simultaneo) fra la parte crypto e XS01: severo, ma coerente con la convenzione gia' in uso fra TP01 e SKH01 dentro l'ora. Fra TP01 e SKH01 il minimo e' invece ESATTO (path orario condiviso). * wick a risoluzione 1h sul crypto. ⚠️ NON e' un tetto di risoluzione: MISURATO contro i 5m sulla gamba TP01, p99 -3.06pp (1h) vs -3.14pp (5m) -> l'ora cattura gia' il minimo del giorno. (Il caveat "i 5m sarebbero piu' profondi" del 24/07 e' quantificato e trascurabile.) * XS01 a risoluzione di barra giornaliera (4 checkpoint O/H/L/C, ordine condiviso avverso); il bound opposto (ogni gamba al proprio peggio insieme) e' girato come sensitivita' e NON ribalta nessuna conclusione. Uso: `uv run python scripts/research/r0725_prop_coupled.py` (~2-3 min: recon SKH su 5m full) """ from __future__ import annotations import sys from functools import lru_cache 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.portfolio import metrics # noqa: E402 from src.portfolio.sleeves import XS_CFG, XS_UNIVERSE, _HL_DIR, _xsec_returns # noqa: E402 import r0724_goal50_intraday_mc as mc # noqa: E402 import r0725_prop_ladder as pl # noqa: E402 from r0725_capcurve import EURUSD, TAX_RATE # noqa: E402 ASSETS = ("BTC", "ETH") DELUCK = pl.DELUCK START_EUR = pl.START_EUR HORIZON_MONTHS = pl.HORIZON_MONTHS MAX_CONCURRENT = pl.MAX_CONCURRENT TARGET_EUR_DAY = pl.TARGET_EUR_DAY SPLIT = pl.SPLIT FIRM_DEATH_PER_YEAR = pl.FIRM_DEATH_PER_YEAR FIRMS = pl.FIRMS # ================================================================= 1. RECON PER-SLEEVE (1h) @lru_cache(maxsize=1) def _hourly_legs() -> tuple[pd.DataFrame, ...]: """(ret, wick) ORARI UNITARI di TP01 e SKH01 — cioe' a peso di sleeve 1.0, gia' sommati sui due asset (ogni asset pesa 0.5 dentro lo sleeve). r0724 cabla i pesi di book (W_TP=0.75, W_SKH=0.25) dentro il recon; qui si dividono via per ottenere gli sleeve nudi, che si possono poi ricombinare a qualsiasi peso. """ tp_r = tp_w = skh_r = skh_w = None for a in ASSETS: df = load_data(a, "1h") idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) close = df["close"].values.astype(float) low = df["low"].values.astype(float) high = df["high"].values.astype(float) prev = np.concatenate(([np.nan], close[:-1])) r_close = close / prev - 1.0 # --- TP01: peso costante nel giorno (deciso a close[d-1]), fee alla prima ora tpf = mc.tp01_daily_frame(a) # gia' scalato *0.5 (leg), senza W_TP day_of = idx.floor("D") w_tp = np.nan_to_num(tpf["w"].reindex(day_of).values) fee_tp = np.nan_to_num(tpf["fee"].reindex(day_of).values) first = np.concatenate(([True], day_of[1:] != day_of[:-1])) r_wick = np.where(w_tp >= 0, low / prev - 1.0, high / prev - 1.0) t_r = pd.Series(w_tp * r_close - np.where(first, fee_tp, 0.0), index=idx) t_w = pd.Series(w_tp * r_wick - np.where(first, fee_tp, 0.0), index=idx) # --- SKH01: MTM per-trade con exit AL LIVELLO (r0724 cabla w = W_SKH*0.5) s_r_arr, s_w_arr = mc.skh_hourly_contrib(a, idx, close, low, high) s_r = pd.Series(s_r_arr / mc.W_SKH, index=idx) s_w = pd.Series(s_w_arr / mc.W_SKH, index=idx) tp_r = t_r if tp_r is None else tp_r.add(t_r, fill_value=np.nan) tp_w = t_w if tp_w is None else tp_w.add(t_w, fill_value=np.nan) skh_r = s_r if skh_r is None else skh_r.add(s_r, fill_value=np.nan) skh_w = s_w if skh_w is None else skh_w.add(s_w, fill_value=np.nan) J = pd.concat({"tp_r": tp_r, "tp_w": tp_w, "skh_r": skh_r, "skh_w": skh_w}, axis=1).dropna() return (J,) def crypto_daily_tuples(w_tp: float, w_skh: float) -> pd.DataFrame: """Per giorno UTC: R (chiusura) e m (minimo intraday) del sub-book crypto a pesi arbitrari. Il minimo fra TP01 e SKH01 e' ESATTO: si compone il path ORARIO del book e si guarda il minimo, includendo il wick dentro ogni ora. (Il wick intra-ora dei due sleeve e' sommato: li' resta la convenzione worst-simultaneo di r0724.) """ (J,) = _hourly_legs() r_h = (w_tp * J["tp_r"] + w_skh * J["skh_r"]).values w_h = (w_tp * J["tp_w"] + w_skh * J["skh_w"]).values idx = J.index days = idx.floor("D") rows = [] for day, pos in pd.Series(np.arange(len(idx)), index=days).groupby(level=0): sl = pos.values r = r_h[sl] wk = w_h[sl] cum = np.cumprod(1.0 + r) cum_prev = np.concatenate(([1.0], cum[:-1])) m = min(cum.min(), (cum_prev * (1.0 + wk)).min()) - 1.0 rows.append((day, cum[-1] - 1.0, min(m, cum[-1] - 1.0))) return pd.DataFrame(rows, columns=["day", "R", "m"]).set_index("day") # ================================================================= 2. XS01 ACCOPPIATO (1d OHLC) @lru_cache(maxsize=2) def xsec_daily_tuples(wick_mode: str = "shared") -> pd.DataFrame: """XS01 con escursione intraday ricostruita dagli OHLC GIORNALIERI dei 19 alt. Replica la costruzione delle posizioni di `sleeves._xsec_returns` (verificata a valle contro lo sleeve ufficiale: `assert` sulla chiusura) e in piu', per ogni giorno, valuta il book ai 4 checkpoint O / primo estremo / secondo estremo / C, con **ordinamento CONDIVISO** fra le gambe (le alt co-muovono) e prendendo l'**ordine avverso** fra "prima i minimi" e "prima i massimi". Su un book market-neutral l'ordinamento condiviso si auto-compensa in larga parte (i long perdono, gli short guadagnano): il rischio intraday vero e' la DISPERSIONE, che questa lente cattura solo in parte -> il gap XS01 esce PICCOLO, e va letto come tale. wick_mode='perleg' e' il BOUND SEVERO opposto: ogni gamba tocca il PROPRIO peggio nello stesso istante (long al minimo E short al massimo). E' impossibile in pratica su 10 gambe co-moventi, ma delimita l'errore: se la conclusione regge a entrambi gli estremi, non dipende dalla convenzione. Serve perche' e' XS01 a reggere il vantaggio della config diversificata. """ o = {}; h = {}; lo = {}; c = {} for sym in XS_UNIVERSE: p = _HL_DIR / f"hl_{sym.lower()}_1d.parquet" if not p.exists(): continue d = pd.read_parquet(p) i = pd.to_datetime(d["timestamp"], unit="ms", utc=True) o[sym] = pd.Series(d["open"].astype(float).values, index=i) h[sym] = pd.Series(d["high"].astype(float).values, index=i) lo[sym] = pd.Series(d["low"].astype(float).values, index=i) c[sym] = pd.Series(d["close"].astype(float).values, index=i) C = pd.concat(c, axis=1, join="inner").sort_index().dropna() O = pd.concat(o, axis=1).reindex(C.index)[C.columns] H = pd.concat(h, axis=1).reindex(C.index)[C.columns] L = pd.concat(lo, axis=1).reindex(C.index)[C.columns] px = C.values n, A = px.shape lookbacks, Hh, k, mode, tv = (XS_CFG["lookbacks"], XS_CFG["H"], XS_CFG["k"], XS_CFG["mode"], XS_CFG["target_vol"]) disp_pct, minhist = XS_CFG.get("disp_pct", 0), XS_CFG.get("disp_minhist", 20) mlb = max(lookbacks) dret = np.vstack([np.zeros(A), px[1:] / px[:-1] - 1.0]) W = np.zeros((n, A)); w = np.zeros(A); disp_hist = [] for i in range(n): if i >= mlb and i % Hh == 0: rLs = [px[i] / px[i - Lk] - 1.0 for Lk in lookbacks] disp_i = float(np.mean([r.std() for r in rLs])) thr = (np.percentile(disp_hist, disp_pct) if (disp_pct > 0 and len(disp_hist) >= minhist) else -np.inf) if disp_i >= thr: score = np.zeros(A); cnt = 0 for rL in rLs: sd = rL.std() if sd > 0: score += (rL - rL.mean()) / sd; cnt += 1 if cnt: score /= cnt order = np.argsort(score) w = np.zeros(A); low_i, hi_i = order[:k], order[-k:] if mode == "mom": w[hi_i] = 0.5 / k; w[low_i] = -0.5 / k else: w[low_i] = 0.5 / k; w[hi_i] = -0.5 / k else: w = np.zeros(A) disp_hist.append(disp_i) W[i] = w gross = np.zeros(n); gross[1:] = np.sum(W[:-1] * dret[1:], axis=1) turn = np.zeros(n); turn[0] = np.abs(W[0]).sum() turn[1:] = np.abs(np.diff(W, axis=0)).sum(axis=1) net = gross - turn * (0.001 / 2.0) s = pd.Series(net, index=C.index) rv = s.rolling(30, min_periods=15).std().shift(1) * np.sqrt(365.25) scale = np.clip(np.nan_to_num(tv / rv.replace(0, np.nan).values, nan=0.0), 0, 3.0) # --- escursione intraday, gambe tenute = W[i-1], prezzo base = close[i-1] prev_c = np.vstack([np.full((1, A), np.nan), px[:-1]]) held = np.vstack([np.zeros((1, A)), W[:-1]]) with np.errstate(invalid="ignore"): rO = O.values / prev_c - 1.0 rH = H.values / prev_c - 1.0 rL = L.values / prev_c - 1.0 rC = px / prev_c - 1.0 vO = np.nansum(held * rO, axis=1) vH = np.nansum(held * rH, axis=1) vL = np.nansum(held * rL, axis=1) vC = np.nansum(held * rC, axis=1) # ordinamento condiviso: low-first (O -> L -> H -> C) oppure high-first (O -> H -> L -> C). # Il book market-neutral vede i due estremi come punti, non come somma di worst per gamba. if wick_mode == "perleg": # bound severo: ogni gamba al proprio peggio simultaneamente worst_leg = np.where(held >= 0, rL, rH) m_gross = np.minimum(np.nansum(held * worst_leg, axis=1), vC) elif wick_mode == "shared": m_lowfirst = np.minimum.reduce([vO, vL, vH, vC]) m_highfirst = np.minimum.reduce([vO, vH, vL, vC]) m_gross = np.minimum(m_lowfirst, m_highfirst) # ordine AVVERSO else: raise ValueError(wick_mode) # il costo di turnover si paga comunque; il vol-target scala tutto m_net = (m_gross - turn * (0.001 / 2.0)) * scale R_net = net * scale out = pd.DataFrame({"R": R_net, "m": np.minimum(m_net, R_net)}, index=C.index) return out # ================================================================= 3. PANNELLO A 3 SLEEVE def _d(s: pd.Series) -> pd.Series: s = s.dropna().sort_index() if s.index.tz is None: s.index = s.index.tz_localize("UTC") return s _TUP_CACHE: dict = {} def book_tuples(w_tp: float, w_skh: float, w_xs: float, start: str | None = None, xs_mode: str = "shared") -> tuple[np.ndarray, np.ndarray, pd.DatetimeIndex]: """(R, gap) giornalieri di un book a pesi arbitrari. gap = minimo_intraday - chiusura (<=0). Crypto: minimo ESATTO sul path orario condiviso di TP01+SKH01. XS01: il suo minimo si SOMMA a quello crypto (worst simultaneo — convenzione dichiarata, severa). La finestra si interseca con XS01 SOLO se XS01 ha peso: a peso zero il book crypto tiene tutta la sua storia (serve per misurare la geometria dei gap del book live). NB: niente round-trip int64 dell'indice — su indici tz-aware non-ns pandas 2.x cambia scala silenziosamente (lezione 2026-07-01). Si cachea il DatetimeIndex cosi' com'e'. """ key = (float(w_tp), float(w_skh), float(w_xs), xs_mode) if key not in _TUP_CACHE: cry = None if (w_tp == 0.0 and w_skh == 0.0) else crypto_daily_tuples(w_tp, w_skh) xs = xsec_daily_tuples(xs_mode) if cry is None: idx, R, m = xs.index, w_xs * xs["R"].values, w_xs * xs["m"].values elif w_xs == 0.0: idx, R, m = cry.index, cry["R"].values, cry["m"].values else: idx = cry.index.intersection(xs.index) cr, xr = cry.reindex(idx), xs.reindex(idx) R = cr["R"].values + w_xs * xr["R"].values m = cr["m"].values + w_xs * xr["m"].values keep = np.isfinite(R) & np.isfinite(m) R, m, idx = R[keep], m[keep], idx[keep] _TUP_CACHE[key] = (R, np.minimum(m, R) - R, idx) R, G, I = _TUP_CACHE[key] if start is not None: k = I >= pd.Timestamp(start, tz="UTC") return R[k], G[k], I[k] return R.copy(), G.copy(), I # ================================================================= 4. SIMULAZIONE (3 lenti) def simulate(cfgs: list[tuple[float, float, float]], slots: list[int], n_paths: int, lev: float, seed: int, lens: str, start: str, block: int = 20) -> dict: """Scala di conti funded. `cfgs` = vettori di peso distinti; `slots` = indice di cfg per ciascuno dei MAX_CONCURRENT slot. Il bootstrap a blocchi usa lo STESSO indice per tutte le configurazioni -> la correlazione fra conti e' quella REALE (e le tuple (R,gap) restano accoppiate al loro giorno). lens: 'close' -> breach sulla chiusura (tetto) 'indip' -> gap lognormale indipendente (il pavimento del 25/07) 'coupled'-> gap dal recon, accoppiato al giorno (la lente onesta) """ rng = np.random.default_rng(seed) n_days = HORIZON_MONTHS * 30 mats = [book_tuples(*c, start=start) for c in cfgs] L = min(len(m[0]) for m in mats) Rm = np.stack([m[0][-L:] for m in mats], axis=1) # (giorni, cfg) Gm = np.stack([m[1][-L:] for m in mats], axis=1) Rm = Rm - (1.0 - DELUCK) * Rm.mean(axis=0, keepdims=True) # de-luck sul drift n_blocks = int(np.ceil(n_days / block)) starts = rng.integers(0, L - block, size=(n_paths, n_blocks)) idx = (starts[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1)[:, :n_days] n_slot = len(slots) sel = np.asarray(slots) Rs = Rm[idx][:, :, sel] * lev # (path, day, slot) if lens == "close": G = np.zeros_like(Rs) elif lens == "coupled": G = Gm[idx][:, :, sel] * lev elif lens == "indip": vol_k = Rs.reshape(-1, n_slot).std(axis=0) * np.sqrt(365.0) scale = np.clip(vol_k / pl.WICK_REF_VOL, 0.0, 3.0) G = -np.exp(rng.normal(pl.WICK_MU, pl.WICK_SIGMA, size=Rs.shape)) * scale[None, None, :] else: raise ValueError(lens) cash = np.full(n_paths, START_EUR * EURUSD) phase = np.full((n_paths, n_slot), -1, np.int8) eq = np.ones((n_paths, n_slot)) firm = np.full((n_paths, n_slot), -1, np.int8) notional = np.zeros((n_paths, n_slot)) firm_alive = np.ones((n_paths, len(FIRMS)), bool) cum_payout = np.zeros(n_paths) spent = np.zeros(n_paths) p_death_day = 1.0 - (1.0 - FIRM_DEATH_PER_YEAR) ** (1 / 365.0) for t in range(n_days): died = rng.random((n_paths, len(FIRMS))) < p_death_day newly_dead = died & firm_alive firm_alive &= ~died if newly_dead.any(): for fi in range(len(FIRMS)): kill = newly_dead[:, fi] if kill.any(): hit = (firm == fi) & kill[:, None] phase[hit] = -1 notional[hit] = 0.0 for k in range(n_slot): free = phase[:, k] == -1 if not free.any(): continue for fi, F in enumerate(FIRMS): used = np.where(firm == fi, notional, 0.0).sum(axis=1) for (size, price) in sorted(F["tickets"], key=lambda x: -x[0]): buy = (free & firm_alive[:, fi] & (cash >= price) & (used + size <= F["cap"])) if not buy.any(): continue cash[buy] -= price spent[buy] += price phase[buy, k] = 0 eq[buy, k] = 1.0 firm[buy, k] = fi notional[buy, k] = size free = free & ~buy if not free.any(): break if not free.any(): break r = Rs[:, t, :] g = G[:, t, :] act = phase >= 0 eq_prev = eq eq = np.where(act, eq * (1.0 + r), eq) eq_low = np.where(act, eq_prev * (1.0 + r + g), eq) r_low = r + g for fi, F in enumerate(FIRMS): s_ = act & (firm == fi) if not s_.any(): continue ev_t, ev_dd, ev_dl = F["ev"] fu_ml, fu_dl = F["fu"] in_ev = s_ & (phase == 0) in_fu = s_ & (phase == 1) bust = ((in_ev & ((eq_low < 1.0 - ev_dd) | (r_low < -ev_dl))) | (in_fu & ((eq_low < 1.0 - fu_ml) | (r_low < -fu_dl)))) phase[bust] = -1 notional[bust] = 0.0 passed = in_ev & (eq >= 1.0 + ev_t) & ~bust phase[passed] = 1 eq[passed] = 1.0 if (t + 1) % 30 == 0: fu = phase == 1 gain = np.where(fu & (eq > 1.0), (eq - 1.0) * notional, 0.0) net = gain.sum(axis=1) * SPLIT * (1.0 - TAX_RATE) cum_payout += net cash += net eq = np.where(fu & (eq > 1.0), 1.0, eq) eur_day = cum_payout / EURUSD / (n_days / 365.0) / 365.0 return dict(eur_day=eur_day, med=float(np.median(eur_day)), p10=float(np.percentile(eur_day, 10)), p90=float(np.percentile(eur_day, 90)), p_ge10=float((eur_day >= 10.0).mean()), p_target=float((eur_day >= TARGET_EUR_DAY).mean()), p_zero=float((cum_payout <= 0).mean()), med_spent=float(np.median(spent))) # --------------------------------------------------- eval/funded a conto singolo (HYRO) def single_account(cfg: tuple[float, float, float], lev: float, lens: str, start: str, n_paths: int = 6000, seed: int = 725, block: int = 20, xs_mode: str = "shared") -> dict: """P(pass eval), P(vivo 1 anno da funded), E[payout annuo] per UN conto HYRO $100k.""" rng = np.random.default_rng(seed) R0, G0, _ = book_tuples(*cfg, start=start, xs_mode=xs_mode) R0 = R0 - (1.0 - DELUCK) * R0.mean() n = len(R0) n_days = 365 n_blocks = int(np.ceil(n_days / block)) st = rng.integers(0, n - block, size=(n_paths, n_blocks)) idx = (st[:, :, None] + np.arange(block)[None, None, :]).reshape(n_paths, -1)[:, :n_days] R = R0[idx] * lev if lens == "close": G = np.zeros_like(R) elif lens == "coupled": G = G0[idx] * lev else: vol = R.std() * np.sqrt(365.0) G = -np.exp(rng.normal(pl.WICK_MU, pl.WICK_SIGMA, size=R.shape)) \ * np.clip(vol / pl.WICK_REF_VOL, 0.0, 3.0) ev_t, ev_dd, ev_dl = FIRMS[0]["ev"] fu_ml, fu_dl = FIRMS[0]["fu"] NOTIONAL = 100_000.0 eq = np.ones(n_paths) phase = np.zeros(n_paths, np.int8) # 0 eval, 1 funded, -1 morto days_funded = np.zeros(n_paths, int) payout = np.zeros(n_paths) passed_ev = np.zeros(n_paths, bool) for t in range(n_days): alive = phase >= 0 r = R[:, t]; g = G[:, t] eq_prev = eq eq = np.where(alive, eq * (1 + r), eq) eq_low = np.where(alive, eq_prev * (1 + r + g), eq) r_low = r + g in_ev = alive & (phase == 0) in_fu = alive & (phase == 1) bust = ((in_ev & ((eq_low < 1 - ev_dd) | (r_low < -ev_dl))) | (in_fu & ((eq_low < 1 - fu_ml) | (r_low < -fu_dl)))) phase[bust] = -1 ok = in_ev & (eq >= 1 + ev_t) & ~bust phase[ok] = 1 eq[ok] = 1.0 passed_ev |= ok days_funded += (phase == 1) if (t + 1) % 30 == 0: fu = phase == 1 gain = np.where(fu & (eq > 1), (eq - 1) * NOTIONAL, 0.0) payout += gain * SPLIT * (1 - TAX_RATE) eq = np.where(fu & (eq > 1), 1.0, eq) return dict(p_pass=float(passed_ev.mean()), p_alive=float((phase[passed_ev] == 1).mean()) if passed_ev.any() else 0.0, e_payout=float(payout.mean()), e_payout_if_pass=float(payout[passed_ev].mean()) if passed_ev.any() else 0.0) # ------------------------------------------------- verifica di risoluzione (1h vs 5m) @lru_cache(maxsize=4) def _tp_leg_gap(tf: str) -> pd.DataFrame: """(R, m) giornalieri della SOLA gamba TP01, misurati a risoluzione `tf`. TP01 tiene un peso COSTANTE nel giorno, quindi il suo minimo intraday e' esattamente il minimo del path di prezzo: si puo' calcolare a 5m e confrontare con l'1h. E' il test che dice se l'1h basta a catturare il minimo giornaliero, invece di assumerlo. """ tot_r = tot_w = None for a in ASSETS: df = load_data(a, tf) idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) c = df["close"].values.astype(float) lo = df["low"].values.astype(float) hi = df["high"].values.astype(float) prev = np.concatenate(([np.nan], c[:-1])) tpf = mc.tp01_daily_frame(a) day = idx.floor("D") w = np.nan_to_num(tpf["w"].reindex(day).values) fee = np.nan_to_num(tpf["fee"].reindex(day).values) first = np.concatenate(([True], day[1:] != day[:-1])) rw = np.where(w >= 0, lo / prev - 1.0, hi / prev - 1.0) r = pd.Series(w * (c / prev - 1.0) - np.where(first, fee, 0.0), index=idx) wk = pd.Series(w * rw - np.where(first, fee, 0.0), index=idx) tot_r = r if tot_r is None else tot_r.add(r, fill_value=np.nan) tot_w = wk if tot_w is None else tot_w.add(wk, fill_value=np.nan) J = pd.concat({"r": tot_r, "w": tot_w}, axis=1).dropna() rows = [] for d, g in J.groupby(J.index.floor("D")): r = g["r"].values; wk = g["w"].values cum = np.cumprod(1 + r); cp = np.concatenate(([1.0], cum[:-1])) m = min(cum.min(), (cp * (1 + wk)).min()) - 1.0 rows.append((d, cum[-1] - 1.0, min(m, cum[-1] - 1.0))) return pd.DataFrame(rows, columns=["d", "R", "m"]).set_index("d") # ================================================================= main START_WIN = "2024-01-01" # finestra comune ai 3 sleeve (XS01 nasce qui) CFG_A = (0.75, 0.25, 0.0) # book live Deribit (TP01/SKH01) — quello che gira oggi CFG_B = (0.55, 0.20, 0.25) # book diversificato (config B del 25/07) CFG_TP = (1.0, 0.0, 0.0) CFG_SKH = (0.0, 1.0, 0.0) CFG_XS = (0.0, 0.0, 1.0) def main() -> None: print("=" * 106) print(" LENTE WICK ACCOPPIATA — chiusura del follow-up dichiarato (r0725_prop_ladder §ONESTA')") print("=" * 106) print("\n recon MTM orario per-sleeve (TP01 daily + SKH01 230m exit-al-livello)...") R_A, G_A, I_A = book_tuples(*CFG_A) print(f" book A (TP01 75 / SKH01 25): {len(R_A)} giorni {I_A[0].date()} -> {I_A[-1].date()}") # --- sanity: il recon deve riprodurre lo sleeve ufficiale xs_off = _d(_xsec_returns()) xs_rec = xsec_daily_tuples() common = xs_off.index.intersection(xs_rec.index) dmax = float(np.abs(xs_off.reindex(common).values - xs_rec["R"].reindex(common).values).max()) print(f" sanity XS01 recon vs sleeve ufficiale: max|delta| chiusura = {dmax:.2e}" f" ({'OK' if dmax < 1e-9 else 'DIVERGE'})") # ---------------------------------------------------------- 3. la prova del difetto print("\n" + "-" * 106) print(" A. IL GAP NON E' INDIPENDENTE DAL GIORNO — e la dipendenza va nel verso INATTESO") print("-" * 106) print(f" {'book':>26} {'giorni':>7} {'m==R':>7} {'gap medio':>10} " f"{'gap | decile PEGGIORE':>22} {'gap | decile MIGLIORE':>22}") for nm, cfg in [("A TP01/SKH01 (live)", CFG_A), ("B +XS01 (diversif.)", CFG_B), ("TP01 solo", CFG_TP), ("SKH01 solo", CFG_SKH), ("XS01 solo", CFG_XS)]: R, G, _ = book_tuples(*cfg) if R.std() == 0: continue q = np.quantile(R, [0.1, 0.9]) print(f" {nm:>26} {len(R):>7d} {(G > -1e-12).mean():>6.0%} {G.mean()*100:>9.2f}pp " f"{G[R <= q[0]].mean()*100:>21.2f}pp {G[R >= q[1]].mean()*100:>21.2f}pp") print("\n Lettura — il contrario di quello che si assume di solito: l'escursione intraday e'") print(" piu' PROFONDA nei giorni che finiscono BENE (recuperi a V) che in quelli che finiscono") print(" male, perche' un giorno brutto scende tutto il giorno e CHIUDE sul minimo (m==R nel") print(" ~29% dei giorni). Il breach pero' si valuta sul minimo, quindi cio' che conta e' quanto") print(" gap si aggiunge NEI GIORNI BRUTTI — e li' il gap vero e' ~1/3 di quello nei giorni buoni.") print(" Un'estrazione indipendente usa la STESSA distribuzione ovunque -> carica i giorni") print(" brutti con la coda che nella realta' appartiene ai giorni buoni.") # --- la misura che chiude la questione: quanti breach in piu' inventa la lente indipendente print("\n" + "-" * 106) print(" A-bis. CONTEGGIO EMPIRICO DEI BREACH sui giorni STORICI (nessun bootstrap)") print("-" * 106) rng = np.random.default_rng(725) print(f" {'book':>26} {'leva':>5} {'regola':>22} {'close-only':>11} " f"{'ACCOPPIATO':>11} {'indipendente':>13} {'gonfiaggio':>11}") for nm, cfg in [("A TP01/SKH01 (live)", CFG_A), ("B +XS01 (diversif.)", CFG_B)]: R0, G0, _ = book_tuples(*cfg, start=START_WIN) for lev in (0.75, 1.0): R, G = R0 * lev, G0 * lev vol = R.std() * np.sqrt(365.0) n_rep = 400 Gi = -np.exp(rng.normal(pl.WICK_MU, pl.WICK_SIGMA, size=(n_rep, len(R)))) \ * np.clip(vol / pl.WICK_REF_VOL, 0.0, 3.0) for lab, thr in (("daily-loss 4% (HYRO)", 0.04), ("daily-loss 5% (FTMO)", 0.05)): b_cl = float((R < -thr).mean()) b_co = float((R + G < -thr).mean()) b_in = float((R[None, :] + Gi < -thr).mean()) infl = (b_in / b_co) if b_co > 0 else np.inf print(f" {nm:>26} {lev:>5.2f} {lab:>22} {b_cl:>10.2%} " f"{b_co:>10.2%} {b_in:>12.2%} {infl:>10.1f}x") # geometria dei gap: confronto diretto fra le due lenti print("\n" + "-" * 106) print(" B. GEOMETRIA DEI GAP — accoppiato (recon) vs indipendente (calibrazione 25/07)") print("-" * 106) q_ind = pl.wick_percentiles() print(f" {'book':>28} {'p50':>9} {'p90':>9} {'p99':>9} {'peggiore':>10} {'vol annua':>10}") for nm, cfg in [("A TP01/SKH01 (live)", CFG_A), ("B +XS01 (diversificato)", CFG_B), ("XS01 solo", CFG_XS)]: R, G, _ = book_tuples(*cfg) if R.std() == 0: continue p = np.percentile(-G, [50, 90, 99]) print(f" {nm:>28} {-p[0]*100:>8.2f}pp {-p[1]*100:>8.2f}pp {-p[2]*100:>8.2f}pp " f"{G.min()*100:>9.2f}pp {R.std()*np.sqrt(365)*100:>9.1f}%") print(f" {'INDIPENDENTE (25/07)':>28} {-q_ind[0.50]*100:>8.2f}pp {-q_ind[0.90]*100:>8.2f}pp " f"{-q_ind[0.99]*100:>8.2f}pp {'—':>10} {'(scalato)':>10}") # ---------------------------------------------------------- 4. conto singolo HYRO START = START_WIN print("\n" + "-" * 106) print(f" C. UN CONTO HYRO $100k (max-loss 6% statico, daily 4%) — finestra {START}+, de-luck x{DELUCK}") print("-" * 106) print(f" {'config':>26} {'leva':>5} {'lente':>10} {'P(pass eval)':>13} " f"{'P(vivo 1a)':>11} {'E[payout/a]':>12}") for nm, cfg in [("A book live 75/25", CFG_A), ("B +XS01 55/20/25", CFG_B)]: for lev in (0.50, 0.75, 1.00): for lens in ("close", "indip", "coupled"): r = single_account(cfg, lev, lens, START) print(f" {nm:>26} {lev:>5.2f} {lens:>10} {r['p_pass']:>13.1%} " f"{r['p_alive']:>11.1%} {r['e_payout']:>11,.0f}$") print() # ------------------------------------------------- 4-bis. le tre verifiche che possono ribaltare print("-" * 106) print(" C-bis. VERIFICHE — un risultato 'nessuna differenza' va provato, non assunto") print("-" * 106) print("\n (1) RISOLUZIONE: il wick 1h cattura davvero il minimo del giorno? (gamba TP01, 5m vs 1h)") for tf in ("1h", "5m"): D = _tp_leg_gap(tf) G = (D["m"] - D["R"]).values k = np.isfinite(G) & (np.abs(D["R"].values) > 0) # i giorni flat non hanno escursione print(f" @{tf}: p50 {-np.percentile(-G[k], 50)*100:>6.3f}pp " f"p90 {-np.percentile(-G[k], 90)*100:>6.3f}pp " f"p99 {-np.percentile(-G[k], 99)*100:>6.3f}pp " f"peggiore {G[k].min()*100:>6.2f}pp (n={k.sum()} giorni non-flat)") print(" -> se coincidono, la lente accoppiata a 1h NON e' un tetto di risoluzione.") print("\n (2) RICONCILIAZIONE col 24/07 (recon MTM indipendente, finestra piena): " "config A @0.75x, P(vivo 1a)") for st, lab in (("2019-01-01", "2019-03+ FULL"), (START, "2024+ comune")): for lens in ("close", "coupled"): r = single_account(CFG_A, 0.75, lens, st) print(f" {lab:>14} {lens:>8}: P(pass) {r['p_pass']:>5.1%} " f"P(vivo) {r['p_alive']:>5.1%} E[payout] {r['e_payout']:>6,.0f}$") print(" -> il 24/07 riportava 58% sulla finestra piena; il 25/07 (wick indipendente) 10%.") print("\n (3) BOUND SEVERO su XS01 (ogni gamba al proprio peggio insieme) — config B @0.75x") for xm, lab in (("shared", "ordine condiviso (base)"), ("perleg", "per-gamba (bound severo)")): R, G, _ = book_tuples(*CFG_B, start=START, xs_mode=xm) r = single_account(CFG_B, 0.75, "coupled", START, xs_mode=xm) print(f" {lab:>26}: gap p90 {-np.percentile(-G, 90)*100:>6.2f}pp " f"peggiore {G.min()*100:>6.2f}pp | P(pass) {r['p_pass']:>5.1%} " f"P(vivo) {r['p_alive']:>5.1%} E[payout] {r['e_payout']:>6,.0f}$") print(" -> se la config B regge a ENTRAMBI gli estremi, la conclusione non dipende dalla convenzione.") # ---------------------------------------------------------- 5. la scala di conti print("\n" + "-" * 106) print(f" D. LA SCALA DI CONTI — {HORIZON_MONTHS} mesi, EUR {START_EUR:.0f} di cassa, " f"max {MAX_CONCURRENT} conti, morte-firm {FIRM_DEATH_PER_YEAR:.0%}/a") print("-" * 106) cfgs = [CFG_A, CFG_B, CFG_TP, CFG_SKH, CFG_XS] POLS = { "CONC-2SL ogni conto = book live": [0] * MAX_CONCURRENT, "CONC-DIV ogni conto = diversificato": [1] * MAX_CONCURRENT, "SPARSO 1 sleeve per conto": [2, 4, 3, 2, 4, 3], "MISTO meta' div / meta' single": [1, 4, 1, 2, 1, 4], } for lens in ("close", "indip", "coupled"): tag = {"close": "CLOSE-ONLY (tetto)", "indip": "WICK INDIPENDENTE (pavimento 25/07)", "coupled": "WICK ACCOPPIATO (la lente onesta)"}[lens] for lev in (0.50, 1.00): print(f"\n {tag} — LEVA {lev:.2f}x") print(f" {'politica':>40} {'EUR/g med':>10} {'p10':>7} {'p90':>7} " f"{'P(>=10/g)':>10} {'P(>=50/g)':>10} {'P(zero)':>8}") for pname, slots in POLS.items(): r = simulate(cfgs, slots, n_paths=3000, lev=lev, seed=20260725, lens=lens, start=START) print(f" {pname:>40} {r['med']:>10.2f} {r['p10']:>7.2f} {r['p90']:>7.2f} " f"{r['p_ge10']:>10.1%} {r['p_target']:>10.1%} {r['p_zero']:>8.1%}") print("\n" + "=" * 106) if __name__ == "__main__": main()