PythagorasGoal/scripts/live/paper_prevday.py

"""FORWARD-MONITOR — PREVDAY RANGE BREAKOUT (lead ortogonale a TP01), forward-only, PAPER.

NON è esecuzione reale. È il monitoraggio forward-only del LEAD validato dall'onda intraday
(src/strategies/prevday_breakout.py, parametri CONGELATI) per vedere se l'edge in-sample regge
FUORI CAMPIONE VERO nei prossimi mesi. Stesso trattamento di XS01 STAT-MODE / STA05.

DESIGN (onesto):
  - Legge i parquet certificati BTC/ETH 1h (data/raw). Segnale a 1h, libro 50/50.
  - Alla prima esecuzione parte dall'ultima barra 1h CHIUSA (forward-only: lo storico NON entra
    nel PnL di paper, si traccia solo da ora in avanti).
  - Ogni run processa le NUOVE barre 1h chiuse: applica il rendimento della posizione tenuta,
    addebita le fee sul turnover, registra i flip di segno, poi ricalcola la posizione-bersaglio.
  - Traccia DUE libri in parallelo per onestà sull'esecuzione (lo scettico ha segnalato che a $600
    il micro-ribilanciamento del vol-target ha un haircut di fill):
      * MODELED  : capitale nominale $2000, ribilanciamento continuo (fee proporzionale su ogni |Δ|).
      * REAL-$600: capitale reale $600, salta i ribilanciamenti di nozionale < min_order ($5) —
        cosa che il conto vero catturerebbe davvero. Il gap MODELED-REAL = l'haircut di fill reale.
  - Per barre fresche, aggiornare prima i dati:
        uv run python scripts/analysis/rebuild_history.py --asset BTC ETH

Stato: data/paper_prevday/{state.json, trades.jsonl, returns.jsonl} (append-only).

    uv run python scripts/live/paper_prevday.py            # avanza col dato disponibile
    uv run python scripts/live/paper_prevday.py --status    # solo stato, non avanza
    uv run python scripts/live/paper_prevday.py --reset      # azzera (riparte da ora)
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.backtest.harness import load                                   # noqa: E402
from src.live import paper_guard as PG                                  # noqa: E402
from src.strategies.prevday_breakout import target as prevday_target    # noqa: E402
from src.strategies import prevday_breakout as pb                       # noqa: E402

STATE_DIR = PROJECT_ROOT / "data" / "paper_prevday"
STATE_FILE = STATE_DIR / "state.json"
TRADES_FILE = STATE_DIR / "trades.jsonl"
RETURNS_FILE = STATE_DIR / "returns.jsonl"
ASSETS = ["BTC", "ETH"]
WEIGHT = 0.5
FEE_SIDE = 0.0005              # 0.05%/side = 0.10% round-trip (Deribit taker)
MODELED_CAPITAL = 2000.0       # nominale, ribilanciamento continuo
REAL_CAPITAL = 600.0           # capitale mainnet reale
MIN_ORDER = 5.0                # min order Deribit -> sotto, il conto vero NON ribilancia

# --- GATE PRE-REGISTRATO `GATE PREVDAY-01` (2026-08-23, RESULTS-0822 §54) -----------------------
# Cablato qui il 2026-09-10 (issue #4): esisteva solo nel testo, e la revisione del 09/09 lo ha
# letto come "gate senza data". Decisione dell'operatore: tenere la data, cablare kill e veto.
GATE_DATE = "2027-06-21"       # decisione piena: 7 condizioni (a)-(g), tutte necessarie
KILL_SHARPE = -0.50            # kill: Sharpe forward < -0,50 ...
KILL_MIN_ACTIVE_DAYS = 180     # ... su >= 180 giorni di barre ATTIVE (posizione != 0)
MIN_RECON_FRAC = 0.80          # veto d'integrita': >= 80% di barre ricostruibili dal feed
RECON_TOL = 1.5e-6             # net_modeled e' registrato a 6 decimali: mezzo ulp di tolleranza
RECENTI_GIORNI = 30            # "i minuti non registrati non devono crescere": ultimi 30g vs tutto
MIN_DIV_CRESCITA = 5           # ... ma UNA barra divergente su 1943 non e' una crescita (P14): la
                               # crescita richiede >= 5 divergenti recenti E tasso doppio del totale
# LENTE: il kill si misura sullo Sharpe GIORNALIERO (somma delle barre orarie per giorno UTC),
# la lente con cui e' misurato ogni altro sleeve del progetto — §54: sulla lente ORARIA lo stesso
# forward vale +2,04 contro +1,56 giornaliero. Al kill la famiglia NON si ri-ottimizza.


def build_bars() -> dict[str, pd.DataFrame]:
    return {a: load(a, "1h").reset_index(drop=True) for a in ASSETS}


def _state_io(write: dict | None = None):
    if write is not None:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        STATE_FILE.write_text(json.dumps(write, indent=2))
        return write
    return json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else None


def _append(path: Path, rec: dict):
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    with open(path, "a") as f:
        f.write(json.dumps(rec) + "\n")


def init_state(dfs) -> dict:
    last_ts = min(int(dfs[a]["timestamp"].iloc[-1]) for a in ASSETS)
    pos = {a: pb.current_target(dfs[a][dfs[a]["timestamp"] <= last_ts]) for a in ASSETS}
    return dict(
        start_ts=last_ts, last_ts=last_ts, n_bars=0,
        pos_modeled=pos, pos_real=dict(pos),
        cap_modeled=MODELED_CAPITAL, cap_real=REAL_CAPITAL,
        peak_modeled=MODELED_CAPITAL, peak_real=REAL_CAPITAL,
        dd_modeled=0.0, dd_real=0.0, n_trades=0,
    )


def advance(st: dict, dfs: dict) -> dict:
    data = {}
    for a in ASSETS:
        df = dfs[a]
        c = df["close"].values.astype(float)
        r = np.zeros(len(c)); r[1:] = c[1:] / c[:-1] - 1.0
        data[a] = dict(ts=df["timestamp"].values.astype("int64"),
                       dt=pd.to_datetime(df["datetime"]).values, r=r,
                       tgt=prevday_target(df))
    common = sorted(set(data["BTC"]["ts"]).intersection(data["ETH"]["ts"]))
    # SOLO barre 1h CHIUSE: il parquet 1h puo' contenere l'ora in corso (1 barra su 24 —
    # r0822_monitor_audit); la si lascia al giro successivo
    new_ts = [t for t in common
              if t > st["last_ts"] and PG.chiusa(int(t), PG.MS_1H)]
    if not new_ts:
        return st
    idx = {a: {int(t): i for i, t in enumerate(data[a]["ts"])} for a in ASSETS}
    pm, pr = dict(st["pos_modeled"]), dict(st["pos_real"])
    cm, cr = st["cap_modeled"], st["cap_real"]
    pkm, pkr = st["peak_modeled"], st["peak_real"]
    ddm, ddr = st["dd_modeled"], st["dd_real"]
    ntr = st.get("n_trades", 0)

    for t in new_ts:
        net_m = net_r = 0.0
        nm, nr = {}, {}
        for a in ASSETS:
            i = idx[a][int(t)]
            r = float(data[a]["r"][i]); tgt = float(data[a]["tgt"][i])
            # MODELED: continuous rebalance
            hm = pm[a]
            net_m += WEIGHT * (hm * r - FEE_SIDE * abs(tgt - hm))
            nm[a] = tgt
            if np.sign(tgt) != np.sign(hm):
                _append(TRADES_FILE, dict(ts=int(t), dt=str(pd.Timestamp(data[a]["dt"][i])),
                        asset=a, action="ENTRY" if tgt != 0 else "EXIT",
                        from_pos=round(hm, 4), to_pos=round(tgt, 4)))
                ntr += 1
            # REAL-$600: skip sub-min_order rebalances
            hr = pr[a]
            leg_cap = cr * WEIGHT
            executed = abs(tgt - hr) * leg_cap >= MIN_ORDER
            new_hr = tgt if executed else hr
            net_r += WEIGHT * (hr * r - FEE_SIDE * abs(new_hr - hr))
            nr[a] = new_hr
        cm *= (1.0 + max(net_m, -0.99)); cr *= (1.0 + max(net_r, -0.99))
        pkm = max(pkm, cm); pkr = max(pkr, cr)
        ddm = max(ddm, (pkm - cm) / pkm if pkm > 0 else 0.0)
        ddr = max(ddr, (pkr - cr) / pkr if pkr > 0 else 0.0)
        pm, pr = nm, nr
        _append(RETURNS_FILE, dict(ts=int(t), dt=str(pd.Timestamp(data["BTC"]["dt"][idx["BTC"][int(t)]])),
                net_modeled=round(net_m, 6), net_real=round(net_r, 6),
                pos_btc=round(pr["BTC"], 4), pos_eth=round(pr["ETH"], 4),
                cap_modeled=round(cm, 2), cap_real=round(cr, 2)))

    st.update(last_ts=int(new_ts[-1]), n_bars=st.get("n_bars", 0) + len(new_ts),
              pos_modeled=pm, pos_real=pr, cap_modeled=cm, cap_real=cr,
              peak_modeled=pkm, peak_real=pkr, dd_modeled=ddm, dd_real=ddr, n_trades=ntr)
    return st


def _returns_df() -> pd.DataFrame | None:
    if not RETURNS_FILE.exists():
        return None
    r = pd.read_json(RETURNS_FILE, lines=True)
    if r.empty:
        return None
    r["dt"] = pd.to_datetime(r["ts"], unit="ms", utc=True)
    return r


def sharpe_giornaliero(r: pd.DataFrame, col: str = "net_modeled") -> tuple[float, int]:
    """PURA. Sharpe annualizzato sui rendimenti GIORNALIERI (somma delle barre orarie per giorno
    UTC) -> (sharpe, n_giorni). nan sotto 30 giorni o a varianza nulla."""
    d = r.groupby(r["dt"].dt.floor("D"))[col].sum()
    if len(d) < 30 or d.std() == 0:
        return float("nan"), int(len(d))
    return float(d.mean() / d.std() * np.sqrt(365.25)), int(len(d))


def giorni_attivi(r: pd.DataFrame) -> int:
    """PURA. Giorni UTC con almeno una barra a posizione != 0 (il kill conta le barre ATTIVE).
    ⚠️ le posizioni registrate sono quelle del libro REAL-$600 (advance scrive `pr`), lo Sharpe
    del kill e' MODELED: divergono solo se un ingresso da 0 vale < $5 di nozionale, cosa che a
    vol-target 20% non succede (10/09: 81/81 giorni). Dichiarato, non riparato (D5)."""
    att = r[(r["pos_btc"] != 0) | (r["pos_eth"] != 0)]
    return int(att["dt"].dt.floor("D").nunique())


def kill_check(sharpe: float, n_attivi: int, today: pd.Timestamp) -> tuple[str, str]:
    """PURA. -> (stato, motivo). KILL solo con >= 180 giorni attivi E Sharpe < -0,50 (§54):
    a quell'orizzonte P(uccidere un edge vivo a Sharpe 1,2) = 9,7%."""
    if n_attivi < KILL_MIN_ACTIVE_DAYS:
        return "NON_MATURO", (f"{n_attivi} giorni attivi < {KILL_MIN_ACTIVE_DAYS}: il kill non e' "
                              f"leggibile (mancano {KILL_MIN_ACTIVE_DAYS - n_attivi} giorni)")
    if not np.isfinite(sharpe):
        return "NON_MISURABILE", "Sharpe forward non misurabile"
    if sharpe < KILL_SHARPE:
        return "KILL", f"Sharpe {sharpe:+.2f} < {KILL_SHARPE:+.2f} su {n_attivi} giorni attivi -> RITIRARE, senza ri-ottimizzare"
    return "SUPERATO", f"Sharpe {sharpe:+.2f} >= {KILL_SHARPE:+.2f} su {n_attivi} giorni attivi"


