PythagorasGoal/scripts/analysis/pairs_zstop_research.py

"""Ricerca PRE-REGISTRATA: disaster-cap z-score (z_stop) per la famiglia PAIRS.

Ipotesi pre-registrata: uscita immediata al close della barra se |z| >= z_stop
dopo l'ingresso taglia la coda da structural-break senza toccare i trade normali
(che vivono fra z_exit e z_in).

Griglia PRE-REGISTRATA (unica, completa — NIENTE varianti a posteriori):
  - 5 coppie 1h (config universale n=50 z_in=2.0 z_exit=0.75 max_bars=72):
      z_stop in {3.0, 3.5, 4.0, 5.0}
  - ETH/BTC 15m flat_skip (n=66 z_in=1.674 z_exit=1.0 max_bars=35):
      z_stop in {2.5, 3.0, 3.5, 4.0}
Split: TRAIN = entry prima del 2023-11-01, OOS = dopo (convenzione progetto).

Engine: copia FEDELE di pairs_research.pairs_sim / pairs_sim_flat (stessa
matematica, fee 2 gambe = 2*fee_rt*lev) + parametro z_stop. Causalita': lo z
usato per l'exit alla barra j e' lo stesso z[j] causale (rolling su r[<=j])
gia' usato dall'exit |z|<=z_exit.

REGRESSION-LOCK obbligatorio (eseguito in main, si ferma se fallisce):
  z_stop=None deve riprodurre ESATTAMENTE pairs_sim (ETH/BTC 1h) e
  pairs_sim_flat (ETH/BTC 15m flat_skip): stesso n trade, stesso ret.
"""
from __future__ import annotations

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 scripts.analysis.pairs_research import (  # noqa: E402
    FEE_RT, LEV, POS, aligned_ohlc, is_flat_ohlc, pairs_sim, pairs_sim_flat,
)

SPLIT_DT = pd.Timestamp("2023-11-01", tz="UTC")


def pairs_sim_zstop(a, b, tf="1h", n=50, z_in=2.0, z_exit=0.75, max_bars=72,
                    jump_max=0.08, fee_rt=FEE_RT, lev=LEV, pos=POS,
                    z_stop=None, t0=None, t1=None,
                    flat_skip=False, scan_buffer=192):
    """Copia fedele dell'engine pairs (pairs_sim_flat, che con flat_skip=False
    e' identico a pairs_sim — regression-lock in main) + disaster-cap z_stop.

    z_stop: se non None, l'exit si arma anche quando |z[jj]| >= z_stop
    (structural break: lo spread diverge oltre l'ingresso). Stessa convenzione
    causale e stesso fill (close della barra) dell'exit |z|<=z_exit.
    t0/t1: finestra sul timestamp della barra di ENTRY (train/OOS split).
    """
    m = aligned_ohlc(a, b, tf)
    ca, cb = m["close_a"].values, m["close_b"].values
    N = len(ca)
    if flat_skip:
        flat = (is_flat_ohlc(m["open_a"].values, m["high_a"].values, m["low_a"].values, ca)
                | is_flat_ohlc(m["open_b"].values, m["high_b"].values, m["low_b"].values, cb))
    else:
        flat = np.zeros(N, dtype=bool)
    r = np.log(ca / cb)
    dr = np.abs(np.diff(r, prepend=r[0]))
    ma = pd.Series(r).rolling(n).mean().values
    sd = pd.Series(r).rolling(n).std().values
    z = (r - ma) / np.where(sd == 0, np.nan, sd)        # causale: usa r[<=i]
    ts = m["dt"]
    tsv = ts.values                                      # datetime64 per filtro finestra
    t0v = np.datetime64(t0.tz_convert(None)) if t0 is not None else None
    t1v = np.datetime64(t1.tz_convert(None)) if t1 is not None else None
    fee = 2 * fee_rt * lev                               # 2 gambe
    cap = peak = 1000.0; dd = 0.0; last = -1
    trades = wins = n_stop = 0
    rets = []; rets_raw = []
    eq_ts, eq_v = [], []
    kmax = max_bars + (scan_buffer if flat_skip else 0)
    for i in range(n + 1, N - 1):
        if np.isnan(z[i]) or dr[i] > jump_max or i <= last:
            continue
        if t0v is not None and tsv[i] < t0v:
            continue
        if t1v is not None and tsv[i] >= t1v:
            continue
        if z[i] <= -z_in:
            d = 1
        elif z[i] >= z_in:
            d = -1
        else:
            continue
        if flat[i]:
            continue                                   # niente ingresso su barra stale
        # exit: |z|<=z_exit, max_bars, o DISASTER-CAP |z|>=z_stop; con flat_skip
        # l'exit si arma e si esce alla prima barra pulita (live-realizable)
        exit_ready = False; stopped = False; j = i
        for k in range(1, kmax + 1):
            jj = i + k
            if jj >= N:
                j = N - 1; break
            if not exit_ready:
                if z_stop is not None and abs(z[jj]) >= z_stop:
                    exit_ready = True; stopped = True
                elif abs(z[jj]) <= z_exit or k >= max_bars:
                    exit_ready = True
            if exit_ready and not flat[jj]:
                j = jj; break
            j = jj
        retA = (ca[j] - ca[i]) / ca[i]
        retB = (cb[j] - cb[i]) / cb[i]
        ret = (retA - retB) * d * lev - fee             # long A / short B (o viceversa)
        cap = max(cap + cap * pos * ret, 10.0)
        peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
        trades += 1; wins += ret > 0; n_stop += stopped
        rets.append(ret * pos); rets_raw.append(ret); last = j
        eq_ts.append(ts.iloc[j]); eq_v.append(cap)
    # span temporale della finestra effettiva (per annualizzare lo Sharpe)
    lo = ts.iloc[0] if t0 is None else max(ts.iloc[0], t0)
    hi = ts.iloc[-1] if t1 is None else min(ts.iloc[-1], t1)
    yrs_span = (hi - lo).days / 365.25 or 1
    sharpe = 0.0
    if len(rets) > 1 and np.std(rets) > 0:
        sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(trades / yrs_span))
    ret_tot = (cap / 1000 - 1) * 100
    worst = min(rets_raw) * 100 if rets_raw else 0.0
    return dict(trades=trades, n_stop=n_stop, win=wins / trades * 100 if trades else 0,
                ret=ret_tot, dd=dd * 100, sharpe=sharpe, worst=worst)


