PythagorasGoal/Old/scripts/analysis/pairs15m_flatcheck.py

"""Check candele FLAT (O=H=L=C, liquidita' zero) sui pairs ETH/BTC a 15m.

Rischio noto (CLAUDE.md): ETH 15m ha 14-30%/anno di candele flat per bassa liquidita'
del perpetuo. Su un pairs, un close stale gonfia lo z-score (l'altra gamba si muove,
questa e' ferma) -> segnale di "reversione" FINTO che rientra solo quando la gamba
stale si sblocca: profitto NON eseguibile dal vivo. Questo gonfierebbe il backtest 15m.

Test:
  [1] prevalenza candele flat per anno (ETH 15m, BTC 15m).
  [2] quanti trade del pairs 15m hanno ENTRY/EXIT su una candela flat (gamba stale).
  [3] re-sim flat-aware: entry/exit SOLO su barre pulite (non-flat in ENTRAMBE le gambe)
      -> quanto sopravvive l'edge? (parita': senza flat-skip == pairs_sim).
  [4] gate PORT06 col 15m flat-filtrato vs baseline 1h.

    uv run python scripts/analysis/pairs15m_flatcheck.py
"""
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 src.data.downloader import load_data
from scripts.analysis.pairs_research import pairs_sim, OOS_FRAC, FEE_RT, LEV, POS, BARS_YEAR
from scripts.analysis.report_families import daily_from
from scripts.analysis.combine_portfolio import metrics, SPLIT, OOS_DATE
from scripts.analysis.pairs15m_port06_gate import port_metrics, eth_btc_daily, UNIV_1H, GAME_15M
from scripts.portfolios._defs import PORTFOLIOS
from src.portfolio.sleeves import all_sleeve_equities


def aligned2(a, b, tf="15m"):
    """Merge con OHLC di ENTRAMBE le gambe (serve per rilevare i flat su entrambe)."""
    da = load_data(a, tf)[["timestamp", "open", "high", "low", "close"]].rename(
        columns=lambda x: x + "_a" if x != "timestamp" else x)
    db = load_data(b, tf)[["timestamp", "open", "high", "low", "close"]].rename(
        columns=lambda x: x + "_b" if x != "timestamp" else x)
    m = da.merge(db, on="timestamp", how="inner").reset_index(drop=True)
    m["dt"] = pd.to_datetime(m["timestamp"], unit="ms", utc=True)
    return m


def is_flat(o, h, l, c):
    return (o == h) & (h == l) & (l == c)


def flat_prevalence(asset, tf="15m"):
    d = load_data(asset, tf)
    d = d.copy()
    d["dt"] = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
    fl = is_flat(d["open"].values, d["high"].values, d["low"].values, d["close"].values)
    d["flat"] = fl
    by = d.groupby(d["dt"].dt.year)["flat"].mean() * 100
    return by, fl.mean() * 100


def pairs_sim_flataware(a, b, tf="15m", n=66, z_in=1.674, z_exit=1.0, max_bars=35,
                        jump_max=0.08, fee_rt=FEE_RT, lev=LEV, pos=POS,
                        split_frac=0.0, skip_flat=True):
    """Come pairs_sim ma: entry/exit consentiti SOLO su barre pulite (se skip_flat).
    Ritorna anche n_entry_flat / n_exit_flat (diagnostica, calcolata sempre)."""
    m = aligned2(a, b, tf)
    ca, cb = m["close_a"].values, m["close_b"].values
    flat_a = is_flat(m["open_a"].values, m["high_a"].values, m["low_a"].values, ca)
    flat_b = is_flat(m["open_b"].values, m["high_b"].values, m["low_b"].values, cb)
    flat = flat_a | flat_b                     # barra "sporca" se una delle due gambe e' flat
    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)
    ts = m["dt"]; N = len(r)
    split = int(N * split_frac)
    fee = 2 * fee_rt * lev
    cap = peak = 1000.0; dd = 0.0; last = -1
    trades = wins = 0; rets = []; yearly = {}
    eq_ts, eq_v = [], []
    n_entry_flat = n_exit_flat = 0
    for i in range(n + 1, N - 1):
        if i < split or np.isnan(z[i]) or dr[i] > jump_max or i <= last:
            continue
        if z[i] <= -z_in:
            d = 1
        elif z[i] >= z_in:
            d = -1
        else:
            continue
        if flat[i]:
            n_entry_flat += 1
            if skip_flat:
                continue                       # non si entra su una gamba stale
        # exit: |z|<=z_exit o max_bars; se skip_flat, salta le barre flat come uscita
        j = min(i + max_bars, N - 1)
        for k in range(1, max_bars + 1):
            jj = i + k
            if jj >= N:
                j = N - 1; break
            if skip_flat and flat[jj]:
                j = jj                          # avanza, non esce su barra stale
                continue
            if abs(z[jj]) <= z_exit:
                j = jj; break
            j = jj
        if flat[j]:
            n_exit_flat += 1
            if skip_flat:
                # spingi all'ultima barra pulita entro l'orizzonte
                back = j
                while back > i and flat[back]:
                    back -= 1
                j = back if back > i else j
        retA = (ca[j] - ca[i]) / ca[i]
        retB = (cb[j] - cb[i]) / cb[i]
        ret = (retA - retB) * d * lev - fee
        cap = max(cap + cap * pos * ret, 10.0)
        peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
        trades += 1; wins += ret > 0; rets.append(ret * pos); last = j
        eq_ts.append(ts.iloc[j]); eq_v.append(cap)
        yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100
    yrs_span = (ts.iloc[-1] - ts.iloc[max(split, 0)]).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
    return dict(trades=trades, win=wins / trades * 100 if trades else 0, ret=ret_tot,
                dd=dd * 100, sharpe=sharpe, yearly=yearly, eq_ts=eq_ts, eq_v=eq_v,
                n_entry_flat=n_entry_flat, n_exit_flat=n_exit_flat)