def ricostruzione(r: pd.DataFrame, dfs: dict, start_ts: int) -> dict:
    """Rigioca la strategia CONGELATA sul feed di OGGI dallo `start_ts` del monitor e confronta
    barra per barra `net_modeled` registrato con quello ricalcolato (ribilanciamento continuo,
    la stessa aritmetica di `advance`). Una barra e' RICOSTRUIBILE se coincide entro RECON_TOL.

    Cosa misura: quante barre registrate sono riproducibili dal dato certificato. Le divergenti
    sono barre calcolate su un feed che il rebuild ha poi rivisto (§54: 67 su 1512, tutte
    all'ora del cron) — e' la misura del veto d'integrita' (>= 80%, e i minuti persi non devono
    crescere: qui, divergenti/giorno negli ultimi 30g contro l'intera finestra)."""
    data = {}
    for a in ASSETS:
        df = dfs[a]
        c = df["close"].values.astype(float)
        rr = np.zeros(len(c)); rr[1:] = c[1:] / c[:-1] - 1.0
        data[a] = dict(ts=df["timestamp"].values.astype("int64"), r=rr, tgt=prevday_target(df))
    idx = {a: {int(t): i for i, t in enumerate(data[a]["ts"])} for a in ASSETS}
    # posizione iniziale come init_state: target all'ultima barra <= start_ts
    pos = {}
    for a in ASSETS:
        i0 = max(i for t, i in idx[a].items() if t <= start_ts)
        pos[a] = float(data[a]["tgt"][i0])
    ricalc, presenti = [], []
    for t in r["ts"].astype("int64"):
        t = int(t)
        if any(t not in idx[a] for a in ASSETS):
            ricalc.append(np.nan); presenti.append(False); continue
        net = 0.0
        for a in ASSETS:
            i = idx[a][t]
            tgt = float(data[a]["tgt"][i])
            net += WEIGHT * (pos[a] * float(data[a]["r"][i]) - FEE_SIDE * abs(tgt - pos[a]))
            pos[a] = tgt
        ricalc.append(net); presenti.append(True)
    ricalc = np.asarray(ricalc, dtype=float)
    reg = r["net_modeled"].values.astype(float)
    ok = np.isfinite(ricalc) & (np.abs(ricalc - reg) <= RECON_TOL)
    n = len(reg)
    giorni = r["dt"].dt.floor("D")
    span_g = max(1, int(giorni.nunique()))
    recenti = r["dt"] >= (r["dt"].max() - pd.Timedelta(days=RECENTI_GIORNI))
    div_rec = int((~ok[recenti.values]).sum())
    g_rec = max(1, int(giorni[recenti].nunique()))
    return dict(n=n, ricostruibili=int(ok.sum()), frac=float(ok.sum() / n) if n else float("nan"),
                assenti_nel_feed=int((~np.asarray(presenti)).sum()),
                div_per_giorno=float((n - ok.sum()) / span_g),
                div_recenti=div_rec, div_per_giorno_recenti=float(div_rec / g_rec),
                divergenti_ts=[int(t) for t in r["ts"].values[~ok]][:20])


