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

research(weekend): ondata ven->lun SCARTATA 4/4 — famiglia calendario SATURA; weekend = 38% del gross TP01 (mai de-esporre)

Browse Source 
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>
This commit is contained in:
Adriano Dal Pastro
2026-07-18 15:03:44 +00:00
parent 67f7b89e4c
commit 7534b08be0
7 changed files with 1584 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/r0717_wk_cond.py
+334
View File
@@ -0,0 +1,334 @@
"""r0717_wk_cond — WK-COND: esposizione weekend (ven->lun) CONDIZIONATA da gate causali.
DOMANDA: esiste un gate causale (noto al venerdi' H_in) che rende la finestra weekend un
edge, dove il drift nudo non lo e' (day-of-week nudo gia' morto, SEA ~0)?
METODO (onesto, vettoriale su 1h, harness altlib):
* Finestra: ven H_in -> lun H_out, H_in in {12,16,20}, H_out in {0,8,16} (9 timing).
* Convenzione eval_weights: target[i] deciso con dati <= close[i], tenuto nella barra
i+1. Le barre 1h sono open-labeled -> il decision time della barra i e' datetime[i]+1h.
Barra d'ingresso = quella il cui CLOSE cade a ven H_in:00 UTC (decisione con dati fino
a ven H_in:00 esatte, nessun leak). Ultima barra target = close lun H_out-1h -> l'
esposizione reale copre esattamente [ven H_in, lun H_out).
* Gate (tutti causali, calcolati al close della barra d'ingresso):
A TSMOM multi-orizzonte 30/90/180g (LO long-se-up / SO short-se-down / LS entrambi)
B FOLLOW del venerdi' (open ven 00:00 -> close H_in), + FADE come controllo
C ritorno settimana (lun 00:00 -> ven H_in), follow + fade
D regime vol: percentile ESPANDENTE causale della RV30g (LOW: pct<=0.30 / HIGH: >=0.70)
E prevday level: close H_in vs high/low del GIOVEDI' -> breakout-follow L/S
F NAKED long (riferimento: il drift nudo NON e' un edge)
* Ogni cella = (gate-variante x timing); valutata su BTC ed ETH e come 50/50 daily
(convenzione candidate_daily). TUTTI i trial contano per il deflated_sharpe.
* Selezione cella SOLO in-sample (Sharpe 50/50 daily pre-2025), poi hold-out + banda
dei 9 timing (lezione anchor timing-luck 2026-07-02).
* Per ogni gate: study_marginal vs TP01 (corr/earns_slot/is_hedge), fee sweep, per-anno,
day_boundary_robust (il segnale E' calendario-dipendente), causality_ok.
TRIAL: 3+2+2+2+1+1 = 11 varianti x 9 timing = 99 combo (x2 asset = 198 stream valutati).
VINCOLI: nessun file esistente modificato; niente rete (dati locali certificati).
"""
from __future__ import annotations
import sys
import numpy as np
import pandas as pd
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al # noqa: E402
H_INS = (12, 16, 20) # venerdi' UTC, ora di ingresso
H_OUTS = (0, 8, 16) # lunedi' UTC, ora di uscita (0 = mezzanotte dom->lun)
ASSETS = tuple(al.CERTIFIED)
HOLDOUT = al.HOLDOUT
# ===========================================================================
# SEGNALI CAUSALI — memo per-df (chiave su timestamp: regge shift di calendario
# di day_boundary_robust e troncamenti di causality_ok senza contaminazioni).
# Tutti i valori all'indice i usano SOLO dati <= close[i].
# ===========================================================================
_SIG_MEMO: dict = {}
def get_signals(df: pd.DataFrame) -> dict:
key = (int(df["timestamp"].iloc[0]), int(df["timestamp"].iloc[-1]), len(df))
if key in _SIG_MEMO:
return _SIG_MEMO[key]
c = df["close"].values.astype(float)
o = df["open"].values.astype(float)
h = df["high"].values.astype(float)
lo_ = df["low"].values.astype(float)
dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
close_dt = dt + pd.Timedelta(hours=1) # decision time della barra i
# (A) TSMOM multi-orizzonte 30/90/180g su close 1h (voto di segni, come TP01)
votes = np.zeros(len(c))
for hd in (30, 90, 180):
hb = hd * 24
s = np.full(len(c), np.nan)
if len(c) > hb:
s[hb:] = np.sign(c[hb:] / c[:-hb] - 1.0)
votes += np.nan_to_num(s)
tsmom = np.sign(votes) # -1 / 0 / +1
# (B) direzione intraday del giorno corrente: close[i] vs OPEN del giorno UTC
day = np.asarray(dt.normalize())
day_open = pd.Series(o).groupby(day).transform("first").values
intraday = np.sign(c / day_open - 1.0)
# (C) direzione della settimana: close[i] vs OPEN del lunedi' 00:00 della settimana
week = np.asarray((dt - pd.to_timedelta(dt.dayofweek, unit="D")).normalize())
week_open = pd.Series(o).groupby(week).transform("first").values
week_dir = np.sign(c / week_open - 1.0)
# (D) percentile espandente CAUSALE della RV 30g (rank del valore corrente vs storia)
r = al.simple_returns(c)
rv = al.realized_vol(r, 30 * 24, 24 * 365.25)
volpct = pd.Series(rv).expanding(min_periods=30 * 24).rank(pct=True).values
# (E) breakout vs high/low del giorno PRECEDENTE completo (per ven = giovedi')
dhi = pd.Series(h).groupby(day).max()
dlo = pd.Series(lo_).groupby(day).min()
prev_hi = dhi.shift(1).reindex(day).values
prev_lo = dlo.shift(1).reindex(day).values
brk = np.where(c > prev_hi, 1.0, np.where(c < prev_lo, -1.0, 0.0))
brk[~np.isfinite(prev_hi)] = 0.0
S = dict(tsmom=tsmom, intraday=intraday, week=week_dir, volpct=volpct, brk=brk,
cdw=close_dt.dayofweek.values, chr=close_dt.hour.values)
if len(_SIG_MEMO) > 48:
_SIG_MEMO.clear()
_SIG_MEMO[key] = S
return S
# ===========================================================================
# GATE (variante -> direzione al momento dell'ingresso, da segnali causali)
# ===========================================================================
GATE_FAMILIES: dict[str, dict] = {
"A-TSMOM": {
"LO": lambda S: np.clip(S["tsmom"], 0.0, 1.0), # long solo se trend up
"SO": lambda S: np.clip(S["tsmom"], -1.0, 0.0), # short solo se trend down
"LS": lambda S: S["tsmom"], # entrambi
},
"B-FRIDIR": {
"FOLLOW": lambda S: S["intraday"],
"FADE": lambda S: -S["intraday"],
},
"C-WEEKDIR": {
"FOLLOW": lambda S: S["week"],
"FADE": lambda S: -S["week"],
},
"D-VOLREG": {
"LOW30": lambda S: np.where(S["volpct"] <= 0.30, 1.0, 0.0),
"HIGH70": lambda S: np.where(S["volpct"] >= 0.70, 1.0, 0.0),
},
"E-PREVDAY": {
"BRKFOLLOW": lambda S: S["brk"],
},
"F-NAKED": {
"LONG": lambda S: np.ones(len(S["tsmom"])),
},
}
def make_target(fam: str, var: str, hin: int, hout: int):
"""target_fn(df) -> posizione per barra. Direzione congelata alla barra d'ingresso
(close = ven H_in) e tenuta su tutta la finestra [ven H_in, lun H_out)."""
gate_fn = GATE_FAMILIES[fam][var]
def target_fn(df: pd.DataFrame) -> np.ndarray:
S = get_signals(df)
dirs = np.nan_to_num(np.asarray(gate_fn(S), float))
dw, hr = S["cdw"], S["chr"]
# barre target = quelle il cui CLOSE cade in [ven H_in, lun H_out)
in_win = ((dw == 4) & (hr >= hin)) | (dw == 5) | (dw == 6)
if hout > 0:
in_win |= (dw == 0) & (hr < hout)
entry = (dw == 4) & (hr == hin)
sig = pd.Series(np.where(entry, dirs, np.nan)).ffill().values
return np.where(in_win, np.nan_to_num(sig), 0.0)
target_fn.__name__ = f"wk_{fam}_{var}_F{hin}_M{hout}"
return target_fn
# ===========================================================================
# VALUTAZIONE DI UNA CELLA (per-asset 1h + 50/50 daily, split in/hold)
# ===========================================================================
def combined_daily(target_fn, fee_side: float = al.FEE_SIDE) -> pd.Series:
series = {}
for a in ASSETS:
df = al.get(a, "1h")
ev = al.eval_weights(df, target_fn(df), fee_side=fee_side)
series[a] = pd.Series(ev["net"], index=ev["idx"])
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
return al._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]])
def run_cell(fam: str, var: str, hin: int, hout: int) -> dict:
tfn = make_target(fam, var, hin, hout)
per_asset = {}
for a in ASSETS:
df = al.get(a, "1h")
ev = al.eval_weights(df, tfn(df))
per_asset[a] = dict(full_sh=ev["full"]["sharpe"], hold_sh=ev["holdout"].get("sharpe", 0.0),
dd=ev["full"]["maxdd"], tim=ev["time_in_market"],
turnover=ev["turnover_per_year"], yearly=ev["yearly"])
D = combined_daily(tfn)
ins, hold = D[D.index < HOLDOUT], D[D.index >= HOLDOUT]
eqf = np.cumprod(1 + D.values)
pk = np.maximum.accumulate(eqf)
return dict(fam=fam, var=var, hin=hin, hout=hout, target_fn=tfn, daily=D,
per_asset=per_asset,
is_sh=round(al._sh(ins), 3), full_sh=round(al._sh(D), 3),
hold_sh=round(al._sh(hold), 3),
dd_comb=round(float(np.max((pk - eqf) / pk)), 4),
ret_hold=round(float(np.prod(1 + hold.values) - 1), 4))
def fee_sweep_cell(cell: dict) -> dict:
out = {}
for f in al.FEE_SWEEP:
D = combined_daily(cell["target_fn"], fee_side=f)
ins, hold = D[D.index < HOLDOUT], D[D.index >= HOLDOUT]
out[f"{2 * f * 100:.2f}%RT"] = dict(is_sh=round(al._sh(ins), 2),
full=round(al._sh(D), 2),
hold=round(al._sh(hold), 2))
return out
def yearly_comb(cell: dict) -> dict:
D = cell["daily"]
out = {}
for y, g in D.groupby(D.index.year):
out[int(y)] = dict(ret=round(float(np.prod(1 + g.values) - 1), 4),
sh=round(al._sh(g), 2))
return out
def gate_selectivity(cell: dict) -> dict:
"""Quota di venerdi' in cui il gate apre una posizione (in-sample / hold-out), BTC."""
df = al.get("BTC", "1h")
S = get_signals(df)
dirs = np.nan_to_num(np.asarray(GATE_FAMILIES[cell["fam"]][cell["var"]](S), float))
entry = (S["cdw"] == 4) & (S["chr"] == cell["hin"])
dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))
m_ins = entry & np.asarray(dt < HOLDOUT)
m_hold = entry & np.asarray(dt >= HOLDOUT)
def frac(m):
return round(float(np.mean(dirs[m] != 0)), 3) if m.sum() else None
def longshare(m):
on = dirs[m][dirs[m] != 0]
return round(float(np.mean(on > 0)), 3) if len(on) else None
return dict(n_fridays=int(entry.sum()), on_ins=frac(m_ins), on_hold=frac(m_hold),
long_share_ins=longshare(m_ins))
# ===========================================================================
# MAIN
# ===========================================================================
def main() -> None:
print("=" * 100)
print("WK-COND — weekend ven->lun condizionato da gate causali | 1h BTC/ETH certificati")
print(f"timing: H_in {H_INS} x H_out {H_OUTS} | fee {2 * al.FEE_SIDE * 100:.2f}%RT | HOLDOUT {HOLDOUT.date()}")
print("=" * 100)
# ---- sweep completo -----------------------------------------------------------
rows = []
for fam, variants in GATE_FAMILIES.items():
for var in variants:
for hin in H_INS:
for hout in H_OUTS:
rows.append(run_cell(fam, var, hin, hout))
all_full = [r["full_sh"] for r in rows]
n_combo = len(rows)
print(f"\nTRIAL: {n_combo} combo (gate-variante x timing) x {len(ASSETS)} asset = "
f"{n_combo * len(ASSETS)} stream valutati. DSR calcolato vs tutte le {n_combo} combo 50/50.")
# ---- tabella compatta di famiglia (mediane per variante) ----------------------
print("\n--- PANORAMICA per variante (mediana e banda sui 9 timing, 50/50 daily) ---")
hdr = f"{'variante':<22}{'IS med [min,max]':>24}{'FULL med':>10}{'HOLD med [min,max]':>26}"
print(hdr)
for fam, variants in GATE_FAMILIES.items():
for var in variants:
sub = [r for r in rows if r["fam"] == fam and r["var"] == var]
iss = [r["is_sh"] for r in sub]; hs = [r["hold_sh"] for r in sub]
fs = [r["full_sh"] for r in sub]
print(f"{fam + '/' + var:<22}"
f"{np.median(iss):>8.2f} [{min(iss):+.2f},{max(iss):+.2f}]"
f"{np.median(fs):>10.2f}"
f"{np.median(hs):>12.2f} [{min(hs):+.2f},{max(hs):+.2f}]")
# ---- selezione IN-SAMPLE-ONLY per gate + verifica cella -----------------------
print("\n" + "=" * 100)
print("SELEZIONE IN-SAMPLE-ONLY (max Sharpe 50/50 daily pre-2025) + verifica per gate")
print("=" * 100)
summary = {}
for fam in GATE_FAMILIES:
sub = [r for r in rows if r["fam"] == fam]
best = max(sub, key=lambda r: r["is_sh"])
# banda dei 9 timing per la variante scelta (lezione anchor timing-luck)
band = sorted(r["hold_sh"] for r in sub if r["var"] == best["var"])
band_is = sorted(r["is_sh"] for r in sub if r["var"] == best["var"])
pctl = float(np.mean([b <= best["hold_sh"] for b in band]))
dsr, sr0 = al.deflated_sharpe(al._sh(best["daily"]), all_full, best["daily"].values)
name = f"{fam}/{best['var']} F{best['hin']}->M{best['hout']}"
print(f"\n### {name} (cella scelta in-sample su {len(sub)} trial del gate)")
print(f" 50/50: IS {best['is_sh']:+.2f} | FULL {best['full_sh']:+.2f} | "
f"HOLD {best['hold_sh']:+.2f} (ret hold {best['ret_hold'] * 100:+.1f}%) | DD {best['dd_comb'] * 100:.1f}%")
pa = best["per_asset"]
for a in ASSETS:
print(f" {a}: full {pa[a]['full_sh']:+.2f} hold {pa[a]['hold_sh']:+.2f} "
f"DD {pa[a]['dd']*100:.0f}% tim {pa[a]['tim']*100:.0f}% turn/y {pa[a]['turnover']:.0f}")
print(f" banda 9 timing (var {best['var']}): HOLD [{band[0]:+.2f} .. med {np.median(band):+.2f} .. {band[-1]:+.2f}] "
f"(cella scelta al {pctl * 100:.0f}° pctl) | IS [{band_is[0]:+.2f}..{band_is[-1]:+.2f}]")
print(f" deflated Sharpe vs {n_combo} trial: DSR={dsr:.3f} (null max atteso {sr0:.2f}) "
f"{'PASS' if dsr >= 0.95 else 'FAIL'}")
sel = gate_selectivity(best)
print(f" selettivita' (BTC, ven {best['hin']}h): on IS {sel['on_ins']} / HOLD {sel['on_hold']} "
f"(quota long IS {sel['long_share_ins']}) su {sel['n_fridays']} venerdi'")
cok = al.causality_ok(best["target_fn"], tf="1h", tail=400)
print(f" causality_ok: {cok['ok']} (max tail diff {cok['max_tail_diff']}, checked {cok['checked']})")
print(f" fee sweep 50/50: " + " ".join(
f"{k}: IS {v['is_sh']:+.2f}/FULL {v['full']:+.2f}/HOLD {v['hold']:+.2f}"
for k, v in fee_sweep_cell(best).items()))
yr = yearly_comb(best)
print(" per-anno 50/50: " + " ".join(f"{y}:{d['ret']*100:+.1f}%({d['sh']:+.1f})" for y, d in yr.items()))
summary[fam] = dict(best=best, dsr=dsr, band=band, name=name,
survivor=(best["is_sh"] >= 0.5))
# ---- marginale vs TP01 + day-boundary per ogni gate a-e -----------------------
print("\n" + "=" * 100)
print("MARGINALE vs TP01 (study_marginal, 1h) + day_boundary_robust — gate A..E")
print("=" * 100)
for fam, info in summary.items():
if fam == "F-NAKED":
continue
best = info["best"]
sm = al.study_marginal(f"WK-{info['name']}", best["target_fn"], tf="1h")
print("\n" + al.fmt_marginal(sm))
dbr = al.day_boundary_robust(best["target_fn"], tf="1h")
print(f" day_boundary: {dbr['verdict']} (per_offset {dbr['per_offset']}, spread {dbr.get('spread')})")
info["sm"] = sm
info["dbr"] = dbr
# ---- sintesi finale ------------------------------------------------------------
print("\n" + "=" * 100)
print("SINTESI")
print("=" * 100)
for fam, info in summary.items():
b = info["best"]
line = (f"{info['name']:<38} IS {b['is_sh']:+.2f} FULL {b['full_sh']:+.2f} "
f"HOLD {b['hold_sh']:+.2f} DSR {info['dsr']:.2f} survivor_IS={info['survivor']}")
if "sm" in info:
m = info["sm"]
line += (f" | marg={m['marginal_verdict']} earns_slot={m['earns_slot']} "
f"corr {m['marginal'].get('corr_full')} hedge={m['marginal'].get('is_hedge')}"
f" | boundary={info['dbr']['verdict']}")
print(line)
if __name__ == "__main__":
main()