PythagorasGoal/scripts/analysis/shape_pivot_research.py

"""Famiglia SHAPE-PIVOT: geometria a punti di svolta (PIP / pivot) -> bias futuro.

Idea (causale, no look-ahead):
  - a ogni barra i comprimo la finestra di L barre terminante a close[i] nei suoi
    P punti percettivamente importanti (PIP, Perceptually Important Points: i punti
    di massima deviazione dalla retta congiungente — Fu et al.);
  - la sequenza di P punti e' una POLILINEA = forma geometrica grezza;
  - la classifico con feature interpretabili e CAUSALI:
      * trend dei pivot interni: higher-highs/higher-lows (HH/HL) vs lower-*  (LH/LL);
      * convergenza/divergenza delle pendenze (triangoli/cunei);
      * distanza % di close[i] dall'ultimo pivot alto/basso (vicino a R / a S);
      * pendenza dell'ultimo segmento (slancio recente);
  - per ogni CLASSE geometrica stimo l'esito medio a H barre usando SOLO occorrenze
    passate il cui esito era gia' realizzato prima di i (statistica causale rolling);
  - entro a close[i] nella direzione del bias di classe se l'edge passato e' netto;
    exit a H barre o TP/SL in ATR.

VINCOLI (CLAUDE.md "metodologia obbligatoria" + "lezione squeeze look-ahead"):
  - PIP/pivot calcolati SOLO su close[i-L+1 .. i]; nessun pivot "confermato dal futuro".
  - ogni statistica per-classe usa solo campioni con esito (entry+H) <= i-1.
  - ingresso eseguibile a close[i]; netto fee (0.10% RT base, sweep a 0.20%); leva 3x,
    pos 0.15; validazione OOS (ultimo 30%) + robustezza griglia + >=2 asset.
  - check di causalita' esplicito (perturbo il futuro: la forma a i non cambia).

Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust).
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.analysis.explore_lab import (  # noqa: E402
    get_df, evaluate, robust, simulate, atr, ema, _dt, OOS_FRAC,
)


# =========================================================================
#  PIP — Perceptually Important Points (causale, solo su close[a..b])
# =========================================================================
def pip_indices(seg: np.ndarray, p: int) -> list[int]:
    """Estrae p indici PIP dalla serie `seg` (inclusi i 2 estremi).

    Algoritmo Fu et al.: parti dai 2 estremi; aggiungi iterativamente il punto a
    massima distanza VERTICALE dalla retta che unisce i due PIP adiacenti, finche'
    non hai p punti. Tutto sul segmento dato -> nessun look-ahead se seg=close[..i].
    """
    n = len(seg)
    if p >= n:
        return list(range(n))
    pts = [0, n - 1]
    while len(pts) < p:
        best_d, best_k = -1.0, -1
        for s in range(len(pts) - 1):
            l, r = pts[s], pts[s + 1]
            if r - l < 2:
                continue
            x1, y1 = l, seg[l]
            x2, y2 = r, seg[r]
            dx = x2 - x1
            # distanza verticale dalla retta (interpolazione lineare in x)
            for k in range(l + 1, r):
                if dx == 0:
                    dist = abs(seg[k] - y1)
                else:
                    yline = y1 + (y2 - y1) * (k - x1) / dx
                    dist = abs(seg[k] - yline)
                if dist > best_d:
                    best_d, best_k = dist, k
        if best_k < 0:
            break
        # inserisci mantenendo l'ordine
        for s in range(len(pts) - 1):
            if pts[s] < best_k < pts[s + 1]:
                pts.insert(s + 1, best_k)
                break
    return pts


# =========================================================================
#  Classe geometrica della polilinea PIP (feature causali interpretabili)
# =========================================================================
def shape_class(seg: np.ndarray, p: int) -> tuple | None:
    """Ritorna una tupla-classe discreta della forma PIP di `seg`, o None se degenere.

    Feature (tutte da seg=close[..i], causali):
      - dir_seq: per ogni pivot interno, segno della variazione vs precedente
                 (sequenza su/giu) -> cattura HH/HL vs LH/LL e zig-zag;
      - conv: convergenza pendenze inizio vs fine (triangolo/cuneo): segno di
              (|slope_last| - |slope_first|) discretizzato;
      - loc: posizione di close[i] nel range della finestra (vicino a max=resistenza,
             vicino a min=supporto), in 3 bucket.
    La classe e' invariante a livello/scala (z-norm implicito su forma).
    """
    idx = pip_indices(seg, p)
    if len(idx) < 3:
        return None
    y = seg[idx]
    rng = y.max() - y.min()
    if rng <= 0:
        return None
    yn = (y - y.min()) / rng                      # forma normalizzata 0..1
    # sequenza direzioni dei segmenti (su=1 / giu=0)
    diffs = np.diff(yn)
    dir_seq = tuple(int(x > 0) for x in diffs)
    # convergenza: pendenza primo vs ultimo segmento
    s_first = abs(diffs[0])
    s_last = abs(diffs[-1])
    if s_last > s_first * 1.3:
        conv = 1            # divergente (slancio finale)
    elif s_last < s_first * 0.77:
        conv = -1           # convergente (compressione, triangolo/cuneo)
    else:
        conv = 0
    # posizione di close[i] (=ultimo punto) nel range: 0..1 in 3 bucket
    last = yn[-1]
    loc = 0 if last < 0.33 else (2 if last > 0.67 else 1)
    return (dir_seq, conv, loc)


# =========================================================================
#  Strategia: bias per-classe stimato CAUSALMENTE (rolling, esito realizzato)
# =========================================================================
def pivot_entries(df, L=48, P=5, H=12, min_lib=1000, min_samples=20,
                  edge=0.0, tp_atr=None, sl_atr=None,
                  trend_max=None, ema_long=200, mode="bias") -> list[dict]:
    """Entries dalla geometria PIP con bias di classe causale.

    L: lunghezza finestra-forma. P: n. punti PIP. H: orizzonte (=max_bars).
    min_lib: barre minime prima di operare. min_samples: campioni minimi per fidarsi
             della statistica di una classe. edge: |rendimento medio classe| minimo
             (frazione, es. 0.002 = 0.2%) per entrare. mode:
       - "bias": entra nel verso del rendimento medio passato della classe (momentum
                 della forma: la classe X storicamente -> su/giu);
       - "fade": entra nel verso OPPOSTO (test mean-reversion della forma).
    Statistica per-classe accumulata SOLO con esiti realizzati < i (causale stretta).
    """
    close = df["close"].values
    high = df["high"].values
    low = df["low"].values
    n = len(close)
    a = atr(df, 14)
    el = ema(close, ema_long) if trend_max is not None else None

    # stato rolling per classe: somma rendimenti e conteggio (solo esiti < i)
    cls_sum: dict[tuple, float] = {}
    cls_cnt: dict[tuple, int] = {}
    # coda di campioni la cui forma e' stata calcolata ma esito non ancora maturo
    # pending[t] = (classe, indice_entry t) -> matura quando t+H <= i-1
    pending: list[tuple] = []           # (mature_at, cls, t)
    pend_ptr = 0

    entries: list[dict] = []

    for i in range(min_lib, n - 1):
        # 1) integra nello storico tutti i campioni il cui esito e' realizzato (< i)
        #    un campione formato a t matura quando t+H <= i-1  => mature_at = t+H+1 <= i
        while pend_ptr < len(pending) and pending[pend_ptr][0] <= i:
            _, cls_p, t = pending[pend_ptr]
            ret_real = (close[t + H] - close[t]) / close[t]
            cls_sum[cls_p] = cls_sum.get(cls_p, 0.0) + ret_real
            cls_cnt[cls_p] = cls_cnt.get(cls_p, 0) + 1
            pend_ptr += 1

        # 2) forma corrente (solo close fino a i)
        seg = close[i - L + 1: i + 1]
        cls = shape_class(seg, P)
        if cls is None:
            continue
        # registra il campione corrente come pending (esito da realizzare in futuro)
        pending.append((i + H + 1, cls, i))

        # 3) decisione con statistica PASSATA della classe
        cnt = cls_cnt.get(cls, 0)
        if cnt < min_samples:
            continue
        mean_ret = cls_sum[cls] / cnt
        if abs(mean_ret) < edge:
            continue
        d = 1 if mean_ret > 0 else -1
        if mode == "fade":
            d = -d
        # filtro trend opzionale
        if trend_max is not None and a[i] > 0:
            if abs(close[i] - el[i]) / a[i] > trend_max:
                continue
        e = {"i": i, "d": d, "max_bars": H}
        if tp_atr is not None and a[i] > 0:
            e["tp"] = close[i] + d * tp_atr * a[i]
        if sl_atr is not None and a[i] > 0:
            e["sl"] = close[i] - d * sl_atr * a[i]
        entries.append(e)
    return entries


# =========================================================================
#  Filone (c): distanza da supporto/resistenza locale (ultimo pivot alto/basso)
# =========================================================================
def sr_entries(df, L=48, P=7, H=12, near=0.5, mode="fade",
               tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]:
    """Filone (c): close[i] vicino all'ultimo pivot alto (R) o basso (S) della forma.

    Usa i PIP per individuare l'ultimo massimo/minimo locale (resistenza/supporto) e
    misura la distanza % di close[i]. Se close e' entro `near`*ATR da R -> bias short
    (mode='fade': rimbalzo da R) o long (mode='break': rottura). Simmetrico per S.
    Tutto causale: PIP su close[..i], decisione a close[i].
    """
    close = df["close"].values
    n = len(close)
    a = atr(df, 14)
    el = ema(close, ema_long) if trend_max is not None else None
    entries: list[dict] = []

    for i in range(L, n - 1):
        seg = close[i - L + 1: i + 1]
        idx = pip_indices(seg, P)
        if len(idx) < 3 or a[i] <= 0:
            continue
        y = seg[idx]
        # pivot interni (escludi i 2 estremi e l'ultimo punto = close[i])
        inner = y[1:-1]
        if len(inner) == 0:
            continue
        res = inner.max()       # resistenza locale
        sup = inner.min()       # supporto locale
        cur = close[i]
        dist_r = (res - cur) / a[i]
        dist_s = (cur - sup) / a[i]
        d = None
        if 0 <= dist_r <= near:          # appena sotto R
            d = -1 if mode == "fade" else 1
        elif 0 <= dist_s <= near:        # appena sopra S
            d = 1 if mode == "fade" else -1
        if d is None:
            continue
        if trend_max is not None and abs(cur - el[i]) / a[i] > trend_max:
            continue
        e = {"i": i, "d": d, "max_bars": H}
        if tp_atr is not None:
            e["tp"] = cur + d * tp_atr * a[i]
        if sl_atr is not None:
            e["sl"] = cur - d * sl_atr * a[i]
        entries.append(e)
    return entries


# =========================================================================
#  Check causalita' esplicito
# =========================================================================
def check_no_lookahead(df, L=48, P=5) -> bool:
    """La classe-forma a i non deve cambiare se perturbo il FUTURO (>i)."""
    close = df["close"].values.copy()
    i = len(close) // 2
    seg0 = close[i - L + 1: i + 1].copy()
    c0 = shape_class(seg0, P)
    close2 = close.copy()
    close2[i + 1:] *= 1.7                       # stravolge il futuro
    seg1 = close2[i - L + 1: i + 1]
    c1 = shape_class(seg1, P)
    ok = (c0 == c1)
    print(f"  no-lookahead classe-forma a i={i}: {'OK' if ok else 'VIOLATO'} "
          f"(c0={c0} c1={c1})")
    # check su PIP indices
    p0 = pip_indices(seg0, P)
    p1 = pip_indices(seg1, P)
    ok2 = (p0 == p1)
    print(f"  no-lookahead indici PIP: {'OK' if ok2 else 'VIOLATO'}")
    return ok and ok2


# =========================================================================
#  run() riproducibile
# =========================================================================
def run():
    print("=" * 100)
    print("  SHAPE-PIVOT RESEARCH — geometria PIP/pivot -> bias futuro | netto fee, OOS")
    print("=" * 100)

    df_btc = get_df("BTC", "1h")
    print("\n[CAUSALITA']")
    check_no_lookahead(df_btc, L=48, P=5)

    assets = ["BTC", "ETH", "SOL", "ADA"]
    dfs = {a: get_df(a, "1h") for a in assets}

    # ---- A) bias di classe PIP (momentum della forma) ----
    print("\n[A] BIAS di classe PIP (entra nel verso del rendimento medio passato della classe)")
    print("    sweep L/P/H, edge=0.002, min_samples=25, time-exit a H")
    A_grid = [(48, 5, 12), (48, 5, 24), (72, 6, 24), (36, 5, 12), (96, 7, 24), (48, 7, 12)]
    for L, P, H in A_grid:
        print(f"  -- L{L} P{P} H{H} --")
        for a in assets:
            ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="bias")
            evaluate(f"{a} bias L{L}P{P}H{H}", ents, dfs[a])

    # ---- B) fade di classe PIP (mean-reversion della forma) ----
    print("\n[B] FADE di classe PIP (entra opposto al bias storico -> test mean-reversion)")
    for L, P, H in A_grid:
        print(f"  -- L{L} P{P} H{H} --")
        for a in assets:
            ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="fade")
            evaluate(f"{a} fade L{L}P{P}H{H}", ents, dfs[a])

    # ---- C) supporto/resistenza locale dai pivot ----
    print("\n[C] S/R locale dai PIP — FADE (rimbalzo da R/S) vs BREAK (rottura)")
    for mode in ("fade", "break"):
        for near in (0.5, 1.0):
            print(f"  -- mode={mode} near={near} ATR, TP/SL 1.5/1.5 ATR, H=12 --")
            for a in assets:
                ents = sr_entries(dfs[a], L=48, P=7, H=12, near=near, mode=mode,
                                  tp_atr=1.5, sl_atr=1.5)
                evaluate(f"{a} SR-{mode} near{near}", ents, dfs[a])

    # ---- D) miglior candidato con TP/SL ATR + filtro trend (se A o B mostra segnali) ----
    print("\n[D] FADE di classe con TP/SL ATR (2.0/1.5) + filtro trend 3.0, L48 P5 H24")
    for a in assets:
        ents = pivot_entries(dfs[a], L=48, P=5, H=24, edge=0.002, min_samples=25,
                             mode="fade", tp_atr=2.0, sl_atr=1.5, trend_max=3.0)
        res = evaluate(f"{a} fadeTPSL L48P5H24", ents, dfs[a])
        if robust(res):
            print(f"       ^^^ {a} ROBUSTO")

    print("\n" + "=" * 100)
    print("  Verdetto: cerca righe con FULL>0 E OOS>0 E fee0.2% OOS>0 su >=2 asset.")
    print("=" * 100)


if __name__ == "__main__":
    run()