Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(tp01-ls): la gamba SHORT del trend a 1d e' SCARTATA — toglie drift a 0/24 ancore

Browse Source 
§42 TP01-LS. Meccanismo TP01 di produzione CONGELATO, un solo grado di liberta': il
pavimento della direzione TSMOM (floor 0 / -1/3 / -2/3 / -1, 4 trial dichiarati prima).
Replica bit-exact 3/3 a 0.0 (floor=0 == CANONICAL long_only=True, floor=-1 == long_only=
False, sleeve 50/50 == src/portfolio/sleeves._tp01_returns) + 4a replica candidate_daily
vs tp01_baseline_daily a 0.0. Controlli positivi 4/4 (causality_ok su variante leaky,
implausible_sharpe su serie senza perdite, gate iso-vol su leva pura, anchor_luck_delta
A-vs-A).

Esito: la short NON paga il proprio costo. dDRIFT appaiato -1.91%/anno positivo in 0/24
ancore, dShFULL -0.354 (0/24), maxDD +8.80pp (24/24); gate iso-volatilita' FAIL (dCAGR
-6.19% a pari vol); senza il 2022 il divario RADDOPPIA (dShFULL -0.406, 0/24); anni
positivi 2/8 (2022 +1.84, 2025 +0.27) e non compensano gli altri sei; marginal_vs_tp01
= NEUTRAL su tutte e tre le celle (multicut False, corr 0.79-0.93, alpha -2.6%/anno);
la cella scelta AL BUIO in-sample e' IL CANONICO mentre quella scelta sull'hold-out e'
floor=-1/3 = firma di selezione-sull-hold-out. Non e' morte-per-fee: a fee ZERO 1.338
contro 0.904. Attesa a priori del coordinatore CONFERMATA e superata.

Unica metrica nel verso della short: dShHOLD +0.292 in 23/24 ancore — ma l'hold-out e'
1.6 anni, non e' selezionabile in-sample, e all'ancora canonica non si vede (-0.008:
caso speculare della lezione 26/07). Libro 75/25 con SKH01: dSh -0.356 in 0/24, e la
gamba short isolata ha Sharpe -0.335 con corr +0.10 a SKH01 (non ridondante, solo
perdente). Trovato per strada: src/live/book.py clippa gia' TP01 con max(tp_frac, 0.0)
-> il long-flat e' cablato anche nell'esecutore, non solo in CANONICAL.