def veto_check(frac: float, div_g_tot: float, div_g_rec: float, div_rec: int) -> tuple[bool, str]:
    """PURA. Veto d'integrita' (blocca, non decide): sotto l'80% di barre ricostruibili, o con le
    divergenze che CRESCONO negli ultimi 30 giorni, la finestra si ESTENDE invece di decidere."""
    motivi = []
    if not np.isfinite(frac) or frac < MIN_RECON_FRAC:
        motivi.append(f"barre ricostruibili {frac:.1%} < {MIN_RECON_FRAC:.0%}")
    if div_rec >= MIN_DIV_CRESCITA and div_g_rec > 2.0 * div_g_tot + 1e-12:
        motivi.append(f"divergenze in crescita: {div_rec} negli ultimi {RECENTI_GIORNI}g "
                      f"({div_g_rec:.2f}/g contro {div_g_tot:.2f}/g sull'intera finestra)")
    return (not motivi), ("; ".join(motivi) if motivi else
                          f"ricostruibili {frac:.1%}, divergenti {div_rec} negli ultimi {RECENTI_GIORNI}g "
                          f"({div_g_rec:.2f}/g contro {div_g_tot:.2f}/g totale)")


def print_gate(st: dict, dfs: dict, today: pd.Timestamp | None = None) -> dict:
    """Stampa lo stato del gate pre-registrato e lo RITORNA (per i test e per chi legge il log).
    Non decide niente: al kill e alla data stampa cosa la regola dice, in maiuscolo."""
    r = _returns_df()
    out = dict(gate_date=GATE_DATE, kill="NON_MISURABILE", veto_ok=None)
    print(f"\n  GATE PRE-REGISTRATO `PREVDAY-01` (23/08, RESULTS-0822 §54) — decisione {GATE_DATE}")
    if r is None or len(r) < 2:
        print("    serie forward assente o troppo corta: niente da misurare")
        return out
    today = today or pd.Timestamp.now(tz="UTC").normalize()
    sh, n_g = sharpe_giornaliero(r)
    n_att = giorni_attivi(r)
    stato, motivo = kill_check(sh, n_att, today)
    out.update(sharpe_giornaliero=sh, n_giorni=n_g, n_attivi=n_att, kill=stato, kill_motivo=motivo)
    print(f"    Sharpe forward GIORNALIERO (MODELED): {sh:+.2f} su {n_g} giorni, {n_att} attivi"
          f"   [lente oraria {_sharpe_orario(r):+.2f}: non si cita, §54]")
    if stato == "KILL":
        print(f"    *** KILL: {motivo} ***")
    else:
        print(f"    kill (Sharpe < {KILL_SHARPE:+.2f} su >= {KILL_MIN_ACTIVE_DAYS}g attivi): {stato} — {motivo}")
    rc = ricostruzione(r, dfs, int(st["start_ts"]))
    ok, vm = veto_check(rc["frac"], rc["div_per_giorno"], rc["div_per_giorno_recenti"], rc["div_recenti"])
    out.update(ricostruzione=rc, veto_ok=ok, veto_motivo=vm)
    print(f"    veto d'integrita': {'ok' if ok else '*** VETO — la finestra si ESTENDE ***'} — {vm}"
          f" ({rc['ricostruibili']}/{rc['n']} barre, {rc['assenti_nel_feed']} assenti nel feed)")
    if today >= pd.Timestamp(GATE_DATE, tz="UTC"):
        print(f"    *** DECISIONE DOVUTA ({GATE_DATE}): (a) DSR >= 0,95 sulla famiglia dichiarata "
              f"(b) cella al buio == congelata (c) delta libro > +0,05 col funding (d) weights_tilt_null "
              f"(e) ADDS+robust_oos+non-hedge sulla cella al buio (f) day_boundary_robust "
              f"(g) Sharpe forward > 0 [{sh:+.2f}] — TUTTE necessarie ***")
    else:
        print(f"    decisione fra {(pd.Timestamp(GATE_DATE, tz='UTC') - today).days} giorni; "
              f"le condizioni (a)-(b) sono STRUTTURALI e il forward non le cambia (§54)")
    return out