def main():
    print("=" * 100)
    print("  CHECK FLAT-CANDLE — ETH/BTC pairs 15m (gate condizionato)")
    print("=" * 100)

    # [1] prevalenza
    print("\n[1] Prevalenza candele flat (O=H=L=C) per anno, 15m:")
    for asset in ("ETH", "BTC"):
        by, tot = flat_prevalence(asset, "15m")
        print(f"    {asset}: media {tot:.1f}%  | " +
              "  ".join(f"{y}:{v:.0f}%" for y, v in by.items()))

    # [2] quanti trade toccano un flat (sim SENZA skip per diagnostica)
    diag = pairs_sim_flataware("ETH", "BTC", **GAME_15M, skip_flat=False)
    tr = diag["trades"]
    print(f"\n[2] Trade 15m totali: {tr} | entry su barra flat: {diag['n_entry_flat']} "
          f"({diag['n_entry_flat']/tr*100:.1f}%) | exit su barra flat: {diag['n_exit_flat']} "
          f"({diag['n_exit_flat']/tr*100:.1f}%)")

    # [3] parita' + edge filtrato
    print("\n[3] Edge 15m: NO-skip (== pairs_sim) vs FLAT-AWARE (entry/exit solo barre pulite):")
    # parita': flataware skip_flat=False deve ~== pairs_sim
    base_ps = pairs_sim("ETH", "BTC", **GAME_15M, pos=POS, lev=LEV)
    print(f"    parita' pairs_sim     : trd {base_ps['trades']:>5d} Sh {base_ps['sharpe']:.2f} "
          f"DD {base_ps['dd']:.0f}% ret {base_ps['ret']:+.0f}%")
    print(f"    flataware (no-skip)   : trd {diag['trades']:>5d} Sh {diag['sharpe']:.2f} "
          f"DD {diag['dd']:.0f}% ret {diag['ret']:+.0f}%")
    filt = pairs_sim_flataware("ETH", "BTC", **GAME_15M, skip_flat=True)
    filt_o = pairs_sim_flataware("ETH", "BTC", **GAME_15M, skip_flat=True, split_frac=1 - OOS_FRAC)
    print(f"    FLAT-AWARE (skip)     : trd {filt['trades']:>5d} Sh {filt['sharpe']:.2f} "
          f"DD {filt['dd']:.0f}% ret {filt['ret']:+.0f}% | OOS Sh {filt_o['sharpe']:.2f} DD {filt_o['dd']:.0f}%")
    drop = (1 - filt['trades'] / diag['trades']) * 100
    sh_keep = filt['sharpe'] / diag['sharpe'] * 100 if diag['sharpe'] else 0
    verdict = "EDGE NON artefatto flat" if sh_keep > 70 else "EDGE in larga parte ARTEFATTO flat"
    print(f"    -> rimossi {drop:.1f}% dei trade; Sharpe trattenuto {sh_keep:.0f}% ({verdict})")

    # [4] gate PORT06 col 15m flat-filtrato
    print("\n[4] GATE PORT06 — ETH/BTC: baseline 1h vs SWAP 15m-FLATAWARE vs BLEND:")
    p = PORTFOLIOS["PORT06"]
    pair_ids = [s.sid for s in p.sleeves if s.sid.startswith("PR_")]
    eq_base = dict(all_sleeve_equities())
    e1h, _ = eth_btc_daily(UNIV_1H)
    e15f = daily_from(filt["eq_ts"], filt["eq_v"])
    # blend 1h + 15m-flataware (50/50 daily-rebalanced)
    from scripts.analysis.pairs15m_port06_gate import blend
    eblend = blend(e1h, e15f, 0.5)
    corr = e1h.pct_change().fillna(0).corr(e15f.pct_change().fillna(0))
    print(f"    corr 1h vs 15m-flataware: {corr:.3f}")
    print(f"    {'variante':<18s} | {'FULL Sh':>8s}{'FULL DD%':>9s}{'CAGR':>6s} | {'OOS Sh':>7s}{'OOS DD%':>8s}")
    print("    " + "-" * 70)
    res = {}
    for tag, eth in [("baseline 1h", e1h), ("SWAP 15m-flat", e15f), ("BLEND 1h+15m-flat", eblend)]:
        members = dict(eq_base); members["PR_ETHBTC"] = eth
        f, o = port_metrics(members, p)
        res[tag] = (f, o)
        print(f"    {tag:<18s} | {f['sharpe']:>8.2f}{f['dd']:>9.2f}{f['cagr']:>5.0f}%"
              f" | {o['sharpe']:>7.2f}{o['dd']:>8.2f}")

    fb, ob = res["baseline 1h"]
    print("\n  VERDETTO (vs baseline 1h, fee backtest): Sharpe non peggiora E DD <= baseline")
    for tag in ("SWAP 15m-flat", "BLEND 1h+15m-flat"):
        f, o = res[tag]
        ok = o["sharpe"] >= ob["sharpe"] - 0.02 and o["dd"] <= ob["dd"] + 1e-9 and f["sharpe"] >= fb["sharpe"] - 0.02 and f["dd"] <= fb["dd"] + 1e-9
        print(f"    {tag:<18s}: OOS {ob['sharpe']:.2f}->{o['sharpe']:.2f} DD {ob['dd']:.2f}->{o['dd']:.2f}"
              f" | FULL {fb['sharpe']:.2f}->{f['sharpe']:.2f} DD {fb['dd']:.2f}->{f['dd']:.2f} => {'PROMOSSO' if ok else 'bocciato'}")


if __name__ == "__main__":
    main()