# ----------------------------------------------------------------------------- lock

def regression_lock():
    """z_stop=None deve riprodurre ESATTAVENTE l'engine canonico."""
    ok = True
    # 1h plain vs pairs_sim (config universale live z_exit=0.75)
    ref = pairs_sim("ETH", "BTC", n=50, z_in=2.0, z_exit=0.75, max_bars=72)
    new = pairs_sim_zstop("ETH", "BTC", n=50, z_in=2.0, z_exit=0.75, max_bars=72,
                          z_stop=None, flat_skip=False)
    m1 = (ref["trades"] == new["trades"]) and abs(ref["ret"] - new["ret"]) < 1e-9
    print(f"  LOCK 1h  ETH/BTC vs pairs_sim:      trades {ref['trades']} vs {new['trades']}, "
          f"ret {ref['ret']:+.6f} vs {new['ret']:+.6f} -> {'OK' if m1 else 'FAIL'}")
    ok &= m1
    # 15m flat_skip vs pairs_sim_flat
    ref = pairs_sim_flat("ETH", "BTC", tf="15m", n=66, z_in=1.674, z_exit=1.0,
                         max_bars=35, flat_skip=True)
    new = pairs_sim_zstop("ETH", "BTC", tf="15m", n=66, z_in=1.674, z_exit=1.0,
                          max_bars=35, z_stop=None, flat_skip=True)
    m2 = (ref["trades"] == new["trades"]) and abs(ref["ret"] - new["ret"]) < 1e-9
    print(f"  LOCK 15m ETH/BTC vs pairs_sim_flat: trades {ref['trades']} vs {new['trades']}, "
          f"ret {ref['ret']:+.6f} vs {new['ret']:+.6f} -> {'OK' if m2 else 'FAIL'}")
    ok &= m2
    return ok


# ----------------------------------------------------------------------------- main

PAIRS_1H = [("ETH", "BTC"), ("LTC", "ETH"), ("ADA", "ETH"), ("BTC", "LTC"), ("ETH", "SOL")]
GRID_1H = [None, 3.0, 3.5, 4.0, 5.0]
GRID_15M = [None, 2.5, 3.0, 3.5, 4.0]


def run_cell(a, b, win, z_stop, **kw):
    t0, t1 = (None, SPLIT_DT) if win == "TRAIN" else (SPLIT_DT, None)
    return pairs_sim_zstop(a, b, z_stop=z_stop, t0=t0, t1=t1, **kw)


def main():
    print("=" * 100)
    print("  PAIRS disaster-cap z_stop — ricerca PRE-REGISTRATA (griglia fissa, tutti i risultati)")
    print(f"  split TRAIN < {SPLIT_DT.date()} <= OOS | fee 2 gambe {2*FEE_RT*LEV*100:.2f}% | lev {LEV:.0f}x pos {POS}")
    print("=" * 100)
    print("\nREGRESSION-LOCK (z_stop=None == engine canonico):")
    if not regression_lock():
        print("\n  LOCK FALLITO — STOP."); sys.exit(1)

    hdr = (f"  {'z_stop':>7s} | {'trd':>5s} {'stop':>5s} {'ret%':>9s} {'Shrp':>6s} "
           f"{'DD%':>6s} {'worst%':>8s}")
    for a, b in PAIRS_1H:
        kw = dict(tf="1h", n=50, z_in=2.0, z_exit=0.75, max_bars=72, flat_skip=False)
        print(f"\n{'-'*100}\n  {a}/{b} 1h  (n=50 z_in=2.0 z_exit=0.75 max_bars=72)")
        for win in ("TRAIN", "OOS"):
            print(f"  [{win}]\n{hdr}")
            for zs in GRID_1H:
                r = run_cell(a, b, win, zs, **kw)
                lab = "None" if zs is None else f"{zs:.1f}"
                print(f"  {lab:>7s} | {r['trades']:>5d} {r['n_stop']:>5d} {r['ret']:>+9.1f} "
                      f"{r['sharpe']:>6.2f} {r['dd']:>6.2f} {r['worst']:>+8.2f}")

    a, b = "ETH", "BTC"
    kw = dict(tf="15m", n=66, z_in=1.674, z_exit=1.0, max_bars=35, flat_skip=True)
    print(f"\n{'-'*100}\n  {a}/{b} 15m flat_skip  (n=66 z_in=1.674 z_exit=1.0 max_bars=35)")
    for win in ("TRAIN", "OOS"):
        print(f"  [{win}]\n{hdr}")
        for zs in GRID_15M:
            r = run_cell(a, b, win, zs, **kw)
            lab = "None" if zs is None else f"{zs:.1f}"
            print(f"  {lab:>7s} | {r['trades']:>5d} {r['n_stop']:>5d} {r['ret']:>+9.1f} "
                  f"{r['sharpe']:>6.2f} {r['dd']:>6.2f} {r['worst']:>+8.2f}")


if __name__ == "__main__":
    main()