Adriano Dal Pastro
7534b08be0
4 agenti (drift/cond/intra/overlay), 375 trial: il drift weekend e' beta B&H, i gate condizionali sono TP01 re-timed, l'intraday weekend e' moneta simmetrica (CME-gap morto anche lordo). Fatto strutturale: TP01-weekend-flat = danno certo (dSh -0.46, P=1.000) -> ogni proposta "risk-off weekend" parte REFUTED salvo null de-levering. Chiusa la famiglia calendario su BTC/ETH (SEA+expiry+event-clock +weekend). Book/pesi INVARIATI. Diario 2026-07-17-weekend-window.md. gitignore: + data/live/ (log esecuzioni book live, stato runtime del conto reale) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
248 lines
13 KiB
Python
248 lines
13 KiB
Python
"""r0717_wk_drift — WK-DRIFT: esposizione weekend pura (ven H_in -> lun H_out), griglia di timing.
|
|
|
|
Domanda: la finestra ven->lun su BTC/ETH contiene drift direzionale sfruttabile NETTO fee,
|
|
robusto al timing? Prior art: SEA (day-of-week) e' gia' morta nel progetto; onere della prova ALTO.
|
|
|
|
Metodo (harness onesto altlib):
|
|
* Famiglia: posizione (long E short = celle separate) da ven H_in a lun H_out,
|
|
H_in, H_out in {0,4,8,12,16,20} UTC, asset in {BTC, ETH} -> 6*6*2*2 = 144 trial.
|
|
* Causalita': eval_weights fa pos[t] = target[t-1] (target deciso a close[i] e' applicato
|
|
al ritorno close[i]->close[i+1]). L'intervallo detenuto da target[i] INIZIA al close
|
|
della barra i => l'indicatore di finestra si valuta su datetime[i] + 1h (close time).
|
|
Il segnale e' puro calendario (noto in anticipo) -> causale per costruzione; verificato
|
|
comunque con causality_ok.
|
|
* Selezione cella con SOLO Sharpe in-sample (pre-2025); poi si guarda l'hold-out.
|
|
* deflated_sharpe su TUTTA la famiglia (144 trial, unita' daily-compounded coerenti).
|
|
* NULL di specialita': stessa finestra (stessa lunghezza, stessi H_in/H_out) ancorata a
|
|
ogni altro giorno della settimana (lun->gio, mar->ven, ...) -> il ven->lun e' speciale
|
|
o qualsiasi finestra ~3g cattura lo stesso beta di B&H a esposizione ridotta?
|
|
+ confronto capture: quota di log-return catturata vs quota di ore esposte.
|
|
* day_boundary_robust sulla cella selezionata (effetto calendario che si inverte
|
|
spostando il confine UTC = artefatto di etichettatura).
|
|
* Sweep fee per-side (0, 0.05%, 0.10%, 0.15%) sulla cella selezionata.
|
|
* Per-anno: la cella vive di 1-2 anni o e' positiva ogni anno?
|
|
|
|
Vincoli: nessun file esistente modificato; dati solo via altlib (BTC/ETH 1h certificati).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
import altlib as al
|
|
|
|
H_GRID = (0, 4, 8, 12, 16, 20)
|
|
ASSETS = ("BTC", "ETH")
|
|
DIRS = (1, -1) # long e short come celle separate
|
|
FRI = 4 # pandas weekday: Mon=0 ... Fri=4
|
|
SPAN_DAYS = 3 # ven -> lun = anchor_day d -> (d+3) % 7
|
|
DAY_NAMES = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri", 5: "Sat", 6: "Sun"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Target di finestra calendario (causale: funzione deterministica del clock)
|
|
# ---------------------------------------------------------------------------
|
|
def make_target(h_in: int, h_out: int, direction: int, day_in: int = FRI):
|
|
"""target[i] = direction se l'intervallo detenuto da target[i] — che inizia al
|
|
CLOSE della barra i (datetime[i] + 1h) — cade dentro [day_in H_in, day_out H_out).
|
|
day_out = (day_in + 3) % 7. Entrata eseguita al close di ven H_in:00, uscita al
|
|
close di lun H_out:00 => esposizione esattamente [ven H_in, lun H_out]."""
|
|
day_out = (day_in + SPAN_DAYS) % 7
|
|
start = day_in * 24 + h_in # hour-of-week di inizio finestra
|
|
end = day_out * 24 + h_out # hour-of-week di fine finestra
|
|
|
|
def fn(df):
|
|
dt = pd.to_datetime(df["datetime"], utc=True) + pd.Timedelta(hours=1) # close time barra i
|
|
wh = (dt.dt.weekday * 24 + dt.dt.hour).values
|
|
if start < end:
|
|
ind = (wh >= start) & (wh < end)
|
|
else: # la finestra scavalca il confine di settimana
|
|
ind = (wh >= start) | (wh < end)
|
|
return direction * ind.astype(float)
|
|
|
|
return fn
|
|
|
|
|
|
def window_hours(h_in: int, h_out: int) -> int:
|
|
return (SPAN_DAYS * 24) - h_in + h_out # ven H_in -> lun H_out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Metriche per cella (unita' PRIMARIA = Sharpe su ritorni daily-compounded,
|
|
# stessa convenzione di candidate_daily/_to_daily del marginal scorer)
|
|
# ---------------------------------------------------------------------------
|
|
def cell_metrics(asset: str, h_in: int, h_out: int, direction: int,
|
|
day_in: int = FRI, fee_side: float = al.FEE_SIDE, keep: bool = False) -> dict:
|
|
df = al.get(asset, "1h")
|
|
fn = make_target(h_in, h_out, direction, day_in)
|
|
ev = al.eval_weights(df, fn(df), fee_side=fee_side)
|
|
s = pd.Series(ev["net"], index=ev["idx"])
|
|
d = al._to_daily(s)
|
|
ins = d[d.index < al.HOLDOUT]
|
|
hold = d[d.index >= al.HOLDOUT]
|
|
out = dict(asset=asset, h_in=h_in, h_out=h_out, dir=direction, day_in=day_in,
|
|
sh_is=al._sh(ins), sh_full=al._sh(d), sh_hold=al._sh(hold))
|
|
if keep:
|
|
out["ev"] = ev
|
|
out["daily"] = d
|
|
return out
|
|
|
|
|
|
def band(vals: list[float]) -> str:
|
|
v = np.asarray(vals, float)
|
|
return f"min {v.min():+.2f} / med {np.median(v):+.2f} / max {v.max():+.2f}"
|
|
|
|
|
|
def main() -> None:
|
|
print("=" * 88)
|
|
print("WK-DRIFT r0717 — esposizione weekend pura ven->lun, griglia di timing (dati 1h certificati)")
|
|
print("=" * 88)
|
|
|
|
# ================= 1) GRIGLIA COMPLETA (144 trial) =================
|
|
rows = []
|
|
for a in ASSETS:
|
|
for dr in DIRS:
|
|
for hi in H_GRID:
|
|
for ho in H_GRID:
|
|
rows.append(cell_metrics(a, hi, ho, dr))
|
|
n_trials = len(rows)
|
|
print(f"\n[1] FAMIGLIA: {len(H_GRID)}x{len(H_GRID)} timing x {len(DIRS)} direzioni x "
|
|
f"{len(ASSETS)} asset = {n_trials} trial (fee {2*al.FEE_SIDE*100:.2f}% RT)")
|
|
|
|
longs = [r for r in rows if r["dir"] == 1]
|
|
shorts = [r for r in rows if r["dir"] == -1]
|
|
print(f" BANDA FULL Sharpe (daily, netto) — tutte le 144 celle : {band([r['sh_full'] for r in rows])}")
|
|
print(f" BANDA HOLD Sharpe (2025+, netto) — tutte le 144 celle : {band([r['sh_hold'] for r in rows])}")
|
|
print(f" solo LONG (72): FULL {band([r['sh_full'] for r in longs])} | "
|
|
f"HOLD {band([r['sh_hold'] for r in longs])}")
|
|
print(f" solo SHORT (72): FULL {band([r['sh_full'] for r in shorts])} | "
|
|
f"HOLD {band([r['sh_hold'] for r in shorts])}")
|
|
n_pos_full = sum(1 for r in rows if r["sh_full"] > 0)
|
|
n_pos_both = sum(1 for r in rows if r["sh_full"] > 0 and r["sh_hold"] > 0)
|
|
print(f" celle FULL>0: {n_pos_full}/{n_trials} | celle FULL>0 E HOLD>0: {n_pos_both}/{n_trials}")
|
|
|
|
# ================= 2) SELEZIONE IN-SAMPLE-ONLY =================
|
|
chosen = max(rows, key=lambda r: r["sh_is"])
|
|
ch = cell_metrics(chosen["asset"], chosen["h_in"], chosen["h_out"], chosen["dir"], keep=True)
|
|
ev = ch["ev"]
|
|
dirname = "LONG" if ch["dir"] == 1 else "SHORT"
|
|
wh = window_hours(ch["h_in"], ch["h_out"])
|
|
print(f"\n[2] CELLA SELEZIONATA IN-SAMPLE (max Sharpe pre-2025 sui 144 trial):")
|
|
print(f" {dirname} {ch['asset']} ven {ch['h_in']:02d}:00 -> lun {ch['h_out']:02d}:00 UTC "
|
|
f"({wh}h esposte = {wh/168*100:.0f}% della settimana)")
|
|
print(f" Sharpe daily netto: IS {ch['sh_is']:+.2f} | FULL {ch['sh_full']:+.2f} | HOLD {ch['sh_hold']:+.2f}")
|
|
print(f" eval_weights (1h): FULL Sh {ev['full']['sharpe']:+.2f} CAGR {ev['full']['cagr']*100:+.1f}% "
|
|
f"maxDD {ev['full']['maxdd']*100:.1f}% ret {ev['full']['ret']*100:+.1f}%")
|
|
print(f" HOLD Sh {ev['holdout'].get('sharpe', 0):+.2f} "
|
|
f"ret {ev['holdout'].get('ret', 0)*100:+.1f}%")
|
|
print(f" time-in-market {ev['time_in_market']*100:.1f}% turnover/anno {ev['turnover_per_year']:.0f} "
|
|
f"=> fee drag ~{ev['turnover_per_year']*al.FEE_SIDE*100:.1f}%/anno a nozionale pieno")
|
|
top5 = sorted(rows, key=lambda r: -r["sh_is"])[:5]
|
|
print(" top-5 in-sample:")
|
|
for r in top5:
|
|
print(f" {'L' if r['dir']==1 else 'S'} {r['asset']} ven{r['h_in']:02d}->lun{r['h_out']:02d}: "
|
|
f"IS {r['sh_is']:+.2f} FULL {r['sh_full']:+.2f} HOLD {r['sh_hold']:+.2f}")
|
|
|
|
# ================= 3) DEFLATED SHARPE sull'intera famiglia =================
|
|
dsr, sr0 = al.deflated_sharpe(ch["sh_full"], [r["sh_full"] for r in rows], ch["daily"].values)
|
|
print(f"\n[3] DEFLATED SHARPE (famiglia {n_trials} trial): DSR = {dsr:.3f} "
|
|
f"(PASS>=0.95) | max Sharpe atteso sotto il null = {sr0:.2f} "
|
|
f"vs osservato {ch['sh_full']:+.2f}")
|
|
|
|
# ================= 4) NULL DI SPECIALITA' — ancore weekday =================
|
|
print(f"\n[4] NULL WEEKDAY — stessa finestra ({ch['h_in']:02d}->+3g {ch['h_out']:02d}, "
|
|
f"{dirname} {ch['asset']}) ancorata a ogni giorno:")
|
|
anchor = {}
|
|
for d0 in range(7):
|
|
m = cell_metrics(ch["asset"], ch["h_in"], ch["h_out"], ch["dir"], day_in=d0)
|
|
anchor[d0] = m
|
|
tag = " <== ven->lun" if d0 == FRI else ""
|
|
print(f" {DAY_NAMES[d0]}->{DAY_NAMES[(d0+3)%7]}: IS {m['sh_is']:+.2f} "
|
|
f"FULL {m['sh_full']:+.2f} HOLD {m['sh_hold']:+.2f}{tag}")
|
|
fri_is, fri_full = anchor[FRI]["sh_is"], anchor[FRI]["sh_full"]
|
|
pct_is = np.mean([anchor[d]["sh_is"] <= fri_is for d in range(7)])
|
|
pct_full = np.mean([anchor[d]["sh_full"] <= fri_full for d in range(7)])
|
|
print(f" percentile ancora-ven fra le 7: IS {pct_is*100:.0f}% FULL {pct_full*100:.0f}%")
|
|
|
|
# null a livello di FAMIGLIA: mediana IS su tutte le 36 celle timing (long, entrambi gli asset)
|
|
print(" null di famiglia (mediana Sharpe IS delle 72 celle LONG BTC+ETH, per ancora):")
|
|
fam = {}
|
|
for d0 in range(7):
|
|
vals = [cell_metrics(a, hi, ho, 1, day_in=d0)["sh_is"]
|
|
for a in ASSETS for hi in H_GRID for ho in H_GRID]
|
|
fam[d0] = float(np.median(vals))
|
|
for d0 in range(7):
|
|
tag = " <== ven" if d0 == FRI else ""
|
|
print(f" anchor {DAY_NAMES[d0]}: med IS {fam[d0]:+.2f}{tag}")
|
|
fam_pct = np.mean([fam[d] <= fam[FRI] for d in range(7)])
|
|
print(f" percentile famiglia ancora-ven: {fam_pct*100:.0f}%")
|
|
|
|
# confronto B&H a pari frazione di tempo: capture di log-return vs quota ore
|
|
print(" capture vs B&H (quota del log-return totale catturata dentro la finestra vs quota ore):")
|
|
for a in ASSETS:
|
|
df = al.get(a, "1h")
|
|
c = df["close"].values.astype(float)
|
|
lr = np.zeros(len(c)); lr[1:] = np.log(c[1:] / c[:-1])
|
|
dt = pd.to_datetime(df["datetime"], utc=True)
|
|
idx = pd.DatetimeIndex(dt)
|
|
ins_mask = np.asarray(idx < al.HOLDOUT)
|
|
bh = al._to_daily(pd.Series(al.simple_returns(c), index=idx))
|
|
for d0 in [FRI] + [d for d in range(7) if d != FRI]:
|
|
fn = make_target(ch["h_in"], ch["h_out"], 1, day_in=d0)
|
|
tgt = fn(df)
|
|
pos = np.zeros(len(tgt)); pos[1:] = tgt[:-1] # convenzione eval_weights
|
|
inw = pos > 0
|
|
t_share = inw.mean()
|
|
cap_full = lr[inw].sum() / lr.sum() if lr.sum() != 0 else np.nan
|
|
cap_is = (lr[inw & ins_mask].sum() / lr[ins_mask].sum()
|
|
if lr[ins_mask].sum() != 0 else np.nan)
|
|
if d0 == FRI:
|
|
print(f" {a}: B&H daily Sharpe FULL {al._sh(bh):+.2f} | finestra ven: "
|
|
f"ore {t_share*100:.0f}% capture FULL {cap_full*100:.0f}% IS {cap_is*100:.0f}%")
|
|
else:
|
|
print(f" anchor {DAY_NAMES[d0]}: ore {t_share*100:.0f}% "
|
|
f"capture FULL {cap_full*100:.0f}% IS {cap_is*100:.0f}%")
|
|
|
|
# ================= 5) DAY-BOUNDARY ROBUST sulla cella selezionata =================
|
|
fn_ch = make_target(ch["h_in"], ch["h_out"], ch["dir"])
|
|
dbr = al.day_boundary_robust(fn_ch, tf="1h")
|
|
print(f"\n[5] DAY_BOUNDARY_ROBUST (uplift marginale vs TP01 per offset del confine UTC):")
|
|
print(f" per_offset = {dbr['per_offset']}")
|
|
print(f" spread {dbr.get('spread')} verdetto = {dbr['verdict']}")
|
|
cok = al.causality_ok(fn_ch, tf="1h", assets=(ch["asset"],))
|
|
print(f" causality_ok: {cok['ok']} (max_tail_diff {cok['max_tail_diff']})")
|
|
|
|
# ================= 6) SWEEP FEE sulla cella selezionata =================
|
|
print(f"\n[6] SWEEP FEE (per-side) sulla cella selezionata:")
|
|
for f in al.FEE_SWEEP:
|
|
m = cell_metrics(ch["asset"], ch["h_in"], ch["h_out"], ch["dir"], fee_side=f, keep=True)
|
|
e = m["ev"]
|
|
print(f" fee {f*100:.2f}%/side ({2*f*100:.2f}% RT): Sh daily FULL {m['sh_full']:+.2f} "
|
|
f"HOLD {m['sh_hold']:+.2f} | ret FULL {e['full']['ret']*100:+.0f}% "
|
|
f"CAGR {e['full']['cagr']*100:+.1f}%")
|
|
|
|
# ================= 7) PER-ANNO della cella selezionata =================
|
|
print(f"\n[7] PER-ANNO (netto {2*al.FEE_SIDE*100:.2f}% RT) della cella selezionata:")
|
|
yl = ev["yearly"]
|
|
n_posy = sum(1 for v in yl.values() if v["ret"] > 0)
|
|
for y, v in yl.items():
|
|
print(f" {y}: ret {v['ret']*100:+6.1f}% dd {v['dd']*100:5.1f}%")
|
|
print(f" anni positivi: {n_posy}/{len(yl)}")
|
|
n_wk = len(ch["daily"]) / 7.02
|
|
arith = float(np.sum(ev["net"]))
|
|
print(f" media aritmetica per weekend (netto, full): {arith / max(n_wk, 1) * 100:+.3f}% "
|
|
f"su ~{n_wk:.0f} finestre")
|
|
|
|
print("\n" + "=" * 88)
|
|
print("FINE r0717_wk_drift — banda, selezione IS-only, DSR, null-weekday, boundary, fee, per-anno")
|
|
print("=" * 88)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|