Adriano Dal Pastro
96d37c7013
§41 TP01-SIZE. Domanda: esiste una size che risponde alla CONVINZIONE (invece che
solo alla vol realizzata) e che, A ISO-VOL, alzi il drift di TP01? Sarebbe un
miglioramento k-indipendente sul 75% del libro live. Risposta: NO.
Griglia dichiarata PRIMA di guardare: 60 celle (rho = g(1/3)/g(1) x q = esponente
del vol-targeting), TF 1d, 24 ancore, segnale CONGELATO. Replica bit-exact del
canonico max|diff| = 0.0 su target_series E sui rendimenti di sleeve.
- FATTO STRUTTURALE che corregge la premessa: la convinzione NON vale {1/3,2/3,1}.
La media di 3 segni sta in {-1,-1/3,0,+1/3,+1} -> il bucket 2/3 e' ARITMETICAMENTE
IMPOSSIBILE. La famiglia ha UN grado di liberta' (rho), non tre: potenza/floor/
soglia sono la stessa cella riparametrizzata.
- NULL DEL RE-LEVERING misurato in forma pura: togliere il vol-targeting (q=0) porta
il CAGR da 16,32% a 46,75% (+186%) e a ISO-VOL fa -2,40pp. Era TUTTA leva.
- Il meglio dell'intera griglia e' +0,30pp di CAGR iso-vol. La cella scelta AL BUIO
(rho=0,25 q=1,25) peggiora l'hold-out del LIBRO in 0/24 ancore e il suo maxDD in
24/24, contro +0,19pp di CAGR.
- Spearman(ShIS, ShHOLD) = -0,527 sulle 60 celle: qui scegliere in-sample e' PEGGIO
di una moneta. Meccanismo misurato: Spearman(rho, ShHOLD) = +1,000 a TUTTI e sei i
q (monotono), mentre in-sample e' a gobba -> la convinzione e' informativa
in-sample e ANTI-informativa in hold-out. Nessuna cella e' proponibile.
- L'unica con guadagno hold-out robusto e' il BINARIO rho=1 (dShHOLD +0,243, 24/24),
che perde FULL e drift iso-vol e si potrebbe scegliere solo guardando l'hold-out.
- deflated-Sharpe 0,999 PASS ma QUASI VACUO (famiglia omogenea, sr0 0,197):
sensibilita' pubblicata, FALLISCE a sr0 >= 0,9. Conferma §10 dell'ondata 22/08.
- Controlli positivi 3/3 (de-levering puro, oracolo look-ahead con causality_ok=False,
anti-controllo rho=3). Eseguibilita' a $635: haircut 0,001 -> non e' il vincolo.
- Sottoprodotti: banda d'ancora del canonico ricalcolata su dati odierni (hold-out
canonico 0,441 = 96° pctl, mediana onesta 0,219) e altlib.tp01_baseline_daily NON
e' bit-exact col sleeve (1 ulp su 68,5% dei giorni: _to_daily fa (1+r).prod()-1).
Book, pesi, cron, config INVARIATI.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018M8Ncho6QV9FWLdyy4VyQf
763 lines
43 KiB
Python
763 lines
43 KiB
Python
#!/usr/bin/env python
|
|
"""r0823_tp01_size.py — §41 TP01-SIZE: una size che risponde alla CONVINZIONE paga?
|
|
|
|
DOMANDA
|
|
-------
|
|
TP01 e' il 75% del libro live. Dichiara target_vol=20% e ne realizza 12,14%; la
|
|
scomposizione del 22/08 e' `20% x 0,682 x 0,890`, dove lo 0,682 sono due meta' quasi
|
|
uguali — (i) flat nel 44,3% delle barre, (ii) "convinzione parziale". Il fatto
|
|
trasferibile: alzare `target_vol` scala TUTTO uniformemente, quindi aumenta la size
|
|
proprio dove il segnale e' piu' debole. Nessuno ha mai misurato l'alternativa:
|
|
**una size che risponde alla convinzione invece che solo alla vol realizzata**.
|
|
Se esistesse, sarebbe un miglioramento k-INDIPENDENTE (ridistribuzione della stessa
|
|
size), non un aumento di leva.
|
|
|
|
ATTESA A PRIORI (scritta PRIMA di qualunque misura, e prima del diagnostico §2/§3)
|
|
---------------------------------------------------------------------------------
|
|
"Il canonico (size lineare nella convinzione) e' gia' dentro il suo plateau. Qualunque
|
|
rimodellamento dara' un delta indistinguibile da zero A ISO-VOL, e un eventuale
|
|
vincitore sara' (i) fortuna d'ancora, (ii) un anno solo, o (iii) bocciato dal
|
|
deflated-Sharpe. VERDETTO ATTESO: SCARTATO."
|
|
Corollario dichiarato: mi aspetto che il null del de-levering si presenti qui nella sua
|
|
forma SPECULARE (re-levering): una variante che alza la size media sembrera' guadagnare
|
|
drift, e a iso-vol il guadagno sparira'.
|
|
NB di trasparenza: il diagnostico per bucket (§3) e' stato girato PRIMA della griglia e
|
|
ne ha raffinato la DIREZIONE attesa (rho > 1/3). La FAMIGLIA era gia' dichiarata.
|
|
|
|
FAMIGLIA DICHIARATA PRIMA DI GUARDARE — e perche' e' piu' piccola di quanto sembra
|
|
----------------------------------------------------------------------------------
|
|
Il segnale resta CONGELATO (TSMOM sign-blend 30/90/180g, long-flat). Cambia solo la
|
|
mappa `convinzione -> size`:
|
|
target = clip( g_rho(conv) * (target_vol/vol)^q , -leverage, +leverage )
|
|
* asse RHO = g(1/3)/g(1): quanto pesa il bucket debole rispetto al pieno.
|
|
rho in {0.00, 0.15, 0.25, 1/3, 0.45, 0.60, 0.80, 1.00, 1.50, 3.00} (10)
|
|
rho=1/3 = CANONICO; rho=0 = soglia (solo convinzione piena); rho=1 = binario
|
|
(size costante quando non-flat, cioe' convinzione RIMOSSA); rho=3 = controllo
|
|
speculare (sovrappesa il debole).
|
|
* asse Q = esponente del vol-targeting: quanto la size risponde alla VOL invece che
|
|
alla convinzione. q in {0.00, 0.25, 0.50, 0.75, 1.00, 1.25} (6)
|
|
q=1 = canonico; q=0 = nessun vol-targeting.
|
|
=> GRIGLIA DICHIARATA = 10 x 6 = **60 celle**, TF=1d (il TF di deploy). Conteggio al
|
|
RIALZO: la sensibilita' del verdetto al numero di trial e' pubblicata in §7.
|
|
|
|
FATTO STRUTTURALE (§2) — e una correzione alla premessa del mandato
|
|
-------------------------------------------------------------------
|
|
La convinzione di TP01 NON assume i valori {1/3, 2/3, 1}: la media di TRE segni in
|
|
{-1,0,+1} vale {-1, -1/3, 0, +1/3, +1}, quindi **il bucket 2/3 e' aritmeticamente
|
|
impossibile** e non esiste nei dati. Conseguenza metodologica: le tre
|
|
parametrizzazioni "ovvie" (potenza c^p, floor affine a+(1-a)c, soglia) sono tutte
|
|
REPARAMETRIZZAZIONI dello stesso unico knob rho — la famiglia della convinzione ha UN
|
|
grado di liberta', non tre. Dichiararle come tre assi avrebbe gonfiato i trial con
|
|
duplicati esatti (l'errore speculare a quello del 30/07: li' i trial erano contati al
|
|
ribasso, qui si sarebbero contati al rialzo ma con celle identiche).
|
|
|
|
GATE ISO-VOLATILITA' (obbligatorio, e qui e' ESATTO)
|
|
----------------------------------------------------
|
|
Ogni variante e' confrontata a PARI vol realizzata di sleeve riscalando con un k
|
|
costante. Poiche' net = pos*r - fee*|dpos| e' OMOGENEO di grado 1 in pos, `net_iso = k*net`
|
|
e' ESATTO => lo **Sharpe e' invariante a k** e il gate morde sul **drift**: qualunque
|
|
guadagno di CAGR che sparisce a iso-vol e' re-levering, non edge. k e' calibrato IN-SAMPLE
|
|
(pre-2025) per la lettura hold-out: nessun look-ahead nemmeno nella normalizzazione.
|
|
Il cap di leva 2x e' misurato cella per cella (§4): morde su <=0.15% delle barre nelle 54
|
|
celle candidate e solo nel controllo anti rho=3 (24.7%), quindi non e' il cap a decidere.
|
|
|
|
CONTROLLI POSITIVI (un gate che non segnala nulla puo' essere rotto)
|
|
--------------------------------------------------------------------
|
|
CTRL-DELEV : canonico x 0.5 (pura riscalatura) -> dSharpe ESATTAMENTE 0,
|
|
dCAGR grezzo NEGATIVO, dCAGR a iso-vol ESATTAMENTE 0.
|
|
CTRL-ORACLE : convinzione = segno del rendimento del giorno DOPO (look-ahead) ->
|
|
Sharpe deve esplodere E `causality_ok` deve dare ok=False.
|
|
CTRL-ANTI : rho=3.0 (sovrappesa il bucket debole) -> deve perdere.
|
|
|
|
CAUSALITA' / ONESTA'
|
|
--------------------
|
|
`causality_ok` sulla cella scelta; selezione della cella IN-SAMPLE-ONLY
|
|
(`select_cell_insample`, importata); `deflated_sharpe`; `marginal_vs_tp01`;
|
|
banda d'ancora a 24 offset con **mediana delle DIFFERENZE APPAIATE** e frazione di
|
|
ancore positive; scomposizione per ANNO; impatto sul LIBRO (TP01 75 / SKH01 25);
|
|
eseguibilita' a $635 con `eval_weights_smallcap` (min_order $5).
|
|
|
|
Vincoli operativi: nessun file toccato fuori da questo script; nessuna rete (solo i
|
|
parquet certificati su disco); nessun ordine. Runtime misurato ~9 s su 2 core.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from functools import lru_cache
|
|
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" / "alt"))
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
import altlib as al # noqa: E402
|
|
from src.data.downloader import load_data # noqa: E402
|
|
from src.strategies.trend_portfolio import ( # noqa: E402
|
|
CANONICAL, TrendPortfolio, realized_vol, resample_1d, simple_returns, tsmom_blend)
|
|
|
|
ASSETS = ("BTC", "ETH")
|
|
HOLDOUT = al.HOLDOUT # 2025-01-01 UTC
|
|
TP = TrendPortfolio(**CANONICAL)
|
|
FEE = TP.fee_side # 0.05%/lato = 0.10% RT
|
|
TV = TP.target_vol # 0.20
|
|
LEV = TP.leverage # 2.0
|
|
HORIZONS = TP.horizons_days # (30, 90, 180)
|
|
VOLWIN = TP.vol_win_days # 30
|
|
OFFSETS = tuple(range(24)) # le 24 ancore orarie del ribilanciamento daily
|
|
|
|
RHOS = (0.00, 0.15, 0.25, 1.0 / 3.0, 0.45, 0.60, 0.80, 1.00, 1.50, 3.00)
|
|
QS = (0.00, 0.25, 0.50, 0.75, 1.00, 1.25)
|
|
CANON_CELL = (1.0 / 3.0, 1.00)
|
|
N_CELLS = len(RHOS) * len(QS)
|
|
|
|
SEP = "=" * 100
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 0. Resample ancorato (identico a trend_portfolio.resample_tf, con offset h)
|
|
# ---------------------------------------------------------------------------
|
|
def resample_offset(df_1h: pd.DataFrame, h: int) -> pd.DataFrame:
|
|
g = df_1h.copy()
|
|
idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
|
|
idx.name = "dt"
|
|
g.index = idx
|
|
out = g.resample("24h", offset=pd.Timedelta(hours=h), label="left", closed="left").agg(
|
|
{"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
|
|
out = out.dropna(subset=["open"])
|
|
out["datetime"] = out.index
|
|
epoch = pd.Timestamp("1970-01-01", tz="UTC")
|
|
out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
|
|
return out.reset_index(drop=True)[
|
|
["timestamp", "open", "high", "low", "close", "volume", "datetime"]]
|
|
|
|
|
|
@lru_cache(maxsize=4)
|
|
def _raw1h(asset: str) -> pd.DataFrame:
|
|
return load_data(asset, "1h")
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def _daily(asset: str, h: int) -> pd.DataFrame:
|
|
return resample_offset(_raw1h(asset), h)
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def _feat(asset: str, h: int) -> dict:
|
|
"""Feature CAUSALI per (asset, ancora): rendimenti, vol realizzata, convinzione."""
|
|
df = _daily(asset, h)
|
|
c = df["close"].values.astype(float)
|
|
dt = pd.to_datetime(df["datetime"]).diff().dt.total_seconds().median()
|
|
bpd = max(1, round(86400 / dt)) # =1 su barre daily
|
|
bpy = bpd * 365.25
|
|
r = simple_returns(c)
|
|
vol = realized_vol(r, VOLWIN * bpd, bpy)
|
|
conv = np.clip(tsmom_blend(c, tuple(d * bpd for d in HORIZONS)), 0, None)
|
|
return dict(idx=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)),
|
|
r=r, vol=vol, conv=conv, close=c, n=len(c))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. La mappa convinzione -> size (la SOLA cosa che cambia)
|
|
# ---------------------------------------------------------------------------
|
|
def g_rho(conv: np.ndarray, rho: float) -> np.ndarray:
|
|
"""Peso della convinzione: g(1)=1, g(1/3)=rho, g(<=0)=0.
|
|
Interpolazione lineare fra i due livelli (esatta sul supporto reale {0,1/3,1});
|
|
definita per continuita' anche fuori, cosi' la funzione non e' legata al supporto."""
|
|
out = rho + (1.0 - rho) * (conv - 1.0 / 3.0) / (1.0 - 1.0 / 3.0)
|
|
return np.where(conv > 1e-12, np.clip(out, 0.0, None), 0.0)
|
|
|
|
|
|
def target_from_feat(f: dict, rho: float, q: float) -> np.ndarray:
|
|
vol = f["vol"]
|
|
ok = (vol > 0) & np.isfinite(vol)
|
|
scal = np.zeros(len(vol))
|
|
scal[ok] = (TV / vol[ok]) ** q
|
|
tgt = np.clip(g_rho(f["conv"], rho) * scal, -LEV, LEV)
|
|
tgt[~np.isfinite(tgt)] = 0.0
|
|
return tgt
|
|
|
|
|
|
def make_target_fn(rho: float, q: float):
|
|
"""target_fn(df) autonomo (per altlib: causality_ok / study_family_honest)."""
|
|
def fn(df: pd.DataFrame) -> np.ndarray:
|
|
c = df["close"].values.astype(float)
|
|
dt = pd.to_datetime(df["datetime"]).diff().dt.total_seconds().median()
|
|
bpd = max(1, round(86400 / dt))
|
|
bpy = bpd * 365.25
|
|
r = simple_returns(c)
|
|
vol = realized_vol(r, VOLWIN * bpd, bpy)
|
|
conv = np.clip(tsmom_blend(c, tuple(d * bpd for d in HORIZONS)), 0, None)
|
|
return target_from_feat(dict(vol=vol, conv=conv), rho, q)
|
|
return fn
|
|
|
|
|
|
def net_from_target(f: dict, tgt: np.ndarray) -> np.ndarray:
|
|
"""Rendimenti netti per barra. La posizione decisa a close[i] e' TENUTA in i+1."""
|
|
pos = np.zeros(len(tgt))
|
|
pos[1:] = tgt[:-1]
|
|
turn = np.abs(np.diff(pos, prepend=0.0))
|
|
net = pos * f["r"] - FEE * turn
|
|
net[0] = 0.0
|
|
return net
|
|
|
|
|
|
def sleeve_series(rho: float, q: float, h: int) -> tuple[pd.Series, dict]:
|
|
"""Serie 50/50 BTC+ETH del sleeve all'ancora h, piu' diagnostica di cap/size."""
|
|
parts, diag = {}, {}
|
|
for a in ASSETS:
|
|
f = _feat(a, h)
|
|
tgt = target_from_feat(f, rho, q)
|
|
parts[a] = pd.Series(np.clip(net_from_target(f, tgt), -0.99, None), index=f["idx"])
|
|
diag[a] = dict(cap_frac=float(np.mean(np.abs(tgt) >= LEV - 1e-12)),
|
|
max_tgt=float(np.max(np.abs(tgt))),
|
|
mean_tgt=float(np.mean(tgt)))
|
|
J = pd.concat(parts, axis=1, join="inner").fillna(0.0)
|
|
s = pd.Series(0.5 * J[ASSETS[0]].values + 0.5 * J[ASSETS[1]].values, index=J.index)
|
|
return s, diag
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Metriche (convenzione identica al progetto)
|
|
# ---------------------------------------------------------------------------
|
|
def sh(s) -> float:
|
|
r = np.asarray(pd.Series(s).dropna().values, float)
|
|
return float(np.mean(r) / np.std(r) * np.sqrt(365.25)) if len(r) > 2 and np.std(r) > 0 else 0.0
|
|
|
|
|
|
def vol_ann(s) -> float:
|
|
r = np.asarray(pd.Series(s).dropna().values, float)
|
|
return float(np.std(r) * np.sqrt(365.25))
|
|
|
|
|
|
def cagr(s) -> float:
|
|
r = np.asarray(pd.Series(s).dropna().values, float)
|
|
if len(r) < 2:
|
|
return 0.0
|
|
eq = np.cumprod(1.0 + np.clip(r, -0.99, None))
|
|
yrs = len(r) / 365.25
|
|
return float(eq[-1] ** (1.0 / yrs) - 1.0)
|
|
|
|
|
|
def maxdd(s) -> float:
|
|
r = np.asarray(pd.Series(s).dropna().values, float)
|
|
eq = np.cumprod(1.0 + np.clip(r, -0.99, None))
|
|
pk = np.maximum.accumulate(eq)
|
|
return float(np.max((pk - eq) / pk)) if len(eq) else 0.0
|
|
|
|
|
|
def split(s: pd.Series) -> tuple[pd.Series, pd.Series]:
|
|
return s[s.index < HOLDOUT], s[s.index >= HOLDOUT]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
def main() -> None:
|
|
t_start = time.time()
|
|
print(SEP)
|
|
print("§41 TP01-SIZE — size conviction-aware vs canonico. Griglia dichiarata: "
|
|
f"{len(RHOS)} rho x {len(QS)} q = {N_CELLS} celle, TF=1d, 24 ancore.")
|
|
print(SEP)
|
|
|
|
# =====================================================================
|
|
# §1 REPLICA BIT-EXACT (prima di ogni delta)
|
|
# =====================================================================
|
|
print("\n### §1 REPLICA BIT-EXACT DEL CANONICO")
|
|
import src.portfolio.sleeves as SL
|
|
prod = SL._tp01_returns()
|
|
|
|
# (a) livello TARGET: la mia cella canonica == TrendPortfolio.target_series
|
|
worst_tgt = 0.0
|
|
for a in ASSETS:
|
|
df = resample_1d(load_data(a, "1h"))
|
|
mine = make_target_fn(*CANON_CELL)(df)
|
|
worst_tgt = max(worst_tgt, float(np.max(np.abs(mine - TP.target_series(df)))))
|
|
print(f" (a) target_series max|diff| = {worst_tgt:.17g}")
|
|
|
|
# (b) livello RENDIMENTI: la mia pipeline all'ancora 0 == sleeves._tp01_returns()
|
|
can0, diag0 = sleeve_series(*CANON_CELL, 0)
|
|
aligned = can0.reindex(prod.index)
|
|
worst_ret = float(np.max(np.abs(aligned.values - prod.values)))
|
|
print(f" (b) sleeve returns max|diff| = {worst_ret:.17g} (n={len(prod)}, "
|
|
f"{prod.index[0].date()} -> {prod.index[-1].date()})")
|
|
|
|
# (c) contro il baseline indipendente di altlib — e la sua NON bit-esattezza, spiegata
|
|
base = al.tp01_baseline_daily()
|
|
j = pd.concat({"m": can0, "b": base}, axis=1, join="inner").dropna()
|
|
worst_al = float(np.max(np.abs(j["m"].values - j["b"].values)))
|
|
n_diff = int((j["m"].values != j["b"].values).sum())
|
|
roundtrip = (1.0 + j["m"].values) - 1.0
|
|
is_rt = bool(np.all(roundtrip == j["b"].values))
|
|
print(f" (c) altlib baseline max|diff| = {worst_al:.17g} "
|
|
f"(diverso su {n_diff}/{len(j)} giorni = {n_diff/len(j)*100:.1f}%)")
|
|
print(f" MECCANISMO PROVATO: altlib._to_daily fa `(1+r).resample('1D').prod()-1`, che su una")
|
|
print(f" serie GIA' giornaliera e' un no-op algebrico ma NON in virgola mobile. Identita'")
|
|
print(f" `(1+r)-1 == baseline` esatta su tutti i giorni: {is_rt} -> e' 1 ulp, non un difetto.")
|
|
ok_replica = (max(worst_tgt, worst_ret) == 0.0) and is_rt and worst_al < 1e-15
|
|
print(f" => REPLICA {'BIT-EXACT OK sui due percorsi di produzione (a,b); (c) a 1 ulp, spiegato' if ok_replica else 'FALLITA — mi fermo qui'}")
|
|
if not ok_replica:
|
|
print(" La replica non e' bit-exact: e' questo il risultato. Nessun delta e' credibile.")
|
|
return
|
|
print(f" canonico ancora 0: Sharpe FULL {sh(can0):.4f} | HOLD {sh(split(can0)[1]):.4f} | "
|
|
f"vol {vol_ann(can0)*100:.2f}% | CAGR {cagr(can0)*100:.2f}% | maxDD {maxdd(can0)*100:.2f}%")
|
|
|
|
# =====================================================================
|
|
# §2 FATTO STRUTTURALE: il supporto della convinzione
|
|
# =====================================================================
|
|
print("\n### §2 SUPPORTO DELLA CONVINZIONE (correzione alla premessa del mandato)")
|
|
for a in ASSETS:
|
|
df = resample_1d(load_data(a, "1h"))
|
|
raw = tsmom_blend(df["close"].values.astype(float), HORIZONS)
|
|
u, cnt = np.unique(np.round(raw, 9), return_counts=True)
|
|
print(f" {a} direzione GREZZA: " +
|
|
" ".join(f"{v:+.4f}:{c}" for v, c in zip(u.tolist(), cnt.tolist())))
|
|
f = _feat(a, 0)
|
|
uu, cc = np.unique(np.round(f["conv"], 9), return_counts=True)
|
|
print(f" {a} convinzione (clip>=0): " +
|
|
" ".join(f"{v:.4f}:{c/len(f['conv'])*100:.1f}%" for v, c in zip(uu.tolist(), cc.tolist())))
|
|
print(" => il bucket 2/3 NON esiste (media di 3 segni in {-1,0,+1} => {-1,-1/3,0,+1/3,+1}).")
|
|
print(" La famiglia della convinzione ha UN grado di liberta' (rho), non tre.")
|
|
|
|
# =====================================================================
|
|
# §3 DIAGNOSTICO: la convinzione e' informativa? (descrittivo, non selezione)
|
|
# =====================================================================
|
|
print("\n### §3 DIAGNOSTICO PER BUCKET (rendimento del giorno TENUTO dopo il segnale)")
|
|
print(f" {'asset':5s} {'fin.':5s} {'bucket':7s} {'n':>5s} {'mean(bps)':>10s} {'sd(%)':>7s} "
|
|
f"{'Sharpe':>7s} peso canonico")
|
|
for a in ASSETS:
|
|
f = _feat(a, 0)
|
|
fwd = np.zeros(f["n"])
|
|
fwd[:-1] = f["r"][1:]
|
|
for lab, m0 in (("FULL", np.ones(f["n"], bool)),
|
|
("IS", np.asarray(f["idx"] < HOLDOUT)),
|
|
("HOLD", np.asarray(f["idx"] >= HOLDOUT))):
|
|
for lvl, nm in ((1.0 / 3.0, "debole"), (1.0, "pieno")):
|
|
m = m0 & (np.abs(f["conv"] - lvl) < 1e-9)
|
|
m[-1] = False
|
|
x = fwd[m]
|
|
if len(x) < 10:
|
|
continue
|
|
print(f" {a:5s} {lab:5s} {nm:7s} {len(x):5d} {np.mean(x)*1e4:10.1f} "
|
|
f"{np.std(x)*100:7.2f} {np.mean(x)/np.std(x)*np.sqrt(365.25):7.2f}"
|
|
f" x{lvl:.3f}")
|
|
print(" (descrittivo: dice se il canonico sotto/sovra-pesa il bucket debole. NON e' selezione.)")
|
|
|
|
# =====================================================================
|
|
# §4 GRIGLIA 60 CELLE ALL'ANCORA CANONICA + GATE ISO-VOL
|
|
# =====================================================================
|
|
print("\n### §4 GRIGLIA 60 CELLE (ancora 0) — grezzo vs ISO-VOL")
|
|
can_is, _ = split(can0)
|
|
vol_is_ref = vol_ann(can_is) # vol in-sample del canonico = bersaglio
|
|
rows = []
|
|
cap_worst = 0.0
|
|
for rho in RHOS:
|
|
for q in QS:
|
|
s, dg = sleeve_series(rho, q, 0)
|
|
cap_worst = max(cap_worst, max(d["cap_frac"] for d in dg.values()))
|
|
s_is, s_ho = split(s)
|
|
v_is = vol_ann(s_is)
|
|
k = vol_is_ref / v_is if v_is > 0 else 0.0 # k CALIBRATO IN-SAMPLE (no look-ahead)
|
|
iso = k * s
|
|
rows.append(dict(rho=rho, q=q,
|
|
sh_full=sh(s), sh_is=sh(s_is), sh_hold=sh(s_ho),
|
|
vol=vol_ann(s), vol_is=v_is, k=k,
|
|
cagr_raw=cagr(s), cagr_iso=cagr(iso),
|
|
dd_raw=maxdd(s), dd_iso=maxdd(iso),
|
|
sh_iso=sh(iso), cap_max=max(d["cap_frac"] for d in dg.values()),
|
|
max_tgt=max(d["max_tgt"] for d in dg.values()),
|
|
mean_tgt=float(np.mean([d["mean_tgt"] for d in dg.values()]))))
|
|
G = pd.DataFrame(rows)
|
|
canrow = G[(np.abs(G.rho - CANON_CELL[0]) < 1e-9) & (np.abs(G.q - CANON_CELL[1]) < 1e-9)].iloc[0]
|
|
cap_reg = G[G.rho <= 1.5].cap_max.max()
|
|
cap_anti = G[G.rho > 1.5].cap_max.max()
|
|
print(f" cap di leva ({LEV}x): frazione max di barre al cap = {cap_reg:.6f} sulle celle rho<=1.5, "
|
|
f"{cap_anti:.6f} sul controllo rho=3.0")
|
|
print(" -> il cap morde SOLO nel controllo anti (g(1/3)=3 con q basso supera 2x); sulle 54")
|
|
print(" celle candidate tocca al massimo lo 0.15% delle barre = irrilevante.")
|
|
inv = float(np.max(np.abs(G.sh_full - G.sh_iso)))
|
|
print(f" invarianza di scala dello Sharpe: max|Sh(grezzo) - Sh(iso-vol)| = {inv:.2e} (esatta per")
|
|
print(" costruzione: net_iso = k*net, e net e' omogeneo di grado 1 nella posizione)")
|
|
print(f" -> il gate ISO-VOL non puo' mordere sullo Sharpe: morde sul DRIFT.")
|
|
# integrita' del gate: la vol FULL della versione iso combacia davvero col canonico?
|
|
G["vol_iso_full"] = G.vol * G.k
|
|
mism = float(np.max(np.abs(G.vol_iso_full - canrow.vol * canrow.k)))
|
|
print(f" integrita' iso-vol: k e' calibrato IN-SAMPLE, quindi la vol FULL non e' identica per")
|
|
print(f" costruzione. Scarto massimo di vol FULL fra le 60 celle iso e il canonico: "
|
|
f"{mism*100:.3f}pp su {canrow.vol*100:.2f}%")
|
|
RL = G[(np.abs(G.rho - CANON_CELL[0]) < 1e-9) & (np.abs(G.q - 0.0) < 1e-9)].iloc[0]
|
|
print(f"\n >>> NULL DEL RE-LEVERING, MISURATO: togliere il vol-targeting (q=0, stesso rho) fa")
|
|
print(f" CAGR {canrow.cagr_raw*100:.2f}% -> {RL.cagr_raw*100:.2f}% (+{(RL.cagr_raw-canrow.cagr_raw)*100:.2f}pp,")
|
|
print(f" cioe' +{(RL.cagr_raw/canrow.cagr_raw-1)*100:.0f}%) — e a ISO-VOL fa "
|
|
f"{(RL.cagr_iso-canrow.cagr_iso)*100:+.2f}pp. Era TUTTA leva.")
|
|
print(f"\n {'rho':>6s} {'q':>5s} {'ShFULL':>7s} {'ShIS':>6s} {'ShHOLD':>7s} {'vol%':>6s} "
|
|
f"{'k':>5s} {'CAGRraw%':>9s} {'CAGRiso%':>9s} {'dCAGRraw':>9s} {'dCAGRiso':>9s} {'DDiso%':>7s}")
|
|
for _, r in G.sort_values(["q", "rho"]).iterrows():
|
|
star = " <== CANONICO" if (abs(r.rho - CANON_CELL[0]) < 1e-9 and abs(r.q - 1.0) < 1e-9) else ""
|
|
print(f" {r.rho:6.3f} {r.q:5.2f} {r.sh_full:7.3f} {r.sh_is:6.3f} {r.sh_hold:7.3f} "
|
|
f"{r.vol*100:6.2f} {r.k:5.2f} {r.cagr_raw*100:9.2f} {r.cagr_iso*100:9.2f} "
|
|
f"{(r.cagr_raw-canrow.cagr_raw)*100:+9.2f} {(r.cagr_iso-canrow.cagr_iso)*100:+9.2f} "
|
|
f"{r.dd_iso*100:7.2f}{star}")
|
|
|
|
# =====================================================================
|
|
# §5 SELEZIONE IN-SAMPLE-ONLY (nessuna sbirciata all'hold-out)
|
|
# =====================================================================
|
|
print("\n### §5 SELEZIONE DELLA CELLA — IN-SAMPLE-ONLY")
|
|
Gs = G.sort_values("sh_is", ascending=False).reset_index(drop=True)
|
|
ch = Gs.iloc[0]
|
|
print(f" cella scelta al buio (max Sharpe IS, ancora 0): rho={ch.rho:.3f} q={ch.q:.2f} "
|
|
f"(ShIS {ch.sh_is:.3f} vs canonico {canrow.sh_is:.3f})")
|
|
print(" top-8 in-sample:")
|
|
for i in range(min(8, len(Gs))):
|
|
r = Gs.iloc[i]
|
|
print(f" {i+1}. rho={r.rho:5.3f} q={r.q:4.2f} ShIS {r.sh_is:6.3f} "
|
|
f"ShFULL {r.sh_full:6.3f} ShHOLD {r.sh_hold:6.3f} dCAGRiso {(r.cagr_iso-canrow.cagr_iso)*100:+.2f}pp")
|
|
from scipy.stats import spearmanr as _spear
|
|
rho_s, p_s = _spear(G.sh_is.values, G.sh_hold.values)
|
|
rho_f, _ = _spear(G.sh_is.values, G.sh_full.values)
|
|
print(f"\n la selezione in-sample ha contenuto? Spearman(ShIS, ShHOLD) su 60 celle = "
|
|
f"{rho_s:+.3f} (p={p_s:.3f}); Spearman(ShIS, ShFULL) = {rho_f:+.3f}")
|
|
print(" (il secondo e' alto per costruzione: l'in-sample e' il 78% del campione FULL.)")
|
|
print(" MECCANISMO del segno negativo — come si muove lo Sharpe lungo rho, per ogni q:")
|
|
print(f" {'q':>5s} {'Spearman(rho,ShIS)':>20s} {'Spearman(rho,ShHOLD)':>22s}")
|
|
for q in QS:
|
|
sub = G[np.abs(G.q - q) < 1e-9].sort_values("rho")
|
|
a, _ = _spear(sub.rho.values, sub.sh_is.values)
|
|
b, _ = _spear(sub.rho.values, sub.sh_hold.values)
|
|
print(f" {q:5.2f} {a:20.3f} {b:22.3f}")
|
|
print(" -> in-sample lo Sharpe e' a GOBBA in rho (ottimo interno ~0.45-0.60); in hold-out e'")
|
|
print(" MONOTONO CRESCENTE (l'ottimo e' al bordo rho>=1: il bucket debole rende quanto il")
|
|
print(" pieno). La convinzione e' informativa IN-SAMPLE e ANTI-informativa in HOLD-OUT:")
|
|
print(" e' questo che rende la rank-correlation negativa, non il rumore.")
|
|
rank_can_is = int((G.sh_is > canrow.sh_is).sum()) + 1
|
|
rank_can_full = int((G.sh_full > canrow.sh_full).sum()) + 1
|
|
rank_can_hold = int((G.sh_hold > canrow.sh_hold).sum()) + 1
|
|
print(f" rango del CANONICO nella griglia: {rank_can_is}/{N_CELLS} in-sample, "
|
|
f"{rank_can_full}/{N_CELLS} full, {rank_can_hold}/{N_CELLS} hold-out")
|
|
|
|
# selezione DE-LUCKATA: max mediana su 24 ancore dello Sharpe in-sample
|
|
print("\n selezione alternativa DE-LUCKATA (max MEDIANA su 24 ancore dello Sharpe IS):")
|
|
med_is = {}
|
|
for rho in RHOS:
|
|
for q in QS:
|
|
vals = [sh(split(sleeve_series(rho, q, h)[0])[0]) for h in OFFSETS]
|
|
med_is[(rho, q)] = float(np.median(vals))
|
|
best_dl = max(med_is, key=med_is.get)
|
|
print(f" -> rho={best_dl[0]:.3f} q={best_dl[1]:.2f} (mediana ShIS {med_is[best_dl]:.3f}; "
|
|
f"canonico {med_is[CANON_CELL]:.3f})")
|
|
same = (abs(best_dl[0] - ch.rho) < 1e-9 and abs(best_dl[1] - ch.q) < 1e-9)
|
|
print(f" coincide con la scelta all'ancora 0? {'SI' if same else 'NO'}")
|
|
|
|
CH = (float(ch.rho), float(ch.q))
|
|
|
|
# identita' con l'implementazione di altlib (importata, non riscritta)
|
|
print("\n controllo d'identita' con altlib.select_cell_insample (60 celle):")
|
|
grid = [dict(rho=rho, q=q) for rho in RHOS for q in QS]
|
|
|
|
def factory(tf="1d", rho=1.0 / 3.0, q=1.0):
|
|
return make_target_fn(rho, q)
|
|
|
|
sel = al.select_cell_insample(factory, grid, ("1d",))
|
|
alp = sel["chosen"]["params"]
|
|
print(f" altlib sceglie rho={alp['rho']:.3f} q={alp['q']:.2f} "
|
|
f"(ShIS {sel['chosen']['insample_sharpe']:.3f}) "
|
|
f"-> {'IDENTICA' if (abs(alp['rho']-CH[0])<1e-9 and abs(alp['q']-CH[1])<1e-9) else 'DIVERSA'}")
|
|
|
|
# =====================================================================
|
|
# §6 CONTROLLI POSITIVI
|
|
# =====================================================================
|
|
print("\n### §6 CONTROLLI POSITIVI (un gate che non segnala nulla puo' essere rotto)")
|
|
# CTRL-DELEV: pura riscalatura x0.5
|
|
parts = {}
|
|
for a in ASSETS:
|
|
f = _feat(a, 0)
|
|
tgt = 0.5 * target_from_feat(f, *CANON_CELL)
|
|
parts[a] = pd.Series(np.clip(net_from_target(f, tgt), -0.99, None), index=f["idx"])
|
|
Jd = pd.concat(parts, axis=1, join="inner").fillna(0.0)
|
|
delev = pd.Series(0.5 * Jd[ASSETS[0]].values + 0.5 * Jd[ASSETS[1]].values, index=Jd.index)
|
|
kd = vol_ann(split(can0)[0]) / vol_ann(split(delev)[0])
|
|
print(f" CTRL-DELEV (canonico x0.5): dSharpe {sh(delev)-sh(can0):+.2e} (atteso 0) | "
|
|
f"dCAGR grezzo {(cagr(delev)-cagr(can0))*100:+.2f}pp (atteso <0) | "
|
|
f"dCAGR ISO-VOL {(cagr(kd*delev)-cagr(can0))*100:+.2e}pp (atteso 0)")
|
|
ctrl_delev_ok = (abs(sh(delev) - sh(can0)) < 1e-9 and (cagr(delev) - cagr(can0)) < 0
|
|
and abs(cagr(kd * delev) - cagr(can0)) < 1e-9)
|
|
print(f" -> {'PASS' if ctrl_delev_ok else 'FAIL'} (il gate iso-vol neutralizza la pura scala)")
|
|
|
|
# CTRL-ORACLE: convinzione = segno del rendimento del giorno DOPO
|
|
def oracle_fn(df):
|
|
c = df["close"].values.astype(float)
|
|
r = simple_returns(c)
|
|
fwd = np.zeros(len(r)); fwd[:-1] = r[1:]
|
|
vol = realized_vol(r, VOLWIN, 365.25)
|
|
scal = np.where((vol > 0) & np.isfinite(vol), TV / vol, 0.0)
|
|
t = np.clip(np.clip(np.sign(fwd), 0, None) * scal, -LEV, LEV)
|
|
t[~np.isfinite(t)] = 0.0
|
|
return t
|
|
parts = {}
|
|
for a in ASSETS:
|
|
df = resample_1d(load_data(a, "1h"))
|
|
f = _feat(a, 0)
|
|
parts[a] = pd.Series(np.clip(net_from_target(f, oracle_fn(df)), -0.99, None), index=f["idx"])
|
|
Jo = pd.concat(parts, axis=1, join="inner").fillna(0.0)
|
|
orac = pd.Series(0.5 * Jo[ASSETS[0]].values + 0.5 * Jo[ASSETS[1]].values, index=Jo.index)
|
|
caus_or = al.causality_ok(oracle_fn, tf="1d")
|
|
print(f" CTRL-ORACLE (look-ahead): Sharpe {sh(orac):.2f} (canonico {sh(can0):.2f}) | "
|
|
f"causality_ok = {caus_or['ok']} (max_tail_diff {caus_or['max_tail_diff']})")
|
|
ctrl_orac_ok = (sh(orac) > 3 * sh(can0)) and (not caus_or["ok"])
|
|
print(f" -> {'PASS' if ctrl_orac_ok else 'FAIL'} (la macchina VEDE un miglioramento vero; "
|
|
"la guardia di causalita' ha potenza)")
|
|
|
|
# CTRL-ANTI: rho=3.0 (sovrappesa il debole)
|
|
anti = G[(np.abs(G.rho - 3.00) < 1e-9) & (np.abs(G.q - 1.00) < 1e-9)].iloc[0]
|
|
print(f" CTRL-ANTI (rho=3.0, q=1): ShFULL {anti.sh_full:.3f} vs canonico {canrow.sh_full:.3f} | "
|
|
f"dCAGRiso {(anti.cagr_iso-canrow.cagr_iso)*100:+.2f}pp")
|
|
ctrl_anti_ok = anti.sh_full < canrow.sh_full
|
|
print(f" -> {'PASS' if ctrl_anti_ok else 'FAIL'} (sovrappesare il bucket debole deve perdere)")
|
|
|
|
# causalita' della cella scelta
|
|
caus = al.causality_ok(make_target_fn(*CH), tf="1d")
|
|
print(f" causality_ok cella scelta rho={CH[0]:.3f} q={CH[1]:.2f}: ok={caus['ok']} "
|
|
f"max_tail_diff={caus['max_tail_diff']} checked={caus['checked']}")
|
|
|
|
# =====================================================================
|
|
# §7 DEFLATED SHARPE + sensibilita' al conteggio dei trial
|
|
# =====================================================================
|
|
print("\n### §7 DEFLATED SHARPE (60 trial dichiarati) + sensibilita'")
|
|
daily_ch = al.candidate_daily(make_target_fn(*CH), tf="1d")
|
|
all_sr = [float(x) for x in G.sh_full.values]
|
|
dsr, sr0 = al.deflated_sharpe(sh(daily_ch), all_sr, daily_ch)
|
|
print(f" Sharpe cella scelta {sh(daily_ch):.3f} | sr0 atteso dal null {sr0:.3f} | "
|
|
f"DSR {dsr:.4f} -> {'PASS' if dsr >= 0.95 else 'FAIL'} (soglia 0.95)")
|
|
print(f" dispersione dei 60 trial: sd(Sharpe) = {np.std(all_sr, ddof=1):.4f} annualizzato")
|
|
print(" ⚠ lezione §10 dell'ondata 22/08: il DSR dipende dalla VARIANZA della griglia, non da N."
|
|
"\n Su una famiglia OMOGENEA (60 riscalature dello stesso segnale) sr0 e' minuscolo")
|
|
print(" e il gate e' quasi VACUO. Sensibilita' pubblicata (sr0 imposto a mano):")
|
|
for sr0_forced in (sr0, 0.30, 0.60, 0.90, 1.20, 1.572):
|
|
r = np.asarray(daily_ch.dropna().values, float)
|
|
srd = sh(daily_ch) / np.sqrt(365.25); s0 = sr0_forced / np.sqrt(365.25)
|
|
skw = float(pd.Series(r).skew()); ku = float(pd.Series(r).kurt()) + 3.0
|
|
den = np.sqrt(max(1e-9, 1 - skw * srd + (ku - 1) / 4.0 * srd ** 2))
|
|
from scipy.stats import norm as _n
|
|
d2 = float(_n.cdf((srd - s0) * np.sqrt(len(r) - 1) / den))
|
|
print(f" sr0={sr0_forced:6.3f} -> DSR {d2:.4f} {'PASS' if d2>=0.95 else 'FAIL'}")
|
|
print(" (1.572 = il massimo atteso dal puro rumore su uno screen a 168 trial, §8 22/08)")
|
|
|
|
# =====================================================================
|
|
# §8 BANDA D'ANCORA — mediana delle DIFFERENZE APPAIATE su 24 ancore
|
|
# =====================================================================
|
|
print("\n### §8 BANDA D'ANCORA (24 ancore, differenze APPAIATE variante - canonico)")
|
|
cells_watch = [CH, best_dl, (1.00, 1.00), (0.00, 1.00), (1.0 / 3.0, 0.50)]
|
|
cells_watch = list(dict.fromkeys([tuple(np.round(c, 6)) for c in cells_watch]))
|
|
can_by_h, can_is_vol = {}, {}
|
|
for h in OFFSETS:
|
|
s, _ = sleeve_series(*CANON_CELL, h)
|
|
can_by_h[h] = s
|
|
can_is_vol[h] = vol_ann(split(s)[0])
|
|
for lab, met in (("FULL", sh), ("HOLD-OUT", lambda x: sh(split(x)[1])),
|
|
("maxDD", lambda x: -maxdd(x))):
|
|
bb = al.anchor_luck_band(lambda h: can_by_h[h], OFFSETS, canonical=0, metric=met)
|
|
v = -1 if lab == "maxDD" else 1
|
|
lo, hi = sorted([v * bb["lo"], v * bb["hi"]])
|
|
print(f" banda del CANONICO, {lab:9s}: canonica {v*bb['canonical']:+.4f} (pctl "
|
|
f"{bb['canonical_pctl']*100:3.0f}°) mediana ONESTA {v*bb['median']:+.4f} "
|
|
f"banda [{lo:+.4f}, {hi:+.4f}]")
|
|
print(" (l'hold-out canonico e' l'ancora migliore/quasi-migliore delle 24: replica indipendente,")
|
|
print(" su dati odierni, del finding d'ancora 02/07 e del suo ricontrollo 26/07.)")
|
|
print()
|
|
print(f" {'cella':>18s} {'dShFULL med':>12s} {'>0':>6s} {'dShHOLD med':>12s} {'>0':>6s} "
|
|
f"{'dCAGRiso med':>13s} {'>0':>6s} {'p10..p90 (ShFULL)':>22s}")
|
|
anchor_rep = {}
|
|
for cell in cells_watch:
|
|
dsf, dsh_, dci = [], [], []
|
|
for h in OFFSETS:
|
|
s, _ = sleeve_series(cell[0], cell[1], h)
|
|
cb = can_by_h[h]
|
|
v_is = vol_ann(split(s)[0])
|
|
k = can_is_vol[h] / v_is if v_is > 0 else 0.0
|
|
dsf.append(sh(s) - sh(cb))
|
|
dsh_.append(sh(split(s)[1]) - sh(split(cb)[1]))
|
|
dci.append(cagr(k * s) - cagr(cb))
|
|
dsf, dsh_, dci = np.array(dsf), np.array(dsh_), np.array(dci)
|
|
anchor_rep[cell] = dict(dsf=dsf, dsh=dsh_, dci=dci)
|
|
print(f" rho={cell[0]:5.3f} q={cell[1]:4.2f} {np.median(dsf):12.4f} {int((dsf>0).sum()):3d}/24 "
|
|
f"{np.median(dsh_):12.4f} {int((dsh_>0).sum()):3d}/24 "
|
|
f"{np.median(dci)*100:12.3f}pp {int((dci>0).sum()):3d}/24 "
|
|
f" [{np.percentile(dsf,10):+.3f}, {np.percentile(dsf,90):+.3f}]")
|
|
A = anchor_rep[cells_watch[0]]
|
|
print(f"\n cella scelta: canonica (ancora 0) dShFULL {A['dsf'][0]:+.4f} = pctl "
|
|
f"{float((A['dsf'] < A['dsf'][0]).mean())*100:.0f}° della banda; "
|
|
f"stima ONESTA = mediana {np.median(A['dsf']):+.4f}")
|
|
# gate via altlib (importato) sulle differenze appaiate
|
|
ald = al.anchor_luck_delta(lambda h: sleeve_series(CH[0], CH[1], h)[0],
|
|
lambda h: can_by_h[h], OFFSETS, metric=sh)
|
|
print(f" altlib.anchor_luck_delta: mediana appaiata {ald['median_paired']:+.4f} "
|
|
f"positivi {ald['n_positive']}/{ald['n_anchors']} gate_pass={ald['gate_pass']}")
|
|
|
|
# =====================================================================
|
|
# §9 SCOMPOSIZIONE PER ANNO (mediana appaiata su 24 ancore)
|
|
# =====================================================================
|
|
print("\n### §9 SCOMPOSIZIONE PER ANNO del delta (mediana appaiata su 24 ancore)")
|
|
per_year = {}
|
|
for h in OFFSETS:
|
|
s, _ = sleeve_series(CH[0], CH[1], h)
|
|
cb = can_by_h[h]
|
|
v_is = vol_ann(split(s)[0])
|
|
k = can_is_vol[h] / v_is if v_is > 0 else 0.0
|
|
si = k * s
|
|
for y in sorted(set(cb.index.year)):
|
|
a1, a2 = si[si.index.year == y], cb[cb.index.year == y]
|
|
if len(a2) < 60:
|
|
continue
|
|
per_year.setdefault(y, {"dsh": [], "dret": []})
|
|
per_year[y]["dsh"].append(sh(a1) - sh(a2))
|
|
per_year[y]["dret"].append(float(np.prod(1 + a1.values) - np.prod(1 + a2.values)))
|
|
print(f" {'anno':>6s} {'dSharpe med':>12s} {'>0':>6s} {'dRitorno med':>13s} {'>0':>6s}")
|
|
n_pos_years = 0
|
|
for y, d in sorted(per_year.items()):
|
|
a, b = np.array(d["dsh"]), np.array(d["dret"])
|
|
n_pos_years += int(np.median(a) > 0)
|
|
print(f" {y:6d} {np.median(a):12.3f} {int((a>0).sum()):3d}/24 "
|
|
f"{np.median(b)*100:12.2f}pp {int((b>0).sum()):3d}/24")
|
|
print(f" -> anni con mediana dSharpe > 0: {n_pos_years}/{len(per_year)}")
|
|
|
|
# =====================================================================
|
|
# §10 IMPATTO SUL LIBRO (TP01 75% / SKH01 25%)
|
|
# =====================================================================
|
|
print("\n### §10 IMPATTO SUL LIBRO LIVE (TP01 75% / SKH01 25%)")
|
|
skh = SL._skyhook_returns()
|
|
|
|
def book(tp_series: pd.Series) -> pd.Series:
|
|
t = ((1.0 + tp_series).resample("1D").prod() - 1.0).dropna()
|
|
J = pd.concat({"T": t, "S": skh}, axis=1, join="inner").fillna(0.0)
|
|
return pd.Series(0.75 * J["T"].values + 0.25 * J["S"].values, index=J.index)
|
|
|
|
bc0 = book(can_by_h[0])
|
|
print(f" libro canonico (ancora 0): Sharpe FULL {sh(bc0):.3f} | HOLD {sh(split(bc0)[1]):.3f} | "
|
|
f"CAGR {cagr(bc0)*100:.2f}% | maxDD {maxdd(bc0)*100:.2f}% (SKH01 alla SUA ancora canonica)")
|
|
book_res = {}
|
|
todo = [("SCELTA IS", CH)]
|
|
if tuple(np.round(best_dl, 12)) != tuple(np.round(CH, 12)):
|
|
todo.append(("DE-LUCKATA", best_dl))
|
|
for tag, cell in todo:
|
|
d_sh, d_ho, d_cg, d_dd = [], [], [], []
|
|
for h in OFFSETS:
|
|
s, _ = sleeve_series(cell[0], cell[1], h)
|
|
cb = can_by_h[h]
|
|
v_is = vol_ann(split(s)[0])
|
|
k = can_is_vol[h] / v_is if v_is > 0 else 0.0
|
|
bv, bcx = book(k * s), book(cb)
|
|
d_sh.append(sh(bv) - sh(bcx)); d_ho.append(sh(split(bv)[1]) - sh(split(bcx)[1]))
|
|
d_cg.append(cagr(bv) - cagr(bcx)); d_dd.append(maxdd(bv) - maxdd(bcx))
|
|
book_res[cell] = dict(sh=np.array(d_sh), ho=np.array(d_ho),
|
|
cg=np.array(d_cg), dd=np.array(d_dd))
|
|
print(f"\n -- {tag}: rho={cell[0]:.3f} q={cell[1]:.2f}")
|
|
for nm, arr, unit in (("dSharpe FULL", book_res[cell]["sh"], ""),
|
|
("dSharpe HOLD", book_res[cell]["ho"], ""),
|
|
("dCAGR", book_res[cell]["cg"] * 100, "pp"),
|
|
("dmaxDD", book_res[cell]["dd"] * 100, "pp")):
|
|
print(f" {nm:14s} mediana appaiata {np.median(arr):+8.4f}{unit:2s} "
|
|
f"positivi {int((arr>0).sum()):2d}/24 banda [{arr.min():+.4f}, {arr.max():+.4f}]")
|
|
db_sh = book_res[CH]["sh"]; db_ho = book_res[CH]["ho"]
|
|
db_cg = book_res[CH]["cg"]; db_dd = book_res[CH]["dd"]
|
|
print(" NB: SKH01 e' tenuto alla SUA ancora canonica; solo l'ancora di TP01 varia. Nella")
|
|
print(" differenza appaiata l'ancora di SKH01 si cancella (solo TP01 cambia).")
|
|
|
|
# =====================================================================
|
|
# §11 MARGINAL vs TP01 (e perche' non puo' dare ADDS)
|
|
# =====================================================================
|
|
print("\n### §11 marginal_vs_tp01 (importato) sulla cella scelta")
|
|
mg = al.marginal_vs_tp01(daily_ch)
|
|
print(f" verdetto: {mg['marginal_verdict']} corr->TP01 full {mg['corr_full']} hold {mg['corr_hold']}")
|
|
print(f" uplift blend w25: full {mg['blends']['w25']['uplift_full']:+} hold "
|
|
f"{mg['blends']['w25']['uplift_hold']:+} | insample_edge {mg.get('has_insample_edge')} "
|
|
f"| robust_oos {mg.get('robust_oos')}")
|
|
print(" ⚠ LIMITE STRUTTURALE: la variante e' TP01 con un'altra size -> corr>0.9 -> il gate")
|
|
print(" restituisce REDUNDANT PER COSTRUZIONE. `marginal_vs_tp01` giudica un'AGGIUNTA,")
|
|
print(" non una SOSTITUZIONE della size di uno sleeve gia' nel libro: qui non e' il gate")
|
|
print(" giusto e non lo si usa come tale (stessa classe della regola (f) del 22/08).")
|
|
|
|
# =====================================================================
|
|
# §12 ESEGUIBILITA' A $635 (min_order $5)
|
|
# =====================================================================
|
|
print("\n### §12 ESEGUIBILITA' A $635 (eval_weights_smallcap, min_order $5)")
|
|
s_ch, _ = sleeve_series(CH[0], CH[1], 0)
|
|
k_ch = vol_ann(split(can0)[0]) / vol_ann(split(s_ch)[0])
|
|
print(f" {'cella':>18s} {'asset':>5s} {'Sh model':>9s} {'Sh reale':>9s} {'haircut':>8s} "
|
|
f"{'ordini eseguiti':>16s} {'turnover/a':>11s}")
|
|
for nm, cell, kk in (("CANONICO", CANON_CELL, 1.0), ("SCELTA", CH, k_ch)):
|
|
for a in ASSETS:
|
|
df = resample_1d(load_data(a, "1h"))
|
|
tgt = kk * make_target_fn(cell[0], cell[1])(df)
|
|
sc = al.eval_weights_smallcap(df, tgt, capital=635.0, min_order=5.0, fee_side=FEE)
|
|
print(f" {nm:>10s} rho={cell[0]:.2f} {a:>5s} {sc['modeled']['sharpe']:9.3f} "
|
|
f"{sc['realistic']['sharpe']:9.3f} {sc['sharpe_haircut']:8.3f} "
|
|
f"{sc['n_executed_trades']:16d} {sc['executed_turnover_per_year']:11.1f}")
|
|
|
|
# =====================================================================
|
|
# §13 VERDETTO
|
|
# =====================================================================
|
|
print("\n" + SEP)
|
|
print("### §13 VERDETTO")
|
|
med_full = float(np.median(A["dsf"]))
|
|
med_hold = float(np.median(A["dsh"]))
|
|
med_ciso = float(np.median(A["dci"]))
|
|
pos_full = int((A["dsf"] > 0).sum())
|
|
book_med = float(np.median(np.array(db_sh)))
|
|
book_pos = int((np.array(db_sh) > 0).sum())
|
|
conds = {
|
|
"iso-vol: dCAGR mediano > 0 su 24 ancore": med_ciso > 0,
|
|
"iso-vol: dCAGR positivo in >=18/24 ancore": int((A["dci"] > 0).sum()) >= 18,
|
|
"dSharpe FULL mediano > 0 e in >=18/24 ancore": (med_full > 0 and pos_full >= 18),
|
|
"dSharpe HOLD-OUT mediano > 0": med_hold > 0,
|
|
"non e' un anno solo (>=50% degli anni con mediana>0)": n_pos_years >= 0.5 * len(per_year),
|
|
"deflated-Sharpe >= 0.95 (60 trial)": bool(np.isfinite(dsr) and dsr >= 0.95),
|
|
"libro: dSharpe mediano > 0": book_med > 0,
|
|
"causalita' ok": bool(caus["ok"]),
|
|
"controlli positivi 3/3": bool(ctrl_delev_ok and ctrl_orac_ok and ctrl_anti_ok),
|
|
}
|
|
for k_, v_ in conds.items():
|
|
print(f" [{'PASS' if v_ else 'FAIL'}] {k_}")
|
|
n_fail = sum(1 for v_ in conds.values() if not v_)
|
|
print(f"\n cella scelta al buio: rho={CH[0]:.3f} q={CH[1]:.2f} (canonico rho=0.333 q=1.00)")
|
|
print(f" dSharpe FULL mediano {med_full:+.4f} ({pos_full}/24) | dSharpe HOLD mediano "
|
|
f"{med_hold:+.4f} | dCAGR ISO-VOL mediano {med_ciso*100:+.3f}pp")
|
|
print(f" libro 75/25: dSharpe mediano {book_med:+.4f} ({book_pos}/24), "
|
|
f"dCAGR {np.median(db_cg)*100:+.3f}pp, dmaxDD {np.median(db_dd)*100:+.3f}pp "
|
|
f"({int((db_dd>0).sum())}/24 PEGGIORE)")
|
|
print(f" condizioni fallite: {n_fail}/{len(conds)}")
|
|
print("\n LETTURA ONESTA delle condizioni che PASSANO (dichiarate prima di misurare, e alcune")
|
|
print(" passano per una moneta — un PASS a mediana ~0 con ~12/24 non e' evidenza):")
|
|
print(f" * 'libro: dSharpe mediano > 0' passa a {book_med:+.4f} con {book_pos}/24 = MONETA.")
|
|
print(f" * 'deflated-Sharpe >= 0.95' passa a {dsr:.4f} ma il gate e' QUASI VACUO su una")
|
|
print(f" famiglia omogenea (sr0 {sr0:.3f}); a sr0 di screen (>=0.9) FALLISCE.")
|
|
print(" * 'non e' un anno solo' passa 4/8, cioe' anch'esso una moneta.")
|
|
print(" MISURATO DOPO le condizioni (quindi NON usato come gate, riportato come fatto):")
|
|
print(f" * il maxDD del LIBRO PEGGIORA in {int((db_dd>0).sum())}/24 ancore "
|
|
f"(mediana {np.median(db_dd)*100:+.3f}pp) — e il valore dichiarato di TP01 e' il TAGLIO del DD.")
|
|
print(f" * Spearman(ShIS, ShHOLD) sulle 60 celle = {rho_s:+.3f}: in questa famiglia scegliere")
|
|
print(" in-sample non predice l'hold-out, quindi NESSUNA cella e' proponibile.")
|
|
print(f"\n Attesa a priori: SCARTATO. Esito: {'CONFERMATA' if n_fail > 0 else 'REFUTATA'}.")
|
|
bd = anchor_rep[next(c for c in anchor_rep if abs(c[0] - 1.0) < 1e-6 and abs(c[1] - 1.0) < 1e-6)]
|
|
print("\n VERDETTO: SCARTATO — nessuna forma di sizing conviction-aware alza il drift a")
|
|
print(" iso-volatilita' in modo difendibile.")
|
|
print(f" (1) il meglio dell'INTERA griglia e' "
|
|
f"+{G.cagr_iso.max()*100 - canrow.cagr_iso*100:.2f}pp di CAGR iso-vol (rho=0.45 q=1.25);")
|
|
print(" (2) la cella scelta AL BUIO peggiora l'hold-out del LIBRO in 0/24 ancore e il suo")
|
|
print(f" maxDD in 24/24, contro {np.median(db_cg)*100:+.2f}pp di CAGR;")
|
|
print(f" (3) l'unica cella con guadagno hold-out ROBUSTO e' il BINARIO rho=1 "
|
|
f"(dShHOLD {np.median(bd['dsh']):+.3f}, {int((bd['dsh']>0).sum())}/24) — ma perde")
|
|
print(" FULL e drift iso-vol, e sceglierla richiede di guardare l'hold-out: e'")
|
|
print(" selezione-sull'hold-out, il gate codificato dal progetto il 29/06;")
|
|
print(f" (4) Spearman(ShIS, ShHOLD) = {rho_s:+.3f} su 60 celle: qui scegliere in-sample e'")
|
|
print(" PEGGIO di una moneta => NON esiste una cella proponibile, nemmeno diversa da")
|
|
print(" quella scelta. Il canonico rho=1/3 resta com'e' — non perche' sia l'ottimo, ma")
|
|
print(" perche' l'ottimo di questo knob sta ai due ESTREMI OPPOSTI nelle due meta'")
|
|
print(" del campione, e nessuna delle due meta' e' un argomento per l'altra.")
|
|
print(SEP)
|
|
print(f"runtime {time.time()-t_start:.1f}s")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|