PythagorasGoal/scripts/analysis/partial_tp_ladder.py

"""Exit 'ladder' per le fade: al tocco di una frazione f del TP esce la quota q,
il runner (1-q) corre con STOP DINAMICO bloccato alla soglia (profitto lockato).

Proposta 2026-06-04 ("e se all'80% del TP usciamo con 80% e mettiamo un SL
dinamico su quella soglia e lo lasciamo correre?"). Confronto ONESTO con l'exit
canonico (TP pieno al livello) sugli STESSI segnali, stesso engine intrabar di
fade_base (ingresso a close[i], SL prioritario nello stesso bar, fee 0.10% RT
x leva su tutto il notional — il ladder NON paga fee extra: due uscite ma
stesso notional totale).

Convenzioni intrabar del ladder (oneste/conservative):
  - SL pieno prioritario sulla soglia nello stesso bar;
  - se il bar che tocca la soglia CHIUDE oltre la soglia (rientro), il runner
    si considera stoppato subito alla soglia (non gli si regala il bar);
  - il runner non ha TP (corre), esce su ri-tocco della soglia o a max_bars.

Gate (metodologia repo): il ladder deve migliorare ret E DD (o chiaramente il
rischio/rendimento) su ENTRAMBI gli asset, full E OOS, per tutte e 3 le fade.

    uv run python scripts/analysis/partial_tp_ladder.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            # noqa: E402
from src.live.strategy_loader import load_strategy   # noqa: E402

LIVE_PARAMS = dict(trend_max=3.0, ema_long=200, hurst_max=0.55, min_tp_frac=0.0015)
OOS_START = pd.Timestamp("2023-11-01", tz="UTC")
LEV, POS, FEE_RT = 3.0, 0.15, 0.001
CODES = ["MR01_bollinger_fade", "MR02_donchian_fade", "MR07_return_reversal"]


def simulate(signals, df, ts, policy: str, f: float = 0.8, q: float = 0.8,
             start=None) -> dict:
    """Replay intrabar degli stessi segnali con exit 'base' o 'ladder'."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c)
    fee = FEE_RT * LEV
    capital = peak = 1000.0
    max_dd = 0.0
    last_exit = -1
    trades = wins = 0
    runner_beyond_tp = runner_stopped = 0
    rets = []

    for sig in signals:
        i, d = sig.idx, sig.direction
        if start is not None and ts.iloc[i] < start:
            continue
        if i <= last_exit or i + 1 >= n:
            continue
        entry = c[i]
        tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"]
        j = i

        if policy == "base":
            exit_p = c[min(i + mb, n - 1)]
            for step in range(1, mb + 1):
                j = i + step
                if j >= n:
                    j = n - 1; exit_p = c[j]; break
                hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
                hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
                if hit_sl:
                    exit_p = sl; break
                if hit_tp:
                    exit_p = tp; break
                if step == mb:
                    exit_p = c[j]
            ret = (exit_p - entry) / entry * d * LEV - fee
        else:  # ladder
            th = entry + f * (tp - entry)            # soglia f del percorso verso il TP
            p1 = p2 = None                            # fill quota q / runner (1-q)
            state = "full"
            for step in range(1, mb + 1):
                j = i + step
                if j >= n:
                    j = n - 1
                    if p1 is None:
                        p1 = c[j]
                    p2 = c[j]
                    break
                if state == "full":
                    hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
                    hit_th = (d == 1 and h[j] >= th) or (d == -1 and l[j] <= th)
                    if hit_sl:                       # SL pieno prioritario (conservativo)
                        p1 = p2 = sl; break
                    if hit_th:
                        p1 = th; state = "runner"
                        # il bar chiude oltre la soglia -> runner stoppato subito
                        if (d == 1 and c[j] < th) or (d == -1 and c[j] > th):
                            p2 = th; runner_stopped += 1; break
                        if step == mb:
                            p2 = c[j]; break
                        continue
                    if step == mb:
                        p1 = p2 = c[j]; break
                else:                                # runner: stop dinamico = soglia
                    hit_stop = (d == 1 and l[j] <= th) or (d == -1 and h[j] >= th)
                    if hit_stop:
                        p2 = th; runner_stopped += 1; break
                    if step == mb:
                        p2 = c[j]; break
            if p2 is not None and (p2 - tp) * d > 0:
                runner_beyond_tp += 1
            ret = (q * (p1 - entry) + (1 - q) * (p2 - entry)) / entry * d * LEV - fee

        capital = max(capital + capital * POS * ret, 10.0)
        peak = max(peak, capital)
        max_dd = max(max_dd, (peak - capital) / peak)
        last_exit = j
        trades += 1
        wins += ret > 0
        rets.append(ret)

    if trades == 0:
        return {}
    rets = np.array(rets)
    return {
        "ret_pct": (capital / 1000.0 - 1) * 100,
        "dd_pct": max_dd * 100,
        "trades": trades,
        "win_pct": wins / trades * 100,
        "avg_ret_bps": rets.mean() * 1e4,
        "sharpe_t": rets.mean() / rets.std() * np.sqrt(len(rets)) if rets.std() else 0,
        "runner_beyond_tp": runner_beyond_tp,
        "runner_stopped": runner_stopped,
    }


def main() -> None:
    variants = [("base", None, None), ("ladder", 0.8, 0.8),
                ("ladder", 0.8, 0.5), ("ladder", 0.5, 0.5)]
    for code in CODES:
        strat = load_strategy(code)
        for asset in ("BTC", "ETH"):
            df = load_data(asset, "1h")
            ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
            signals = strat.generate_signals(df, ts, **LIVE_PARAMS)
            print(f"\n=== {code} {asset} 1h — {len(signals)} segnali (params live) ===")
            print(f"{'policy':<16}{'periodo':<6}{'ret%':>10}{'DD%':>8}{'win%':>7}"
                  f"{'avg bps':>9}{'Sh(t)':>7}{'n':>6}{'run>TP':>8}{'run-stop':>9}")
            for policy, f, q in variants:
                tag = "base" if policy == "base" else f"ladder f{f} q{q}"
                for label, start in (("FULL", None), ("OOS", OOS_START)):
                    r = simulate(signals, df, ts, policy, f or 0.8, q or 0.8, start)
                    if not r:
                        continue
                    print(f"{tag:<16}{label:<6}{r['ret_pct']:>10.0f}{r['dd_pct']:>8.1f}"
                          f"{r['win_pct']:>7.1f}{r['avg_ret_bps']:>9.1f}{r['sharpe_t']:>7.2f}"
                          f"{r['trades']:>6}{r['runner_beyond_tp']:>8}{r['runner_stopped']:>9}")


if __name__ == "__main__":
    main()