live(monitor): prevday-breakout in FORWARD-MONITOR (paper, non deploy)

Il lead ortogonale a TP01 sopravvissuto all'onda intraday entra in forward-monitor (stesso
trattamento di XS01 STAT-MODE / STA05), NON in esecuzione reale.

- src/strategies/prevday_breakout.py: segnale CONGELATO (params fissi anchor=1, k=0.30, simmetrico,
  vol-target 0.20/30/2.0), self-contained. Bit-identico all'agent di ricerca (max diff 0.0):
  BTC full Sh 1.18/hold 0.92, ETH 1.09/1.42; marginal ADDS, earns_slot, corr_hold -0.01, non-hedge.
- scripts/live/paper_prevday.py: forward-only paper, traccia DUE libri — MODELED ($2000 continuo)
  e REAL-$600 (salta i ribilanciamenti < min-order $5) -> il gap = haircut di fill reale che lo
  scettico aveva segnalato. Inizializzato forward-only da oggi.
- cron_daily.sh: avanza il monitor ogni giorno.
- test: param congelati + causale + bounded + long-short. Suite intera verde.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-21 15:37:41 +00:00
parent d5dd6f4b72
commit 5cce7acfe1
5 changed files with 326 additions and 0 deletions
+107
View File
@@ -0,0 +1,107 @@
"""PREVDAY RANGE BREAKOUT — LEAD ortogonale a TP01, in FORWARD-MONITOR (NON deployato).
Stato (2026-06-21): unico segnale sopravvissuto alla verifica avversariale dell'onda intraday
(diario `2026-06-21-intraday-microstructure.md`). Lo scettico d'esecuzione l'ha chiamato "the only
honest candidate, real and conditionally executable". NON è in esecuzione reale: è un LEAD che
teniamo in paper forward-only per vedere se l'edge in-sample regge fuori campione VERO nei prossimi
mesi (stesso trattamento di XS01 STAT-MODE / STA05). Questo modulo CONGELA il segnale (parametri
fissi) così il track record forward è contro una strategia immutabile.
IL SEGNALE (1h, per-asset, posizione continua vol-targeted):
* Direzione +1 quando close[i] perfora in modo DECISIVO il MAX del giorno UTC precedente
(livello + buffer = k * range di ieri); -1 quando perfora il MIN; si CARRYA 24/7 tra i break
(stop-and-stay, non stop-and-flat -> turnover ~50/anno, non ~500). Long-short SIMMETRICO: la
gamba SHORT è ciò che decorrela da TP01 (TP01 è long-flat, dominato dal beta del toro).
* Vol-target TP01-style (causale, vol trailing) -> la size deriva con la vol.
ONESTÀ (giudice indurito 2026-06-21): abs PASS, marginal ADDS, earns_slot TRUE, NON-hedge,
multi-cut persistente, leak-free (causality_ok max_tail_diff 0), ROBUST allo shift del confine-giorno
(day_boundary_robust -> non è un artefatto di calendario come open_drive). corr a TP01 ~0.15 full /
~0 hold; Sharpe in-sample ~1.2; hold-out positivo su BTC ~0.9 e ETH ~1.4. CAVEAT: il lift dell'hold-out
della gamba short si appoggia ai regimi down/chop 2025-26; e a $600 il micro-ribilanciamento del
vol-target ha un haircut di fill reale (vedi eval_weights_smallcap). Per questo: FORWARD-MONITOR,
non deploy. Parametri scelti su un plateau (k 0.20-0.30; anchor=1 only).
CAUSALE: il max/min di ieri usa SOLO barre di giorni STRETTAMENTE precedenti (groupby giorno UTC ->
shift(1)); close[i] è confrontato col livello -> il break è noto a close[i]. Vol trailing. Nessun fit
full-sample. Self-contained (nessuna dipendenza da scripts/research).
"""
from __future__ import annotations
import numpy as np
import pandas as pd
# --- parametri CONGELATI (plateau-interior, vedi diario) -------------------------------
ANCHOR_DAYS = 1 # range = max/min del giorno UTC precedente (1 = "ieri")
BUFFER_K = 0.30 # buffer di break decisivo = k * range di ieri (plateau 0.20-0.30)
ALLOW_SHORT = True # libro SIMMETRICO: la gamba short è ciò che decorrela da TP01
TARGET_VOL = 0.20
VOL_WIN_DAYS = 30
LEV_CAP = 2.0
def _bars_per_day(df: pd.DataFrame) -> int:
dt = pd.to_datetime(df["datetime"]).diff().dt.total_seconds().median()
return max(1, round(86400 / dt)) if dt and dt > 0 else 24
def _vol_target(direction: np.ndarray, df: pd.DataFrame, target_vol: float,
vol_win_days: int, lev_cap: float) -> np.ndarray:
"""Scala una direzione in [-1,1] a posizione vol-targeted. Causale (vol trailing)."""
c = df["close"].values.astype(float)
bpd = _bars_per_day(df)
bpy = bpd * 365.25
r = np.zeros(len(c)); r[1:] = c[1:] / c[:-1] - 1.0
win = max(2, vol_win_days * bpd)
vol = pd.Series(r).rolling(win, min_periods=max(2, win // 2)).std().values * np.sqrt(bpy)
scal = np.where((vol > 0) & np.isfinite(vol), target_vol / vol, 0.0)
tgt = np.clip(direction * scal, -lev_cap, lev_cap)
tgt[~np.isfinite(tgt)] = 0.0
return tgt
def _prior_day_hilo(df: pd.DataFrame, anchor_days: int):
"""MAX(high)/MIN(low) dei `anchor_days` giorni UTC PRECEDENTI, allineati a ogni barra,
noti causalmente (groupby giorno -> rolling -> shift(1) = strettamente < oggi)."""
dt = pd.to_datetime(df["datetime"], utc=True)
day = dt.dt.floor("1D")
g = pd.DataFrame({"day": day.values,
"high": df["high"].values.astype(float),
"low": df["low"].values.astype(float)})
per_day = g.groupby("day").agg(dh=("high", "max"), dl=("low", "min"))
dh = per_day["dh"].rolling(anchor_days, min_periods=1).max().shift(1)
dl = per_day["dl"].rolling(anchor_days, min_periods=1).min().shift(1)
mapped = pd.DataFrame({"dh": dh, "dl": dl}).reindex(g["day"].values)
return mapped["dh"].values, mapped["dl"].values
def _breakout_direction(df: pd.DataFrame, anchor_days: int, buffer_k: float,
allow_short: bool) -> np.ndarray:
c = df["close"].values.astype(float)
pdh, pdl = _prior_day_hilo(df, anchor_days)
rng = pdh - pdl
up_lvl = pdh + buffer_k * rng
dn_lvl = pdl - buffer_k * rng
n = len(c)
dirn = np.zeros(n)
cur = 0.0
low_state = -1.0 if allow_short else 0.0
for i in range(n):
if np.isfinite(up_lvl[i]) and c[i] > up_lvl[i]:
cur = 1.0
elif np.isfinite(dn_lvl[i]) and c[i] < dn_lvl[i]:
cur = low_state
dirn[i] = cur
return dirn
def target(df: pd.DataFrame) -> np.ndarray:
"""Posizione continua per-asset (1h) in [-LEV_CAP, LEV_CAP], decisa a close[i]."""
direction = _breakout_direction(df, ANCHOR_DAYS, BUFFER_K, ALLOW_SHORT)
pos = _vol_target(direction, df, TARGET_VOL, VOL_WIN_DAYS, LEV_CAP)
return np.nan_to_num(pos, nan=0.0)
def current_target(df: pd.DataFrame) -> float:
"""Posizione-bersaglio sull'ultima barra (decisa con dati <= ultima barra chiusa)."""
return float(target(df)[-1])