Nessun file di produzione toccato. Libro, pesi, cron, config INVARIATI.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018M8Ncho6QV9FWLdyy4VyQf
This commit is contained in:
Adriano Dal Pastro
2026-08-23 00:54:17 +00:00
parent 0ff03babae
commit 3e3b41d842
1 changed files with 594 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/r0823_tp01_ls.py
+594
View File
@@ -0,0 +1,594 @@
#!/usr/bin/env python
"""r0823_tp01_ls.py — §42 TP01-LS: la gamba SHORT del trend a 1d paga il proprio costo?
DOMANDA. TP01 e' long-flat PER SCELTA: nei ribassi va a 0x invece di girarsi short. Il progetto
lo descrive come ASSICURAZIONE (taglia il DD ~6x contro buy&hold, contributo hold-out negativo
nel 99,1% delle ancore, LOO 26/07). L'obiettivo dichiarato e' il DRIFT, non lo Sharpe. La
direzione mai presa al TF di deploy e' il TSMOM LONG-SHORT a 1d: il progetto ha UNA riga
("trend 1h L/S ~1.0" fra gli edge deboli), misurata a 1h e senza l'harness onesto di oggi.
Decidibile in quattro pezzi:
(a) la short aggiunge DRIFT?
(b) sopravvive al GATE ISO-VOLATILITA'? (se aggiunge drift solo alzando l'esposizione e'
leva travestita — null del de-levering al contrario, 5 occorrenze nel progetto)
(c) NON e' semplicemente il beta del 2022? (si toglie il 2022 e si rifa')
(d) resta ADDS quando la si giudica contro TP01 stesso (`marginal_vs_tp01`)?
ATTESA A PRIORI DEL COORDINATORE, dichiarata PRIMA di misurare: il crypto ha drift positivo
strutturale e la short di trend perde quasi sempre fuori dal 2022 -> il guadagno sara' UN ANNO
SOLO (2022), lo stesso schema di SOL (22/08: 24/24 ancore positive, un anno solo). Se e' cosi'
la conclusione non e' "la short funziona" ma "il 2022 e' un evento nel campione".
MECCANISMO CONGELATO — UN SOLO GRADO DI LIBERTA'. Si parte dal TP01 di PRODUZIONE
(src/strategies/trend_portfolio.TrendPortfolio + CANONICAL) e si cambia SOLO il pavimento della
direzione TSMOM, da 0 (clip long-flat) a `floor`. Tutto il resto e' importato, non riscritto:
orizzonti 30/90/180g, vol_win 30g, target_vol 20%, leva cap 2x, fee 0.05%/lato, 50/50 BTC+ETH,
lo shift causale e la contabilita' fee stanno in `TrendPortfolio.net_returns` EREDITATA.
La direzione del blend a 3 orizzonti vive in {-1,-1/3,+1/3,+1}, quindi il pavimento e' un vero
knob discreto:
floor= 0 -> {0,0,1/3,1} = CANONICO (long-flat)
floor=-1/3 -> {-1/3,-1/3,1/3,1} = short "timida" (solo 1/3 di size al ribasso pieno)
floor=-2/3 -> {-2/3,-1/3,1/3,1}
floor=-1 -> {-1,-1/3,1/3,1} = LONG-SHORT pieno (== long_only=False di produzione)
GRIGLIA DICHIARATA PRIMA DI GUARDARE: 4 celle (i 4 floor) su UN solo TF (1d, quello di deploy).
Nessun altro parametro riaperto -> 4 trial, contati nel deflated-Sharpe. La griglia di 24 ANCORE
non e' una griglia di parametri (il DSR non la conta): si giudica con `anchor_luck_band` /
`anchor_luck_delta`, come prescritto dal gate del 26/07.
REPLICA BIT-EXACT PRIMA DI OGNI DELTA (tre vie, richiesta max|diff| == 0):
(1) floor=0 -> target_series == TrendPortfolio(**CANONICAL).target_series
(2) floor=-1 -> target_series == TrendPortfolio(long_only=False).target_series
(cioe' la mia unica riga nuova coincide con la produzione a ENTRAMBI gli estremi: il
codice davvero nuovo sono solo i due floor intermedi)
(3) serie 50/50 a floor=0 vs src.portfolio.sleeves._tp01_returns() (lo sleeve del libro)
Onesta': altlib (`marginal_vs_tp01`, `deflated_sharpe`, `causality_ok`, `implausible_sharpe`,
`eval_weights_smallcap`, `anchor_luck_*`) importato, non riscritto. Controlli positivi
OBBLIGATORI su ogni gate (un gate che non segnala puo' essere semplicemente rotto).
Vincoli: NESSUN file toccato fuori da questo script, nessun ordine inviato, sola lettura su
data/raw. Runtime ~1-2 min su 2 core.
"""
from __future__ import annotations
import sys
import time
from dataclasses import dataclass
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.strategies.trend_portfolio import ( # noqa: E402
CANONICAL, TrendPortfolio, realized_vol, simple_returns, tsmom_blend,
)
ASSETS = ("BTC", "ETH")
HOLDOUT = al.HOLDOUT
FLOORS = (0.0, -1.0 / 3.0, -2.0 / 3.0, -1.0) # LA GRIGLIA, dichiarata prima
FLOOR_LBL = {0.0: "0 (canonico LF)", -1.0 / 3.0: "-1/3", -2.0 / 3.0: "-2/3", -1.0: "-1 (LS pieno)"}
OFFSETS = tuple(range(24)) # banda d'ancora giornaliera
W_TP01_BOOK, W_SKH_BOOK = 0.75, 0.25 # src/live/book.py
CAPITAL = 635.0 # conto reale dichiarato nel briefing
# ===========================================================================
# IL MECCANISMO — TP01 di produzione con UN grado di liberta' in piu'
# ===========================================================================
@dataclass
class TPFloor(TrendPortfolio):
"""TP01 con pavimento della direzione parametrico. UNICA riga diversa dalla produzione:
`np.clip(direction, self.floor, None)` al posto di `np.clip(direction, 0, None)`.
Tutto il resto (vol realizzata, blend TSMOM, scaling, cap di leva, net_returns con lo
shift causale e le fee sul turnover) e' ereditato/importato dal modulo di produzione."""
floor: float = 0.0
def target_series(self, df: pd.DataFrame) -> np.ndarray:
c = df["close"].values.astype(float)
bpd = self._bpd(df)
bpy = bpd * 365.25
r = simple_returns(c)
vol = realized_vol(r, self.vol_win_days * bpd, bpy)
horizons = tuple(d * bpd for d in self.horizons_days)
direction = np.clip(tsmom_blend(c, horizons), self.floor, None) # <-- IL GRADO DI LIBERTA'
scal = np.where((vol > 0) & np.isfinite(vol), self.target_vol / vol, 0.0)
tgt = np.clip(direction * scal, -self.leverage, self.leverage)
tgt[~np.isfinite(tgt)] = 0.0
return tgt
_CFG = {k: v for k, v in CANONICAL.items() if k != "long_only"}
def tp_floor(floor: float) -> TPFloor:
return TPFloor(long_only=False, floor=floor, **_CFG)
# ===========================================================================
# DATI — resample 1h -> 24h con ANCORA h (stessa convenzione di resample_tf)
# ===========================================================================
@lru_cache(maxsize=8)
def _h1(asset: str) -> pd.DataFrame:
return al.get(asset, "1h")
@lru_cache(maxsize=128)
def bars(asset: str, off: int) -> pd.DataFrame:
g = _h1(asset).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=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"]]
@lru_cache(maxsize=512)
def sleeve(floor: float, off: int = 0, fee_side: float | None = None) -> pd.Series:
"""Serie 50/50 BTC+ETH dei rendimenti NETTI per barra giornaliera, ancora `off`.
Usa `TrendPortfolio.net_returns` EREDITATA -> shift causale e fee di produzione."""
tp = tp_floor(floor)
if fee_side is not None:
tp.fee_side = fee_side
ser = {}
for a in ASSETS:
df = bars(a, off)
net, ts = tp.net_returns(df)
ser[a] = pd.Series(net, index=pd.DatetimeIndex(pd.to_datetime(ts.values, utc=True)))
J = pd.concat(ser, axis=1, join="inner").fillna(0.0)
return pd.Series(0.5 * J[ASSETS[0]].values + 0.5 * J[ASSETS[1]].values, index=J.index)
# ===========================================================================
# METRICHE
# ===========================================================================
def sh(s: pd.Series) -> float:
return al._sh(s)
def dd(s: pd.Series) -> float:
return al._dd_ret(s)
def drift(s: pd.Series) -> float:
"""Drift annualizzato in LOG (la grandezza su cui vivono i muri di capitale; il carry si
misura sul log, mai sulla media aritmetica delle differenze — regola 23/08)."""
r = np.asarray(s.dropna().values, float)
return float(np.mean(np.log1p(np.clip(r, -0.99, None))) * 365.25) if len(r) else 0.0
def cagr(s: pd.Series) -> float:
r = np.asarray(s.dropna().values, float)
if len(r) < 2:
return 0.0
return float(np.exp(np.mean(np.log1p(np.clip(r, -0.99, None))) * 365.25) - 1.0)
def vol(s: pd.Series) -> float:
r = np.asarray(s.dropna().values, float)
return float(np.std(r) * np.sqrt(365.25)) if len(r) else 0.0
def hold(s: pd.Series) -> pd.Series:
return s[s.index >= HOLDOUT]
def no2022(s: pd.Series) -> pd.Series:
return s[s.index.year != 2022]
def row(s: pd.Series) -> dict:
return dict(sh=sh(s), sh_hold=sh(hold(s)), cagr=cagr(s), drift=drift(s),
vol=vol(s), dd=dd(s), n=len(s))
def turnover_year(floor: float, off: int = 0) -> float:
tot = 0.0
for a in ASSETS:
df = bars(a, off)
tgt = tp_floor(floor).target_series(df)
held = np.zeros(len(tgt)); held[1:] = tgt[:-1]
turn = np.abs(np.diff(held, prepend=0.0)).sum()
yrs = (df["datetime"].iloc[-1] - df["datetime"].iloc[0]).total_seconds() / 86400 / 365.25
tot += turn / yrs
return tot / len(ASSETS)
def time_in_market(floor: float, off: int = 0) -> tuple[float, float]:
lo, sho = [], []
for a in ASSETS:
tgt = tp_floor(floor).target_series(bars(a, off))
lo.append(float(np.mean(tgt > 0))); sho.append(float(np.mean(tgt < 0)))
return float(np.mean(lo)), float(np.mean(sho))
def hr(t=""):
print("\n" + "=" * 78)
if t:
print(t)
print("=" * 78)
# ===========================================================================
def main() -> None:
t_start = time.time()
print(__doc__.split("\n\n")[0])
print(f"\nAvvio {pd.Timestamp.now('UTC')} | griglia dichiarata: {len(FLOORS)} floor x 1 TF (1d) = "
f"{len(FLOORS)} trial | ancore: {len(OFFSETS)} (NON trial: gate d'ancora)")
# ------------------------------------------------------------------ 1. REPLICA BIT-EXACT
hr("1. REPLICA BIT-EXACT DEL CANONICO (prima di qualunque delta)")
ok_all = True
canon = TrendPortfolio(**CANONICAL)
ls_prod = TrendPortfolio(**{**CANONICAL, "long_only": False})
for a in ASSETS:
df = bars(a, 0)
d0 = float(np.max(np.abs(tp_floor(0.0).target_series(df) - canon.target_series(df))))
d1 = float(np.max(np.abs(tp_floor(-1.0).target_series(df) - ls_prod.target_series(df))))
print(f" {a}: max|floor=0 - CANONICAL(long_only=True) | = {d0:.1e} "
f"max|floor=-1 - CANONICAL(long_only=False)| = {d1:.1e}")
ok_all &= (d0 == 0.0) and (d1 == 0.0)
from src.portfolio.sleeves import _tp01_returns
prod = _tp01_returns()
mine = sleeve(0.0, 0)
JJ = pd.concat({"p": prod, "m": mine}, axis=1, join="inner").dropna()
d2 = float(np.max(np.abs(JJ["p"] - JJ["m"])))
print(f" sleeve 50/50 vs src.portfolio.sleeves._tp01_returns(): n={len(JJ)} max|diff| = {d2:.1e}")
ok_all &= (d2 == 0.0)
print(f" ==> REPLICA {'OK (3/3 a 0.0)' if ok_all else 'FALLITA — NON procedere'}")
if not ok_all:
sys.exit(1)
# ------------------------------------------------------------------ 2. CONTROLLI POSITIVI
hr("2. CONTROLLI POSITIVI DEI GATE (un gate che non segnala puo' essere rotto)")
def tgt_fn(floor):
return lambda df: tp_floor(floor).target_series(df)
def leaky(df): # guarda il futuro: DEVE fallire causality_ok
t = tp_floor(-1.0).target_series(df)
out = np.zeros_like(t); out[:-1] = t[1:]
return out
c_ok = al.causality_ok(tgt_fn(-1.0), tf="1d")
c_bad = al.causality_ok(leaky, tf="1d")
print(f" causality_ok(floor=-1) -> ok={c_ok['ok']} max_tail_diff={c_ok['max_tail_diff']}")
print(f" causality_ok(variante LEAKY) -> ok={c_bad['ok']} max_tail_diff={c_bad['max_tail_diff']}"
f" [controllo positivo: deve essere False]")
ctrl1 = c_ok["ok"] and not c_bad["ok"]
rng = np.random.default_rng(20260823)
idx = pd.date_range("2020-01-01", periods=900, freq="1D", tz="UTC")
fake_good = pd.Series(np.abs(rng.normal(0.002, 0.0005, 900)), index=idx) # 0 perdite
fake_noise = pd.Series(rng.normal(0.0004, 0.012, 900), index=idx)
ip_g = al.implausible_sharpe(fake_good)
ip_n = al.implausible_sharpe(fake_noise)
print(f" implausible_sharpe(serie senza perdite) -> implausible={ip_g['implausible']} "
f"[deve essere True] (rumore Sh {ip_n['sharpe']:.2f} -> {ip_n['implausible']} [deve essere False])")
ctrl2 = ip_g["implausible"] and not ip_n["implausible"]
base0 = sleeve(0.0, 0)
lev13 = base0 * 1.3 # LEVA PURA: il gate iso-vol deve bocciarla
k_lev = vol(base0) / vol(lev13)
print(f" GATE ISO-VOL, controllo positivo su LEVA PURA (canonico x1.3):")
print(f" grezzo dCAGR = {cagr(lev13) - cagr(base0):+.4f} (positivo: sembra 'piu' drift')")
print(f" iso-vol dCAGR = {cagr(lev13 * k_lev) - cagr(base0):+.6f} [deve essere ~0 -> e' leva]")
ctrl3 = abs(cagr(lev13 * k_lev) - cagr(base0)) < 1e-6
ad_self = al.anchor_luck_delta(lambda o: sleeve(0.0, o), lambda o: sleeve(0.0, o), OFFSETS, metric=sh)
print(f" anchor_luck_delta(A vs A) -> mediana {ad_self['median_paired']:+.6f}, "
f"positivi {ad_self['n_positive']}/{ad_self['n_anchors']} [deve essere 0 e 0/24]")
ctrl4 = (ad_self["median_paired"] == 0.0) and (ad_self["n_positive"] == 0)
print(f" ==> CONTROLLI POSITIVI: {sum([ctrl1, ctrl2, ctrl3, ctrl4])}/4 superati")
# ------------------------------------------------------------------ 3. TABELLA PRINCIPALE
hr("3. TABELLA PRINCIPALE (ancora canonica h=0, fee 0.10% RT, 50/50 BTC+ETH)")
print(f" {'floor':<16} {'ShFULL':>7} {'ShHOLD':>7} {'CAGR':>8} {'drift':>8} {'vol':>7} "
f"{'maxDD':>7} {'turn/a':>7} {'%long':>6} {'%short':>7}")
rows = {}
for f in FLOORS:
s = sleeve(f, 0)
r = row(s); rows[f] = r
tl, ts = time_in_market(f)
print(f" {FLOOR_LBL[f]:<16} {r['sh']:>7.3f} {r['sh_hold']:>7.3f} {r['cagr']:>7.2%} "
f"{r['drift']:>7.2%} {r['vol']:>6.2%} {r['dd']:>6.2%} {turnover_year(f):>7.2f} "
f"{tl:>5.1%} {ts:>6.1%}")
print(f" (n barre = {rows[0.0]['n']}, dal {base0.index[0].date()} al {base0.index[-1].date()})")
# ------------------------------------------------------------------ 4. GATE ISO-VOL
hr("4. GATE ISO-VOLATILITA' (null del de-levering AL CONTRARIO: la variante ALZA la vol)")
print(" Convenzione: si DE-LEVA la variante fino alla vol del canonico (k<1, gratis: nessuna")
print(" leva richiesta). Se a pari vol il canonico regge, la short era leva travestita.")
print(f" {'floor':<16} {'k de-lev':>9} {'CAGR iso':>9} {'dCAGR iso':>10} {'dSharpe':>9} "
f"{'maxDD iso':>10} {'esito':>10}")
isovol = {}
for f in FLOORS:
s = sleeve(f, 0)
k = vol(base0) / vol(s) if vol(s) > 0 else 1.0
si = s * k
dc = cagr(si) - cagr(base0)
isovol[f] = dict(k=k, dcagr=dc, dsh=sh(s) - sh(base0), dd=dd(si))
v = "—" if f == 0.0 else ("PASS" if dc > 0 else "FAIL")
print(f" {FLOOR_LBL[f]:<16} {k:>9.4f} {cagr(si):>8.2%} {dc:>+10.2%} "
f"{sh(s) - sh(base0):>+9.3f} {dd(si):>9.2%} {v:>10}")
# ------------------------------------------------------------------ 5. BANDA D'ANCORA
hr("5. BANDA D'ANCORA — 24 ore, MEDIANA DELLE DIFFERENZE APPAIATE (variante - canonico)")
print(" (un Delta misurato a una sola ancora eredita la fortuna di quell'ancora: regola 26/07)")
band = {}
for f in FLOORS[1:]:
d_sh = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=sh)
d_ho = al.anchor_luck_delta(lambda o, f=f: hold(sleeve(f, o)), lambda o: hold(sleeve(0.0, o)),
OFFSETS, metric=sh)
d_dr = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=drift)
d_dd = al.anchor_luck_delta(lambda o, f=f: sleeve(f, o), lambda o: sleeve(0.0, o), OFFSETS, metric=dd)
band[f] = dict(sh=d_sh, ho=d_ho, dr=d_dr, dd=d_dd)
print(f" floor {FLOOR_LBL[f]:<14} dShFULL {d_sh['median_paired']:+.3f} "
f"[{d_sh['lo']:+.3f},{d_sh['hi']:+.3f}] {d_sh['n_positive']:>2}/24 | "
f"dShHOLD {d_ho['median_paired']:+.3f} {d_ho['n_positive']:>2}/24 | "
f"dDRIFT {d_dr['median_paired']:+.2%} {d_dr['n_positive']:>2}/24 | "
f"dMAXDD {d_dd['median_paired']:+.2%} {d_dd['n_positive']:>2}/24")
print("\n ⚠️ L'UNICA metrica che va nel verso della short e' l'HOLD-OUT, e all'ancora")
print(" CANONICA non si vede: e' il caso speculare della lezione 26/07 (li' l'ancora")
print(" canonica INVENTAVA un danno, qui NASCONDE un vantaggio).")
print(f" {'floor':<16} {'dShHOLD @h=0':>13} {'mediana 24':>11} {'p10':>8} {'p90':>8}")
for f in FLOORS[1:]:
v = np.array(list(band[f]['ho']['per_offset'].values()), float)
print(f" {FLOOR_LBL[f]:<16} {band[f]['ho']['per_offset'][0]:>+13.3f} "
f"{np.median(v):>+11.3f} {np.percentile(v, 10):>+8.3f} {np.percentile(v, 90):>+8.3f}")
b_can = al.anchor_luck_band(lambda o: sleeve(0.0, o), OFFSETS, metric=sh)
b_ls = al.anchor_luck_band(lambda o: sleeve(-1.0, o), OFFSETS, metric=sh)
print(f" livelli: canonico ShFULL canonica {b_can['canonical']:.3f} (pctl {b_can['canonical_pctl']:.2f}) "
f"mediana {b_can['median']:.3f} | LS pieno canonica {b_ls['canonical']:.3f} "
f"(pctl {b_ls['canonical_pctl']:.2f}) mediana {b_ls['median']:.3f}")
# ------------------------------------------------------------------ 6. PER ANNO / SENZA 2022
hr("6. SCOMPOSIZIONE PER ANNO — e la domanda che decide: E SENZA IL 2022?")
years = sorted(set(base0.index.year))
print(f" Sharpe per anno (ancora h=0)")
print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS))
for y in years:
line = f" {y:<6}"
for f in FLOORS:
s = sleeve(f, 0); sy = s[s.index.year == y]
line += f"{sh(sy):>18.2f}"
print(line)
print(f"\n Rendimento per anno")
print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS))
for y in years:
line = f" {y:<6}"
for f in FLOORS:
s = sleeve(f, 0); sy = s[s.index.year == y]
line += f"{float(np.prod(1 + sy.values) - 1):>17.1%} "
print(line)
print(f"\n dSharpe ANNUALE della variante vs canonico (positivo = la short paga QUELL'anno)")
print(f" {'anno':<6}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS[1:]))
npos = {f: 0 for f in FLOORS[1:]}
for y in years:
line = f" {y:<6}"
s0 = sleeve(0.0, 0)
for f in FLOORS[1:]:
s = sleeve(f, 0)
d = sh(s[s.index.year == y]) - sh(s0[s0.index.year == y])
npos[f] += int(d > 0)
line += f"{d:>+18.2f}"
print(line)
print(" anni positivi: " + " ".join(f"{FLOOR_LBL[f]} {npos[f]}/{len(years)}" for f in FLOORS[1:]))
print(f"\n --- SENZA IL 2022 (ancora h=0) ---")
print(f" {'floor':<16} {'ShFULL':>7} {'dSh':>7} {'CAGR':>8} {'dCAGR':>8} {'iso-vol dCAGR':>14}")
b0n = no2022(base0)
for f in FLOORS:
sn = no2022(sleeve(f, 0))
k = vol(b0n) / vol(sn) if vol(sn) > 0 else 1.0
print(f" {FLOOR_LBL[f]:<16} {sh(sn):>7.3f} {sh(sn) - sh(b0n):>+7.3f} {cagr(sn):>7.2%} "
f"{cagr(sn) - cagr(b0n):>+7.2%} {cagr(sn * k) - cagr(b0n):>+14.2%}")
print(f"\n --- BANDA D'ANCORA SENZA IL 2022 (mediana differenze appaiate) ---")
for f in FLOORS[1:]:
d_sh = al.anchor_luck_delta(lambda o, f=f: no2022(sleeve(f, o)),
lambda o: no2022(sleeve(0.0, o)), OFFSETS, metric=sh)
d_dr = al.anchor_luck_delta(lambda o, f=f: no2022(sleeve(f, o)),
lambda o: no2022(sleeve(0.0, o)), OFFSETS, metric=drift)
print(f" floor {FLOOR_LBL[f]:<14} dShFULL {d_sh['median_paired']:+.3f} "
f"[{d_sh['lo']:+.3f},{d_sh['hi']:+.3f}] {d_sh['n_positive']:>2}/24 | "
f"dDRIFT {d_dr['median_paired']:+.2%} {d_dr['n_positive']:>2}/24")
# ------------------------------------------------------------------ 7. PROTEZIONE DD
hr("7. LA PROTEZIONE — il valore DICHIARATO di TP01 e' il taglio del DD: la short lo rompe?")
bh = {}
for a in ASSETS:
df = bars(a, 0)
bh[a] = pd.Series(simple_returns(df["close"].values.astype(float)),
index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)))
JB = pd.concat(bh, axis=1, join="inner").fillna(0.0)
bh50 = pd.Series(0.5 * JB[ASSETS[0]].values + 0.5 * JB[ASSETS[1]].values, index=JB.index)
bh50 = bh50.reindex(base0.index).fillna(0.0)
print(f" buy&hold 50/50: Sharpe {sh(bh50):.2f} CAGR {cagr(bh50):.1%} maxDD {dd(bh50):.1%}")
print(f" Criterio (B) di edge_watch: in un anno con DD del buy&hold > 10%, il DD dello sleeve")
print(f" deve restare sotto il 75% di quello. Rapporto DD_sleeve/DD_B&H per anno di sinistro:")
print(f" {'anno':<6} {'DD B&H':>8}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS))
fails = {f: 0 for f in FLOORS}; sinistri = 0
for y in years:
by = bh50[bh50.index.year == y]
if dd(by) <= 0.10:
continue
sinistri += 1
line = f" {y:<6} {dd(by):>7.1%}"
for f in FLOORS:
s = sleeve(f, 0); r_ = dd(s[s.index.year == y]) / dd(by)
fails[f] += int(r_ >= 0.75)
line += f"{r_:>17.2f}x" if False else f"{r_:>18.2f}"
print(line)
print(f" anni di sinistro: {sinistri}. Violazioni del criterio (rapporto >= 0.75): " +
" ".join(f"{FLOOR_LBL[f]} {fails[f]}" for f in FLOORS))
print(f" maxDD full: " + " ".join(f"{FLOOR_LBL[f]} {rows[f]['dd']:.2%}" for f in FLOORS))
# ------------------------------------------------------------------ 8. COSTI
hr("8. COSTI — la short raddoppia il turnover? curva fee 0-20 bps/lato")
print(f" {'fee/lato':>9} {'%RT':>7}" + "".join(f"{FLOOR_LBL[f]:>18}" for f in FLOORS))
for bps in (0, 2.5, 5, 10, 15, 20):
line = f" {bps:>8.1f}b {bps * 2 / 100:>6.2f}%"
for f in FLOORS:
line += f"{sh(sleeve(f, 0, bps / 10000.0)):>18.3f}"
print(line)
print(" turnover/anno (media asset): " + " ".join(f"{FLOOR_LBL[f]} {turnover_year(f):.2f}" for f in FLOORS))
# ------------------------------------------------------------------ 9. LIBRO 75/25 + SKH01
hr("9. IMPATTO SUL LIBRO 75/25 CON SKH01 (che e' GIA' long/short: ridondanza?)")
from src.portfolio.sleeves import _skyhook_returns
skh = _skyhook_returns()
skh.index = pd.DatetimeIndex(skh.index).tz_convert("UTC") if skh.index.tz else \
pd.DatetimeIndex(skh.index).tz_localize("UTC")
def book(floor: float, off: int = 0) -> pd.Series:
s = sleeve(floor, off).copy()
s.index = s.index.floor("1D") # ancora h -> data di calendario (1 barra per data)
J = pd.concat({"t": s, "k": skh}, axis=1, join="inner").dropna()
return pd.Series(W_TP01_BOOK * J["t"].values + W_SKH_BOOK * J["k"].values, index=J.index)
print(f" {'floor':<16} {'ShFULL':>7} {'ShHOLD':>7} {'CAGR':>8} {'maxDD':>7}")
for f in FLOORS:
b = book(f)
print(f" {FLOOR_LBL[f]:<16} {sh(b):>7.3f} {sh(hold(b)):>7.3f} {cagr(b):>7.2%} {dd(b):>6.2%}")
print(f" banda d'ancora del LIBRO (24 ancore su TP01, SKH01 all'ancora canonica):")
for f in FLOORS[1:]:
d_b = al.anchor_luck_delta(lambda o, f=f: book(f, o), lambda o: book(0.0, o), OFFSETS, metric=sh)
d_c = al.anchor_luck_delta(lambda o, f=f: book(f, o), lambda o: book(0.0, o), OFFSETS, metric=drift)
print(f" floor {FLOOR_LBL[f]:<14} dSh libro {d_b['median_paired']:+.3f} "
f"[{d_b['lo']:+.3f},{d_b['hi']:+.3f}] {d_b['n_positive']:>2}/24 | "
f"dDRIFT {d_c['median_paired']:+.2%} {d_c['n_positive']:>2}/24")
print(f"\n RIDONDANZA con SKH01 — la 'gamba short' isolata = variante meno canonico:")
for f in FLOORS[1:]:
leg = (sleeve(f, 0) - sleeve(0.0, 0))
leg.index = leg.index.floor("1D")
JL = pd.concat({"l": leg, "k": skh}, axis=1, join="inner").dropna()
print(f" floor {FLOOR_LBL[f]:<14} corr(gamba short, SKH01) = {JL['l'].corr(JL['k']):+.3f} "
f"Sharpe della gamba isolata {sh(JL['l']):+.3f} drift {drift(JL['l']):+.2%}")
# ------------------------------------------------------------------ 10. MARGINALE + DSR
hr("10. MARGINALE vs TP01 STESSO + DEFLATED SHARPE (4 trial dichiarati)")
cd0 = al.candidate_daily(tgt_fn(0.0), tf="1d")
bl = al.tp01_baseline_daily()
JC = pd.concat({"a": cd0, "b": bl}, axis=1, join="inner").dropna()
print(f" 4a replica: candidate_daily(floor=0) vs tp01_baseline_daily() max|diff| "
f"{float(np.max(np.abs(JC['a'] - JC['b']))):.1e} (n={len(JC)})")
# trial set del DSR: si prende dal percorso CODIFICATO (select_cell_insample), non a mano.
# ⚠️ errore mio catturato: la prima stesura costruiva la lista dei trial da solo. Due percorsi
# per lo stesso numero e' esattamente cio' che il gate codificato serve a evitare -> qui li
# confronto invece di sostituirne uno in silenzio.
sel = al.select_cell_insample(lambda tf, floor: tgt_fn(floor),
[dict(floor=f) for f in FLOORS], ("1d",))
all_sh = sel["all_full_sharpe"]
mine_sh = [round(sh(sleeve(f, 0)), 3) for f in FLOORS]
print(f" trial set DSR da select_cell_insample: {all_sh}")
print(f" stesso conto per via indipendente (sleeve): {mine_sh} -> "
f"{'COINCIDONO' if all_sh == mine_sh else 'DIVERGONO (fermarsi)'}")
for f in FLOORS[1:]:
cd = al.candidate_daily(tgt_fn(f), tf="1d")
m = al.marginal_vs_tp01(cd)
d, sr0 = al.deflated_sharpe(sh(sleeve(f, 0)), all_sh, sleeve(f, 0))
ip = al.implausible_sharpe(sleeve(f, 0))
print(f"\n floor {FLOOR_LBL[f]}: marginal_verdict = {m['marginal_verdict']}")
print(f" corr->TP01 {m['corr_full']} beta {m['beta_to_tp01']} alpha/a {m['alpha_ann']:+.4f} "
f"resid Sharpe {m['resid_sharpe_full']}")
print(f" blend w25 uplift full {m['blends']['w25']['uplift_full']:+.3f} / hold "
f"{m['blends']['w25']['uplift_hold']:+.3f} w50 full "
f"{m['blends']['w50']['uplift_full']:+.3f} / hold {m['blends']['w50']['uplift_hold']:+.3f}")
print(f" robust_oos={m['robust_oos']} multicut={m['multicut_persistent']} "
f"insample_edge={m['has_insample_edge']} (Sh IS {m['cand_insample_sharpe']}) "
f"is_hedge={m['is_hedge']}")
print(f" HEDGE vs ALPHA: corr(Sharpe-TP01 annuo, uplift annuo) = {m['hedge_yearly_corr']} "
f"uplift quando TP01 SALE {m['uplift_tp01_up']} / quando TP01 SCENDE {m['uplift_tp01_down']}")
print(f" multicut per anno: {m['multicut_uplift']}")
print(f" deflated-Sharpe {d:.3f} (sr0 nullo {sr0:.3f}, N={len(all_sh)}) "
f"implausible={ip['implausible']}")
print(f"\n --- GATE SELECTION-ON-HOLDOUT (codificato 29/06): quale floor si sceglie ")
print(f" guardando SOLO il pre-2025? Se e' il canonico, il guadagno hold-out NON e'")
print(f" selezionabile e sceglierlo sarebbe selezione-sull'hold-out. ---")
print(f" {'floor':<16} {'Sh IN-SAMPLE (<2025)':>21} {'Sh HOLD-OUT':>12} {'Sh FULL':>9}")
for f in FLOORS:
s_ = sleeve(f, 0)
print(f" {FLOOR_LBL[f]:<16} {sh(s_[s_.index < HOLDOUT]):>21.3f} {sh(hold(s_)):>12.3f} {sh(s_):>9.3f}")
fb = sel["chosen"]["params"]["floor"]
fh = max(FLOORS, key=lambda f: sh(hold(sleeve(f, 0))))
print(f" cella scelta AL BUIO (max Sharpe in-sample) = floor {fb:+.4f} -> "
f"{'IL CANONICO' if abs(fb) < 1e-9 else 'NON il canonico'}")
print(f" cella scelta SULL'HOLD-OUT (max Sharpe hold-out) = floor {fh:+.4f} -> "
f"{'diversa: firma di selezione-sull-hold-out' if abs(fh - fb) > 1e-9 else 'la stessa'}")
print(f"\n ⚠️ NOTA ONESTA sul DSR: con 4 celle quasi identiche sr0 e' minuscolo -> il DSR e'")
print(f" alto per COSTRUZIONE (dipende dalla VARIANZA della griglia, lezione §10 del 22/08).")
print(f" Qui non e' il gate che decide: e' un contorno. Il gate che decide e' l'iso-vol +")
print(f" la banda d'ancora + il per-anno.")
# ------------------------------------------------------------------ 11. ESEGUIBILITA'
hr(f"11. ESEGUIBILITA' A ${CAPITAL:.0f} (min_order $5) e vincolo di NETTING del libro")
for f in (0.0, -1.0):
for a in ASSETS:
df = bars(a, 0)
sc = al.eval_weights_smallcap(df, tp_floor(f).target_series(df),
capital=CAPITAL, min_order=5.0)
print(f" floor {FLOOR_LBL[f]:<14} {a}: Sharpe modellato {sc['modeled']['sharpe']:.3f} -> "
f"reale {sc['realistic']['sharpe']:.3f} haircut {sc['sharpe_haircut']:+.3f} "
f"trade eseguiti {sc['n_executed_trades']}")
src_book = (ROOT / "src" / "live" / "book.py").read_text()
clipped = "max(tp_frac, 0.0)" in src_book
print(f"\n VERIFICA SUL CODICE CHE GIRA: src/live/book.py contiene 'max(tp_frac, 0.0)' -> {clipped}")
print(f" Cioe' l'esecutore del libro CLIPPA a zero la frazione di TP01: il long-flat non e' solo")
print(f" in CANONICAL, e' cablato anche in `book_net_target`. Attivare la short richiederebbe")
print(f" di toccare src/live/book.py, cioe' CODICE su un percorso con soldi veri; e TP01 e SKH01")
print(f" si nettano su UNA sola posizione per strumento, quindi la short di TP01 andrebbe a")
print(f" sommarsi/cancellarsi con quella di SKH01 invece di essere una gamba a se'.")
hr("12. VERDETTO E COSA NON E' STATO MISURATO")
dsh_ls = band[-1.0]["sh"]["median_paired"]; ddr_ls = band[-1.0]["dr"]["median_paired"]
print(f" floor=-1 (LS pieno): dShFULL appaiato {dsh_ls:+.3f} ({band[-1.0]['sh']['n_positive']}/24), "
f"dDRIFT {ddr_ls:+.2%} ({band[-1.0]['dr']['n_positive']}/24), "
f"dMAXDD {band[-1.0]['dd']['median_paired']:+.2%} ({band[-1.0]['dd']['n_positive']}/24)")
print(f" iso-vol dCAGR {isovol[-1.0]['dcagr']:+.2%} | anni positivi {npos[-1.0]}/{len(years)} | "
f"cella scelta al buio = il canonico")
print(f"\n >>> VERDETTO: SCARTATO. La gamba short NON paga il proprio costo: toglie drift")
print(f" ({ddr_ls:+.2%}/anno a 0/24 ancore), FALLISCE il gate iso-vol, peggiora il maxDD")
print(f" ({band[-1.0]['dd']['median_paired']:+.2%} a 24/24) e non e' selezionabile in-sample.")
print(f" L'attesa a priori del coordinatore (il guadagno e' il 2022) e' CONFERMATA e")
print(f" superata: il 2022 c'e' ({npos[-1.0]}/8 anni positivi, dSharpe 2022 +1.84) ma NON")
print(f" compensa gli altri sei nemmeno con se stesso dentro il campione.")
print(f" E NON e' morte-per-fee: a fee ZERO il canonico fa {sh(sleeve(0.0, 0, 0.0)):.3f} contro "
f"{sh(sleeve(-1.0, 0, 0.0)):.3f} -> l'edge lordo non c'e'.")
print(f"\n COSA NON HO MISURATO (dichiarato, non nascosto):")
print(f" (a) FUNDING. Nessun backtest del progetto lo modella (§35, 22/08) e qui nemmeno.")
print(f" E' l'unica voce che potrebbe muovere il numero nel verso della short: una")
print(f" posizione corta INCASSA il funding quando il tasso e' positivo. MA il")
print(f" meccanismo misurato il 22/08 e' che esposizione e funding sono POSITIVAMENTE")
print(f" correlati, e un trend-short entra quando il mercato scende, cioe' quando il")
print(f" funding e' tipicamente NEGATIVO (e li' e' lo short a pagare). Argomento, non")
print(f" misura: si chiude con la macchineria di r0822d_funding.py, non a parole.")
print(f" Ordine di grandezza in gioco (~1-2%/anno) contro un drift perso di "
f"{-ddr_ls:.2%}/anno.")
print(f" (b) TF diversi dal 1d. Riaprire il TF riapre la FAMIGLIA (griglia 4 -> 4xN celle)")
print(f" e imporrebbe study_family_honest; qui il TF e' congelato a quello di deploy.")
print(f" (c) La banda d'ancora del LIBRO tiene SKH01 alla sua ancora canonica (le 23 fasi")
print(f" di SKH01 non sono state incrociate): il delta di libro e' appaiato solo sulle")
print(f" 24 ancore di TP01. Non cambia il segno (0/24 su tutte e tre le celle).")
print(f" (d) La lente WICK accoppiata (25/07): tutto qui e' close-only. Irrilevante per il")
print(f" verdetto (non c'e' barriera di rovina in gioco), rilevante se qualcuno")
print(f" riproponesse la variante sul canale funded.")
print(f"\n Tempo totale {time.time() - t_start:.1f}s")
if __name__ == "__main__":
main()