PythagorasGoal/scripts/research/intraday/prevday_bootstrap.py

"""prevday_bootstrap — l'edge di PREVDAY è coda-fortuna o persistente? (blocker #2/#3)

CHIARIMENTO: il "top-5 giorni = 76-83% del PnL" del diario intraday era sulle GAMBE REVERT del combo
a 5 segnali (vol_event/volume_spike/gap_fill), poi SCARTATE. Il sopravvissuto in forward-monitor è
PREVDAY (breakout-continuation). Qui testiamo la concentrazione e la robustezza di PREVDAY STESSO —
e in particolare della sua GAMBA SHORT, che (prevday_turnover) è l'intero valore di portafoglio.

Due test:
  A) CONCENTRAZIONE — quota del PnL nei top-K giorni (riproduce la metrica del diario su PREVDAY,
     full / short-only / long-only, vs TP01 come riferimento: PREVDAY è PIÙ concentrato di ciò che
     già deployamo?). + giorni per arrivare al 50% del guadagno cumulato.
  B) CIRCULAR BLOCK BOOTSTRAP (blocchi da 20g, preserva autocorrelazione/regime) — distribuzione di:
     standalone Sharpe (full + hold-out) e dell'UPLIFT hold-out del blend 80%TP01+20%PREVDAY (la
     metrica-soldi). %>0 e 5° percentile = quanto l'edge dipende da quali blocchi sono capitati.

    uv run python scripts/research/intraday/prevday_bootstrap.py
"""
import sys
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(ROOT))
from src.backtest.harness import load                       # noqa: E402
from src.strategies import prevday_breakout as pb           # noqa: E402
from src.portfolio.portfolio import to_daily                # noqa: E402
from src.portfolio.sleeves import _tp01_returns             # noqa: E402

HOLD = pd.Timestamp("2025-01-01", tz="UTC")
FEE_SIDE = 0.0005
WEIGHT = 0.5
ASSETS = ["BTC", "ETH"]
RNG = np.random.default_rng(12345)
B = 3000
BLOCK = 20


def _sh(x):
    x = np.asarray(x, float); x = x[np.isfinite(x)]
    return float(x.mean() / x.std() * np.sqrt(365.25)) if len(x) > 2 and x.std() > 0 else 0.0


def _leg_daily(dfs, leg):
    """Ritorni daily 50/50 di PREVDAY restringendo la direzione: 'full'|'short'|'long'."""
    out = None
    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
        d = pb._breakout_direction(df, pb.ANCHOR_DAYS, pb.BUFFER_K, True)
        if leg == "short":
            d = np.minimum(d, 0.0)
        elif leg == "long":
            d = np.maximum(d, 0.0)
        tgt = np.nan_to_num(pb._vol_target(d, df, pb.TARGET_VOL, pb.VOL_WIN_DAYS, pb.LEV_CAP), nan=0.0)
        held = np.zeros(len(tgt)); held[1:] = tgt[:-1]
        net = held * r - FEE_SIDE * np.abs(np.diff(tgt, prepend=tgt[0]))
        s = pd.Series(net, index=pd.to_datetime(df["datetime"], utc=True))
        dd = s.groupby(s.index.floor("1D")).sum()
        out = dd if out is None else out.add(dd, fill_value=0)
    return WEIGHT * out


def concentration(daily, label):
    s = daily.dropna()
    total = s.sum()
    pos = s[s > 0].sum()
    topk = {k: s.nlargest(k).sum() / total if total != 0 else float("nan") for k in (5, 10, 20)}
    # giorni (in ordine decrescente) per arrivare al 50% del guadagno lordo positivo
    cum = s.sort_values(ascending=False).cumsum()
    d50 = int((cum < 0.5 * pos).sum()) + 1 if pos > 0 else -1
    n = len(s)
    print(f"  {label:<22s} n={n}  totRet {total*100:+6.0f}%  "
          f"top5 {topk[5]*100:4.0f}%  top10 {topk[10]*100:4.0f}%  top20 {topk[20]*100:4.0f}%  "
          f"giorni->50% gain: {d50} ({d50/n*100:.1f}% dei giorni)")