def _sharpe_orario(r: pd.DataFrame) -> float:
    x = r["net_modeled"].astype(float)
    return float(x.mean() / x.std() * np.sqrt(24 * 365.25)) if x.std() > 0 else float("nan")


def print_status(st: dict, dfs: dict):
    days = (max(int(dfs[a]["timestamp"].iloc[-1]) for a in ASSETS) - st["start_ts"]) / 86400_000
    rm = st["cap_modeled"] / MODELED_CAPITAL - 1
    rr = st["cap_real"] / REAL_CAPITAL - 1
    print(f"\n  PREVDAY-BREAKOUT forward-monitor (PAPER, lead ortogonale a TP01 — non deploy)")
    print(f"  forward da {pd.Timestamp(st['start_ts'], unit='ms', tz='UTC').date()}  "
          f"({st['n_bars']} barre 1h ~{days:.0f}g)   trade(flip): {st['n_trades']}")
    print(f"  posizione corrente: BTC {st['pos_real']['BTC']:+.3f}  ETH {st['pos_real']['ETH']:+.3f}")
    print(f"  MODELED  ($2000 nominale): {rm*100:+6.2f}%   eq ${st['cap_modeled']:.2f}   maxDD {st['dd_modeled']*100:.1f}%")
    print(f"  REAL-$600 (min-order $5) : {rr*100:+6.2f}%   eq ${st['cap_real']:.2f}   maxDD {st['dd_real']*100:.1f}%")
    print(f"  -> fill-haircut MODELED-REAL: {(rm-rr)*100:+.2f} pp  (lo scettico l'aveva segnalato)")
    print(f"  log: {RETURNS_FILE}")
    print_gate(st, dfs)
    print()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--status", action="store_true")
    ap.add_argument("--reset", action="store_true")
    args = ap.parse_args()
    dfs = build_bars()
    if args.reset:
        for p in (STATE_FILE, TRADES_FILE, RETURNS_FILE):
            if p.exists():
                p.unlink()
        st = init_state(dfs); _state_io(st)
        print("forward-monitor inizializzato (forward-only da ora).")
        print_status(st, dfs); return
    st = _state_io()
    if st is None:
        st = init_state(dfs); _state_io(st)
        print("forward-monitor inizializzato (forward-only da ora).")
        print_status(st, dfs); return
    if not args.status:
        st = advance(st, dfs); _state_io(st)
    print_status(st, dfs)


if __name__ == "__main__":
    main()