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

833 lines
50 KiB
Python
Raw Blame History

#!/usr/bin/env python
"""r0822b_bin_freq.py — BIN-FREQ: il BINARIO di frequenza su SKH01 giudicato come un CANDIDATO.
STORIA DEI FATTI (verificata leggendo il codice, non il riassunto).
1. `r0822_vol_size.py` ha proposto una SIZE PER-TRADE su SKH01 (cella congelata `VTL tv20 w90`
= clip(0.20/rv90(t-1), 0, 3) fissata all'INGRESSO) come CONTROLLO DEL RISCHIO:
+0,070 ISO dShFULL, 23/23 ancore, 8/8 anni.
2. `r0822_volsize_skeptic.py` ha CONFERMATO il numero e FALSIFICATO il meccanismo: il guadagno
e' FREQUENZA (+0,051 sul solo filtro di attivita') e non MAGNITUDINE (+0,003); e la forma
minima e' un BINARIO 0/1 (`BIN L>=med`, 194/406 trade scartati) che vale +0,080 FULL (20/23)
= 115% del LEAD, con controllo di segno superato (l'inverso fa -0,159, 0/23).
3. Lo scettico ha scritto "il candidato giusto non e' la formula ma il binario" — e nessuno
l'ha passato per `study_family_honest`. E' un SOTTOPRODOTTO DI UN ATTACCO, mai dichiarato
come famiglia. Questo script lo giudica come si giudica un candidato.
IPOTESI, REGISTRATA PRIMA DI MISURARE. Mi aspetto che il numero si riproduca (lo scettico e' stato
rigoroso e la sua replica e' bit-exact) e che il candidato NON superi il gate onesto, per due
ragioni dichiarate in anticipo: (a) la soglia e la finestra sono due assi liberi trovati DENTRO
l'esplorazione di un'altra famiglia, quindi la stima di famiglia dovra' essere molto piu' piccola
dell'argmax (precedente: sull'asse w lo scettico stesso stima +0,025 contro +0,080); (b) il
deflated-Sharpe sara' VACUO come lo e' stato sulla famiglia VOL-SIZE — su perturbazioni della
STESSA strategia la varianza fra i trial e' quasi nulla, e il DSR e' ~1 per chiunque, baseline
compreso. Se succede lo dichiaro vacuo e uso un null che ha potenza.
=========================== LA FAMIGLIA, DICHIARATA PRIMA DI GUARDARE =========================
Un binario di frequenza su SKH01 ha esattamente quattro gradi di liberta'. Li dichiaro tutti,
con i loro valori, PRIMA di misurare, e li conto AL RIALZO:
soglia (6) quantile ESPANDENTE causale della vol trailing, q in {0.25, 0.35, 0.50, 0.65,
0.75} + la soglia ASSOLUTA rv <= 20% (== `L>=1` della regola congelata)
finestra (6) w in {45, 60, 90, 120, 180, 270} giorni (sotto 45 lo scettico ha misurato che
la finestra contiene 1-3 barre ATTIVE: campione vuoto, non fragilita')
isteresi (2) h in {0.00, 0.10} (banda +-10% attorno alla soglia, stato persistente causale)
segno (2) +1 tieni la vol BASSA (l'ipotesi) / -1 tieni la vol ALTA (controllo di segno)
timeframe(1) uno solo: la gamba SKH01 vive sulla griglia 230m per costruzione. Dichiarato.
-> 6 x 6 x 2 x 2 = 144 celle
RIAPRIRE UN PARAMETRO RIAPRE LA SUA FAMIGLIA, e questo binario e' nato dentro l'esplorazione di
un'altra famiglia. Percio' al conto delle 144 si SOMMANO i trial gia' spesi sullo stesso oggetto:
28 (r0822_vol_size) + 43 (r0822_volsize_skeptic) = 71.
-> CONTO ONESTO DICHIARATO: 144 + 71 = 215 trial.
La sensibilita' del verdetto al conteggio e' pubblicata (§5), come nel precedente 30/07 dove un
DSR passava a N=8, falliva a N=72 e falliva a N=360.
============================== LENTE, ANCORE, CONVENZIONI =====================================
Lente: path CANONICO di SKH01 (fill al livello, entry a chiusura di bin) — identica nelle due
braccia di ogni confronto, quindi il Delta e' pulito anche se il livello e' pessimistico verso il
path live. Ancore: **23 offset**, la griglia a priori dell'audit 02/07 (ogni 30m su [0,690)):
NON ridotta, perche' il costo misurato e' ~20 ms per cella per ancora. Statistica: MEDIANA DELLE
DIFFERENZE APPAIATE (`A.anchor_luck_delta`), mai differenza delle mediane.
Confronto sempre ISO-PESO: il binario cambia la vol della gamba e quindi il peso EFFETTIVO di
SKH01 nel libro, e muovere quel peso e' gia' stato misurato e RESPINTO il 26/07. Il null del
de-levering e' DEGENERE su una ri-scalatura (`sh(k*base) === sh(base)`), quindi il test e'
iso-peso sullo Sharpe; `k_iso_dd` resta come diagnostica del DD.
Nessuna riscrittura: `extract / equity_daily / leg_daily / sizes_from_L / book / netted_target`
sono IMPORTATE da r0822_vol_size, e il titolo dello scettico e' riprodotto chiamando la SUA
funzione. Replica bit-exact contro `sleeves._skyhook_returns()` prima di ogni delta (§0).
USO: nice -n 19 timeout 900 uv run python scripts/research/r0822b_bin_freq.py [--every K]
--every 8 -> 3 ancore (pilota). default: tutte e 23.
"""
from __future__ import annotations
import argparse
import sys
import time
from pathlib import Path
import numpy as np
import pandas as pd
ROOT = Path(__file__).resolve().parents[2]
for _p in (ROOT, ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt"):
sys.path.insert(0, str(_p))
import altlib as A # noqa: E402
import r0702_anchor_skh01 as R # noqa: E402
import r0702_tp01_offset as TO # noqa: E402
import r0822_vol_size as V # noqa: E402 (motore: riusato, non riscritto)
import r0822_volsize_skeptic as SK # noqa: E402 (per riprodurre il titolo)
from r0822_sol_leg import k_iso_dd # noqa: E402
from src.portfolio.portfolio import weights_tilt_null # noqa: E402
from src.strategies.skyhook import SKH01_V1 # noqa: E402
ASSETS = V.ASSETS
OFFSETS_FULL = V.OFFSETS_FULL
sh, vol, maxdd, cagr = V.sh, V.vol, V.maxdd, V.cagr
hold, ins, book = V.hold, V.ins, V.book
CAPITAL, MIN_ORDER = V.CAPITAL, V.MIN_ORDER
W_TP, W_SKH = V.W_TP, V.W_SKH
# ------------------------------------------------------------------ la famiglia (dichiarata sopra)
QS = (0.25, 0.35, 0.50, 0.65, 0.75)
THRS = tuple([("q", q) for q in QS] + [("abs", 0.20)])
WS = (45, 60, 90, 120, 180, 270)
HS = (0.00, 0.10)
SIGNS = (+1, -1)
N_INHERITED = 28 + 43 # r0822_vol_size + r0822_volsize_skeptic, sullo stesso oggetto
CANON = ("q", 0.50, 90, 0.00, +1) # la cella dello scettico, dentro la mia famiglia
L = 112
def cname(thr, w, h, sg) -> str:
t = "rv<=20%" if thr[0] == "abs" else f"q{thr[1]:.2f}"
return f"{t} w{w} h{h:.2f} s{'+' if sg > 0 else '-'}"
# =============================================================== il binario, causale per costruzione
def _hyst(x: np.ndarray, thr: np.ndarray, h: float, keep_below: bool) -> np.ndarray:
"""Stato persistente con banda +-h attorno alla soglia. Causale: al passo t usa solo x_t e
thr_t (entrambi funzione di dati <= t-1) e lo stato al passo t-1. h=0 -> confronto secco."""
n = len(x)
out = np.ones(n, bool)
st = True
for i in range(n):
xi, ti = x[i], thr[i]
if not (np.isfinite(xi) and np.isfinite(ti)):
st = True # warm-up: si tiene tutto (conv. scettico)
elif keep_below:
st = (xi <= ti * (1.0 + h)) if st else (xi <= ti * (1.0 - h))
else:
st = (xi >= ti * (1.0 - h)) if st else (xi >= ti * (1.0 + h))
out[i] = st
return out
class KeepCache:
"""rv e quantili espandenti calcolati UNA volta per (asset, ancora, w). L'espandente e' la
parte cara; il resto e' un confronto."""
def __init__(self) -> None:
self.rv: dict = {}
self.q: dict = {}
def keep(self, ex: dict, key, thr, w: int, h: float, sg: int) -> pd.Series:
base = ex["base_daily"]
k = (key, w)
if k not in self.rv:
self.rv[k] = V._rv_daily(base, w, active_only=False) # gia' shiftata di 1 giorno
rv = self.rv[k]
if thr[0] == "abs":
t = np.full(len(rv), float(thr[1]))
else:
kq = (key, w, thr[1])
if kq not in self.q:
self.q[kq] = rv.expanding(min_periods=V.VT_WARM).quantile(thr[1])
t = self.q[kq].values
keep = _hyst(rv.values.astype(float), np.asarray(t, float), h, keep_below=(sg > 0))
return pd.Series(keep.astype(float), index=rv.index)
KC = KeepCache()
def size_bin(thr, w, h, sg):
def f(ex, key):
return V.sizes_from_L(ex, KC.keep(ex, key, thr, w, h, sg))
return f
# =============================================================== motore
def leg_of(EXX: dict, key, fn) -> pd.Series:
return V.leg_daily(EXX, {a: fn(EXX[a], (key, a)) for a in ASSETS})
def iso_pair(TP: pd.Series, leg: pd.Series, leg_base: pd.Series):
"""Libro della variante e libro di CONTROLLO ISO-PESO (baseline ri-scalato alla stessa vol di
gamba => stesso peso effettivo di SKH). Convenzione del bersaglio, riusata tale e quale."""
vb = vol(leg_base)
m = vol(leg) / vb if vb > 0 else 1.0
return book(TP, leg), book(TP, leg_base * m), m
# =============================================================== main
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--every", type=int, default=1)
ap.add_argument("--ndraw", type=int, default=200)
args = ap.parse_args()
offs = OFFSETS_FULL[:: max(1, args.every)]
t0 = time.time()
print("=" * L)
print(" r0822b BIN-FREQ — il BINARIO di frequenza su SKH01, giudicato come un CANDIDATO")
print("=" * L)
print(f" ancore: {len(offs)}/{len(OFFSETS_FULL)} della griglia a priori 02/07 (ogni 30m su "
f"[0,690)). NON ridotta.")
print(" lente: path CANONICO, identica nelle due braccia. Confronto sempre ISO-PESO.")
# ---------------------------------------------------------- §1 famiglia, PRIMA di misurare
print("\n" + "-" * L)
print(" 1. GRIGLIA DICHIARATA — scritta prima di guardare i risultati, contata AL RIALZO")
print("-" * L)
cells = [(t, w, h, s) for t in THRS for w in WS for h in HS for s in SIGNS]
print(f" soglia {len(THRS):>3} {['q%.2f' % q for q in QS] + ['rv<=20% (== L>=1)']}")
print(f" finestra {len(WS):>3} {list(WS)} giorni")
print(f" isteresi {len(HS):>3} {list(HS)} (banda +-h attorno alla soglia)")
print(f" segno {len(SIGNS):>3} +1 tieni vol BASSA (ipotesi) / -1 tieni vol ALTA (controllo)")
print(f" timeframe 1 la gamba SKH01 vive su 230m per costruzione")
print(f" -> celle di QUESTA famiglia: {len(cells)}")
print(f" -> trial gia' spesi sullo STESSO oggetto: {N_INHERITED} "
f"(28 r0822_vol_size + 43 r0822_volsize_skeptic)")
print(f" -> CONTO ONESTO DICHIARATO: {len(cells)} + {N_INHERITED} = "
f"{len(cells) + N_INHERITED} trial. La sensibilita' al conteggio e' in §5.")
assert len(cells) == 144
# ---------------------------------------------------------- estrazione
print("\n estrazione tabelle trade (V2-DD) ...", flush=True)
EX = {o: {a: V.extract(a, o) for a in ASSETS} for o in offs}
ntr = {a: int(np.mean([len(EX[o][a]["i_ent"]) for o in offs])) for a in ASSETS}
print(f" fatto in {time.time()-t0:.0f}s — trade medi/ancora: " +
", ".join(f"{a} {ntr[a]}" for a in ASSETS))
TP = A.tp01_baseline_daily()
can = offs[0]
# ---------------------------------------------------------- §0 SANITY
print("\n" + "-" * L)
print(" 0. SANITY — bit-exact prima di qualunque delta")
print("-" * L)
from src.portfolio.sleeves import _skyhook_returns
base_leg = {o: V.leg_daily(EX[o], {a: V.size_flat(EX[o][a]) for a in ASSETS}) for o in offs}
off = _skyhook_returns()
d0 = float(np.max(np.abs(base_leg[can].values - off.values))) if len(base_leg[can]) == len(off) \
else float("nan")
print(f" gamba 50/50 off{can} vs sleeves._skyhook_returns(): max|dif| = {d0:.2e} "
f"({len(base_leg[can])} giorni)")
assert d0 < 1e-15, "replica NON bit-exact"
for a in ASSETS:
ref = R.run_asset(a, can)[0]
R._CACHE.clear()
dd = float(np.max(np.abs(V.equity_daily(EX[can][a], V.size_flat(EX[can][a])).values
- ref.values)))
print(f" {a} vs r0702_anchor_skh01.run_asset: max|dif| = {dd:.2e}")
assert dd < 1e-15
# ---------------------------------------------------------- §2 riproduzione del titolo
print("\n" + "-" * L)
print(" 2. RIPRODUZIONE DEL TITOLO DELLO SCETTICO — chiamando la SUA funzione, non la mia")
print(" (pubblicato: BIN L>=med +0,080 FULL 20/23 ; INVERSO -0,159 0/23)")
print("-" * L)
REP = {}
for tag, mode in (("BIN L>=med (scettico)", "gemed"), ("BIN L<1 INVERSO (scettico)", "lt1"),
("BIN L>=1 (scettico)", "ge1")):
fn = SK.size_bin(SK.FROZEN, mode)
bk, ct = {}, {}
for o in offs:
lg = V.leg_daily(EX[o], {a: fn(EX[o][a]) for a in ASSETS})
bk[o], ct[o], _ = iso_pair(TP, lg, base_leg[o])
REP[tag] = (bk, ct)
dF = A.anchor_luck_delta(lambda o: bk[o], lambda o: ct[o], offs, metric=sh)
dH = A.anchor_luck_delta(lambda o: hold(bk[o]), lambda o: hold(ct[o]), offs, metric=sh)
print(f" {tag:<28} ISO dShFULL {dF['median_paired']:+.3f} "
f"({dF['n_positive']}/{len(offs)}) banda [{dF['lo']:+.3f},{dF['hi']:+.3f}] "
f"ISO dShHOLD {dH['median_paired']:+.3f} ({dH['n_positive']}/{len(offs)})")
med_F = float(np.median([sh(book(TP, base_leg[o])) for o in offs]))
med_H = float(np.median([sh(hold(book(TP, base_leg[o]))) for o in offs]))
med_D = float(np.median([maxdd(book(TP, base_leg[o])) for o in offs]))
ban_F = [sh(book(TP, base_leg[o])) for o in offs]
print(f" libro BASE, mediana della banda: ShFULL {med_F:.3f} / ShHOLD {med_H:.3f} / "
f"maxDD {med_D*100:.1f}% [pubblicato 1,626 / 1,037 / 10,4%]")
print(f" banda d'ancora del LIVELLO del libro: ShFULL [{min(ban_F):.3f},{max(ban_F):.3f}] "
f"= ampiezza {max(ban_F)-min(ban_F):.3f} (il termine di paragone di §12)")
# ---------------------------------------------------------- §3 causalita' per troncamento
print("\n" + "-" * L)
print(" 3. CAUSALITA' — la decisione tenere/scartare ricalcolata TRONCANDO i dati al giorno")
print(" d'ingresso. `A.causality_ok` NON E' GIRATO (vuole un target_fn(df) e qui il")
print(" segnale nasce da 5m->230m: ri-eseguirlo su prefissi costa quanto tutto lo")
print(" script). Sostituito da un troncamento diretto, piu' stringente sul punto esatto.")
print("-" * L)
thr_c, w_c, h_c, s_c = ("q", CANON[1]), CANON[2], CANON[3], CANON[4]
bad = tested = 0
for a in ASSETS:
ex = EX[can][a]
kfull = KC.keep(ex, (can, a), thr_c, w_c, h_c, s_c)
rv_full = V._rv_daily(ex["base_daily"], w_c, False)
days = list(dict.fromkeys(ex["day_ent"]))
for d in days[::7]: # 1 ingresso su 7, su tutta la storia
if d not in kfull.index:
continue
b_tr = ex["base_daily"][ex["base_daily"].index <= d]
if len(b_tr) < V.VT_WARM + w_c:
continue
rv_t = V._rv_daily(b_tr, w_c, False)
q_t = rv_t.expanding(min_periods=V.VT_WARM).quantile(thr_c[1])
k_t = _hyst(rv_t.values.astype(float), q_t.values.astype(float), h_c, s_c > 0)[-1]
tested += 1
if bool(k_t) != bool(kfull.loc[d] > 0.5):
bad += 1
assert abs(float(rv_t.iloc[-1]) - float(rv_full.loc[d])) < 1e-12 or \
not np.isfinite(rv_full.loc[d])
print(f" decisioni ricalcolate su prefisso: {tested} DIVERGENTI: {bad}")
print(" NB l'isteresi ha memoria: troncando si perde lo stato, quindi il test e' esatto")
print(" per h=0 (la cella canonica) e conservativo per h>0.")
assert bad == 0, "il binario NON e' causale"
# ---------------------------------------------------------- §4 tabella di famiglia
print("\n" + "-" * L)
print(" 4. LA FAMIGLIA INTERA — ISO dShFULL, mediana delle differenze appaiate su "
f"{len(offs)} ancore")
print("-" * L)
# DUE PASSATE, per la RAM (7 GB sulla VPS, ~3 liberi, e siamo in tanti): la 1a calcola le
# metriche per-ancora di tutte le 144 celle SENZA conservare le serie (144 x 23 x 3 serie da
# 2717 punti sarebbero ~500 MB); la 2a ricalcola e CONSERVA le serie delle sole 3-4 celle che
# servono dopo (canonica / scelta al buio / argmax): 4 x 23 gambe, ~2 s.
BK, CT, LEG, MS = {}, {}, {}, {}
def eval_cell(thr, w, h, sg, store: bool = False) -> dict:
nm = cname(thr, w, h, sg)
fn = size_bin(thr, w, h, sg)
dF, dH, dI, lF, lI, lH, ms, kp, ddv, ddc, dC_ = ([] for _ in range(11))
yr: dict = {}
for o in offs:
lg = leg_of(EX[o], o, fn)
b, c, m = iso_pair(TP, lg, base_leg[o])
dF.append(sh(b) - sh(c)); dH.append(sh(hold(b)) - sh(hold(c)))
dI.append(sh(ins(b)) - sh(ins(c))); dC_.append(cagr(b) - cagr(c))
lF.append(sh(lg)); lI.append(sh(ins(lg))); lH.append(sh(hold(lg)))
ms.append(m); ddv.append(maxdd(b)); ddc.append(maxdd(c))
kp.append(float(np.mean(np.concatenate([fn(EX[o][a], (o, a)) for a in ASSETS]))))
for y in {int(t) for t in b.index.year}:
by, cy = b[b.index.year == y], c[c.index.year == y]
if len(by) > 20:
yr.setdefault(y, []).append(sh(by) - sh(cy))
if store:
BK.setdefault(nm, {})[o] = b
CT.setdefault(nm, {})[o] = c
LEG.setdefault(nm, {})[o] = lg
MS.setdefault(nm, {})[o] = m
a_ = np.asarray(dF, float)
return dict(nm=nm, thr=thr, w=w, h=h, sg=sg,
dF=float(np.median(a_)), nF=int((a_ > 0).sum()),
lo=float(a_.min()), hi=float(a_.max()), arrF=a_,
dH=float(np.median(dH)), nH=int((np.asarray(dH) > 0).sum()),
isoI=float(np.median(dI)), dC=float(np.median(dC_)),
legF=float(np.median(lF)), legI=float(np.median(lI)),
legH=float(np.median(lH)), kept=float(np.median(kp)),
weff=float(np.median([W_SKH * x for x in ms])),
ddv=float(np.median(ddv)), ddc=float(np.median(ddc)),
yr={y: float(np.median(v)) for y, v in yr.items()})
RES = {}
for (thr, w, h, sg) in cells:
r = eval_cell(thr, w, h, sg)
RES[r["nm"]] = r
print(f" ({len(cells)} celle x {len(offs)} ancore valutate in {time.time()-t0:.0f}s "
f"dall'avvio, prima passata senza conservare le serie)")
def matrix(sg: int, h: float) -> None:
print(f"\n ISO dShFULL — segno {'+1 (tieni vol BASSA)' if sg>0 else '-1 (CONTROLLO)'}"
f", isteresi h={h:.2f}")
print(" soglia " + "".join(f"{('w%d' % w):>11}" for w in WS))
for thr in THRS:
t = "rv<=20%" if thr[0] == "abs" else f"q{thr[1]:.2f}"
row = ""
for w in WS:
r = RES[cname(thr, w, h, sg)]
row += f"{r['dF']:>+8.3f}{r['nF']:>3}"
print(f" {t:<10}" + row)
for sg in SIGNS:
for h in HS:
matrix(sg, h)
print("\n (ogni cella: mediana appaiata e n. di ancore positive su "
f"{len(offs)})")
POS = [RES[cname(t, w, h, +1)] for t in THRS for w in WS for h in HS]
NEG = [RES[cname(t, w, h, -1)] for t in THRS for w in WS for h in HS]
ALLC = [RES[cname(t, w, h, s)] for t in THRS for w in WS for h in HS for s in SIGNS]
nmcan = cname(("q", CANON[1]), CANON[2], CANON[3], CANON[4])
argmax = max(ALLC, key=lambda r: r["dF"])
print(f"\n ARGMAX della famiglia: {cname(argmax['thr'], argmax['w'], argmax['h'], argmax['sg']):<26}"
f" dShFULL {argmax['dF']:+.3f} ({argmax['nF']}/{len(offs)})")
print(f" cella dello scettico: {nmcan:<26} dShFULL {RES[nmcan]['dF']:+.3f} "
f"({RES[nmcan]['nF']}/{len(offs)}) rango "
f"{1 + sum(r['dF'] > RES[nmcan]['dF'] for r in ALLC)}/{len(ALLC)}")
print(f" STIMA ONESTA se la cella si sceglie a caso dentro la famiglia (mediana):")
print(f" tutta la famiglia (144) {float(np.median([r['dF'] for r in ALLC])):+.3f} "
f"positive {sum(r['dF']>0 for r in ALLC)}/{len(ALLC)}")
print(f" solo segno +1 (72) {float(np.median([r['dF'] for r in POS])):+.3f} "
f"positive {sum(r['dF']>0 for r in POS)}/{len(POS)}")
print(f" solo segno -1 (72) {float(np.median([r['dF'] for r in NEG])):+.3f} "
f"positive {sum(r['dF']>0 for r in NEG)}/{len(NEG)} <- controllo di segno")
# seconda passata: conserva le serie delle sole celle che servono dopo
sel_iso0 = max(ALLC, key=lambda r: r["isoI"])
for r in dict((x["nm"], x) for x in (RES[nmcan], argmax, sel_iso0,
max(ALLC, key=lambda z: z["legI"]))).values():
eval_cell(r["thr"], r["w"], r["h"], r["sg"], store=True)
dchk = A.anchor_luck_delta(lambda o: BK[nmcan][o], lambda o: CT[nmcan][o], offs, metric=sh)
assert abs(dchk["median_paired"] - RES[nmcan]["dF"]) < 1e-12 and \
dchk["n_positive"] == RES[nmcan]["nF"], "la statistica appaiata di §4 non coincide con A.anchor_luck_delta"
print(f" controllo: la statistica appaiata calcolata in §4 coincide con "
f"`A.anchor_luck_delta` ({dchk['median_paired']:+.6f})")
print(f" quota di trade TENUTI, mia cella q0.50 w90: {RES[nmcan]['kept']*100:.1f}% "
f"(mediana fra le ancore)")
fn_can = size_bin(("q", CANON[1]), CANON[2], CANON[3], CANON[4])
fn_sk = SK.size_bin(SK.FROZEN, "gemed")
ov, k_me, k_sk = [], [], []
for o in offs:
for a in ASSETS:
x = fn_can(EX[o][a], (o, a)) > 0.5
y = fn_sk(EX[o][a]) > 0.5
ov.append(float((x == y).mean())); k_me.append(float(x.mean())); k_sk.append(float(y.mean()))
dleg = float(np.max(np.abs(BK[nmcan][can].values - REP["BIN L>=med (scettico)"][0][can].values)))
print(f" ⚠ la MIA parametrizzazione NON e' bit-identica a quella dello scettico: lui prende")
print(f" la mediana espandente di L = clip(0.20/rv,0,3) DOPO un `.fillna(1.0)` che inietta")
print(f" dei 1.0 nel warm-up; io prendo il quantile espandente di rv, che salta i NaN. Il")
print(f" trasformo e' monotono, quindi le due decisioni coincidono nel "
f"{float(np.mean(ov))*100:.1f}% dei trade")
print(f" (tenuti {float(np.mean(k_me))*100:.1f}% io / {float(np.mean(k_sk))*100:.1f}% lui); "
f"max|dif| fra i due libri = {dleg:.2e}.")
# ---------------------------------------------------------- §5 i tre componenti del gate
print("\n" + "-" * L)
print(" 5. `study_family_honest` — NON chiamabile direttamente (e' cablato sui candidati")
print(" DIREZIONALI: factory -> target_fn via candidate_daily, e qui il candidato e' un")
print(" overlay su uno sleeve). Come il 30/07 sul tenore VRP01, uso i suoi TRE")
print(" COMPONENTI REALI: selezione in-sample-only -> deflated_sharpe -> marginal_vs_tp01.")
print("-" * L)
print(" (a) SELEZIONE IN-SAMPLE-ONLY — due criteri, entrambi dichiarati, entrambi sulla")
print(" MEDIANA fra le ancore (l'ancora non e' un hold-out, e' una dimensione di luck)")
sel_leg = max(ALLC, key=lambda r: r["legI"]) # come `select_cell_insample`
sel_iso = max(ALLC, key=lambda r: r["isoI"]) # cio' su cui si decide davvero
hop_leg = max(ALLC, key=lambda r: r["legH"])
hop_iso = max(ALLC, key=lambda r: r["dH"])
for tag, r in (("Sharpe IS standalone della gamba (come select_cell_insample)", sel_leg),
("ISO dShFULL IN-SAMPLE del libro (cio' che decide)", sel_iso)):
n = cname(r["thr"], r["w"], r["h"], r["sg"])
print(f" scelta al buio su {tag}")
print(f" -> {n:<26} ISO dShFULL {r['dF']:+.3f} ({r['nF']}/{len(offs)}) "
f"ISO dShHOLD {r['dH']:+.3f} ({r['nH']}/{len(offs)}) banda "
f"[{r['lo']:+.3f},{r['hi']:+.3f}]")
for tag, r in (("Sharpe HOLD-OUT standalone", hop_leg), ("ISO dShHOLD del libro", hop_iso)):
print(f" controllo (scegliendo SULL'HOLD-OUT, cio' che NON si fa) su {tag}:")
print(f" -> {cname(r['thr'], r['w'], r['h'], r['sg']):<26} dShFULL {r['dF']:+.3f}"
f" dShHOLD {r['dH']:+.3f}")
CHOSEN = sel_iso
nmc = cname(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"])
print(f" CELLA SCELTA AL BUIO (criterio primario = ISO dShFULL in-sample): {nmc}")
print("\n (b) DEFLATED SHARPE — sullo Sharpe FULL standalone della gamba (la grandezza che")
print(" `select_cell_insample` usa), mediana fra le ancore; `daily` della barra canonica")
print(" serve solo per T/asimmetria/curtosi. Sensibilita' al CONTEGGIO ottenuta")
print(" REPLICANDO la distribuzione dei trial osservati (tiling), non riempiendola con")
print(" la mediana: riempire con la mediana ABBASSA la varianza dei trial e fa SALIRE il")
print(" DSR all'aumentare di N, che e' il verso sbagliato. (Errore mio, catturato nel")
print(" pilota e corretto: il primo output dava DSR 0,966 -> 0,986 -> 0,998 da N=144 a")
print(" N=500, cioe' 'piu' trial = piu' significativo'.)")
all_sr = [r["legF"] for r in ALLC]
pos_sr = [r["legF"] for r in POS]
base_sr = float(np.median([sh(base_leg[o]) for o in offs]))
dly = LEG[nmc][can]
def tile(v, N):
return (list(v) * (N // len(v) + 1))[:N]
for lab, sr, d in (("cella scelta al buio", CHOSEN["legF"], dly),
("BASELINE (size fissa)", base_sr, base_leg[can])):
for N, tagN in ((len(ALLC), f"{len(ALLC)} di QUESTA famiglia"),
(len(ALLC) + N_INHERITED,
f"{len(ALLC)+N_INHERITED} = conto onesto (+{N_INHERITED} ereditati)"),
(500, "500 (ipotetico, al rialzo)")):
dsr, sr0 = A.deflated_sharpe(sr, tile(all_sr, N), d)
print(f" {lab:<22} N={tagN:<44} DSR {dsr:.3f} max atteso sotto il null {sr0:+.3f}")
lo_n, hi_n = len(ALLC), 20000
if A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, hi_n), dly)[0] >= 0.95:
cross = f">{hi_n}"
elif A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, lo_n), dly)[0] < 0.95:
cross = f"<{lo_n}"
else:
for _ in range(40):
mid = (lo_n + hi_n) // 2
if A.deflated_sharpe(CHOSEN["legF"], tile(all_sr, mid), dly)[0] >= 0.95:
lo_n = mid
else:
hi_n = mid
cross = f"~{lo_n}"
print(f" -> IL VERDETTO SI RIBALTA A N {cross} trial: sotto quel conteggio il DSR passa,")
print(f" sopra fallisce. Il conto onesto dichiarato ({len(ALLC)+N_INHERITED}) sta dal lato")
print(f" del PASS. Precedente 30/07: un DSR passava a N=8, falliva a N=72 e a N=360; qui")
print(f" la stessa sensibilita' e' pubblicata invece di essere taciuta.")
dsp, sr0p = A.deflated_sharpe(CHOSEN["legF"], tile(pos_sr, len(POS) + N_INHERITED), dly)
print(f" sensibilita' alla COMPOSIZIONE della famiglia: escludendo il segno -1 (che e' un")
print(f" controllo e non un candidato) i trial diventano {len(POS)}+{N_INHERITED} e la loro")
print(f" dispersione cala -> DSR {dsp:.3f} (max atteso {sr0p:+.3f}). Includere il segno -1")
print(f" e' la scelta CONSERVATIVA e resta quella dichiarata.")
print(f" dispersione dei trial: sd(Sharpe FULL) = {float(np.std(all_sr, ddof=1)):.3f} sulle")
print(f" {len(ALLC)} celle, {float(np.std(pos_sr, ddof=1)):.3f} sulle {len(POS)} di segno +1.")
print(f" -> il DSR qui NON e' vacuo (il baseline lo FALLISCE dove il candidato lo passa):")
print(f" la famiglia contiene celle davvero diverse fra loro, al contrario della famiglia")
print(f" VOL-SIZE dove erano tutte perturbazioni della stessa size continua.")
print(f" ❌ QUESTO REFUTA LA MIA IPOTESI (b), registrata nel docstring PRIMA di misurare")
print(f" ('il DSR sara' vacuo come su VOL-SIZE'). La ragione per cui mi sbagliavo e'")
print(f" strutturale e vale come regola: un DSR e' vacuo quando i trial si somigliano")
print(f" (var_tr -> 0 => sr0 -> 0 => DSR ~ 1 per chiunque). Una famiglia BINARIA contiene")
print(f" celle che scartano insiemi di trade MOLTO diversi — compreso il segno inverso —")
print(f" quindi la dispersione dei trial e' reale e il test riacquista potenza. Non e' il")
print(f" binario a essere piu' solido della formula: e' la FAMIGLIA a essere piu' larga.")
print("\n (c) MARGINAL vs TP01 — sulla gamba modificata e sulla gamba base")
for lab, s in (("BASE (size fissa)", base_leg[can]), (nmc, LEG[nmc][can]),
("cella dello scettico", LEG[nmcan][can])):
m = A.marginal_vs_tp01(s)
b25 = m.get("blends", {}).get("w25", {})
print(f" {lab:<26} {str(m.get('marginal_verdict')):<10} corr {m.get('corr_full')} "
f"uplift w25 full {b25.get('uplift_full')} / hold {b25.get('uplift_hold')} "
f"robust_oos={m.get('robust_oos')} insample_edge={m.get('has_insample_edge')} "
f"hedge={m.get('is_hedge')}")
imp = A.implausible_sharpe(
LEG[nmc][can],
n_trades=int(sum(len(EX[can][a]["net"]) for a in ASSETS)),
n_losing_trades=int(sum((EX[can][a]["net"] < 0).sum() for a in ASSETS)))
print(f" implausible_sharpe (gamba scelta): implausible={imp['implausible']} "
f"attive {imp.get('active_frac', 0)*100:.1f}% perdite/attive "
f"{imp.get('loss_frac', 0)*100:.1f}% Calmar {imp.get('calmar', 0):.1f}")
# ---------------------------------------------------------- §6 null con potenza
print("\n" + "-" * L)
print(" 6. IL NULL CHE HA POTENZA — permutazione della SEQUENZA tenuti/scartati.")
print(" Stessa quota di trade tenuti, riassegnata a caso: separa 'conta QUALE trade si")
print(" scarta' (informazione) da 'conta scartarne meta'' (aritmetica). Una mediana del")
print(" null lontana da 0 direbbe che il null non e' un null (lezione 22/08 sul null")
print(f" invertito). {args.ndraw} estrazioni su TUTTE le {len(offs)} ancore, non su 3.")
print("-" * L)
rng = np.random.default_rng(20260822)
for lab, r in (("cella scelta al buio", CHOSEN), ("cella dello scettico", RES[nmcan]),
("argmax di famiglia", argmax)):
nm = cname(r["thr"], r["w"], r["h"], r["sg"])
real = float(np.median([sh(BK[nm][o]) - sh(CT[nm][o]) for o in offs]))
fn = size_bin(r["thr"], r["w"], r["h"], r["sg"])
sz0 = {o: {a: fn(EX[o][a], (o, a)) for a in ASSETS} for o in offs}
nul = []
for _ in range(args.ndraw):
v = []
for o in offs:
szp = {a: rng.permutation(sz0[o][a]) for a in ASSETS}
lg = V.leg_daily(EX[o], szp)
b, c, _ = iso_pair(TP, lg, base_leg[o])
v.append(sh(b) - sh(c))
nul.append(float(np.median(v)))
nul = np.asarray(nul)
pc = float((nul < real).mean())
print(f" {lab:<24} {nm:<24} reale {real:+.3f} null mediana {np.median(nul):+.3f} "
f"p90 {np.percentile(nul, 90):+.3f} max {nul.max():+.3f} -> p ~ {1-pc:.3f}")
print(f" (seme 20260822; {args.ndraw} estrazioni x {len(offs)} ancore)")
print("\n ⚠ LA MEDIANA DEL NULL NON E' ~0, E NON E' UN DIFETTO DEL NULL: E' UN FATTO.")
print(" Scartare a caso meta' dei trade DANNEGGIA, perche' lo Sharpe ANNUALIZZATO cresce")
print(" con la radice del numero di scommesse per anno: dimezzare i trade sullo stesso")
print(" arco di calendario costa un fattore ~sqrt(2) di Sharpe di gamba. Misurato:")
nmv = cname(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"])
fnv = size_bin(CHOSEN["thr"], CHOSEN["w"], CHOSEN["h"], CHOSEN["sg"])
rng2 = np.random.default_rng(20260823)
shb = float(np.median([sh(base_leg[o]) for o in offs]))
shv = float(np.median([sh(LEG[nmv][o]) for o in offs]))
kp = float(np.median([np.mean(np.concatenate([fnv(EX[o][a], (o, a)) for a in ASSETS]))
for o in offs]))
shn = []
for _ in range(min(40, args.ndraw)):
sz0 = {o: {a: fnv(EX[o][a], (o, a)) for a in ASSETS} for o in offs}
shn.append(float(np.median([sh(V.leg_daily(EX[o], {a: rng2.permutation(sz0[o][a])
for a in ASSETS})) for o in offs])))
print(f" Sharpe della GAMBA: baseline (tutti i trade) {shb:.3f} -> permutata "
f"{float(np.median(shn)):.3f} -> filtrata {shv:.3f}")
print(f" quota tenuta {kp*100:.0f}% -> attesa dalla sola perdita di scommesse "
f"{shb*np.sqrt(kp):.3f}; osservata sotto permutazione {float(np.median(shn)):.3f}")
print(" Conseguenza per la lettura: il null NON e' centrato su zero, quindi il p-value")
print(" risponde alla domanda 'la SELEZIONE batte una selezione casuale della stessa")
print(" taglia?' e non alla domanda 'il filtro batte il non filtrare?'. La seconda e' il")
print(" confronto con la BASE (§4), che e' quello su cui si decide. Il filtro deve")
print(" superare una penale meccanica prima di guadagnare qualcosa: e' un test SEVERO.")
# ---------------------------------------------------------- §7 iso-peso / de-levering
print("\n" + "-" * L)
print(" 7. ISO-PESO e NULL DEL DE-LEVERING")
print("-" * L)
print(f" {'cella':<26}{'wSKH eff':>10}{'dShFULL RAW':>13}{'ISO':>9}{'maxDD var':>11}"
f"{'maxDD ctrl':>11}{'k_iso_dd':>10}")
for lab, r in (("BASE", None), ("cella dello scettico", RES[nmcan]),
("cella scelta al buio", CHOSEN)):
if r is None:
print(f" {'BASE (size fissa)':<26}{W_SKH:>10.3f}{0.0:>+13.3f}{0.0:>+9.3f}"
f"{maxdd(book(TP, base_leg[can]))*100:>10.1f}%{'-':>11}{'-':>10}")
continue
nm = cname(r["thr"], r["w"], r["h"], r["sg"])
raw = A.anchor_luck_delta(lambda o, n=nm: BK[n][o],
lambda o: book(TP, base_leg[o]), offs, metric=sh)
dv = float(np.median([maxdd(BK[nm][o]) for o in offs]))
dc = float(np.median([maxdd(CT[nm][o]) for o in offs]))
ks = (f"{k_iso_dd(CT[nm][can], BK[nm][can]):.3f}" if dv < dc
else "n/a") # n/a = non riduce il DD: nulla da negare
print(f" {nm:<26}{r['weff']:>10.3f}{raw['median_paired']:>+13.3f}{r['dF']:>+9.3f}"
f"{dv*100:>10.1f}%{dc*100:>10.1f}%{ks:>10}")
print(" Il null del de-levering e' DEGENERE su una ri-scalatura (sh(k*base) === sh(base)):")
print(" qui il test e' l'iso-peso sullo Sharpe. `k_iso_dd` e' riportato come diagnostica del")
print(" DD ed e' calcolato CONTRO il controllo iso-peso (non contro il baseline nudo).")
print(" NB il binario ABBASSA il peso effettivo di SKH01, non lo alza: la preoccupazione del")
print(" 26/07 ('alzare SKH01') non morde, ma l'iso-peso resta obbligatorio nell'altro verso.")
print(f"\n LIVELLI del libro (mediana fra le {len(offs)} ancore), netti fee:")
print(f" {'configurazione':<30}{'ShFULL':>9}{'ShHOLD':>9}{'ShIS':>8}{'maxDD':>9}{'CAGR':>9}"
f"{'vol':>8}")
lev = [("BASE (libro live 75/25)", {o: book(TP, base_leg[o]) for o in offs})]
for lab, r in (("BIN, peso eff. lasciato cadere", RES[nmcan]),
("BIN, controllo ISO-PESO", None)):
if r is None:
lev.append((lab, {o: CT[nmcan][o] for o in offs}))
else:
lev.append((lab, {o: BK[nmcan][o] for o in offs}))
for lab, dd_ in lev:
f_ = float(np.median([sh(dd_[o]) for o in offs]))
h_ = float(np.median([sh(hold(dd_[o])) for o in offs]))
i_ = float(np.median([sh(ins(dd_[o])) for o in offs]))
d_ = float(np.median([maxdd(dd_[o]) for o in offs]))
c_ = float(np.median([cagr(dd_[o]) for o in offs]))
v_ = float(np.median([vol(dd_[o]) for o in offs]))
print(f" {lab:<30}{f_:>9.3f}{h_:>9.3f}{i_:>8.3f}{d_*100:>8.1f}%{c_*100:>8.1f}%"
f"{v_*100:>7.1f}%")
print(" ⚠ la DIFFERENZA di queste mediane NON e' la mediana delle differenze appaiate (§4):")
print(" qui le mediane cadono su ancore diverse. La tabella serve a dare i LIVELLI netti fee")
print(" richiesti dal contratto; il numero che decide resta quello appaiato.")
# ---------------------------------------------------------- §8 weights_tilt_null
print("\n" + "-" * L)
print(" 8. `weights_tilt_null` — una modulazione binaria della size E' un cambio di pesi nel")
print(" tempo. Il gate confronta vettori STATICI: cattura la sola componente di SCALA")
print(" (limite dichiarato, come nel bersaglio). k_seen = tutti i trial dichiarati.")
print("-" * L)
cols = {"TP01": TP, "SKH01": base_leg[can]}
for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN),
("argmax di famiglia", argmax)):
nm = cname(r["thr"], r["w"], r["h"], r["sg"])
m = MS[nm][can]
wp = {"TP01": W_TP, "SKH01": W_SKH * m}
g = weights_tilt_null(cols, {"TP01": W_TP, "SKH01": W_SKH}, wp, caps={"SKH01": 0.5},
floor=0.05, n=300, k_seen=len(ALLC) + N_INHERITED)
print(f" {lab:<24} {nm:<24} peso equiv SKH {wp['SKH01']/sum(wp.values()):.3f} "
f"d_IS {g['delta_insample']:+.4f} d_HOLD {g['delta_hold']:+.4f} "
f"pctl {g['pctl_hold']:.1f} gate_pass={g['gate_pass']}")
print(" ⚠ LIMITE, e va detto perche' e' aggirabile: il binario ABBASSA il peso effettivo di")
print(" SKH01, e il gate misura proprio quella caduta. Nella forma di deploy naturale (si")
print(" ri-scala la gamba per riportare il peso effettivo a 0,25) il vettore proposto e'")
print(" IDENTICO a quello corrente -> d_IS = d_HOLD = 0 e il gate diventa INERTE, non")
print(" superato. Quindi `weights_tilt_null` NON puo' essere l'evidenza decisiva qui: e'")
print(" negativo nella forma nuda e vacuo nella forma compensata. Cio' che decide resta la")
print(" banda APPAIATA a iso-peso (§4) e la stima di famiglia.")
# ---------------------------------------------------------- §9 eseguibilita'
print("\n" + "-" * L)
print(f" 9. ESEGUIBILITA' a ${CAPITAL:.0f} — il vantaggio DICHIARATO dal filone (dimezza gli")
print(" ordini). Target NETTATO 0.75*TP01 + 0.25*SKH sul grid 1h = cio' che il cron manda.")
print("-" * L)
print(f" {'configurazione':<26}{'Sh model':>10}{'Sh reale':>10}{'haircut':>9}"
f"{'ordini eseg':>13}{'sotto-min':>11}{'turnover/a':>12}")
EXEC = {}
for lab, r in (("BASE (size fissa)", None), ("cella dello scettico", RES[nmcan]),
("cella scelta al buio", CHOSEN)):
nm = None if r is None else "SEL"
tm = tr = tu = 0.0
no = ns = 0
for a in ASSETS:
df1h = TO.get1h(a)
SZa = {} if r is None else {
"SEL": (lambda ex, rr=r, aa=a: size_bin(rr["thr"], rr["w"], rr["h"],
rr["sg"])(ex, (can, aa)))}
tgt = V.netted_target(a, EX[can][a], nm, SZa, {}, {}, {}, {}, TP, can, df1h)
ev = A.eval_weights_smallcap(df1h, tgt, capital=CAPITAL, min_order=MIN_ORDER)
tm += 0.5 * ev["modeled"]["sharpe"]
tr += 0.5 * ev["realistic"]["sharpe"]
tu += ev["executed_turnover_per_year"]
no += ev["n_executed_trades"]
dw = np.abs(np.diff(np.nan_to_num(tgt), prepend=0.0))
ns += int(((dw > 1e-12) & (dw * CAPITAL < MIN_ORDER)).sum())
EXEC[lab] = (tm, tr, no, ns, tu)
print(f" {lab:<26}{tm:>10.3f}{tr:>10.3f}{tm-tr:>9.3f}{no:>13}{ns:>11}{tu:>12.1f}")
print(" 'ordini eseg' e 'sotto-min' sono sui due asset sommati, su tutta la storia 1h.")
ntr_base = int(sum(len(EX[can][a]['i_ent']) for a in ASSETS))
ntr_sel = int(sum((size_bin(CHOSEN['thr'], CHOSEN['w'], CHOSEN['h'], CHOSEN['sg'])
(EX[can][a], (can, a)) > 0).sum() for a in ASSETS))
print(f" trade SKH eseguiti (off{can}): BASE {ntr_base} -> cella scelta {ntr_sel} "
f"({ntr_sel/ntr_base*100:.0f}%)")
print(" ⚠ IL RISPARMIO DI FEE E' GIA' DENTRO i numeri sopra: `extract` netta ogni trade di")
print(f" FEE_RT={V.FEE_RT:.4f} (10 bps RT) all'ingresso, quindi un trade scartato non paga.")
print(" Convertirlo di nuovo con la curva del 26/07 (-0,017 Sh/bps di libro, SKH 4x TP01)")
print(" sarebbe un DOPPIO CONTEGGIO. Cio' che NON e' dentro e' (a) il min-order a $635,")
print(" misurato qui sopra come haircut, e (b) il fatto che la fee reale e' 3,5 bps/lato =")
print(" 7 bps RT contro i 10 modellati -> i backtest sono CONSERVATIVI, in entrambe le braccia.")
# ---------------------------------------------------------- §9b fee vs informazione
print("\n" + "-" * L)
print(" 9b. QUANTO DELL'EFFETTO E' INFORMAZIONE E QUANTO E' SOLO FEE RISPARMIATA.")
print(" Scartare meta' dei trade risparmia meta' delle fee: parte del +0,080 e' quella,")
print(" ed e' un guadagno vero ma NON e' un'ipotesi sulla vol. Lo stesso confronto,")
print(" identico in tutto, ricalcolato a fee diverse (le size si ricalcolano da zero,")
print(" perche' la vol trailing della gamba dipende anche dalle fee).")
print("-" * L)
print(f" {'lente':<40}{'cella scettico':>17}{'pos/n':>8}{'cella al buio':>17}{'pos/n':>8}")
FEEDEC = {}
keep_fee = V.FEE_RT
for fee, tag in ((keep_fee, f"MODELLATA {keep_fee*1e4:.0f} bps RT (i backtest)"),
(0.0007, "REALE 3,5 bps/lato = 7 bps RT"),
(0.0, "fee ZERO (informazione pura)")):
V.FEE_RT = fee
if fee == keep_fee:
EXf, basef = EX, base_leg
else:
EXf = {o: {a: V.extract(a, o) for a in ASSETS} for o in offs}
basef = {o: V.leg_daily(EXf[o], {a: V.size_flat(EXf[o][a]) for a in ASSETS})
for o in offs}
row = ""
for r in (RES[nmcan], CHOSEN):
fn = size_bin(r["thr"], r["w"], r["h"], r["sg"])
bk, ct = {}, {}
for o in offs:
lg = V.leg_daily(EXf[o], {a: fn(EXf[o][a], (tag, o, a)) for a in ASSETS})
bk[o], ct[o], _ = iso_pair(TP, lg, basef[o])
d = A.anchor_luck_delta(lambda o: bk[o], lambda o: ct[o], offs, metric=sh)
FEEDEC[(tag, cname(r["thr"], r["w"], r["h"], r["sg"]))] = d["median_paired"]
row += f"{d['median_paired']:>+17.3f}{d['n_positive']:>5}/{len(offs):<3}"
print(f" {tag:<40}{row}")
if fee != keep_fee:
del EXf, basef
V.FEE_RT = keep_fee
assert V.FEE_RT == 0.001
f_mod = FEEDEC[(f"MODELLATA {keep_fee*1e4:.0f} bps RT (i backtest)", nmcan)]
f_zero = FEEDEC[("fee ZERO (informazione pura)", nmcan)]
print(f" -> a fee ZERO resta {f_zero:+.3f} di {f_mod:+.3f} = "
f"{100*f_zero/f_mod if f_mod else float('nan'):.0f}% dell'effetto: la parte restante")
print(" e' fee risparmiata. La fee vera oggi e' 3,5 bps/lato, meno di quella modellata,")
print(" quindi nel LIVE la componente-fee del vantaggio e' PIU' PICCOLA di quella misurata")
print(" sui backtest — il verso in cui il numero pubblicato e' ottimista per il deploy.")
# ---------------------------------------------------------- §10 trasferimento su V1
print("\n" + "-" * L)
print(" 10. TRASFERIMENTO su SKH01_V1 — seconda sequenza di trade, geometria d'uscita")
print(" DIVERSA (stop/target in ATR invece che a percentuale fissa). Non e' un hold-out")
print(" temporale, ma e' l'unico campione indipendente disponibile a questo costo.")
print("-" * L)
EX1 = {o: {a: V.extract(a, o, SKH01_V1) for a in ASSETS} for o in offs}
base1 = {o: V.leg_daily(EX1[o], {a: V.size_flat(EX1[o][a]) for a in ASSETS}) for o in offs}
print(f" {'cella':<26}{'ISO dShFULL':>13}{'pos/n':>8}{'ISO dShHOLD':>14}{'pos/n':>8}"
f"{'tenuti':>9}")
for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)):
fn = size_bin(r["thr"], r["w"], r["h"], r["sg"])
b1, c1, kp = {}, {}, []
for o in offs:
lg = V.leg_daily(EX1[o], {a: fn(EX1[o][a], ("V1", o, a)) for a in ASSETS})
b1[o], c1[o], _ = iso_pair(TP, lg, base1[o])
kp.append(float(np.mean(np.concatenate(
[fn(EX1[o][a], ("V1", o, a)) for a in ASSETS]))))
dF = A.anchor_luck_delta(lambda o: b1[o], lambda o: c1[o], offs, metric=sh)
dH = A.anchor_luck_delta(lambda o: hold(b1[o]), lambda o: hold(c1[o]), offs, metric=sh)
print(f" {cname(r['thr'], r['w'], r['h'], r['sg']):<26}{dF['median_paired']:>+13.3f}"
f"{dF['n_positive']:>5}/{len(offs):<3}{dH['median_paired']:>+14.3f}"
f"{dH['n_positive']:>5}/{len(offs):<3}{float(np.median(kp))*100:>8.0f}%")
# ---------------------------------------------------------- §11 per anno
print("\n" + "-" * L)
print(" 11. SCOMPOSIZIONE PER ANNO (regola 22/08 dopo SOL: un contributo positivo si")
print(" scompone per ANNO prima di crederci — 24/24 ancore possono essere un anno solo)")
print("-" * L)
yrs = sorted({int(y) for y in BK[nmcan][can].index.year})
print(f" {'cella':<26}" + "".join(f"{y:>8}" for y in yrs))
for lab, r in (("cella dello scettico", RES[nmcan]), ("cella scelta al buio", CHOSEN)):
nm = cname(r["thr"], r["w"], r["h"], r["sg"])
cc = []
for y in yrs:
v = [sh(BK[nm][o][BK[nm][o].index.year == y]) - sh(CT[nm][o][CT[nm][o].index.year == y])
for o in offs if (BK[nm][o].index.year == y).sum() > 20]
cc.append(float(np.median(v)) if v else float("nan"))
print(f" {nm:<26}" + "".join(f"{c:>+8.2f}" for c in cc) +
f" positivi {int(np.nansum(np.asarray(cc) > 0))}/{len(yrs)}")
# ---------------------------------------------------------- §12 la domanda che decide
print("\n" + "-" * L)
print(" 12. LA DOMANDA CHE DECIDE — un effetto che vale un terzo della propria incertezza")
print(" merita di cambiare un sistema che gira con soldi veri?")
print("-" * L)
amp = max(ban_F) - min(ban_F)
nm = nmcan
dC = float(np.median([cagr(BK[nm][o]) - cagr(CT[nm][o]) for o in offs]))
cag_band = [cagr(book(TP, base_leg[o])) for o in offs]
print(f" (i) IL CONFRONTO GIUSTO NON E' QUELLO DELLA DOMANDA. La banda d'ancora del")
print(f" LIVELLO ({amp:.3f} di ampiezza, +-{amp/2:.3f}) e' incertezza sul livello, e")
print(f" si CANCELLA in un confronto APPAIATO: le due braccia girano sulla STESSA")
print(f" ancora. Cio' che va confrontato con l'effetto e' la dispersione della")
print(f" DIFFERENZA appaiata, che e' un ordine di grandezza piu' piccola.")
print(f" cella scettico: effetto {RES[nm]['dF']:+.3f}, banda appaiata "
f"[{RES[nm]['lo']:+.3f},{RES[nm]['hi']:+.3f}], "
f"{RES[nm]['nF']}/{len(offs)} ancore positive.")
print(f" (ii) MA la banda del livello torna a mordere DUE volte, e in modo decisivo:")
print(f" (a) la SCELTA della cella e' fatta guardando quella banda -> la stima onesta")
print(f" non e' l'argmax ma la MEDIANA DI FAMIGLIA "
f"({float(np.median([r['dF'] for r in POS])):+.3f} sul segno +1);")
print(f" (b) l'effetto e' misurato sugli STESSI 8 anni su cui SKH01 e' stato")
print(f" selezionato, e non esiste un forward capace di misurarlo (SE(Sharpe)")
print(f" ~1,4 su 6 mesi contro un effetto di 0,08).")
print(f" (iii) IL CRITERIO, dichiarato: si tocca il libro live solo se TUTTE E QUATTRO —")
print(f" (1) l'effetto sopravvive alla stima ONESTA di famiglia (mediana, non argmax)")
print(f" (2) `weights_tilt_null` passa (e' il gate codificato per i cambi di peso)")
print(f" (3) il gate marginale + un null CON POTENZA lo sostengono")
print(f" (4) l'effetto e' grande abbastanza da pagare un modo di fallire in piu'.")
print(f" (iv) ORDINI DI GRANDEZZA a ${CAPITAL:.0f} (regola 26/07: quando le leve hanno")
print(f" ordini di grandezza diversi, DIRLO):")
print(f" effetto della cella (ISO, mediana appaiata): dCAGR {dC*100:+.2f}%/anno")
print(f" = ${dC*CAPITAL:+.2f}/anno = EUR {dC*CAPITAL/365*0.86:+.4f}/giorno")
print(f" banda d'ancora del CAGR del libro: "
f"[{min(cag_band)*100:.1f}%, {max(cag_band)*100:.1f}%] "
f"= ${(max(cag_band)-min(cag_band))*CAPITAL:.0f}/anno di incertezza")
print(f" versare EUR 500/mese (il piano dell'operatore) = EUR 16,44/giorno di capitale")
print(f" nuovo, cioe' ~{16.44/max(1e-9, abs(dC*CAPITAL/365*0.86)):.0f}x l'effetto misurato qui.")
print(f" (v) IL CRITERIO APPLICATO, riga per riga:")
med_pos = float(np.median([r['dF'] for r in POS]))
c1 = med_pos > 0
c3 = True
print(f" (1) stima ONESTA di famiglia (mediana segno +1) = {med_pos:+.3f} "
f"contro argmax {argmax['dF']:+.3f} e cella pubblicata {RES[nmcan]['dF']:+.3f}"
f" -> {'PASS' if c1 else 'FAIL'}")
print(f" (2) `weights_tilt_null`: gate_pass=False nella forma nuda, INERTE nella forma")
print(f" compensata -> il gate non lo sostiene in nessuna delle due -> FAIL/vacuo")
print(f" (3) marginale ADDS + null di permutazione con potenza (§6) -> vedi sopra")
print(f" (4) taglia contro il modo di fallire in piu': "
f"EUR {dC*CAPITAL/365*0.86:+.4f}/giorno a ${CAPITAL:.0f} -> FAIL")
print(f" Di cui, per giunta, solo il {100*f_zero/f_mod if f_mod else float('nan'):.0f}% "
f"e' informazione e il resto e' fee risparmiata (§9b),")
print(f" e la fee vera e' piu' bassa di quella modellata.")
print("\n" + "=" * L)
print(f" fatto in {time.time()-t0:.0f}s")
print("=" * L)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink