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

958 lines
52 KiB
Python
Raw Blame History

#!/usr/bin/env python
"""r0822b_bocpd.py — IL FALSIFICATORE NOMINATO E MAI ESEGUITO del filone ADAPTIVE-HORIZON.
CONTESTO (perche' questo file esiste)
-------------------------------------
Il 22/08 `r0822_adaptive_horizon.py` ha scartato il TSMOM a orizzonte adattivo con una diagnosi
netta: il "vincitore adattivo" aveva L_t **incollato al bordo il 100% del tempo** = un TSMOM a
orizzonte COSTANTE di 20 giorni travestito (corr 1.000, dSh 0.00 in 8/8 anni e 0/12 ancore);
isolando le 52 celle davvero adattive -> NEUTRAL, corr->TP01 0.90 = TP01 travestito.
Ma quell'agente ha dichiarato **cosa lo smentirebbe** e non ha potuto eseguirlo:
"Un rilevatore il cui L_t si muove DAVVERO e batte il miglior lookback COSTANTE (non TP01) in
>=6/8 anni appaiati con corr->TP01 < 0.6. La mia famiglia copre 3 rilevatori a soglia/finestra
fissa — un run-length posterior bayesiano (BOCPD/PELT vero) o un L scelto per walk-forward sul
PnL realizzato non sono stati provati."
Questo file esegue **quel** test. Oggi il filone e' chiuso su UNA famiglia (3 proxy a soglia
fissa), non sul MECCANISMO. Una seconda chiusura, con rilevatori veri, rende il "no" definitivo.
PERCHE' BOCPD E' STRUTTURALMENTE DIVERSO DAI 3 PROXY DEL PRIMO TENTATIVO
-----------------------------------------------------------------------
I 3 proxy (CUSUM a soglia k, variance-ratio, Hurst) producevano una STATISTICA che veniva poi
MAPPATA linearmente dentro [Lmin, Lmax]: la mappatura e' arbitraria, e infatti la statistica
saturava contro il clip (bordo 100%). Un run-length posterior NON ha bisogno di mappatura:
la lunghezza di run **e' gia' un lookback**, nelle stesse unita' (barre). "Quanto indietro posso
guardare" = "da quanto tempo non cambia il regime" e' un'identita', non una scelta di scala.
* BOCPD (Adams & MacKay 2007): posterior online su r_t con hazard costante e verosimiglianza
Normale-InverseGamma coniugata (predittiva Student-t). Causale PER COSTRUZIONE: e' un filtro
in avanti, ogni P(r_t | x_{1:t}) usa solo x_{1:t}.
* OP/PELT causale: PELT e' un'accelerazione (pruning) dell'**optimal partitioning**
F(t) = min_{s<t} [F(s) + C(y_{s+1:t}) + beta]. La ricorsione E' online. ⚠️ La maggior parte
delle implementazioni di change-point (ruptures.Pelt().fit_predict(intera_serie)) e'
**retrospettiva** e restituisce la segmentazione dell'INTERO campione = look-ahead colossale.
Qui F(t) e la sua backtracking-partition sono calcolati SOLO su y[0:t] per ogni t.
Verificato con `A.causality_ok` **e** con un troncamento manuale esplicito (sezione 0-bis).
* WF-PnL: L scelto per walk-forward sullo Sharpe NETTO realizzato dei candidati costanti nella
finestra all'indietro. Non usa alcun modello di regime: e' il "cosa avrebbe funzionato finora".
IL NULL GIUSTO NON E' TP01 — E' IL MIGLIOR LOOKBACK COSTANTE
------------------------------------------------------------
E' il controllo che ha ucciso il primo tentativo, e va rimesso PRIMA, non dopo: senza di esso
il null confonde "adattare paga" con "un orizzonte piu' corto paga su questo campione".
Percio' il confronto principale e': adattivo (cella scelta in-sample) vs miglior L COSTANTE
(cella scelta in-sample, stessa geometria, stesso TF, stesse fee), **appaiato per ANNO e per
ANCORA** (mediana delle DIFFERENZE APPAIATE, mai differenza delle mediane — lezione 26/07).
FAMIGLIA DICHIARATA **PRIMA** DI GUARDARE (conteggio al RIALZO)
--------------------------------------------------------------
A) rilevatori veri 2 (bocpd | op)
parametro 3 (bocpd: hazard lambda 50/100/250 g ; op: penalita' k 2/5/10)
statistica 2 (bocpd: E[r_t] | argmax r_t ; op: eta' dall'ultimo cp | mediana
delle lunghezze di segmento gia' concluse)
mappatura 2 (direct | inverse) [il verso non e' ovvio a priori => e' un asse]
blend 2 (single L | triple L/3,L,2L — con L=90 il triple E' TP01)
timeframe 2 (1d | 12h) [>=12h: sotto, costi+overfit dominano, 19/06]
= 2*3*2*2*2*2 = **96 trial**
B) controllo L COSTANTE 12 L x 2 blend x 2 TF = **48 trial**
C) WF-PnL (L da walk-forward) 3 finestre x 2 blend x 2 TF = **12 trial**
D) taratura della SCALA 8 penalita' OP + 5 hazard BOCPD = **13** (sezione 1-pre)
---------------------------------------------------------------- TOTALE DICHIARATO **169**
⚠️ (D) e' una scansione fatta guardando SOLO il numero di change-point rilevati — una proprieta'
del RILEVATORE — e mai uno Sharpe: nessuna di quelle 13 configurazioni produce un rendimento. Si
dichiara lo stesso, al rialzo, perche' ha ristretto la griglia di (A). Il suo esito e' gia' un
risultato: a penalita' OP >= 5 l'eta' resta incollata al tetto della finestra il 44% del tempo,
cioe' **l'artefatto del 22/08 riprodotto su un rilevatore vero** — e la griglia (A) usa solo la
regione in cui il rilevatore rileva davvero (0.5/1.0/2.0 = 18-40 change-point l'anno).
Costanti FISSATE A PRIORI (gradi di liberta' non cercati, dichiarati perche' esistono):
Lmin=20g, Lmax=200g (contengono 30..180 del canonico e sono gli stessi del primo tentativo);
BOCPD su rendimenti log STANDARDIZZATI da vol rolling causale 30g (il bersaglio e' il DRIFT, non
la vol); prior NIG mu0=0, kappa0=1, alpha0=1, beta0=1; troncamento del posterior a Lmax barre;
costo OP = mean-shift gaussiano su dati standardizzati, penalita' beta = k*log(t) causale.
Il deflated-Sharpe e' riportato a N=169 (mia famiglia), N=281 (mia + le 112 del primo tentativo:
**stessa domanda**, i trial si sommano) e N=449 (di SCREEN — lezione 22/08 del filone ORTHO:
su BTC/ETH direzionale il massimo atteso dal rumore su 168 trial e' Sharpe 1.572, SOPRA il
soffitto misurato ~1.3, quindi il conteggio di screen e' quello che decide).
ORDINE DI ESECUZIONE (imposto dalla missione, ed e' l'ordine giusto)
--------------------------------------------------------------------
La diagnostica che ha smascherato il primo tentativo va PER PRIMA: quanto si muove davvero L_t?
Se sta al bordo, l'artefatto e' riprodotto e il filone si chiude di nuovo senza guardare i
rendimenti. Solo se L_t si muove ha senso misurare se il movimento PAGA.
Runtime atteso ~6-10 min su 2 core. Nessun file toccato fuori da questo.
"""
from __future__ import annotations
import sys
import time
from pathlib import Path
import numpy as np
import pandas as pd
from scipy.special import gammaln
from scipy.stats import norm
ROOT = Path("/opt/docker/PythagorasGoal")
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
sys.path.insert(0, str(ROOT))
import altlib as A # noqa: E402
from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio # noqa: E402
ASSETS = A.CERTIFIED
HOLDOUT = A.HOLDOUT
# ---- costanti FISSATE A PRIORI (dichiarate sopra, non cercate) -------------------------
LMIN_D, LMAX_D = 20, 200 # banda dell'orizzonte, IDENTICA al primo tentativo
FIXED_H = (30, 90, 180) # == (L/3, L, 2L) con L=90 -> TP01 canonico
TGT_VOL, LEV, VOLWIN = 0.20, 2.0, 30
NIG_MU0, NIG_KAP0, NIG_AL0, NIG_BE0 = 0.0, 1.0, 1.0, 1.0
STD_WIN_D = 30 # finestra della standardizzazione causale dei rendimenti
TFS = ("1d", "12h")
DETECTORS = ("bocpd", "op")
PARAMS = {"bocpd": (50, 100, 250), "op": (0.5, 1.0, 2.0)}
# TARATURA DELLA SCALA (sezione 1-pre): scansione ESPLORATIVA fatta guardando SOLO il numero di
# change-point rilevati — una proprieta' del RILEVATORE — e mai lo Sharpe. Un rilevatore che non
# rileva nulla non e' un rilevatore: a pen_k>=5 l'eta' resta incollata al tetto della finestra
# (44% del tempo) = ESATTAMENTE l'artefatto del 22/08, riprodotto qui su un rilevatore vero.
CAL_PEN = (0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0)
CAL_LAM = (25, 50, 100, 250, 500)
STATS = {"bocpd": ("mean", "map"), "op": ("age", "segmed")}
MAPPINGS = ("direct", "inverse")
BLENDS = ("single", "triple")
CONST_L = (5, 7, 10, 15, 20, 30, 45, 60, 90, 120, 150, 200)
WF_WIN_D = (180, 365, 730)
N_DECLARED_A = len(DETECTORS) * 3 * 2 * len(MAPPINGS) * len(BLENDS) * len(TFS) # 96
N_DECLARED_B = len(CONST_L) * len(BLENDS) * len(TFS) # 48
N_DECLARED_C = len(WF_WIN_D) * len(BLENDS) * len(TFS) # 12
N_DECLARED_CAL = len(CAL_PEN) + len(CAL_LAM) # 13 (taratura)
N_DECLARED = N_DECLARED_A + N_DECLARED_B + N_DECLARED_C + N_DECLARED_CAL # 169
N_PRIOR_FAMILY = 112 # trial dichiarati da r0822_adaptive_horizon (stessa domanda)
N_SCREEN = 168 # pool di screen direzionale BTC/ETH dell'ondata (filone 12)
# =======================================================================================
# RILEVATORI VERI
# =======================================================================================
def _logret(c: np.ndarray) -> np.ndarray:
r = np.zeros(len(c))
r[1:] = np.log(c[1:] / c[:-1])
return r
def _std_returns(c: np.ndarray, bpd: int) -> np.ndarray:
"""Rendimenti log standardizzati da vol rolling CAUSALE (min_periods -> warmup a 0).
Il bersaglio del rilevatore e' il cambio di DRIFT: senza standardizzare, un cambio di
regime di VOLATILITA' dominerebbe la verosimiglianza e il rilevatore misurerebbe la vol."""
lr = _logret(c)
w = max(5, STD_WIN_D * bpd)
sig = pd.Series(lr).rolling(w, min_periods=max(5, w // 4)).std().values
y = np.where((sig > 0) & np.isfinite(sig), lr / np.where(sig > 0, sig, 1.0), 0.0)
return np.nan_to_num(np.clip(y, -8.0, 8.0))
def _student_logpdf(x: float, mu: np.ndarray, kap: np.ndarray,
al: np.ndarray, be: np.ndarray) -> np.ndarray:
"""Predittiva Student-t della Normale-InverseGamma coniugata (Murphy 2007, eq. 99-100):
x | mu,kappa,alpha,beta ~ t_{2 alpha}(mu, beta(kappa+1)/(alpha kappa))."""
nu = 2.0 * al
s2 = be * (kap + 1.0) / (al * kap)
z = (x - mu) ** 2 / (nu * s2)
return (gammaln(0.5 * (nu + 1.0)) - gammaln(0.5 * nu)
- 0.5 * np.log(nu * np.pi * s2) - 0.5 * (nu + 1.0) * np.log1p(z))
def _bocpd(y: np.ndarray, lam_bars: float, rmax: int) -> tuple[np.ndarray, np.ndarray]:
"""BOCPD (Adams & MacKay 2007) con hazard COSTANTE H = 1/lam_bars e verosimiglianza NIG.
Ritorna (E[r_t], argmax_r P(r_t)) in BARRE. Filtro in avanti: ogni valore usa solo il
passato -> causale per costruzione, e stabile su prefisso (nessuna statistica full-sample).
Il posterior e' troncato a rmax barre e rinormalizzato (il troncamento e' un limite di
memoria, NON un clip sul valore: se la massa si accumula a rmax lo si vede in diagnostica)."""
n = len(y)
haz = 1.0 / max(2.0, float(lam_bars))
R = np.array([1.0])
mu = np.array([NIG_MU0]); kap = np.array([NIG_KAP0])
al = np.array([NIG_AL0]); be = np.array([NIG_BE0])
rl = np.arange(1)
mean_rl = np.zeros(n); map_rl = np.zeros(n)
for t in range(n):
x = float(y[t])
lp = _student_logpdf(x, mu, kap, al, be)
lp -= lp.max()
pred = np.exp(lp)
gr = R * pred * (1.0 - haz)
cp = float(np.sum(R * pred * haz))
m_new = min(len(R) + 1, rmax + 1)
newR = np.empty(m_new)
newR[0] = cp
newR[1:] = gr[:m_new - 1]
s = newR.sum()
if not np.isfinite(s) or s <= 0:
newR = np.zeros(m_new); newR[0] = 1.0; s = 1.0
newR /= s
# aggiornamento delle statistiche sufficienti (run r -> run r+1), prior in testa
mu_f = np.concatenate(([NIG_MU0], (kap * mu + x) / (kap + 1.0)))[:m_new]
kap_f = np.concatenate(([NIG_KAP0], kap + 1.0))[:m_new]
al_f = np.concatenate(([NIG_AL0], al + 0.5))[:m_new]
be_f = np.concatenate(([NIG_BE0], be + kap * (x - mu) ** 2 / (2.0 * (kap + 1.0))))[:m_new]
R, mu, kap, al, be = newR, mu_f, kap_f, al_f, be_f
if len(rl) != m_new:
rl = np.arange(m_new, dtype=float)
mean_rl[t] = float(np.dot(R, rl))
map_rl[t] = float(rl[int(np.argmax(R))])
return mean_rl, map_rl
def _op_causal(y: np.ndarray, pen_k: float, smax: int) -> tuple[np.ndarray, np.ndarray]:
"""OPTIMAL PARTITIONING **online** (l'obiettivo esatto che PELT accelera con il pruning):
F(t) = min_{s in [t-smax, t-1]} [ F(s) + C(y[s:t]) + beta(t) ]
con C = costo gaussiano di mean-shift su dati standardizzati e beta(t) = pen_k*log(t).
⚠️ Il termine sum(y^2) del costo gaussiano e' INVARIANTE alla partizione (ogni punto sta in
esattamente un segmento) -> si omette dall'argmin. C(s:t) = -(S[t]-S[s])^2/(t-s).
Ritorna (eta' dall'ultimo change-point, mediana delle lunghezze dei segmenti CONCLUSI),
entrambe in barre. Per ogni t si usa SOLO y[0:t]: e' la differenza fra questo e un
`ruptures.Pelt().fit_predict(y)` retrospettivo, che guarda tutto il campione."""
n = len(y)
S = np.concatenate(([0.0], np.cumsum(y)))
F = np.full(n + 1, np.inf)
F[0] = 0.0
prev = np.zeros(n + 1, dtype=int)
age = np.zeros(n); segmed = np.zeros(n)
for t in range(1, n + 1):
a = max(0, t - smax)
s_idx = np.arange(a, t)
seg = (S[t] - S[a:t])
ln = (t - s_idx).astype(float)
cand = F[a:t] - seg * seg / ln
j = int(np.argmin(cand))
beta = pen_k * np.log(max(2.0, float(t)))
F[t] = cand[j] + beta
prev[t] = int(s_idx[j])
# backtracking della partizione VALIDA A t (cap a 40 segmenti: ci serve la mediana)
last = prev[t]
age[t - 1] = t - last
lens = []
u = last
while u > 0 and len(lens) < 40:
p = prev[u]
lens.append(u - p)
u = p
segmed[t - 1] = float(np.median(lens)) if lens else float(t)
return age, segmed
_STAT_CACHE: dict = {}
def _stat_key(df: pd.DataFrame, det: str, par) -> tuple:
"""Chiave chiusa sul CONTENUTO del frame (lunghezza + primo/ultimo timestamp): un PREFISSO
non deve prendere la cache del campione pieno, altrimenti `causality_ok` verrebbe ingannata
dalla cache invece che dal codice."""
ts = df["timestamp"].values
return (det, par, len(df), int(ts[0]), int(ts[-1]))
def detector_stats(df: pd.DataFrame, det: str, par) -> dict:
key = _stat_key(df, det, par)
if key in _STAT_CACHE:
return _STAT_CACHE[key]
bpd = A.bars_per_day(df)
c = df["close"].values.astype(float)
y = _std_returns(c, bpd)
rmax = max(4, LMAX_D * bpd)
if det == "bocpd":
m, mp = _bocpd(y, float(par) * bpd, rmax)
out = {"mean": m, "map": mp}
elif det == "op":
ag, sm = _op_causal(y, float(par), rmax)
out = {"age": ag, "segmed": sm}
else:
raise ValueError(det)
_STAT_CACHE[key] = out
return out
def lookback_bars(df: pd.DataFrame, det: str, par, stat: str, mapping: str) -> np.ndarray:
"""L_t in BARRE dentro [Lmin, Lmax] giorni. `direct` = la lunghezza di run E' il lookback
(dopo un change-point si puo' guardare indietro solo fin la'); `inverse` e' lo specchio,
dichiarato come asse perche' il verso non e' ovvio a priori."""
bpd = A.bars_per_day(df)
lo, hi = max(2, LMIN_D * bpd), max(3, LMAX_D * bpd)
v = detector_stats(df, det, par)[stat]
base = np.clip(np.nan_to_num(v, nan=float(lo)), lo, hi)
L = base if mapping == "direct" else (lo + hi - base)
return np.clip(np.round(L), 2, hi).astype(int)
# =======================================================================================
# DIREZIONE TSMOM (orizzonte costante o variabile) + vol-target — IDENTICA al primo tentativo
# =======================================================================================
def _sign_at(c: np.ndarray, h: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
n = len(c)
i = np.arange(n)
j = i - h
ok = j >= 0
s = np.zeros(n)
jj = np.where(ok, j, 0)
s[ok] = np.sign(c[i[ok]] / c[jj[ok]] - 1.0)
return s, ok
def _dir_from_L(c: np.ndarray, L: np.ndarray, blend: str) -> np.ndarray:
hs = [L] if blend == "single" else [
np.maximum(2, np.round(L / 3.0).astype(int)), L, np.round(2.0 * L).astype(int)]
acc = np.zeros(len(c)); cnt = np.zeros(len(c))
for h in hs:
s, ok = _sign_at(c, np.asarray(h, int))
acc[ok] += s[ok]; cnt[ok] += 1
out = np.zeros(len(c)); nz = cnt > 0
out[nz] = acc[nz] / cnt[nz]
return out
def fixed_target(df: pd.DataFrame) -> np.ndarray:
"""BASELINE: TSMOM fisso 30/90/180, long-flat, vol-target 20%, leva 2x = TP01 canonico."""
c = df["close"].values.astype(float)
bpd = A.bars_per_day(df)
acc = np.zeros(len(c)); cnt = np.zeros(len(c))
for hd in FIXED_H:
s, ok = _sign_at(c, np.full(len(c), hd * bpd, int))
acc[ok] += s[ok]; cnt[ok] += 1
d = np.zeros(len(c)); nz = cnt > 0
d[nz] = acc[nz] / cnt[nz]
return A.vol_target(np.clip(d, 0, None), df, TGT_VOL, VOLWIN, LEV)
def const_factory(tf: str, L: int, blend: str):
def fn(df: pd.DataFrame) -> np.ndarray:
c = df["close"].values.astype(float)
bpd = A.bars_per_day(df)
Lb = np.full(len(c), max(2, int(round(L * bpd))), int)
return A.vol_target(np.clip(_dir_from_L(c, Lb, blend), 0, None), df, TGT_VOL, VOLWIN, LEV)
fn.__name__ = f"const_L{L}_{blend}"
return fn
def factory(tf: str, det: str, par, stat: str, mapping: str, blend: str):
def fn(df: pd.DataFrame) -> np.ndarray:
c = df["close"].values.astype(float)
L = lookback_bars(df, det, par, stat, mapping)
return A.vol_target(np.clip(_dir_from_L(c, L, blend), 0, None), df, TGT_VOL, VOLWIN, LEV)
fn.__name__ = f"adapt_{det}{par}_{stat}_{mapping}_{blend}"
return fn
_WF_CACHE: dict = {}
def wf_picks(df: pd.DataFrame, win_d: int, blend: str):
"""(C) L scelto per WALK-FORWARD sul PnL NETTO realizzato: a ogni barra si prende l'L che,
fra i candidati COSTANTI, ha lo Sharpe netto migliore nella finestra all'indietro `win_d`.
Nessun modello di regime: e' il "cosa avrebbe funzionato finora".
Causale: il PnL della barra i si conosce solo a fine barra i, quindi per decidere a i si usa
la finestra che finisce a i-1; `eval_weights` shifta poi la posizione di un'altra barra."""
cands = [L for L in CONST_L if LMIN_D <= L <= LMAX_D]
key = (win_d, blend) + _stat_key(df, "wf", 0)[2:]
if key in _WF_CACHE:
return _WF_CACHE[key]
c = df["close"].values.astype(float)
bpd = A.bars_per_day(df)
n = len(c)
r = A.simple_returns(c)
w = max(20, win_d * bpd)
tgts, sh = [], []
for L in cands:
Lb = np.full(n, max(2, int(round(L * bpd))), int)
t_ = A.vol_target(np.clip(_dir_from_L(c, Lb, blend), 0, None), df, TGT_VOL, VOLWIN, LEV)
pos = np.zeros(n); pos[1:] = t_[:-1]
net = pos * r - A.FEE_SIDE * np.abs(np.diff(pos, prepend=0.0))
ss = pd.Series(net)
mu = ss.rolling(w, min_periods=w // 2).mean()
sd = ss.rolling(w, min_periods=w // 2).std()
with np.errstate(invalid="ignore", divide="ignore"):
sr = (mu / sd).values
sr = np.concatenate(([np.nan], sr[:-1])) # PnL noto solo a fine barra
tgts.append(t_); sh.append(np.where(np.isfinite(sr), sr, -np.inf))
SH = np.vstack(sh); TG = np.vstack(tgts)
idx = np.argmax(SH, axis=0)
idx[~np.isfinite(SH).any(axis=0)] = cands.index(90) # warmup -> il canonico, non il migliore
out = (np.asarray([cands[i] for i in idx]), TG[idx, np.arange(n)], cands)
_WF_CACHE[key] = out
return out
def wf_factory(tf: str, win_d: int, blend: str):
def fn(df: pd.DataFrame) -> np.ndarray:
return wf_picks(df, win_d, blend)[1]
fn.__name__ = f"wfpnl_{win_d}_{blend}"
return fn
GRID_A = [dict(det=d, par=p, stat=s, mapping=m, blend=b)
for d in DETECTORS for p in PARAMS[d] for s in STATS[d]
for m in MAPPINGS for b in BLENDS]
GRID_B = [dict(L=L, blend=b) for L in CONST_L for b in BLENDS]
GRID_C = [dict(win_d=w, blend=b) for w in WF_WIN_D for b in BLENDS]
# =======================================================================================
# UTILITA'
# =======================================================================================
def _dd(s: pd.Series) -> float:
eq = np.cumprod(1.0 + np.asarray(s.dropna().values, float))
pk = np.maximum.accumulate(eq)
return float(np.max((pk - eq) / pk)) if len(eq) else 0.0
def _cagr(s: pd.Series) -> float:
s = s.dropna()
if len(s) < 5:
return float("nan")
eq = float(np.prod(1.0 + s.values))
yrs = (s.index[-1] - s.index[0]).total_seconds() / (86400 * 365.25)
return eq ** (1 / max(yrs, 1e-6)) - 1.0 if eq > 0 else -1.0
def resample_offset(df_1h: pd.DataFrame, hours: int, off: int) -> pd.DataFrame:
"""Stessa convenzione di trend_portfolio.resample_tf, ancora spostata di `off` ore.
NB `offset=`, non `origin=` (pandas 2.x ignora origin in silenzio — lezione 02/07)."""
g = df_1h.copy()
idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
idx.name = "dt"
g.index = idx
out = g.resample(f"{hours}h", offset=pd.Timedelta(hours=off), 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"]]
def paired_by_year(cand: pd.Series, base: pd.Series) -> list[dict]:
J = pd.concat({"C": cand, "B": base}, axis=1, join="inner").dropna()
rows = []
for y, g in J.groupby(J.index.year):
if len(g) < 40:
continue
rows.append(dict(year=int(y), n=len(g),
sh_cand=round(A._sh(g["C"]), 2), sh_base=round(A._sh(g["B"]), 2),
d_sh=round(A._sh(g["C"]) - A._sh(g["B"]), 2)))
return rows
def delever_null(cand: pd.Series, base: pd.Series) -> dict:
"""Esiste k<1 che da' al BASELINE lo stesso maxDD del candidato con Sharpe >= ?
Se si', il candidato non ha comprato protezione: ha solo tolto leva."""
J = pd.concat({"C": cand, "B": base}, axis=1, join="inner").dropna()
dd_c, dd_b = _dd(J["C"]), _dd(J["B"])
if dd_c >= dd_b or dd_b <= 0:
return dict(applicabile=False, dd_cand=round(dd_c, 4), dd_base=round(dd_b, 4),
nota="il candidato NON riduce il maxDD: il null non si applica")
lo, hi = 0.0, 1.0
for _ in range(60):
k = 0.5 * (lo + hi)
if _dd(k * J["B"]) < dd_c:
lo = k
else:
hi = k
k = 0.5 * (lo + hi)
return dict(applicabile=True, k=round(k, 4), dd_cand=round(dd_c, 4), dd_base=round(dd_b, 4),
sharpe_cand=round(A._sh(J["C"]), 3),
sharpe_base_delev=round(A._sh(k * J["B"]), 3),
superato=bool(A._sh(J["C"]) > A._sh(k * J["B"])))
_ANCHOR_CACHE: dict = {}
def series_at_anchor(fn, name: str, hours: int, off: int) -> pd.Series:
key = (name, hours, off)
if key not in _ANCHOR_CACHE:
ser = {}
for a in ASSETS:
dfo = resample_offset(A.get(a, "1h"), hours, off)
ev = A.eval_weights(dfo, A._call_target(fn, dfo, a))
ser[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
_ANCHOR_CACHE[key] = A._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
return _ANCHOR_CACHE[key]
def adaptivity(tf: str, det: str, par, stat: str, mapping: str) -> dict:
"""LA DIAGNOSTICA CHE HA SMASCHERATO IL PRIMO TENTATIVO — quanto si MUOVE davvero L_t.
Criterio (identico a quello del 22/08, per confrontabilita'): sta a un bordo <50% del tempo
E l'IQR e' >= 10 giorni. In piu': |dL|/giorno e numero di RESET (cadute di L > 5 giorni),
che sono la firma di un vero rilevatore di change-point."""
fl, fh, iq, md, dl, nrs, sat = [], [], [], [], [], [], []
for a in ASSETS:
df = A.get(a, tf)
bpd = A.bars_per_day(df)
L = lookback_bars(df, det, par, stat, mapping) / bpd
raw = np.nan_to_num(detector_stats(df, det, par)[stat], nan=0.0) / bpd
fl.append(float(np.mean(L <= LMIN_D + 1)))
fh.append(float(np.mean(L >= LMAX_D - 1)))
iq.append(float(np.percentile(L, 75) - np.percentile(L, 25)))
md.append(float(np.median(L)))
dl.append(float(np.median(np.abs(np.diff(L)))))
d = np.diff(L)
nrs.append(float(np.sum(d < -5.0) / (len(L) / (365.25)))) # reset all'anno
sat.append(float(np.mean(raw >= LMAX_D - 1))) # saturazione del GREZZO
return dict(frac_lo=float(np.mean(fl)), frac_hi=float(np.mean(fh)),
iqr_days=float(np.mean(iq)), med_days=float(np.mean(md)),
dL_med=float(np.mean(dl)), reset_yr=float(np.mean(nrs)),
sat_raw=float(np.mean(sat)),
adattiva=bool(np.mean(fl) < 0.5 and np.mean(fh) < 0.5 and np.mean(iq) >= 10.0))
def hr(t: str = "") -> None:
print("\n" + "=" * 86)
if t:
print(t)
print("=" * 86)
def dsr_given_sr0(daily: pd.Series, sr_ann: float, sr0_ann: float,
dpy: float = 365.25) -> float:
"""Deflated-Sharpe con il massimo-atteso-dal-rumore IMPOSTO dall'esterno.
Serve perche' `A.deflated_sharpe` stima sr0 dalla VARIANZA degli Sharpe della griglia: su
una famiglia OMOGENEA (qui sono tutte varianti di TSMOM, quindi molto correlate) quella
varianza e' piccola, sr0 esce basso e il DSR esce alto **per costruzione**. Il conteggio che
decide, sulla ricerca direzionale BTC/ETH, e' quello di SCREEN: il filone ORTHO ha misurato
che su 168 trial il massimo atteso dal puro rumore e' Sharpe 1.572."""
r = np.asarray(pd.Series(daily).dropna().values, float)
T = len(r)
sr, sr0 = sr_ann / np.sqrt(dpy), sr0_ann / np.sqrt(dpy)
sk = float(pd.Series(r).skew())
ku = float(pd.Series(r).kurt()) + 3.0
den = np.sqrt(max(1e-9, 1 - sk * sr + (ku - 1) / 4.0 * sr ** 2))
return float(norm.cdf((sr - sr0) * np.sqrt(T - 1) / den))
def dsr_line(sr: float, allsr: list, daily: pd.Series, label: str, n_decl: int) -> str:
pad = list(allsr) + [float(np.mean(allsr))] * max(0, n_decl - len(allsr))
d, s0 = A.deflated_sharpe(sr, pad, daily)
return (f" N={n_decl:>4} ({label:<24}) DSR={d:.3f} {'PASS' if d >= 0.95 else 'FAIL'}"
f" max atteso dal rumore {s0:+.3f}")
# =======================================================================================
def main() -> None:
t0 = time.time()
print(__doc__.split("Runtime atteso")[0])
# -- 0. SANITY -----------------------------------------------------------------------
hr("0. SANITY — il baseline 'fisso' e' TP01 canonico (se non lo e', ogni delta e' inventato)")
tp = TrendPortfolio(**CANONICAL)
worst = 0.0
for a in ASSETS:
df = A.get(a, "1d")
worst = max(worst, float(np.max(np.abs(fixed_target(df) - tp.target_series(df)))))
base_1d = A.candidate_daily(fixed_target, tf="1d")
tp01 = A.tp01_baseline_daily()
JJ = pd.concat({"a": base_1d, "b": tp01}, axis=1, join="inner").dropna()
print(f" max|target_mio - TrendPortfolio.target_series| (1d, BTC+ETH) = {worst:.3e}")
print(f" max|serie_giornaliera_mia - A.tp01_baseline_daily()| = "
f"{float(np.max(np.abs(JJ['a'] - JJ['b']))):.3e} (n={len(JJ)})")
assert worst < 1e-12, "il baseline NON riproduce TP01: fermarsi qui"
# -- 0-bis. IL RILEVATORE E' CAUSALE? (troncamento MANUALE, prima dei gate) -----------
hr("0-bis. TRONCAMENTO MANUALE — la trappola del change-point RETROSPETTIVO")
print(" Un `Pelt().fit_predict(serie_intera)` restituisce la segmentazione dell'INTERO")
print(" campione: al tempo t 'saprebbe' dei cambi futuri. Qui BOCPD e OP sono ricorsioni in")
print(" avanti. Prova diretta: ricalcolo su un PREFISSO e confronto la coda con il pieno.")
for det, par in (("bocpd", 100), ("op", 5.0)):
for stat in STATS[det]:
wmax = 0.0
for a in ASSETS:
df = A.get(a, "1d")
full = lookback_bars(df, det, par, stat, "direct")
for frac in (0.70, 0.85):
cut = int(len(df) * frac)
sub = df.iloc[:cut].reset_index(drop=True)
pre = lookback_bars(sub, det, par, stat, "direct")
wmax = max(wmax, float(np.max(np.abs(pre[-100:] - full[cut - 100:cut]))))
print(f" {det:<6} {str(par):>5} {stat:<7}: max|L(prefisso)-L(pieno)| sulla coda "
f"= {wmax:.3e} barre -> {'CAUSALE' if wmax < 1e-9 else 'SOSPETTO'}")
# -- 1-pre. TARATURA DELLA SCALA — guardando il RILEVATORE, mai lo Sharpe -----------
hr("1-pre. TARATURA DELLA SCALA DEI PARAMETRI (guardando SOLO quanti change-point rileva)")
print(" Un rilevatore che non rileva nulla non e' un rilevatore. Questa scansione guarda il")
print(" NUMERO di change-point e la distribuzione di L_t — mai un rendimento. Dichiarata")
print(" come 13 trial in piu' (al rialzo), benche' nessuno produca uno Sharpe.")
dfc = A.get("BTC", "1d")
yc = _std_returns(dfc["close"].values.astype(float), 1)
print(f"\n OP (BTC 1d) {'pen_k':>6} {'cp/anno':>8} {'age med':>8} {'IQR':>7} {'@tetto':>8}")
for k in CAL_PEN:
ag, _ = _op_causal(yc, k, LMAX_D)
ncp = int(np.sum(np.diff(ag) < 0))
flag = " <= USATO" if k in PARAMS["op"] else (" <= degenere (artefatto 22/08)"
if np.mean(ag >= LMAX_D - 1) > 0.30 else "")
print(f" {k:>6} {ncp/(len(yc)/365.25):>8.1f} {np.median(ag):>8.1f} "
f"{np.percentile(ag,75)-np.percentile(ag,25):>7.1f} "
f"{np.mean(ag>=LMAX_D-1)*100:>7.1f}%{flag}")
print(f"\n BOCPD (BTC 1d) {'lambda':>6} {'E[r] med':>9} {'IQR':>7} {'@tetto':>8} "
f"{'MAP med':>8} {'@tetto':>8}")
for lam in CAL_LAM:
mn, mp = _bocpd(yc, float(lam), LMAX_D)
flag = " <= USATO" if lam in PARAMS["bocpd"] else ""
print(f" {lam:>6} {np.median(mn):>9.1f} "
f"{np.percentile(mn,75)-np.percentile(mn,25):>7.1f} "
f"{np.mean(mn>=LMAX_D-1)*100:>7.1f}% {np.median(mp):>8.1f} "
f"{np.mean(mp>=LMAX_D-1)*100:>7.1f}%{flag}")
# -- 1. LA DIAGNOSTICA DI MECCANISMO, PER PRIMA --------------------------------------
hr("1. L_t SI MUOVE DAVVERO? (se sta al bordo l'artefatto e' riprodotto: si chiude e basta)")
print(" criterio (identico al 22/08, per confrontabilita'): a un bordo <50% del tempo E")
print(" IQR >= 10 giorni. `reset/anno` = cadute di L > 5 giorni = firma di un change-point.")
print(f"\n {'TF':>4} {'det':>6} {'par':>5} {'stat':>7} {'map':<7} {'bordo-lo':>9} "
f"{'bordo-hi':>9} {'IQR(g)':>7} {'med(g)':>7} {'|dL|(g)':>8} {'reset/a':>8} "
f"{'sat.grezzo':>10} esito")
diag: dict = {}
for tf in TFS:
for d in DETECTORS:
for p in PARAMS[d]:
for st in STATS[d]:
for mp in MAPPINGS:
v = adaptivity(tf, d, p, st, mp)
diag[(tf, d, p, st, mp)] = v
print(f" {tf:>4} {d:>6} {str(p):>5} {st:>7} {mp:<7} "
f"{v['frac_lo']*100:8.1f}% {v['frac_hi']*100:8.1f}% "
f"{v['iqr_days']:7.1f} {v['med_days']:7.1f} {v['dL_med']:8.2f} "
f"{v['reset_yr']:8.1f} {v['sat_raw']*100:9.1f}% "
f"{'ADATTIVA' if v['adattiva'] else 'costante-travestito'}")
n_ad = sum(1 for v in diag.values() if v["adattiva"])
print(f"\n configurazioni di rilevatore DAVVERO adattive: {n_ad}/{len(diag)}")
if n_ad == 0:
hr("STOP — ARTEFATTO RIPRODOTTO")
print(" Nessun rilevatore muove L_t: ogni 'vincitore' sarebbe un orizzonte COSTANTE")
print(" travestito, esattamente come il 22/08. Il filone si chiude per la seconda volta")
print(" senza bisogno di guardare i rendimenti. FINE.")
return
# -- 2. GRIGLIA (A): cella scelta IN-SAMPLE + deflated-Sharpe -------------------------
hr(f"2. study_family_honest (A) — {len(GRID_A)} celle x {len(TFS)} TF = "
f"{len(GRID_A)*len(TFS)} trial")
rep = A.study_family_honest("BOCPD-H", factory, GRID_A, TFS)
ch = rep["chosen"]
print(f" celle valutate : {rep['n_cells']}")
print(f" cella scelta IN-SAMPLE : tf={ch['tf']} {ch['params']}")
print(f" Sharpe IS {ch['insample_sharpe']} FULL {ch['full_sharpe']}")
print(f" deflated-Sharpe (N={rep['n_cells']}): {rep['deflated_sharpe']} "
f"(max atteso sotto il null {rep['expected_null_max']}) PASS={rep['dsr_pass']}")
print(f" earns_slot_marginal : {rep['earns_slot_marginal']}")
print(f" EARNS_SLOT_HONEST : {rep['earns_slot_honest']}")
print("\n top-10 IN-SAMPLE (come si sceglie onestamente):")
for r in rep["rows"][:10]:
ad = diag[(r["tf"], r["params"]["det"], r["params"]["par"],
r["params"]["stat"], r["params"]["mapping"])]["adattiva"]
print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} "
f"{r['params']} {'[adattiva]' if ad else '[costante-travestito]'}")
fn_ch = factory(tf=ch["tf"], **ch["params"])
d_ch = A.candidate_daily(fn_ch, tf=ch["tf"])
# -- 3. CONTROLLO (B): il MIGLIOR ORIZZONTE COSTANTE ---------------------------------
hr(f"3. CONTROLLO L COSTANTE (B) — {len(GRID_B)} celle x {len(TFS)} TF = "
f"{len(GRID_B)*len(TFS)} trial [IL NULL GIUSTO, non TP01]")
csel = A.select_cell_insample(const_factory, GRID_B, TFS)
cch = csel["chosen"]
print(f" miglior COSTANTE scelto in-sample: tf={cch['tf']} {cch['params']} "
f"IS {cch['insample_sharpe']} FULL {cch['full_sharpe']}")
print(" top-8 costanti in-sample (L=90 triple 1d == TP01 canonico):")
for r in csel["rows"][:8]:
tag = " <== TP01" if (r["params"]["L"] == 90 and r["params"]["blend"] == "triple"
and r["tf"] == "1d") else ""
print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} "
f"L={r['params']['L']:>3}g {r['params']['blend']}{tag}")
fn_const = const_factory(tf=cch["tf"], **cch["params"])
d_const = A.candidate_daily(fn_const, tf=cch["tf"])
# -- 4. WF-PnL (C) --------------------------------------------------------------------
hr(f"4. WALK-FORWARD SUL PnL REALIZZATO (C) — {len(GRID_C)} celle x {len(TFS)} TF = "
f"{len(GRID_C)*len(TFS)} trial")
wsel = A.select_cell_insample(wf_factory, GRID_C, TFS)
wch = wsel["chosen"]
print(f" miglior WF scelto in-sample: tf={wch['tf']} {wch['params']} "
f"IS {wch['insample_sharpe']} FULL {wch['full_sharpe']}")
for r in wsel["rows"]:
print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} {r['tf']:>3} "
f"{r['params']}")
fn_wf = wf_factory(tf=wch["tf"], **wch["params"])
d_wf = A.candidate_daily(fn_wf, tf=wch["tf"])
# quanto si muove L nella variante WF? (stesso metro della sezione 1)
for a in ASSETS:
df = A.get(a, wch["tf"]); bpd = A.bars_per_day(df)
pick, _, cands = wf_picks(df, **wch["params"])
print(f" {a}: L_WF giorni mediana {np.median(pick):.0f} IQR "
f"{np.percentile(pick,75)-np.percentile(pick,25):.0f} "
f"bordo basso {np.mean(pick<=min(cands))*100:.1f}% "
f"bordo alto {np.mean(pick>=max(cands))*100:.1f}% "
f"cambi/anno {np.sum(np.diff(pick)!=0)/(len(pick)/(bpd*365.25)):.0f}")
# -- 5. IL CONFRONTO CHE DECIDE ------------------------------------------------------
hr("5. ADATTIVO vs MIGLIOR COSTANTE (entrambi scelti IN-SAMPLE) — e vs TP01")
base_ch = A.candidate_daily(fixed_target, tf=ch["tf"])
for nome, ser in (("BOCPD/OP (A)", d_ch), ("WF-PnL (C)", d_wf),
("miglior COSTANTE (B)", d_const), ("TP01 30/90/180", base_1d)):
h = ser[ser.index >= HOLDOUT]
print(f" {nome:<22} FULL Sh {A._sh(ser):+.3f} HOLD Sh {A._sh(h):+.3f} "
f"maxDD {_dd(ser)*100:5.2f}% CAGR {_cagr(ser)*100:6.2f}% "
f"corr->TP01 {pd.concat({'a':ser,'b':tp01},axis=1,join='inner').dropna().corr().iloc[0,1]:+.3f}")
for lab, cand, ref, refname in (("A vs B", d_ch, d_const, "miglior costante"),
("C vs B", d_wf, d_const, "miglior costante"),
("A vs TP01", d_ch, base_1d, "TP01")):
J = pd.concat({"C": cand, "B": ref}, axis=1, join="inner").dropna()
JH = J[J.index >= HOLDOUT]
rows = paired_by_year(cand, ref)
ds = [r["d_sh"] for r in rows]
print(f"\n --- {lab} ({refname}) --- corr {J['C'].corr(J['B']):+.3f}")
print(f" dSh FULL {A._sh(J['C'])-A._sh(J['B']):+.3f} "
f"dSh HOLD {A._sh(JH['C'])-A._sh(JH['B']):+.3f}")
print(" differenza APPAIATA per anno:")
print(" " + " ".join(f"{r['year']}:{r['d_sh']:+.2f}" for r in rows))
print(f" -> anni con dSh>0: {sum(1 for x in ds if x > 0)}/{len(ds)} "
f"mediana {np.median(ds):+.2f} media {np.mean(ds):+.2f} "
f"[criterio del falsificatore: >=6/8 anni]")
# decomposizione: quanto del vantaggio su TP01 e' ADATTIVITA' e quanto e' "L piu' corto"?
JT = pd.concat({"C": d_ch, "K": d_const, "B": base_1d}, axis=1, join="inner").dropna()
v_tp = A._sh(JT["C"]) - A._sh(JT["B"])
v_ct = A._sh(JT["K"]) - A._sh(JT["B"])
print(f"\n DECOMPOSIZIONE del vantaggio dell'adattivo su TP01 (FULL, finestra comune):")
print(f" adattivo - TP01 = {v_tp:+.3f}")
print(f" miglior COSTANTE - TP01 = {v_ct:+.3f} <- quota spiegata da 'un L piu' corto'")
print(f" adattivo - miglior COSTANTE= {v_tp - v_ct:+.3f} <- quota spiegata dall'ADATTARE")
print(f" => l'adattivita' spiega il {100*(v_tp-v_ct)/v_tp if abs(v_tp)>1e-9 else float('nan'):+.0f}% "
f"del vantaggio su TP01")
# la superficie di L costante e' una CURVA o un CRINALE DI RUMORE? (replica del 22/08)
rows12 = {r["params"]["L"]: r["insample_sharpe"] for r in csel["rows"]
if r["tf"] == cch["tf"] and r["params"]["blend"] == cch["params"]["blend"]}
Ls = sorted(rows12)
adj = float(np.median(np.abs(np.diff([rows12[L] for L in Ls]))))
print(f"\n RISOLUZIONE della griglia costante ({cch['tf']}, {cch['params']['blend']}): salto IS "
f"mediano fra L ADIACENTI {adj:+.3f}")
print(f" contro il vantaggio del miglior costante su TP01 ({v_ct:+.3f}) -> rapporto "
f"{adj/max(1e-9, abs(v_ct)):.2f} [replica del 'crinale di rumore' del 22/08]")
# -- 6. GATE --------------------------------------------------------------------------
hr("6. GATE sul candidato (A) scelto in-sample")
print(A.fmt_marginal(rep["marginal"]))
cz = A.causality_ok(fn_ch, tf=ch["tf"])
print(f"\n causality_ok : ok={cz['ok']} max_tail_diff={cz['max_tail_diff']} "
f"(prefissi controllati {cz['checked']})")
imp = A.implausible_sharpe(d_ch)
print(f" implausible_sharpe : implausible={imp['implausible']} Sh {imp.get('sharpe')} "
f"maxDD {imp.get('maxdd')} perdite/attive {imp.get('loss_frac')} "
f"attive {imp.get('active_frac')}")
for rr in imp["reasons"]:
print(f" - {rr}")
print(f" null de-levering vs COSTANTE : {delever_null(d_ch, d_const)}")
print(f" null de-levering vs TP01 : {delever_null(d_ch, base_1d)}")
print(" haircut a $600 (min-order $5):")
for a in ASSETS:
df = A.get(a, ch["tf"])
sc = A.eval_weights_smallcap(df, A._call_target(fn_ch, df, a), capital=600.0)
print(f" {a}: modellato {sc['modeled']['sharpe']:+.3f} -> reale "
f"{sc['realistic']['sharpe']:+.3f} (haircut {sc['sharpe_haircut']:+.3f}, "
f"{sc['n_executed_trades']} trade)")
# DSR a piu' conteggi: famiglia, famiglia+112 del primo tentativo, di SCREEN
all_sr = ([r["full_sharpe"] for r in rep["rows"]]
+ [r["full_sharpe"] for r in csel["rows"]]
+ [r["full_sharpe"] for r in wsel["rows"]])
hr("6-bis. DEFLATED-SHARPE — sensibilita' del verdetto al CONTEGGIO (regola 30/07 + 22/08)")
print(f" Sharpe FULL del candidato (A) = {A._sh(d_ch):+.3f}")
print(dsr_line(A._sh(d_ch), all_sr, d_ch, "mia famiglia A+B+C", N_DECLARED))
print(dsr_line(A._sh(d_ch), all_sr, d_ch, "+ le 112 del 22/08", N_DECLARED + N_PRIOR_FAMILY))
print(dsr_line(A._sh(d_ch), all_sr, d_ch, "di SCREEN (ondata)",
N_DECLARED + N_PRIOR_FAMILY + N_SCREEN))
print("\n ⚠️ QUEL 'PASS' NON CERTIFICA NIENTE, e va detto: `deflated_sharpe` stima il massimo")
print(" atteso dal rumore dalla VARIANZA degli Sharpe della griglia. Qui la griglia e'")
print(" OMOGENEA (tutte varianti di TSMOM, molto correlate fra loro) -> varianza piccola")
print(" -> sr0 basso -> DSR alto per COSTRUZIONE. Aumentare N padding con la media alza N")
print(" ma non la varianza. Il conteggio che decide e' quello di SCREEN, e il filone ORTHO")
print(" lo ha gia' misurato: su 168 trial di ricerca direzionale BTC/ETH il massimo atteso")
print(" dal puro rumore e' Sharpe 1.572. Riscritto con QUEL sr0:")
for nome, ser in (("adattivo (A)", d_ch), ("miglior costante (B)", d_const),
("TP01 30/90/180", base_1d)):
print(f" {nome:<22} Sharpe FULL {A._sh(ser):+.3f} vs sr0_screen 1.572 -> "
f"DSR {dsr_given_sr0(ser, A._sh(ser), 1.572):.3f} "
f"{'PASS' if dsr_given_sr0(ser, A._sh(ser), 1.572) >= 0.95 else 'FAIL'}")
# PROVA DI IDENTITA': la cella "adattiva" al 100% sul bordo E' il costante, bit per bit
hr("6-ter. PROVA DI IDENTITA' — la cella al bordo NON somiglia a un costante: LO E'")
deg = [r for r in rep["rows"]
if diag[(r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["stat"],
r["params"]["mapping"])]["frac_lo"] >= 0.999]
if deg:
dr = max(deg, key=lambda r: r["insample_sharpe"])
s_deg = A.candidate_daily(factory(tf=dr["tf"], **dr["params"]), tf=dr["tf"])
s_con = A.candidate_daily(const_factory(tf=dr["tf"], L=LMIN_D,
blend=dr["params"]["blend"]), tf=dr["tf"])
Jd = pd.concat({"a": s_deg, "b": s_con}, axis=1, join="inner").dropna()
print(f" cella 100% al bordo basso: {dr['tf']} {dr['params']} IS {dr['insample_sharpe']}"
f" FULL {dr['full_sharpe']}")
print(f" costante L={LMIN_D}g {dr['params']['blend']} {dr['tf']}: "
f"FULL {A._sh(s_con):+.3f}")
print(f" max|serie_adattiva - serie_costante| = {float(np.max(np.abs(Jd['a']-Jd['b']))):.3e}"
f" (n={len(Jd)}) -> {'SONO LA STESSA STRATEGIA' if float(np.max(np.abs(Jd['a']-Jd['b']))) < 1e-12 else 'diverse'}")
else:
print(" nessuna cella e' al 100% sul bordo basso (la prova non si applica)")
# -- 7. FAMIGLIA RISTRETTA ALLE SOLE CELLE DAVVERO ADATTIVE --------------------------
hr("7. GATE SULLA SOLA FAMIGLIA DAVVERO ADATTIVA (la domanda della missione, isolata)")
ad_rows = [r for r in rep["rows"]
if diag[(r["tf"], r["params"]["det"], r["params"]["par"],
r["params"]["stat"], r["params"]["mapping"])]["adattiva"]]
d_ad_keep = fn_ad_keep = best_keep = None
if not ad_rows:
print(" NESSUNA cella della famiglia e' davvero adattiva -> vedi sezione 1.")
else:
best = max(ad_rows, key=lambda r: r["insample_sharpe"])
print(f" celle adattive: {len(ad_rows)}/{len(rep['rows'])} miglior IN-SAMPLE: "
f"tf={best['tf']} {best['params']} IS {best['insample_sharpe']} "
f"FULL {best['full_sharpe']}")
fn_ad = factory(tf=best["tf"], **best["params"])
d_ad = A.candidate_daily(fn_ad, tf=best["tf"])
sm = A.study_marginal("BOCPD-H (sole adattive)", fn_ad, tf=best["tf"])
dsr_ad, sr0_ad = A.deflated_sharpe(A._sh(d_ad),
[r["full_sharpe"] for r in ad_rows], d_ad)
print(f" marginal={sm['marginal_verdict']} earns_slot={sm['earns_slot']} "
f"corr->TP01 {sm['marginal'].get('corr_full')}")
print(f" deflated-Sharpe (N={len(ad_rows)}) = {dsr_ad:.3f} "
f"{'PASS' if dsr_ad >= 0.95 else 'FAIL'} (null max atteso {sr0_ad:+.3f})")
JA = pd.concat({"C": d_ad, "K": d_const, "B": base_1d}, axis=1, join="inner").dropna()
JH = JA[JA.index >= HOLDOUT]
print(f" FULL adattiva-vera {A._sh(JA['C']):+.3f} | costante {A._sh(JA['K']):+.3f} "
f"| TP01 {A._sh(JA['B']):+.3f}")
print(f" HOLD-OUT adattiva-vera {A._sh(JH['C']):+.3f} | costante {A._sh(JH['K']):+.3f} "
f"| TP01 {A._sh(JH['B']):+.3f}")
r3 = paired_by_year(d_ad, d_const)
d3 = [r["d_sh"] for r in r3]
print(" vs miglior costante, differenza APPAIATA per anno:")
print(" " + " ".join(f"{r['year']}:{r['d_sh']:+.2f}" for r in r3))
print(f" -> dSh>0 in {sum(1 for x in d3 if x>0)}/{len(d3)} mediana {np.median(d3):+.2f}"
f" corr al costante {JA['C'].corr(JA['K']):+.3f}")
print(" ⚠️ l'hold-out sono gli ULTIMI DUE anni di questa stessa riga: se il vantaggio")
print(" hold-out sta li' e non nel resto, e' un evento, non un meccanismo (regola:")
print(" un vantaggio concentrato in un anno non e' un vantaggio).")
print(f" null de-levering vs costante: {delever_null(d_ad, d_const)}")
d_ad_keep, fn_ad_keep, best_keep = d_ad, fn_ad, best
# -- 8. BANDA D'ANCORA (delta APPAIATO adattivo - costante) --------------------------
hours = 24 if ch["tf"] == "1d" else 12
offs = list(range(hours))
hr(f"8. BANDA D'ANCORA — {len(offs)} ancore. Stima onesta = MEDIANA; il confronto fra due "
f"varianti\n e' la MEDIANA DELLE DIFFERENZE APPAIATE (mai la differenza delle mediane).")
if cch["tf"] != ch["tf"]:
print(f" ⚠️ adattivo su {ch['tf']} e miglior costante su {cch['tf']}: per il delta "
f"appaiato uso il miglior costante RIVALUTATO su {ch['tf']} (stesse ancore = coppie).")
crow = max([r for r in csel["rows"] if r["tf"] == ch["tf"]],
key=lambda r: r["insample_sharpe"])
fn_const_anch = const_factory(tf=ch["tf"], **crow["params"])
print(f" miglior costante su {ch['tf']}: {crow['params']} IS {crow['insample_sharpe']}")
else:
fn_const_anch = fn_const
s0 = series_at_anchor(fixed_target, "fixed", hours, 0)
K = pd.concat({"a": s0, "b": base_ch}, axis=1, join="inner").dropna()
print(f" sanity ancora 0 vs serie canonica del fisso: max|Δ| "
f"{float(np.max(np.abs(K['a']-K['b']))):.3e} (n={len(K)})")
for lab, mtr in (("Sharpe FULL", A._sh),
("Sharpe HOLD", lambda s: A._sh(s[s.index >= HOLDOUT]))):
b_c = A.anchor_luck_band(lambda o: series_at_anchor(fn_ch, "cand", hours, o), offs,
metric=mtr)
b_k = A.anchor_luck_band(lambda o: series_at_anchor(fn_const_anch, "const", hours, o),
offs, metric=mtr)
print(f" {lab:<12} adattivo: canonica {b_c['canonical']:+.3f} "
f"(pctl {b_c['canonical_pctl']:.2f}) MEDIANA {b_c['median']:+.3f} "
f"banda [{b_c['lo']:+.3f},{b_c['hi']:+.3f}]")
print(f" {'':<12} costante: canonica {b_k['canonical']:+.3f} "
f"(pctl {b_k['canonical_pctl']:.2f}) MEDIANA {b_k['median']:+.3f} "
f"banda [{b_k['lo']:+.3f},{b_k['hi']:+.3f}]")
dl = A.anchor_luck_delta(lambda o: series_at_anchor(fn_ch, "cand", hours, o),
lambda o: series_at_anchor(fn_const_anch, "const", hours, o),
offs, metric=mtr)
print(f" {'':<12} DELTA APPAIATO (adatt - cost): mediana {dl['median_paired']:+.3f} "
f"positivo in {dl['n_positive']}/{dl['n_anchors']} "
f"banda [{dl['lo']:+.3f},{dl['hi']:+.3f}] gate_pass={dl['gate_pass']}")
# -- 8-bis. ANCORA APPAIATA per la cella DAVVERO ADATTIVA ---------------------------
if fn_ad_keep is not None and best_keep["tf"] == ch["tf"]:
hr("8-bis. ANCORA APPAIATA — cella DAVVERO ADATTIVA vs miglior costante")
for lab, mtr in (("Sharpe FULL", A._sh),
("Sharpe HOLD", lambda s_: A._sh(s_[s_.index >= HOLDOUT]))):
dl = A.anchor_luck_delta(
lambda o: series_at_anchor(fn_ad_keep, "adatt", hours, o),
lambda o: series_at_anchor(fn_const_anch, "const", hours, o), offs, metric=mtr)
b = A.anchor_luck_band(lambda o: series_at_anchor(fn_ad_keep, "adatt", hours, o),
offs, metric=mtr)
print(f" {lab:<12} adattiva-vera: canonica {b['canonical']:+.3f} "
f"(pctl {b['canonical_pctl']:.2f}) MEDIANA {b['median']:+.3f} "
f"banda [{b['lo']:+.3f},{b['hi']:+.3f}]")
print(f" {'':<12} DELTA APPAIATO vs costante: mediana {dl['median_paired']:+.3f} "
f"positivo in {dl['n_positive']}/{dl['n_anchors']} "
f"banda [{dl['lo']:+.3f},{dl['hi']:+.3f}] gate_pass={dl['gate_pass']}")
elif fn_ad_keep is not None:
print(f"\n (8-bis non girata: la cella adattiva sta su {best_keep['tf']} e il costante "
f"su {ch['tf']} -> le ancore non sarebbero coppie)")
# -- 8-ter. IL CRITERIO DEL FALSIFICATORE, SU **TUTTE** LE CELLE ADATTIVE ------------
hr("8-ter. IL CRITERIO DEL FALSIFICATORE APPLICATO A **OGNI** CELLA DAVVERO ADATTIVA")
print(" Il 22/08 dichiarava: «un rilevatore il cui L_t si muove DAVVERO e batte il miglior")
print(" lookback COSTANTE in >=6/8 anni appaiati con corr->TP01 < 0.6». Non basta che")
print(" fallisca il VINCITORE: si applica il criterio a tutte e", len(ad_rows), "le celle adattive.")
print(f"\n {'tf':>4} {'cella':<58} {'corr':>6} {'anni>0':>7} {'dShFULL':>8} {'dShHOLD':>8} esito")
passed = []
for r in sorted(ad_rows, key=lambda x: -x["insample_sharpe"]):
fnr = factory(tf=r["tf"], **r["params"])
dr = A.candidate_daily(fnr, tf=r["tf"])
JJ2 = pd.concat({"a": dr, "b": tp01}, axis=1, join="inner").dropna()
cor = float(JJ2["a"].corr(JJ2["b"]))
yr = paired_by_year(dr, d_const)
nyp = sum(1 for x in yr if x["d_sh"] > 0)
Jr = pd.concat({"C": dr, "B": d_const}, axis=1, join="inner").dropna()
JrH = Jr[Jr.index >= HOLDOUT]
dF = A._sh(Jr["C"]) - A._sh(Jr["B"])
dH = A._sh(JrH["C"]) - A._sh(JrH["B"])
ok = bool(cor < 0.6 and nyp >= 6)
if ok:
passed.append((r, cor, nyp, dF, dH))
lab = (f"{r['params']['det']} {r['params']['par']} {r['params']['stat']} "
f"{r['params']['mapping']} {r['params']['blend']}")
print(f" {r['tf']:>4} {lab:<58} {cor:>6.3f} {nyp:>5}/8 {dF:>+8.3f} {dH:>+8.3f} "
f"{'PASSA' if ok else ''}")
print(f"\n celle che soddisfano ENTRAMBE le condizioni del falsificatore: "
f"{len(passed)}/{len(ad_rows)}")
cors = []
for r in ad_rows:
dr = A.candidate_daily(factory(tf=r["tf"], **r["params"]), tf=r["tf"])
JJ2 = pd.concat({"a": dr, "b": tp01}, axis=1, join="inner").dropna()
cors.append(float(JJ2["a"].corr(JJ2["b"])))
print(f" corr->TP01 sulle celle adattive: min {min(cors):.3f} mediana "
f"{np.median(cors):.3f} max {max(cors):.3f} (soglia del falsificatore 0.60)")
# -- 9. FEE SWEEP ---------------------------------------------------------------------
hr("9. FEE SWEEP — l'adattivo trada di piu': e' li' che muore o no")
for nm, f_, tfx in (("adattivo (A)", fn_ch, ch["tf"]), ("WF-PnL (C)", fn_wf, wch["tf"]),
("miglior costante (B)", fn_const, cch["tf"]),
("TP01 30/90/180", fixed_target, "1d")):
line = []
for fee in (0.0, 0.0005, 0.001, 0.0015):
line.append(f"{2*fee*100:.2f}%RT {A._sh(A.candidate_daily(f_, tf=tfx, fee_side=fee)):+.3f}")
tt = np.mean([A.eval_weights(A.get(a, tfx), A._call_target(f_, A.get(a, tfx), a))
["turnover_per_year"] for a in ASSETS])
print(f" {nm:<22} " + " ".join(line) + f" | turnover {tt:.0f}x/anno")
hr(f"FINE — {time.time()-t0:.0f}s")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink