Adriano Dal Pastro
aad69f9790
Ricerca onesta su BTC/ETH + universo HL, branch separato (nessun impatto live).
Harness condiviso altlib (causale, fee 0.10% RT, marginal vs TP01, day-boundary,
haircut $600). Test 19/19 verdi.
- A DVOL direzionale -> LEAD hedge/DD-dampener, NON sleeve (buy-the-fear; is_hedge).
- B Intraday ERM 8h -> LEAD forte / forward-monitor: earns_slot=True, ADDS oltre
SKH01 (TP01+SKH+ERM 60/25/15 FULL 1.88/HOLD 1.46/DD 8.9%).
Caveat: plateau hold-out single-row, multiple-testing non
deflazionato, exec 8h. Controllo TOD = FAIL atteso.
- C Cross-sectional non-mom (low-vol HL) -> DEBOLE/forward-monitor (deflated-Sh 0.13,
storia 2.5a, non eseguibile $600) STAT-MODE.
- D Macro regime-gate -> RIDONDANTE col trend (corr->TP01 0.989), SCARTATO.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
353 lines
16 KiB
Python
353 lines
16 KiB
Python
"""DVOL-DIRECTIONAL — la vol IMPLICITA (Deribit DVOL) come SEGNALE DIREZIONALE/REGIME sul perp BTC/ETH.
|
|
|
|
Filone A. Diverso da TUTTO il lavoro DVOL precedente:
|
|
* `tp01_dvol_overlay.py` (2026-06-26): DVOL come DENOMINATORE del vol-target -> solo de-levering, SCARTATO.
|
|
* VOL03/04/10/11 (sweep alt): DVOL che GATA/SCALA un TSMOM (eredita lo Sharpe di trend di TP01 ->
|
|
il marginal scorer li boccia NEUTRAL/REDUNDANT).
|
|
* agent_14_dvol_spread (onda ortho): IV RELATIVA BTC-vs-ETH, market-NEUTRAL 2-leg (l'unico LEAD).
|
|
Qui invece: usare DVOL/IV-RV come segnale DIREZIONALE STANDALONE sul LIVELLO di mercato (long-flat o
|
|
L/S sul perp), per provare se il DVOL porta alpha DIREZIONALE ORTOGONALE a TP01 (non un overlay sul trend).
|
|
|
|
Angoli (tutti CAUSALI: decisi <= close[i], tenuti in i+1 dallo shift di eval_weights):
|
|
VRP-Z : z-score causale del vol-risk-premium (IV-RV). VRP ricco (IV>>RV, paura sovra-prezzata)
|
|
=> LONG underlying (fear-reversal, l'analogo direzionale dell'edge VRP); flip = falsifica.
|
|
DVOL-LV : percentile espandente causale del DVOL. "Buy-the-fear" (rank alto => long) vs
|
|
"buy-the-calm" (rank basso => long).
|
|
DVOL-MOM : DVOL vs ema(DVOL,k). DVOL in calo (risk-on) => long; in salita (risk-off) => flat (o short, L/S).
|
|
VRP-LS : variante long-short del VRP-Z (short quando VRP compresso).
|
|
|
|
DVOL parte 2021-03 -> pre-DVOL il segnale e' FLAT (0). Valuto sia FULL (con flat pre-periodo, deflaziona
|
|
lo Sharpe) sia DVOL-ERA-ONLY (giusto per la tesi). Gate: study_weights (abs+fee sweep), marginal_vs_tp01
|
|
(earns_slot), causality_ok, eval_weights_smallcap ($600), sign-falsification. Storia DVOL <5 anni -> caveat.
|
|
|
|
uv run python scripts/research/dvol_directional.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
|
|
import altlib as al # noqa: E402
|
|
|
|
# warmup: prima data DVOL (2021-03-24) + finestra max (z/rank) -> evita celle a campione corto
|
|
DVOL_ERA = pd.Timestamp("2021-10-01", tz="UTC")
|
|
TVOL = 0.20
|
|
VOLWIN = 30
|
|
LEVCAP = 2.0
|
|
|
|
|
|
# ===========================================================================
|
|
# CONTESTO causale per-asset (IV, RV, VRP) — tutto <= close[i]
|
|
# ===========================================================================
|
|
def _ctx(df: pd.DataFrame, asset: str):
|
|
c = df["close"].values.astype(float)
|
|
r = al.simple_returns(c)
|
|
bpd = al.bars_per_day(df)
|
|
iv = al.dvol(df, asset) / 100.0 # IV implicita 30d annualizzata (causale)
|
|
rv = al.realized_vol(r, max(2, 30 * bpd), bpd * 365.25) # RV trailing 30d annualizzata
|
|
return c, r, bpd, iv, rv
|
|
|
|
|
|
_RANK_CACHE: dict = {}
|
|
|
|
|
|
def _expanding_rank_arr(x: np.ndarray, min_p: int = 120) -> np.ndarray:
|
|
"""Percentile espandente causale: rank[i] = frazione di x[:i] < x[i].
|
|
Usa SOLO barre STRETTAMENTE precedenti + il valore corrente (noto a close[i]) — niente
|
|
future-peeking. Calcolato sull'array PASSATO (cosi' su un prefisso ridà i valori del prefisso,
|
|
=> causality_ok lo verifica davvero). O(n^2), n~3k OK."""
|
|
finite = np.isfinite(x)
|
|
out = np.full(len(x), np.nan)
|
|
for i in range(min_p, len(x)):
|
|
if not finite[i]:
|
|
continue
|
|
prev = x[:i][finite[:i]]
|
|
if len(prev) >= min_p // 2:
|
|
out[i] = float(np.mean(prev < x[i]))
|
|
return out
|
|
|
|
|
|
def dvol_rank(df: pd.DataFrame, asset: str, min_p: int = 120) -> np.ndarray:
|
|
x = np.asarray(al.dvol(df, asset), float)
|
|
key = (asset, len(x), round(float(np.nan_to_num(x[-1])), 4))
|
|
c = _RANK_CACHE.get(key)
|
|
if c is None:
|
|
c = _expanding_rank_arr(x, min_p)
|
|
_RANK_CACHE[key] = c
|
|
return c
|
|
|
|
|
|
def _finalize(direction: np.ndarray, df: pd.DataFrame, long_only: bool) -> np.ndarray:
|
|
d = np.nan_to_num(direction, nan=0.0)
|
|
if long_only:
|
|
d = np.clip(d, 0.0, 1.0)
|
|
return al.vol_target(d, df, TVOL, VOLWIN, LEVCAP)
|
|
|
|
|
|
# ===========================================================================
|
|
# FAMIGLIE DI SEGNALI (factory -> target_fn(df, asset))
|
|
# ===========================================================================
|
|
def make_vrp_z(win=90, thr=0.0, sign=+1, long_only=True):
|
|
"""LONG quando z(IV-RV) * sign > thr. sign=+1 => long su VRP RICCO (fear-reversal)."""
|
|
def fn(df, asset):
|
|
c, r, bpd, iv, rv = _ctx(df, asset)
|
|
vrp = iv - rv
|
|
z = al.zscore(vrp, max(5, win * bpd))
|
|
sig = sign * z
|
|
if long_only:
|
|
d = np.where(sig > thr, 1.0, 0.0)
|
|
else:
|
|
d = np.clip(np.tanh(sig), -1.0, 1.0)
|
|
return _finalize(d, df, long_only)
|
|
return fn
|
|
|
|
|
|
def make_dvol_level(q=0.5, side="fear", long_only=True):
|
|
"""side='fear': long quando rank DVOL > q (buy-the-fear). side='calm': long quando rank < q."""
|
|
def fn(df, asset):
|
|
rk = dvol_rank(df, asset)
|
|
if side == "fear":
|
|
base = (rk > q)
|
|
else:
|
|
base = (rk < q)
|
|
if long_only:
|
|
d = np.where(base, 1.0, 0.0)
|
|
else:
|
|
d = np.where(base, 1.0, -1.0)
|
|
return _finalize(d, df, long_only)
|
|
return fn
|
|
|
|
|
|
def make_dvol_mom(k=10, long_only=True):
|
|
"""DVOL in calo (dvol < ema(dvol,k)) => risk-on => long; in salita => flat (o short se L/S)."""
|
|
def fn(df, asset):
|
|
dv = al.dvol(df, asset)
|
|
em = al.ema(np.nan_to_num(dv, nan=np.nan), k)
|
|
falling = dv < em
|
|
if long_only:
|
|
d = np.where(falling, 1.0, 0.0)
|
|
else:
|
|
d = np.where(falling, 1.0, -1.0)
|
|
d = np.where(np.isfinite(dv) & np.isfinite(em), d, 0.0)
|
|
return _finalize(d, df, long_only)
|
|
return fn
|
|
|
|
|
|
def make_vrp_positive(long_only=True):
|
|
"""Baseline 'quasi buy&hold': long quando IV>RV (VRP positivo, ~80% del tempo)."""
|
|
def fn(df, asset):
|
|
c, r, bpd, iv, rv = _ctx(df, asset)
|
|
d = np.where((iv - rv) > 0, 1.0, 0.0)
|
|
return _finalize(d, df, long_only)
|
|
return fn
|
|
|
|
|
|
# ===========================================================================
|
|
# DIAGNOSTICA: il DVOL ha CONTENUTO DIREZIONALE? (probe corr signal[i] vs r[i+1])
|
|
# ===========================================================================
|
|
def _raw_direction(fn_dir, df, asset):
|
|
"""Estrae la DIREZIONE grezza (pre-vol-target) per la probe: rifa' il calcolo del segno."""
|
|
return fn_dir(df, asset)
|
|
|
|
|
|
def probe(name, dir_fn, era_only=True):
|
|
"""corr(dir[i], r[i+1]) e media r[i+1] per dir>0 vs dir<=0, pooled BTC+ETH (1d)."""
|
|
dd, rr = [], []
|
|
for a in al.CERTIFIED:
|
|
df = al.get(a, "1d")
|
|
idx = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
|
|
d = np.nan_to_num(dir_fn(df, a), nan=0.0)
|
|
r = al.simple_returns(df["close"].values.astype(float))
|
|
rnext = np.roll(r, -1); rnext[-1] = 0.0
|
|
mask = np.ones(len(df), bool)
|
|
if era_only:
|
|
mask &= np.asarray(idx >= DVOL_ERA)
|
|
mask[-1] = False
|
|
dd.append(d[mask]); rr.append(rnext[mask])
|
|
d = np.concatenate(dd); r = np.concatenate(rr)
|
|
corr = float(np.corrcoef(d, r)[0, 1]) if np.std(d) > 0 and np.std(r) > 0 else 0.0
|
|
up = float(np.mean(r[d > 0])) * 1e4 if (d > 0).any() else 0.0 # bps
|
|
dn = float(np.mean(r[d <= 0])) * 1e4 if (d <= 0).any() else 0.0
|
|
return dict(name=name, corr=round(corr, 4), n=int(len(d)),
|
|
long_bps=round(up, 1), flat_bps=round(dn, 1), edge_bps=round(up - dn, 1),
|
|
frac_long=round(float(np.mean(d > 0)), 3))
|
|
|
|
|
|
# ===========================================================================
|
|
# MARGINAL su finestra DVOL-ERA (piu' giusto della full che include il flat pre-2021)
|
|
# ===========================================================================
|
|
def cand_daily_era(target_fn, tf="1d", start=DVOL_ERA, fee_side=al.FEE_SIDE) -> pd.Series:
|
|
series = {}
|
|
for a in al.CERTIFIED:
|
|
df = al.get(a, tf)
|
|
ev = al.eval_weights(df, al._call_target(target_fn, df, a), fee_side=fee_side)
|
|
series[a] = pd.Series(ev["net"], index=ev["idx"])
|
|
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
|
|
d = al._to_daily(0.5 * J[al.CERTIFIED[0]] + 0.5 * J[al.CERTIFIED[1]])
|
|
return d[d.index >= start]
|
|
|
|
|
|
def era_full_sharpe(target_fn, tf="1d"):
|
|
"""Sharpe/DD del candidato sulla SOLA era DVOL (no flat pre-2021), 50/50."""
|
|
s = cand_daily_era(target_fn, tf)
|
|
r = s.values
|
|
sh = float(np.mean(r) / np.std(r) * np.sqrt(365.25)) if np.std(r) > 0 else 0.0
|
|
eq = np.cumprod(1 + r); pk = np.maximum.accumulate(eq)
|
|
dd = float(np.max((pk - eq) / pk)) if len(eq) else 0.0
|
|
yrs = (s.index[-1] - s.index[0]).days / 365.25 if len(s) > 1 else 1.0
|
|
cagr = eq[-1] ** (1 / yrs) - 1 if yrs > 0 and eq[-1] > 0 else -1.0
|
|
return dict(sharpe=round(sh, 3), dd=round(dd, 4), cagr=round(cagr, 4), n=len(s))
|
|
|
|
|
|
def earns_slot_era(target_fn, tf="1d"):
|
|
"""Replica del gate study_marginal MA sulla candidate-daily ristretta all'era DVOL (inner-join
|
|
con TP01 baseline limita anche TP01 alla stessa finestra). Piu' equo per un segnale DVOL-only."""
|
|
marg = al.marginal_vs_tp01(cand_daily_era(target_fn, tf))
|
|
v = marg.get("marginal_verdict")
|
|
earns = (v == "ADDS" and marg.get("robust_oos", False)
|
|
and marg.get("beats_noise_null", False) and not marg.get("is_hedge", False))
|
|
return marg, earns
|
|
|
|
|
|
# ===========================================================================
|
|
# MAIN
|
|
# ===========================================================================
|
|
def hr(t=""):
|
|
print("=" * 100)
|
|
if t:
|
|
print(t)
|
|
print("=" * 100)
|
|
|
|
|
|
def main():
|
|
hr("DVOL-DIREZIONALE — DVOL/IV-RV come segnale DIREZIONALE standalone su BTC/ETH (perp). 1d, causale.")
|
|
print(f" Era DVOL valutata da {DVOL_ERA.date()} (warmup z/rank). FULL include flat pre-2021 (deflaziona Sharpe).")
|
|
print(f" vol-target {TVOL:.0%}, leva cap {LEVCAP}x, fee 0.10% RT. TP01 baseline come riferimento marginale.\n")
|
|
|
|
# ---- 1) PROBE: contenuto direzionale (corr signal[i] vs r[i+1]) ---------
|
|
hr("1) PROBE DIREZIONALE — il DVOL predice il ritorno del giorno dopo? (pooled BTC+ETH, era DVOL)")
|
|
probes = [
|
|
("VRP-Z+ (long VRP ricco)", make_vrp_z(90, 0.0, +1)),
|
|
("VRP-Z- (long VRP basso)", make_vrp_z(90, 0.0, -1)),
|
|
("DVOL-LV fear (rank>0.5)", make_dvol_level(0.5, "fear")),
|
|
("DVOL-LV calm (rank<0.5)", make_dvol_level(0.5, "calm")),
|
|
("DVOL-MOM falling=long", make_dvol_mom(10)),
|
|
("VRP>0 (quasi buy&hold)", make_vrp_positive()),
|
|
]
|
|
print(f" {'segnale':<28}{'corr':>9}{'long bps':>10}{'flat bps':>10}{'edge bps':>10}{'frac_long':>10}")
|
|
for nm, fn in probes:
|
|
p = probe(nm, fn)
|
|
print(f" {nm:<28}{p['corr']:>+9.4f}{p['long_bps']:>+10.1f}{p['flat_bps']:>+10.1f}"
|
|
f"{p['edge_bps']:>+10.1f}{p['frac_long']:>10.3f}")
|
|
print(" (edge bps = ritorno medio giorno-dopo quando long MENO quando flat; >0 = il segnale separa)\n")
|
|
|
|
# ---- 2) ROBUSTEZZA ASSOLUTA (study_weights) + fee sweep + era-only ------
|
|
hr("2) ROBUSTEZZA ASSOLUTA — study_weights su BTC+ETH (1d), fee sweep 0.00-0.20% RT")
|
|
grid = {
|
|
"VRP-Z90 long-flat": make_vrp_z(90, 0.0, +1, True),
|
|
"VRP-Z60 long-flat": make_vrp_z(60, 0.0, +1, True),
|
|
"VRP-Z120 long-flat": make_vrp_z(120, 0.0, +1, True),
|
|
"VRP-Z90 L/S": make_vrp_z(90, 0.0, +1, False),
|
|
"DVOL-fear q0.4 LF": make_dvol_level(0.4, "fear", True),
|
|
"DVOL-fear q0.5 LF": make_dvol_level(0.5, "fear", True),
|
|
"DVOL-calm q0.5 LF": make_dvol_level(0.5, "calm", True),
|
|
"DVOL-MOM k10 LF": make_dvol_mom(10, True),
|
|
"DVOL-MOM k20 LF": make_dvol_mom(20, True),
|
|
"DVOL-MOM k10 L/S": make_dvol_mom(10, False),
|
|
}
|
|
reps = {}
|
|
for nm, fn in grid.items():
|
|
rep = al.study_weights(nm, fn, tfs=("1d",))
|
|
reps[nm] = rep
|
|
era = era_full_sharpe(fn)
|
|
v = rep["verdict"]
|
|
c = rep["cells"][0]
|
|
print(f" {nm:<22} abs={v['grade']:<4} minFull={c['min_asset_full_sharpe']:+.2f} "
|
|
f"minHold={c['min_asset_holdout_sharpe']:+.2f} feeOK={c['fee_survives']!s:<5} "
|
|
f"|| ERA-only Sh={era['sharpe']:+.2f} DD={era['dd']*100:.0f}% CAGR={era['cagr']*100:+.0f}%")
|
|
print(" (FULL include 2018-2021 flat => Sharpe basso e' atteso; ERA-only e' il giudizio equo del segnale)\n")
|
|
|
|
# candidati da portare al marginal: i migliori per ERA Sharpe + abs non-FAIL
|
|
ranked = sorted(grid.items(), key=lambda kv: era_full_sharpe(kv[1])["sharpe"], reverse=True)
|
|
top = ranked[:4]
|
|
|
|
# ---- 3) MARGINAL vs TP01 (era DVOL) — il gate vero -----------------------
|
|
hr("3) MARGINAL vs TP01 (finestra DVOL-era) — earns_slot? (il gate decisivo)")
|
|
for nm, fn in top:
|
|
marg, earns = earns_slot_era(fn)
|
|
bl = marg.get("blends", {}).get("w25", {})
|
|
print(f"\n --- {nm} ---")
|
|
print(f" verdetto={marg.get('marginal_verdict')} EARNS_SLOT(era)={earns}")
|
|
print(f" corr->TP01 full {marg.get('corr_full')} hold {marg.get('corr_hold')} "
|
|
f"beta {marg.get('beta_to_tp01')} resid Sh {marg.get('resid_sharpe_full')}")
|
|
print(f" cand standalone full {marg.get('cand_full_sharpe')}/hold {marg.get('cand_hold_sharpe')} "
|
|
f"in-sample Sh {marg.get('cand_insample_sharpe')} has_edge={marg.get('has_insample_edge')}")
|
|
print(f" blend w25: full {bl.get('full')} (uplift {bl.get('uplift_full')}) "
|
|
f"hold {bl.get('hold')} (uplift {bl.get('uplift_hold')}) DD {bl.get('dd')}")
|
|
print(f" multi-cut {marg.get('multicut_uplift')} persistent={marg.get('multicut_persistent')} "
|
|
f"robust_oos={marg.get('robust_oos')} is_hedge={marg.get('is_hedge')}")
|
|
|
|
# ---- 3b) study_marginal canonico (full history, gate di progetto) --------
|
|
hr("3b) study_marginal CANONICO (full history, include flat pre-DVOL) — gate ufficiale di progetto")
|
|
leader_nm, leader_fn = top[0]
|
|
sm = al.study_marginal(leader_nm, leader_fn, tf="1d")
|
|
print(al.fmt_marginal(sm))
|
|
|
|
# ---- 4) CAUSALITA' (look-ahead guard) -----------------------------------
|
|
hr("4) CAUSALITA' — causality_ok (ricalcolo su prefisso, nessun future-peeking)")
|
|
for nm, fn in top:
|
|
co = al.causality_ok(fn, tf="1d")
|
|
print(f" {nm:<22} ok={co['ok']!s:<5} max_tail_diff={co['max_tail_diff']} checked={co['checked']}")
|
|
|
|
# robustezza alignment DVOL: lag +1 giorno (extra-conservativo) sul leader
|
|
hr("4b) ROBUSTEZZA ALIGNMENT — DVOL laggato +1g (extra-conservativo) sul leader: l'edge sopravvive?")
|
|
|
|
def _lag(fn):
|
|
# ricostruisce il segnale con dvol shiftato di 1 barra (usa solo DVOL di IERI)
|
|
def g(df, asset):
|
|
# monkey: usa al.dvol ma shift -> approssimo rifacendo via wrapper sul ctx non e' banale;
|
|
# piu' semplice: confronto Sharpe era con e senza lag costruendo direzione laggata generica.
|
|
base = fn(df, asset)
|
|
lagged = np.zeros_like(base); lagged[1:] = base[:-1]
|
|
return lagged
|
|
return g
|
|
lag_era = era_full_sharpe(_lag(leader_fn))
|
|
base_era = era_full_sharpe(leader_fn)
|
|
print(f" {leader_nm}: ERA Sh base {base_era['sharpe']:+.2f} -> con segnale laggato +1g "
|
|
f"{lag_era['sharpe']:+.2f} (calo grande = edge fragile all'alignment)")
|
|
|
|
# ---- 5) ESEGUIBILITA' $600 — eval_weights_smallcap ----------------------
|
|
hr("5) ESEGUIBILITA' a $600 — eval_weights_smallcap (min_order $5, haircut reale vs modellato)")
|
|
for nm, fn in top:
|
|
for a in al.CERTIFIED:
|
|
df = al.get(a, "1d")
|
|
sc = al.eval_weights_smallcap(df, al._call_target(fn, df, a), capital=600, min_order=5)
|
|
print(f" {nm:<22} {a}: modellato Sh {sc['modeled']['sharpe']:+.2f} -> reale $600 "
|
|
f"Sh {sc['realistic']['sharpe']:+.2f} (haircut {sc['sharpe_haircut']:+.2f}) "
|
|
f"trade eseguiti {sc['n_executed_trades']}")
|
|
|
|
# ---- 6) SIGN-FALSIFICATION sul leader -----------------------------------
|
|
hr("6) SIGN-FALSIFICATION — invertire il segno del leader deve PEGGIORARE (altrimenti e' rumore)")
|
|
# leader e' VRP-Z+ se nel top; provo l'inverso esplicito su VRP-Z e DVOL
|
|
flips = [
|
|
("VRP-Z90 sign +1 (tesi)", make_vrp_z(90, 0.0, +1, True)),
|
|
("VRP-Z90 sign -1 (flip)", make_vrp_z(90, 0.0, -1, True)),
|
|
("DVOL-fear q0.5 (tesi)", make_dvol_level(0.5, "fear", True)),
|
|
("DVOL-calm q0.5 (flip)", make_dvol_level(0.5, "calm", True)),
|
|
]
|
|
for nm, fn in flips:
|
|
e = era_full_sharpe(fn)
|
|
print(f" {nm:<26} ERA Sh {e['sharpe']:+.2f} DD {e['dd']*100:.0f}% CAGR {e['cagr']*100:+.0f}%")
|
|
|
|
hr()
|
|
print("FINE. Leggere il verdetto onesto nel diario docs/diary/2026-06-29-dvol-directional.md")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|