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

729 lines
36 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python
"""r0822_adaptive_horizon.py — ADAPTIVE-HORIZON TSMOM: stimare *quanto indietro guardare*
dai dati (change-point / persistenza) invece di fissarlo a 30/90/180 giorni.
DOMANDA
-------
Un TSMOM a lookback ADATTIVO batte il TSMOM a orizzonti FISSI 30/90/180 (TP01), in modo
MARGINALE e ONESTO?
COSA NON E' (i due vicini gia' bocciati, per non rifarli)
---------------------------------------------------------
* 02/07 "velocita' trend regime-condizionata": pesava i TRE orizzonti fissi in funzione del
percentile di vol -> risulto' un tilt-30d statico travestito (pctl 0.71 vs il null dei pesi
statici casuali). Li' gli orizzonti restavano 30/90/180: cambiavano solo i PESI.
* 29/06 "meta-allocazione dinamica" fra sleeve: peggiore dei pesi fissi. Li' l'oggetto adattato
erano i pesi di PORTAFOGLIO, non un parametro della strategia.
Qui l'oggetto adattato e' **l'orizzonte stesso**: L_t stimato da un rilevatore di change-point
(CUSUM) o di persistenza (variance-ratio, Hurst). Mai provato nel progetto.
FORMA FUNZIONALE (scelta perche' rende il confronto strutturalmente pulito)
--------------------------------------------------------------------------
TP01 canonico e' il blend di sign(c[i]/c[i-h]-1) su h = (30, 90, 180) giorni, long-flat,
vol-target 20%, leva cap 2x. Ma (30, 90, 180) == (L/3, L, 2L) con **L = 90 giorni**.
Quindi il candidato e' LA STESSA FUNZIONE con L che varia nel tempo:
blend='triple' -> horizons_t = (L_t/3, L_t, 2*L_t) (TP01 = L_t costante a 90g)
blend='single' -> horizons_t = (L_t,) (un solo orizzonte adattivo)
Tutto il resto (long-flat, target_vol 20%, vol_win 30g, leva 2x, fee 0.05%/lato, cadenza,
50/50 BTC+ETH) e' IDENTICO al baseline: **l'unico grado di liberta' e' l'orizzonte**.
=> il null non e' "meglio di zero", e' "meglio del TSMOM fisso a PARI GEOMETRIA".
FAMIGLIA DICHIARATA **PRIMA** DI GUARDARE (conteggio al RIALZO)
--------------------------------------------------------------
rilevatore (3): cusum | vratio | hurst
parametro (3): cusum k in {2,4,8} sigma-cumulate; vratio/hurst finestra W in {60,120,250}g
mappatura (2): direct (piu' persistenza / piu' tempo dall'ultimo change-point -> L piu' LUNGO)
inverse (specchiata) [il verso NON e' ovvio a priori => e' un asse, non una scelta]
blend (2): single | triple
timeframe (2): 1d | 12h [>= 12h: sotto, costi+overfit dominano (19/06)]
------------------------------------------------------------------
= 3 x 3 x 2 x 2 = 36 celle x 2 TF = **72 valutazioni**.
Costanti FISSATE A PRIORI e non cercate (dichiarate perche' sono gradi di liberta' non contati
nella griglia): Lmin=20g, Lmax=200g (contengono 30..180 del canonico); q=5g del variance-ratio;
scale del Hurst k in {1,2,4,8}g; finestra della percentile causale M=250g. Per questo il
deflated-Sharpe e' riportato ANCHE a N=144 e N=216 trial (sensibilita' del verdetto al conteggio,
regola del 30/07): un verdetto che si ribalta col conteggio si cita come tale.
GATE (tutti dell'harness del progetto, nessuno riscritto)
---------------------------------------------------------
A.study_family_honest -> cella scelta IN-SAMPLE-ONLY + deflated-Sharpe (NO selezione su hold-out)
A.marginal_vs_tp01 -> ADDS/HEDGE/NOISE/REDUNDANT/DILUTES + corr al baseline
A.causality_ok -> ricalcolo su prefisso (nessun look-ahead nella costruzione del segnale)
A.anchor_luck_band -> stima onesta = MEDIANA delle 24 ancore, non l'ancora 00:00
A.anchor_luck_delta -> confronto adattivo-vs-fisso: MEDIANA delle DIFFERENZE APPAIATE
A.implausible_sharpe -> Sharpe troppo bello = rischio fuori dal campione
A.eval_weights_smallcap-> haircut reale a $600 (min-order $5)
null del de-levering -> esiste k<1 che da' al FISSO lo stesso maxDD con Sharpe >=?
differenza APPAIATA PER ANNO -> un vantaggio concentrato in un anno e' un evento, non un meccanismo
Causalita': ogni rolling ha min_periods e finestra all'indietro; la CUSUM e' un loop in avanti
con reset (prefix-stable); la percentile e' rolling causale; nessuna statistica full-sample.
Runtime atteso ~3-6 min su 2 core. Nessun file toccato fuori da questo.
"""
from __future__ import annotations
import sys
import warnings
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 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 adattivo, in giorni
FIXED_H = (30, 90, 180) # == (L/3, L, 2L) con L=90 -> il baseline
TGT_VOL, LEV, VOLWIN = 0.20, 2.0, 30
VR_Q_DAYS = 5 # passo del variance-ratio
HURST_K_DAYS = (1, 2, 4, 8) # scale del Hurst
PCTL_M_DAYS = 250 # finestra della percentile causale
TFS = ("1d", "12h")
DETECTORS = ("cusum", "vratio", "hurst")
PARAMS = {"cusum": (2.0, 4.0, 8.0), "vratio": (60, 120, 250), "hurst": (60, 120, 250)}
MAPPINGS = ("direct", "inverse")
BLENDS = ("single", "triple")
# ===========================================================================
# RILEVATORI -> lookback adattivo L_t (in barre). Tutti causali.
# ===========================================================================
def _logret(c: np.ndarray) -> np.ndarray:
r = np.zeros(len(c))
r[1:] = np.log(c[1:] / c[:-1])
return r
def _roll_pctl(x: np.ndarray, m: int) -> np.ndarray:
"""Percentile CAUSALE del valore corrente nella sua finestra all'indietro (rolling rank).
Warmup -> 0.5 (neutro): non si inventa un regime dove non c'e' campione."""
s = pd.Series(x)
u = s.rolling(m, min_periods=max(20, m // 4)).rank(pct=True).values
return np.where(np.isfinite(u), u, 0.5)
def _cusum_age(c: np.ndarray, k: float, bpd: int) -> np.ndarray:
"""Eta' (in barre) dall'ultimo change-point CUSUM a due code su rendimenti standardizzati.
Loop in avanti con reset: dipende solo dal passato -> stabile su prefisso."""
lr = _logret(c)
sig = pd.Series(lr).rolling(30 * bpd, min_periods=max(5, 15 * bpd)).std().values
s = np.where((sig > 0) & np.isfinite(sig), lr / np.where(sig > 0, sig, 1.0), 0.0)
s = np.nan_to_num(s)
n = len(c)
age = np.zeros(n)
sp = sm = 0.0
last = 0
for i in range(n):
sp = max(0.0, sp + s[i])
sm = min(0.0, sm + s[i])
if max(sp, -sm) > k:
sp = sm = 0.0
last = i
age[i] = i - last
return age
def _vratio(c: np.ndarray, w_days: int, bpd: int) -> np.ndarray:
"""Variance ratio VR = Var(r_q) / (q * Var(r_1)) su finestra w_days. >1 = persistente."""
lr = _logret(c)
qb = max(2, VR_Q_DAYS * bpd)
lg = np.log(c)
lrq = np.zeros(len(c))
lrq[qb:] = lg[qb:] - lg[:-qb]
lrq[:qb] = np.nan
wb = max(20, w_days * bpd)
v1 = pd.Series(lr).rolling(wb, min_periods=wb // 2).var().values
vq = pd.Series(lrq).rolling(wb, min_periods=wb // 2).var().values
with np.errstate(invalid="ignore", divide="ignore"):
vr = vq / (qb * v1)
return vr
def _hurst(c: np.ndarray, w_days: int, bpd: int) -> np.ndarray:
"""Hurst da scaling della volatilita': sd(r_k) ~ k^H. Regressione di log sd su log k
(x fisso -> forma chiusa). Tutto rolling causale."""
lg = np.log(c)
wb = max(20, w_days * bpd)
xs, ys = [], []
for kd in HURST_K_DAYS:
kb = max(1, kd * bpd)
d = np.full(len(c), np.nan)
d[kb:] = lg[kb:] - lg[:-kb]
sd = pd.Series(d).rolling(wb, min_periods=wb // 2).std().values
with np.errstate(invalid="ignore", divide="ignore"):
ys.append(np.log(np.where(sd > 0, sd, np.nan)))
xs.append(np.log(kb))
X = np.asarray(xs, float)
Y = np.vstack(ys) # (n_scale, n_bar)
xm = X.mean()
denom = float(((X - xm) ** 2).sum())
with warnings.catch_warnings(): # warmup = colonne tutte-NaN
warnings.simplefilter("ignore", RuntimeWarning)
ym = np.nanmean(Y, axis=0)
H = ((X - xm)[:, None] * (Y - ym[None, :])).sum(axis=0) / denom
return H
_STAT_CACHE: dict = {}
def _stat_key(df: pd.DataFrame, det: str, par) -> tuple:
ts = df["timestamp"].values
return (det, par, len(df), int(ts[0]), int(ts[-1]))
def lookback_bars(df: pd.DataFrame, det: str, par, mapping: str) -> np.ndarray:
"""L_t in BARRE, dentro [Lmin, Lmax] giorni. Cache chiusa sul CONTENUTO del frame
(lunghezza + primo/ultimo timestamp) -> un prefisso NON prende la cache del pieno,
altrimenti causality_ok verrebbe ingannata dalla cache stessa."""
bpd = A.bars_per_day(df)
lo, hi = max(2, LMIN_D * bpd), max(3, LMAX_D * bpd)
c = df["close"].values.astype(float)
key = _stat_key(df, det, par)
if key not in _STAT_CACHE:
if det == "cusum":
_STAT_CACHE[key] = _cusum_age(c, float(par), bpd)
elif det == "vratio":
_STAT_CACHE[key] = _roll_pctl(_vratio(c, int(par), bpd), PCTL_M_DAYS * bpd)
elif det == "hurst":
_STAT_CACHE[key] = _roll_pctl(_hurst(c, int(par), bpd), PCTL_M_DAYS * bpd)
else:
raise ValueError(det)
stat = _STAT_CACHE[key]
if det == "cusum":
base = np.clip(stat, lo, hi)
L = base if mapping == "direct" else (lo + hi - base)
else:
u = np.clip(stat, 0.0, 1.0)
L = lo + u * (hi - lo) if mapping == "direct" else hi - u * (hi - lo)
return np.clip(np.round(L), 2, hi).astype(int)
# ===========================================================================
# DIREZIONE TSMOM (orizzonte costante o variabile) + vol-target
# ===========================================================================
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):
"""TSMOM a orizzonte COSTANTE L giorni, STESSA geometria del candidato adattivo.
blend='triple' -> (L/3, L, 2L): con L=90 e' ESATTAMENTE TP01 canonico 30/90/180."""
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)
d = np.clip(_dir_from_L(c, Lb, blend), 0, None)
return A.vol_target(d, df, TGT_VOL, VOLWIN, LEV)
fn.__name__ = f"const_L{L}_{blend}"
return fn
def factory(tf: str, det: str, par, mapping: str, blend: str):
"""target_fn(df) -> posizione per barra, decisa con dati <= close[i]."""
def fn(df: pd.DataFrame) -> np.ndarray:
c = df["close"].values.astype(float)
L = lookback_bars(df, det, par, mapping)
d = np.clip(_dir_from_L(c, L, blend), 0, None)
return A.vol_target(d, df, TGT_VOL, VOLWIN, LEV)
fn.__name__ = f"adapt_{det}{par}_{mapping}_{blend}"
return fn
GRID = [dict(det=d, par=p, mapping=m, blend=b)
for d in DETECTORS for p in PARAMS[d] for m in MAPPINGS for b in BLENDS]
# CONTROLLO dichiarato DOPO aver visto che il vincitore ha L_t incollato al bordo (sezione 3):
# il null "TSMOM fisso 30/90/180" NON basta piu', perche' confonde due cose diverse —
# "adattare l'orizzonte paga" e "un orizzonte piu' CORTO paga su questo campione".
# Il controllo giusto e' il migliore fra gli orizzonti COSTANTI della stessa banda, scelto
# in-sample-only. Trial aggiuntivi, contati al rialzo.
CONST_L = (20, 30, 45, 60, 90, 120, 150, 200)
CONST_GRID = [dict(L=L, blend=b) for L in CONST_L for b in BLENDS]
# ESTENSIONE dichiarata DOPO aver visto che l'ottimo costante cade sul BORDO BASSO della banda
# (L=20g = LMIN_D): un ottimo al bordo non e' un ottimo, e' il punto in cui si e' smesso di
# guardare. Si estende SOTTO e si guarda se e' un plateau o una scivolata verso il rumore.
# +4 valori x 1 blend x 2 TF = 8 trial in piu' (contati al rialzo).
CONST_L_EXT = (5, 7, 10, 15)
# ===========================================================================
# UTILITA' DI MISURA
# ===========================================================================
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
def resample_offset(df_1h: pd.DataFrame, hours: int, off: int) -> pd.DataFrame:
"""Stessa convenzione di trend_portfolio.resample_tf, con ancora spostata di `off` ore
(r0702_tp01_offset.py). NB: si usa `offset=`, non `origin=` (pandas ignora origin su 7D)."""
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 daily_from_target(fn, tf: str) -> pd.Series:
return A.candidate_daily(fn, tf=tf)
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),
ret_cand=round(float(np.prod(1 + g["C"].values) - 1) * 100, 1),
ret_base=round(float(np.prod(1 + g["B"].values) - 1) * 100, 1)))
return rows
def delever_null(cand: pd.Series, base: pd.Series) -> dict:
"""Il fisso, DE-LEVERATO a pari maxDD del candidato, che Sharpe fa?
Scalare le POSIZIONI per k scala gross e fee insieme -> il netto scala per k e lo Sharpe
e' INVARIANTE: percio' un candidato che compra solo meno DD non e' un miglioramento."""
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),
dd_base_delev=round(_dd(k * J["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:
"""Serie giornaliera 50/50 BTC+ETH della strategia `fn` ribilanciata sull'ancora `off`
(resample del 1h certificato). Cache per (name, hours, off)."""
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, mapping: str) -> dict:
"""Quanto si MUOVE davvero L_t (diagnostica di MECCANISMO, non di rendimento).
Una cella con L_t incollato a un bordo NON e' adattiva: e' un TSMOM a orizzonte
COSTANTE travestito, e non appartiene alla famiglia che si sta testando."""
fr_lo, fr_hi, iqr, med = [], [], [], []
for a in ASSETS:
df = A.get(a, tf)
bpd = A.bars_per_day(df)
L = lookback_bars(df, det, par, mapping) / bpd
fr_lo.append(float(np.mean(L <= LMIN_D + 1)))
fr_hi.append(float(np.mean(L >= LMAX_D - 1)))
iqr.append(float(np.percentile(L, 75) - np.percentile(L, 25)))
med.append(float(np.median(L)))
return dict(frac_lo=float(np.mean(fr_lo)), frac_hi=float(np.mean(fr_hi)),
iqr_days=float(np.mean(iqr)), med_days=float(np.mean(med)),
adattiva=bool(np.mean(fr_lo) < 0.5 and np.mean(fr_hi) < 0.5
and np.mean(iqr) >= 10.0))
def hr(t: str = "") -> None:
print("\n" + "=" * 78)
if t:
print(t)
print("=" * 78)
# ===========================================================================
def main() -> None:
print(__doc__.split("Runtime atteso")[0])
# -- 0. SANITY: il mio baseline DEVE essere TP01 canonico, bit-per-bit ----------------
hr("0. SANITY — il baseline 'fisso' e' TP01 canonico (se non lo e', ogni Δ 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 = daily_from_target(fixed_target, "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 - al.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"
# -- 1. GRIGLIA ONESTA: cella scelta IN-SAMPLE + deflated-Sharpe ----------------------
hr(f"1. study_family_honest — {len(GRID)} celle x {len(TFS)} TF = {len(GRID)*len(TFS)} trial")
rep = A.study_family_honest("ADAPT-H", factory, GRID, TFS)
ch = rep["chosen"]
print(f" celle valutate : {rep['n_cells']}")
print(f" cella scelta IN-SAMPLE : tf={ch['tf']} {ch['params']} "
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-8 IN-SAMPLE (come si sceglie onestamente):")
for r in rep["rows"][:8]:
print(f" IS {r['insample_sharpe']:+.3f} FULL {r['full_sharpe']:+.3f} "
f"{r['tf']:>3} {r['params']}")
# sensibilita' del verdetto al CONTEGGIO dei trial (regola 30/07)
fn_ch = factory(tf=ch["tf"], **ch["params"])
d_ch = daily_from_target(fn_ch, ch["tf"])
allsr = [r["full_sharpe"] for r in rep["rows"]]
print("\n sensibilita' del DSR al numero di trial dichiarati:")
for n_decl in (len(GRID) * len(TFS), 144, 216):
pad = allsr + [float(np.mean(allsr))] * max(0, n_decl - len(allsr))
dsr, _ = A.deflated_sharpe(A._sh(d_ch), pad, d_ch)
print(f" N={n_decl:>4} DSR={dsr:.3f} {'PASS' if dsr >= 0.95 else 'FAIL'}")
# cosa avrebbe scelto un disonesto (diagnostica, NON una decisione)
diag = []
for r in rep["rows"][:12]:
s_all = daily_from_target(factory(tf=r["tf"], **r["params"]), r["tf"])
diag.append({**r, "hold": round(A._sh(s_all[s_all.index >= HOLDOUT]), 3)})
diag.sort(key=lambda r: -r["hold"])
print("\n (diagnostica) le stesse 12 celle riordinate per HOLD-OUT — cio' che NON si fa:")
for r in diag[:4]:
print(f" HOLD {r['hold']:+.3f} IS {r['insample_sharpe']:+.3f} {r['tf']:>3} {r['params']}")
# -- 2. NUMERI del candidato vs il FISSO a pari geometria -----------------------------
hr("2. CANDIDATO vs TSMOM FISSO 30/90/180 — stessa geometria, stesso TF, stesse fee")
base_ch = daily_from_target(fixed_target, ch["tf"])
J = pd.concat({"C": d_ch, "B": base_ch}, axis=1, join="inner").dropna()
for lab, sl in (("FULL", J), ("HOLD-OUT", J[J.index >= HOLDOUT])):
if len(sl) < 30:
continue
print(f" {lab:<9} adattivo Sh {A._sh(sl['C']):+.3f} DD {_dd(sl['C'])*100:5.2f}% "
f"CAGR {_cagr(sl['C'])*100:6.2f}% | fisso Sh {A._sh(sl['B']):+.3f} "
f"DD {_dd(sl['B'])*100:5.2f}% CAGR {_cagr(sl['B'])*100:6.2f}% "
f"| dSh {A._sh(sl['C'])-A._sh(sl['B']):+.3f}")
print(f" corr(adattivo, fisso) = {J['C'].corr(J['B']):.3f}")
print("\n DIFFERENZA APPAIATA PER ANNO (un vantaggio in un anno solo e' un evento):")
rows = paired_by_year(d_ch, base_ch)
for r in rows:
print(f" {r['year']} adatt Sh {r['sh_cand']:+.2f} ({r['ret_cand']:+7.1f}%) "
f"fisso Sh {r['sh_base']:+.2f} ({r['ret_base']:+7.1f}%) dSh {r['d_sh']:+.2f}")
ds = [r["d_sh"] for r in rows]
print(f" -> anni con dSh>0: {sum(1 for x in ds if x > 0)}/{len(ds)} "
f"mediana dSh {np.median(ds):+.2f} media {np.mean(ds):+.2f}")
# -- 3. LOOKBACK: si muove davvero? ---------------------------------------------------
hr("3. IL LOOKBACK ADATTIVO SI MUOVE? (se sta al bordo e' un TSMOM COSTANTE travestito)")
for a in ASSETS:
df = A.get(a, ch["tf"])
bpd = A.bars_per_day(df)
L = lookback_bars(df, ch["params"]["det"], ch["params"]["par"],
ch["params"]["mapping"]) / bpd
lo, hi = LMIN_D, LMAX_D
print(f" {a}: L_t giorni min {L.min():.0f} p25 {np.percentile(L,25):.0f} "
f"mediana {np.median(L):.0f} p75 {np.percentile(L,75):.0f} max {L.max():.0f} "
f"| al bordo basso {np.mean(L<=lo+1)*100:.1f}% al bordo alto {np.mean(L>=hi-1)*100:.1f}%"
f" | |dL|/giorno mediano {np.median(np.abs(np.diff(L))):.1f}g")
# -- 4. GATE ---------------------------------------------------------------------------
hr("4. GATE")
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}")
dl = delever_null(d_ch, base_ch)
print(f" null del de-levering : {dl}")
# small-cap a $600
print(" haircut a $600 (min-order $5):")
for a in ASSETS:
df = A.get(a, ch["tf"])
sc_c = A.eval_weights_smallcap(df, A._call_target(fn_ch, df, a), capital=600.0)
sc_b = A.eval_weights_smallcap(df, fixed_target(df), capital=600.0)
print(f" {a}: adattivo modellato {sc_c['modeled']['sharpe']:+.3f} -> reale "
f"{sc_c['realistic']['sharpe']:+.3f} (haircut {sc_c['sharpe_haircut']:+.3f}, "
f"{sc_c['n_executed_trades']} trade) | fisso {sc_b['modeled']['sharpe']:+.3f} -> "
f"{sc_b['realistic']['sharpe']:+.3f} (haircut {sc_b['sharpe_haircut']:+.3f}, "
f"{sc_b['n_executed_trades']} trade)")
# -- 5. BANDA D'ANCORA -----------------------------------------------------------------
hours = 24 if ch["tf"] == "1d" else 12
offs = list(range(hours))
hr(f"5. BANDA D'ANCORA — {len(offs)} ancore, stima onesta = MEDIANA (mai l'ancora 00:00)")
def _series_at(off: int, which: str) -> pd.Series:
return series_at_anchor(fn_ch if which == "cand" else fixed_target, which, hours, off)
# sanity: l'ancora 0 riproduce la serie canonica
s0 = _series_at(0, "fixed")
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(o, "cand"), offs, metric=mtr)
b_f = A.anchor_luck_band(lambda o: _series_at(o, "fixed"), offs, metric=mtr)
print(f" {lab:<12} adattivo: canonica {b_c['canonical']:+.3f} (pctl {b_c['canonical_pctl']:.2f}) "
f"MEDIANA {b_c['median']:+.3f} banda [{b_c['lo']:+.3f},{b_c['hi']:+.3f}]")
print(f" {'':<12} fisso : canonica {b_f['canonical']:+.3f} (pctl {b_f['canonical_pctl']:.2f}) "
f"MEDIANA {b_f['median']:+.3f} banda [{b_f['lo']:+.3f},{b_f['hi']:+.3f}]")
dlt = A.anchor_luck_delta(lambda o: _series_at(o, "cand"),
lambda o: _series_at(o, "fixed"), offs, metric=mtr)
print(f" {'':<12} DELTA APPAIATO: mediana {dlt['median_paired']:+.3f} "
f"positivo in {dlt['n_positive']}/{dlt['n_anchors']} ancore "
f"banda [{dlt['lo']:+.3f},{dlt['hi']:+.3f}] gate_pass={dlt['gate_pass']}")
# -- 6. ADATTIVITA' EFFETTIVA DI TUTTA LA FAMIGLIA ------------------------------------
hr("6. QUANTE DELLE 72 CELLE SONO DAVVERO ADATTIVE? (filtro di MECCANISMO, non di resa)")
print(" criterio: L_t sta a un bordo <50% del tempo E l'IQR di L_t e' >= 10 giorni")
seen, adatt_rows = {}, []
for r in rep["rows"]:
k = (r["tf"], r["params"]["det"], r["params"]["par"], r["params"]["mapping"])
if k not in seen:
seen[k] = adaptivity(*k)
r["_adapt"] = seen[k]
if seen[k]["adattiva"]:
adatt_rows.append(r)
n_ad_cells = sum(1 for r in rep["rows"] if r["_adapt"]["adattiva"])
print(f" celle DAVVERO adattive: {n_ad_cells}/{len(rep['rows'])} "
f"(le altre sono TSMOM a orizzonte COSTANTE travestito)")
for k, v in sorted(seen.items()):
print(f" {k[0]:>3} {k[1]:>6} {str(k[2]):>5} {k[3]:<7} bordo-basso {v['frac_lo']*100:5.1f}% "
f"bordo-alto {v['frac_hi']*100:5.1f}% IQR {v['iqr_days']:5.1f}g "
f"mediana {v['med_days']:5.1f}g -> {'ADATTIVA' if v['adattiva'] else 'costante'}")
# -- 7. IL CONTROLLO CHE MANCAVA: il MIGLIOR ORIZZONTE COSTANTE ------------------------
hr(f"7. CONTROLLO L COSTANTE — {len(CONST_GRID)} celle x {len(TFS)} TF = "
f"{len(CONST_GRID)*len(TFS)} trial IN PIU'")
print(" la domanda diventa: il guadagno viene dall'ADATTARE, o dal fatto che su questo")
print(" campione un orizzonte piu' CORTO batte 30/90/180 anche se lo si tiene fisso?")
csel = A.select_cell_insample(lambda tf, L, blend: const_factory(tf, L, blend),
CONST_GRID, TFS)
cch = csel["chosen"]
print(f"\n miglior COSTANTE scelto in-sample: tf={cch['tf']} {cch['params']} "
f"Sharpe IS {cch['insample_sharpe']} FULL {cch['full_sharpe']}")
print(" griglia costante, in-sample (L=90 triple == TP01 canonico):")
for r in csel["rows"]:
tag = " <== TP01 canonico" 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(**cch["params"], tf=cch["tf"])
d_const = daily_from_target(fn_const, cch["tf"])
JC = pd.concat({"C": d_ch, "K": d_const}, axis=1, join="inner").dropna()
print(f"\n ADATTIVO vs MIGLIOR COSTANTE (entrambi scelti in-sample):")
for lab, sl in (("FULL", JC), ("HOLD-OUT", JC[JC.index >= HOLDOUT])):
if len(sl) < 30:
continue
print(f" {lab:<9} adattivo Sh {A._sh(sl['C']):+.3f} DD {_dd(sl['C'])*100:5.2f}% | "
f"costante Sh {A._sh(sl['K']):+.3f} DD {_dd(sl['K'])*100:5.2f}% | "
f"dSh {A._sh(sl['C'])-A._sh(sl['K']):+.3f}")
print(f" corr(adattivo, miglior costante) = {JC['C'].corr(JC['K']):.3f}")
print("\n differenza APPAIATA per anno vs il miglior COSTANTE:")
rows2 = paired_by_year(d_ch, d_const)
for r in rows2:
print(f" {r['year']} adatt {r['sh_cand']:+.2f} costante {r['sh_base']:+.2f} "
f"dSh {r['d_sh']:+.2f}")
d2 = [r["d_sh"] for r in rows2]
print(f" -> anni con dSh>0: {sum(1 for x in d2 if x > 0)}/{len(d2)} "
f"mediana {np.median(d2):+.2f}")
dl2 = delever_null(d_ch, d_const)
print(f" null del de-levering vs costante: {dl2}")
# ancora appaiata adattivo vs miglior costante (se stesso TF, altrimenti dichiarato)
if cch["tf"] == ch["tf"]:
for lab, mtr in (("Sharpe FULL", A._sh),
("Sharpe HOLD", lambda s: A._sh(s[s.index >= HOLDOUT]))):
dd2 = A.anchor_luck_delta(lambda o: series_at_anchor(fn_ch, "cand", hours, o),
lambda o: series_at_anchor(fn_const, "const", hours, o),
offs, metric=mtr)
print(f" ancora {lab}: DELTA APPAIATO mediana {dd2['median_paired']:+.3f} "
f"positivo in {dd2['n_positive']}/{dd2['n_anchors']} "
f"banda [{dd2['lo']:+.3f},{dd2['hi']:+.3f}] gate_pass={dd2['gate_pass']}")
else:
print(f" (ancora appaiata non girata: TF diversi, adattivo {ch['tf']} vs "
f"costante {cch['tf']} -> le ancore non sono coppie)")
# fee sweep: l'adattivo tradà di piu', il conto va fatto
print("\n FEE SWEEP (il candidato ha piu' turnover: e' li' che muore o no)")
for nm, f_ in (("adattivo", fn_ch), ("costante", fn_const), ("TP01 30/90/180", fixed_target)):
tfx = ch["tf"] if nm != "costante" else cch["tf"]
line = []
for fee in (0.0, 0.0005, 0.001, 0.0015):
dser = A.candidate_daily(f_, tf=tfx, fee_side=fee)
line.append(f"{2*fee*100:.2f}%RT {A._sh(dser):+.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:<16} " + " ".join(line) + f" | turnover {tt:.0f}x/anno")
# -- 8. LA FAMIGLIA RISTRETTA ALLE CELLE DAVVERO ADATTIVE -----------------------------
hr("8. GATE ONESTO SULLA SOLA FAMIGLIA ADATTIVA (la domanda della missione, isolata)")
if not adatt_rows:
print(" NESSUNA cella della famiglia e' davvero adattiva -> la domanda non e'")
print(" misurabile su questa famiglia: ogni 'vincitore' e' un orizzonte costante.")
else:
best_ad = max(adatt_rows, key=lambda r: r["insample_sharpe"])
print(f" celle adattive: {len(adatt_rows)} miglior IN-SAMPLE: tf={best_ad['tf']} "
f"{best_ad['params']} IS {best_ad['insample_sharpe']} FULL {best_ad['full_sharpe']}")
fn_ad = factory(tf=best_ad["tf"], **best_ad["params"])
d_ad = daily_from_target(fn_ad, best_ad["tf"])
sm_ad = A.study_marginal("ADAPT-H(solo adattive)", fn_ad, tf=best_ad["tf"])
dsr_ad, sr0_ad = A.deflated_sharpe(A._sh(d_ad),
[r["full_sharpe"] for r in adatt_rows], d_ad)
print(f" marginal={sm_ad['marginal_verdict']} earns_slot={sm_ad['earns_slot']} "
f"corr->TP01 {sm_ad['marginal'].get('corr_full')}")
print(f" deflated-Sharpe (N={len(adatt_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()
for lab, sl in (("FULL", JA), ("HOLD-OUT", JA[JA.index >= HOLDOUT])):
if len(sl) < 30:
continue
print(f" {lab:<9} adattiva-vera Sh {A._sh(sl['C']):+.3f} | miglior costante "
f"{A._sh(sl['K']):+.3f} | TP01 {A._sh(sl['B']):+.3f}")
rows3 = paired_by_year(d_ad, d_const)
d3 = [r["d_sh"] for r in rows3]
print(f" vs miglior costante, per anno: dSh>0 in {sum(1 for x in d3 if x>0)}/{len(d3)}"
f" mediana {np.median(d3):+.2f}")
# -- 9. L'OTTIMO COSTANTE E' AL BORDO: si estende la banda ----------------------------
hr("9. L'OTTIMO COSTANTE CADE SUL BORDO — plateau o scivolata verso il rumore?")
print(" un argmax al bordo della griglia non e' un ottimo: e' il punto in cui si e' smesso")
print(" di guardare. Si estende SOTTO i 20 giorni (8 trial in piu', contati al rialzo).")
ext = sorted(set(CONST_L) | set(CONST_L_EXT))
tab: dict = {}
for tfx in TFS:
print(f"\n --- TF {tfx}, blend triple (L/3, L, 2L) ---")
print(f" {'L(g)':>5} {'orizzonti':>16} {'IS':>7} {'FULL':>7} {'HOLD':>7} "
f"{'maxDD':>7} {'CAGR':>7} {'turn/a':>7} {'Sh@0.30%RT':>11}")
for L in ext:
fnx = const_factory(tfx, L, "triple")
dx = daily_from_target(fnx, tfx)
dx_h = dx[dx.index >= HOLDOUT]
dx_i = dx[dx.index < HOLDOUT]
d030 = A.candidate_daily(fnx, tf=tfx, fee_side=0.0015)
tt = np.mean([A.eval_weights(A.get(a, tfx), A._call_target(fnx, A.get(a, tfx), a))
["turnover_per_year"] for a in ASSETS])
tag = " <== TP01" if L == 90 else ""
tab[(tfx, L)] = dict(IS=A._sh(dx_i), FULL=A._sh(dx), HOLD=A._sh(dx_h))
print(f" {L:>5} {f'{L//3}/{L}/{2*L}':>16} {A._sh(dx_i):>7.3f} {A._sh(dx):>7.3f} "
f"{A._sh(dx_h):>7.3f} {_dd(dx)*100:>6.2f}% {_cagr(dx)*100:>6.2f}% "
f"{tt:>7.0f} {A._sh(d030):>11.3f}{tag}")
# --- e' una CURVA con un ottimo, o un crinale di RUMORE? --------------------------
print("\n RISOLUZIONE della griglia (regola 07/08: un criterio che decide su una frazione")
print(" dello spread della griglia e' rumore anche quando 'passa'):")
for tfx in TFS:
v = [tab[(tfx, L)]["IS"] for L in ext]
jump = float(np.median(np.abs(np.diff(v))))
arg = ext[int(np.argmax(v))]
arg_orig = max(CONST_L, key=lambda L: tab[(tfx, L)]["IS"])
print(f" {tfx:>3}: argmax IS con la griglia ORIGINALE L={arg_orig}g -> con la griglia "
f"ESTESA L={arg}g ({'SPOSTATO' if arg != arg_orig else 'stabile'})")
print(f" salto IS mediano fra L ADIACENTI {jump:+.3f} contro il vantaggio del "
f"vincitore su TP01 {max(v) - tab[(tfx, 90)]['IS']:+.3f} "
f"(rapporto {jump / max(1e-9, max(v) - tab[(tfx, 90)]['IS']):.2f})")
allc = [(tf, L) for tf in TFS for L in ext]
for met in ("IS", "FULL", "HOLD"):
vals = sorted((tab[k][met] for k in allc), reverse=True)
r1d = 1 + sum(1 for x in vals if x > tab[("1d", 90)][met])
print(f" rango di TP01 (1d, L=90g) fra le {len(allc)} celle su {met}: "
f"{r1d}/{len(allc)} (suo {tab[('1d',90)][met]:+.3f}, migliore {vals[0]:+.3f})")
hr("FINE")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink