PythagorasGoal/scripts/analysis/_port06_gate_common.py

"""REGRESSION-LOCK COMUNE dei gate PORT06 live (exit16 / trendmax / dip01).

Queste funzioni erano copiate quasi-verbatim in exit16_port06_impact.py,
trendmax_port06_impact.py e dip01_exit16_impact.py. Sono il regression-lock
delle DECISIONI LIVE (EXIT-16, swap hurst->trend, DIP01 EXIT-16): la copy-drift
fra le copie avrebbe corrotto i verdetti, quindi vivono qui in un'unica copia.

NON cambiare la matematica: i gate devono restare riproducibili byte-a-byte.
Se un nuovo gate richiede un comportamento diverso, PARAMETRIZZARE (come fu
fatto per hurst_mask/trend_max), mai biforcare una copia.

Contenuto:
  build_trades_variant : replay ESATTO di risk_management.build_trades sulle
                         fade (mode="orig" == canonico), con i rami varianti
                         EXIT-16 (mode="exit16"), filtro trend (trend_max) e
                         loss-guard Hurst (hurst_mask) parametrici.
  equity_from_trades   : trade -> equity giornaliera normalizzata su IDX
                         (stesso flusso di combine_portfolio.fade_daily_equity).
  port_metrics         : metriche FULL/OOS del portafoglio con la STESSA
                         matematica pesi di Portfolio.backtest (weight_vector
                         su tutti gli sleeve, ribilancio come port_returns).
  dd                   : max drawdown % di una equity.

NB: l'engine DIP01 (dip_trades in dip01_exit16_impact.py) NON e' una copia di
build_trades_variant ma un sibling deliberatamente diverso (long-only, mode
"orig_gap" gap-aware, j clampato a n-1 a fine serie, niente filtri trend/hurst)
-> resta nel suo script.
"""
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.strategy_research import atr
from scripts.analysis.risk_management import FEE_RT, LEV, POS, INIT
from scripts.analysis.combine_portfolio import (
    _norm, IDX, port_returns, metrics, SPLIT,
)
from src.portfolio import weighting as W

BUFFER = 0.5        # EXIT-16 close-confirm (come in produzione)
EMA_LONG = 200


def build_trades_variant(ents, df, mode, trend_max, hurst_mask=None,
                         buffer=BUFFER, lev=LEV, fee_rt=FEE_RT, ema_long=EMA_LONG):
    """Replica ESATTA di risk_management.build_trades, con i rami varianti.

    mode="orig"   : SL intrabar al livello (SL prima del TP) == canonico.
    mode="exit16" : SL intrabar OFF; TP intrabar al livello (priorita' nel bar);
                    SL solo se il CLOSE sfonda sl0 -/+ buffer*ATR14[j], fill a close[j].
    trend_max     : None = filtro OFF; 3.0 = config live.
    hurst_mask    : bool[i]=True -> salta l'ingresso (loss-guard storico).
    """
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c)
    a = atr(df, 14)
    el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
    fee = fee_rt * lev
    out = []
    last = -1
    for e in ents:
        i, d = e["i"], e["d"]
        if i <= last or i + 1 >= n:
            continue
        if hurst_mask is not None and hurst_mask[i]:
            continue
        if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max:
            continue
        entry = c[i]
        tp, sl0, mb = e["tp"], e["sl"], e["max_bars"]
        exit_p = c[min(i + mb, n - 1)]
        j = min(i + mb, n - 1)
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                exit_p = c[n - 1]
                break
            if mode == "orig":
                hs = (d == 1 and l[j] <= sl0) or (d == -1 and h[j] >= sl0)
                ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
                if hs:
                    exit_p = sl0
                    break
                if ht:
                    exit_p = tp
                    break
                if k == mb:
                    exit_p = c[j]
            else:  # exit16
                ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
                if ht:
                    exit_p = tp
                    break
                aj = a[j] if np.isfinite(a[j]) else 0.0
                confirm = (d == 1 and c[j] < sl0 - buffer * aj) or \
                          (d == -1 and c[j] > sl0 + buffer * aj)
                if confirm:
                    exit_p = c[j]
                    break
                if k == mb:
                    exit_p = c[j]
        ret = (exit_p - entry) / entry * d * lev - fee
        out.append((i, j, ret))
        last = j
    return out


def equity_from_trades(df, trades):
    """Trade -> equity giornaliera su IDX (flusso di combine_portfolio.fade_daily_equity)."""
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    n = len(df)
    eq = np.full(n, INIT, dtype=float)
    cap = INIT
    for i, j, ret in sorted(trades, key=lambda t: t[1]):
        cap = max(cap + cap * POS * ret, 10.0)
        eq[j:] = cap
    s = pd.Series(eq, index=ts).resample("1D").last().reindex(IDX).ffill().bfill()
    return _norm(s)


def port_metrics(members: dict[str, pd.Series], p):
    """Metriche (FULL, OOS) del portafoglio p con la STESSA matematica pesi cap
    di Portfolio.backtest."""
    ids = p.sleeve_ids
    dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids})
    w = W.weight_vector(p.weighting, ids, dr, weights=p.weights,
                        caps=p.caps, clusters=p.clusters, lookback=p.vol_lookback)
    drp = port_returns({i: members[i] for i in ids}, w)
    return metrics(drp), metrics(drp, lo=SPLIT)


def dd(s):
    """Max drawdown % di una serie equity."""
    pk = s.cummax()
    return float(((pk - s) / pk).max() * 100)