def block_boot_joint(tp, pv, n_iter=B, block=BLOCK):
    """Bootstrap a blocchi circolari della serie CONGIUNTA (tp,pv) allineata. Ritorna i campioni di
    (sh_pv, uplift_blend_80_20)."""
    J = pd.concat({"TP": tp, "PV": pv}, axis=1, sort=True).dropna()
    a = J["TP"].values; b = J["PV"].values
    n = len(a)
    nblocks = int(np.ceil(n / block))
    sh_pv, upl = [], []
    base_tp = _sh(a)
    for _ in range(n_iter):
        starts = RNG.integers(0, n, size=nblocks)
        idx = np.concatenate([(np.arange(s, s + block) % n) for s in starts])[:n]
        ta, tb = a[idx], b[idx]
        sh_pv.append(_sh(tb))
        blend = 0.8 * ta + 0.2 * tb
        upl.append(_sh(blend) - _sh(ta))
    return np.array(sh_pv), np.array(upl), base_tp


def report_boot(name, sh, upl):
    def q(x, p):
        return float(np.percentile(x, p))
    print(f"  {name}")
    print(f"    PREVDAY Sharpe   : mediana {np.median(sh):+.2f}  [5°,95°]=[{q(sh,5):+.2f},{q(sh,95):+.2f}]  %>0 {np.mean(sh>0)*100:.0f}%")
    print(f"    blend 80/20 UPLIFT: mediana {np.median(upl):+.2f}  [5°,95°]=[{q(upl,5):+.2f},{q(upl,95):+.2f}]  "
          f"%>0 {np.mean(upl>0)*100:.0f}%  %>+0.10 {np.mean(upl>0.10)*100:.0f}%")


def main():
    print("=" * 100)
    print("  PREVDAY bootstrap — l'edge è coda-fortuna o persistente? (blocco da 20g, B=%d)" % B)
    print("=" * 100)
    dfs = {a: load(a, "1h").reset_index(drop=True) for a in ASSETS}
    pv_full = _leg_daily(dfs, "full")
    pv_short = _leg_daily(dfs, "short")
    pv_long = _leg_daily(dfs, "long")
    tp = to_daily(_tp01_returns())

    print("\n[A] CONCENTRAZIONE del PnL nei top-K giorni (più alto = più coda-fortuna):")
    concentration(pv_full, "PREVDAY full")
    concentration(pv_short, "PREVDAY short-only")
    concentration(pv_long, "PREVDAY long-only")
    concentration(tp, "TP01 (riferimento)")

    print(f"\n[B] CIRCULAR BLOCK BOOTSTRAP — FULL ({pv_full.dropna().index.min().date()}->{pv_full.dropna().index.max().date()}):")
    sh, upl, base = block_boot_joint(tp, pv_full)
    print(f"    [TP01 base full Sharpe {base:+.2f}; uplift osservato +0.28 a w20]")
    report_boot("full sample:", sh, upl)

    print(f"\n[B] HOLD-OUT (2025+):")
    tph = tp[tp.index >= HOLD]; pvh = pv_full[pv_full.index >= HOLD]
    shH, uplH, baseH = block_boot_joint(tph, pvh)
    print(f"    [TP01 base hold Sharpe {baseH:+.2f}; uplift osservato +0.56 a w20]")
    report_boot("hold-out:", shH, uplH)

    print(f"\n[B] SHORT-ONLY hold-out (la gamba che è tutto il valore):")
    shS, uplS, _ = block_boot_joint(tph, pv_short[pv_short.index >= HOLD])
    report_boot("short-only hold-out:", shS, uplS)
    print("=" * 100)


if __name__ == "__main__":
    main()