PythagorasGoal/Old/scripts/analysis/shape_template_research.py

"""SHAPE_TEMPLATE_RESEARCH — edge nella FORMA del prezzo: distanze alternative e template canonici.

Due filoni, sull'harness ONESTO condiviso (shape_lab + explore_lab), netto-fee e OOS:

  1. ANALOG con distanza di FORMA alternativa (DTW warping-invariant, correlazione/coseno)
     confrontata HEAD-TO-HEAD con l'euclidea a PARITA' di selettivita' (stessa libreria,
     stesso K, stessa soglia di accordo). DTW e' O(W^2): si usa una libreria SOTTOCAMPIONATA
     (uno start ogni `step` barre) + W ridotto + banda di Sakoe-Chiba.

  2. TEMPLATE di forma canonici (doppio top/bottom, testa-spalle, V-reversal, salita/discesa
     lineare, U). A ogni i misuro la similarita' (correlazione di Pearson sulla finestra
     z-normalizzata) fra forma recente e ogni template; se supera soglia, entro a close[i]
     nella DIREZIONE ATTESA del template stimata SOLO sul passato (esito medio causale delle
     occorrenze gia' concluse di quel template), exit H barre o tp/sl ATR.

VINCOLI anti-look-ahead (verificati esplicitamente):
  - la forma/match a i usa SOLO close fino a i (z-norm causale);
  - la direzione attesa di ogni template e la libreria analog usano SOLO occorrenze il cui
    esito a H barre e' gia' realizzato PRIMA di i (end + H <= i-1);
  - ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H.

Netto fee 0.10% RT baseline + sweep fino a 0.20%. Leva 3x, pos 0.15. OOS ultimo 30%.

Run riproducibile:  uv run python scripts/analysis/shape_template_research.py
DTW e' costoso: usa run_in_background per gli sweep larghi (vedi --sweep).
"""
from __future__ import annotations

import sys
import time
from pathlib import Path

import numpy as np
from numpy.lib.stride_tricks import sliding_window_view

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

from scripts.analysis.explore_lab import get_df, evaluate, robust, simulate, atr, ema, OOS_FRAC  # noqa: E402
from scripts.analysis.shape_lab import znorm_windows, fwd_return  # noqa: E402

RNG_SEED = 7
SUBC_ASSETS = ["BTC", "ETH", "SOL"]


# =========================================================================================
#  DISTANZE DI FORMA
# =========================================================================================
def _euclid(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
    """Distanza euclidea fra q (W,) e ogni riga di lib (M,W). Forme gia' z-normalizzate."""
    return np.sqrt(((lib - q) ** 2).sum(axis=1))


def _corr_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
    """Distanza = 1 - correlazione di Pearson (q,lib gia' z-norm: corr = q.lib / W)."""
    # forme z-norm hanno media 0 std 1 -> dot/W e' la correlazione di Pearson
    corr = (lib @ q) / q.shape[0]
    return 1.0 - corr


def _cosine_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
    """Distanza = 1 - coseno fra q e ogni riga di lib."""
    qn = q / (np.linalg.norm(q) + 1e-12)
    ln = lib / (np.linalg.norm(lib, axis=1, keepdims=True) + 1e-12)
    return 1.0 - (ln @ qn)


def _dtw_one(a: np.ndarray, b: np.ndarray, band: int) -> float:
    """DTW 1D con banda di Sakoe-Chiba (|i-j|<=band). a,b stessa lunghezza W."""
    n = len(a)
    INF = 1e18
    prev = np.full(n + 1, INF)
    prev[0] = 0.0
    for i in range(1, n + 1):
        cur = np.full(n + 1, INF)
        jlo = max(1, i - band)
        jhi = min(n, i + band)
        ai = a[i - 1]
        for j in range(jlo, jhi + 1):
            cost = abs(ai - b[j - 1])
            m = prev[j]
            if prev[j - 1] < m:
                m = prev[j - 1]
            if cur[j - 1] < m:
                m = cur[j - 1]
            cur[j] = cost + m
        prev = cur
    return float(prev[n])


def _dtw_dist(q: np.ndarray, lib: np.ndarray, band: int) -> np.ndarray:
    """DTW di q contro ogni riga di lib. O(M * W * band)."""
    out = np.empty(lib.shape[0])
    for k in range(lib.shape[0]):
        out[k] = _dtw_one(q, lib[k], band)
    return out


DIST_FUNCS = {"euclid": _euclid, "corr": _corr_dist, "cosine": _cosine_dist}


# =========================================================================================
#  FILONE 1 — ANALOG con distanza configurabile (libreria sottocampionata, causale)
# =========================================================================================
def analog_dist_entries(df, dist="euclid", W=24, H=12, K=40, step=5, rebuild=500,
                        min_lib=2000, agree=0.62, dtw_band=4, dtw_prefilter=200,
                        decide_step=1, tp_atr=None, sl_atr=None,
                        trend_max=None, ema_long=200) -> list[dict]:
    """Analog kNN sulla FORMA con metrica `dist` ('euclid'|'corr'|'cosine'|'dtw').

    Libreria SOTTOCAMPIONATA: si considerano solo finestre che terminano a indici
    multipli di `step` (riduce N e rende DTW trattabile). Causalita': la libreria a
    decisione i contiene solo finestre con end<=i-1-H (esito gia' realizzato).
    Ricostruita ogni `rebuild` barre. Stessa firma per tutte le metriche -> confronto
    head-to-head a parita' di selettivita' (stesso W,H,K,agree).

    DTW (costoso, O(W*band) per coppia in Python): si PREFILTRA con la correlazione ai
    `dtw_prefilter` candidati piu' simili, poi si fa DTW-rerank solo su quelli (approccio
    standard lower-bound/rerank). `decide_step`>1 valuta una barra ogni decide_step (non
    cambia la causalita', riduce solo il numero di query DTW).
    """
    close = df["close"].values
    n = len(close)
    a = atr(df, 14)
    M, ends = znorm_windows(close, W)
    end_pos = {int(e): k for k, e in enumerate(ends)}
    fr = fwd_return(close, H)
    el = ema(close, ema_long) if trend_max is not None else None

    # candidati di libreria: solo end multipli di step (sottocampionamento causale fisso)
    base_ends = ends[(ends % step == 0)]

    entries: list[dict] = []
    lib_M = None
    lib_idx = None
    next_rebuild = 0

    for i in range(min_lib, n - 1):
        if i % decide_step != 0:
            continue
        if lib_M is None or i >= next_rebuild:
            elig = base_ends[(base_ends <= i - 1 - H) & (base_ends >= W - 1)]
            elig = elig[~np.isnan(fr[elig])]
            if len(elig) < max(K * 3, 200):
                next_rebuild = i + rebuild
                continue
            lib_M = M[[end_pos[int(e)] for e in elig]]
            lib_idx = elig
            next_rebuild = i + rebuild

        if lib_M is None:
            continue
        q = M[end_pos[i]]
        if not np.isfinite(q).all():
            continue
        if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max:
            continue

        if dist == "dtw":
            # prefiltro corr (cheap, vettoriale) -> DTW-rerank solo sui top dtw_prefilter
            pre = _corr_dist(q, lib_M)
            npre = min(dtw_prefilter, len(lib_idx))
            cand = np.argpartition(pre, npre - 1)[:npre]
            dd_cand = _dtw_dist(q, lib_M[cand], dtw_band)
            kk = min(K, len(cand))
            sub = np.argpartition(dd_cand, kk - 1)[:kk]
            nn = cand[sub]
            outs = fr[lib_idx[nn]]
            outs = outs[~np.isnan(outs)]
            if len(outs) < 5:
                continue
            mean_out = float(outs.mean())
            d = 1 if mean_out > 0 else -1
            frac = float(np.mean(np.sign(outs) == d))
            if frac < agree:
                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)
            continue

        dd = DIST_FUNCS[dist](q, lib_M)
        kk = min(K, len(lib_idx))
        nn = np.argpartition(dd, kk - 1)[:kk]
        outs = fr[lib_idx[nn]]
        outs = outs[~np.isnan(outs)]
        if len(outs) < 5:
            continue
        mean_out = float(outs.mean())
        d = 1 if mean_out > 0 else -1
        frac = float(np.mean(np.sign(outs) == d))
        if frac < agree:
            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 2 — TEMPLATE di forma canonici
# =========================================================================================
def make_templates(W: int) -> dict[str, np.ndarray]:
    """Template parametrici z-normalizzati di lunghezza W (forma pura, no scala/livello).

    Sono solo descrittori di FORMA recente (gli ultimi W close). La direzione attesa NON
    e' decisa a priori: viene stimata causalmente sul passato (vedi template_entries).
    """
    t = np.linspace(0, 1, W)
    s = 0.012  # ampiezza gaussiana scalata sulla finestra (W-indipendente in t in [0,1])
    g = lambda c: np.exp(-((t - c) ** 2) / s)
    raw = {
        # estremi di reversione a DOPPIO picco (due massimi / minimi simmetrici)
        "double_top":     g(0.25) + g(0.75),                       # M: due cime
        "double_bottom": -(g(0.25) + g(0.75)),                     # W: due fondi
        # testa-spalle: spalla-testa-spalla (centro piu' alto)
        "head_shoulders": g(0.2) + 1.7 * g(0.5) + g(0.8),
        "inv_head_shoulders": -(g(0.2) + 1.7 * g(0.5) + g(0.8)),
        # singola reversione
        "v_bottom":       np.abs(t - 0.5),
        "inv_v_top":     -np.abs(t - 0.5),
        "u_bottom":       (t - 0.5) ** 2,
        "arch_top":      -((t - 0.5) ** 2),
        # trend lineari
        "ramp_up":        t,
        "ramp_down":     -t,
    }
    out = {}
    for k, v in raw.items():
        v = np.asarray(v, dtype=float)
        sd = v.std()
        out[k] = (v - v.mean()) / (sd if sd > 0 else 1.0)
    return out


def template_entries(df, W=24, H=12, corr_min=0.85, dir_min=0.10, min_lib=2000,
                     rebuild=300, tp_atr=None, sl_atr=None, trend_max=None, ema_long=200,
                     templates=None) -> list[dict]:
    """Entries da match con template canonici, DIREZIONE stimata SOLO sul passato.

    A ogni i, per ogni template calcolo la correlazione di Pearson fra la forma recente
    z-norm (close[i-W+1..i]) e il template. Prendo il template a correlazione massima; se
    >= corr_min lo considero "attivo". La DIREZIONE in cui entrare e' il segno del rendimento
    forward MEDIO storico delle occorrenze gia' concluse (end+H<=i-1) di quel template
    (stesso criterio di match), purche' |media| in barre-equivalenti superi dir_min*media_atr-ish
    -> qui dir_min e' una soglia sulla |media forward| relativa (frazione). NIENTE direzione a
    priori: se il passato non e' coerente (occorrenze<min o segno debole) si salta.
    """
    close = df["close"].values
    n = len(close)
    a = atr(df, 14)
    M, ends = znorm_windows(close, W)
    end_pos = {int(e): k for k, e in enumerate(ends)}
    fr = fwd_return(close, H)
    el = ema(close, ema_long) if trend_max is not None else None
    tps = templates if templates is not None else make_templates(W)
    names = list(tps.keys())
    T = np.stack([tps[k] for k in names])             # (NT, W), gia' z-norm

    # match-history: per ogni end di libreria, quale template e con che corr
    # (precalcolo causale: per ogni end, corr con ogni template)
    # corr Pearson fra forme z-norm = dot/W
    lib_ends = ends[ends >= W - 1]
    lib_M = M[[end_pos[int(e)] for e in lib_ends]]    # (L, W)
    corr_mat = (lib_M @ T.T) / W                       # (L, NT)
    best_tpl = np.argmax(corr_mat, axis=1)
    best_corr = corr_mat[np.arange(len(lib_ends)), best_tpl]
    lib_fr = fr[lib_ends]
    lib_end_arr = lib_ends

    entries: list[dict] = []
    # cache direzione per template, ricostruita ogni rebuild barre
    dir_cache: dict[int, int] = {}
    next_rebuild = 0

    for i in range(min_lib, n - 1):
        q = M[end_pos[i]]
        if not np.isfinite(q).all():
            continue
        cq = (T @ q) / W                               # corr con ogni template
        bt = int(np.argmax(cq))
        if cq[bt] < corr_min:
            continue
        if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max:
            continue

        # direzione attesa: media forward causale delle occorrenze concluse dello stesso template
        if i >= next_rebuild:
            dir_cache = {}
            next_rebuild = i + rebuild
        if bt not in dir_cache:
            mask = (lib_end_arr <= i - 1 - H) & (best_tpl == bt) & (best_corr >= corr_min) & (~np.isnan(lib_fr))
            outs = lib_fr[mask]
            if len(outs) < 30:
                dir_cache[bt] = 0
            else:
                m = float(outs.mean())
                # soglia: |media| forward deve superare dir_min volte la std forward (edge vs rumore)
                sd = float(outs.std()) + 1e-12
                dir_cache[bt] = (1 if m > 0 else -1) if abs(m) / sd >= dir_min else 0
        d = dir_cache[bt]
        if d == 0:
            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


# =========================================================================================
#  CHECK CAUSALITA' espliciti
# =========================================================================================
def check_causality_analog(df, **kw) -> bool:
    """Le entries non devono cambiare se perturbo il FUTURO oltre l'ultima barra usata.
    Tronco il df a una certa lunghezza L e verifico che le entries con i<L-H-1 siano
    identiche a quelle calcolate sul df completo (la coda futura non le tocca)."""
    L = int(len(df) * 0.55)
    H = kw.get("H", 12)
    full = analog_dist_entries(df, **kw)
    trunc = analog_dist_entries(df.iloc[:L].reset_index(drop=True), **kw)
    horizon = L - H - 2
    f = {e["i"]: e["d"] for e in full if e["i"] < horizon}
    t = {e["i"]: e["d"] for e in trunc if e["i"] < horizon}
    ok = (f == t)
    print(f"  causalita' analog ({kw.get('dist','euclid')}): {'OK' if ok else 'VIOLATO'} "
          f"({len(f)} entries confrontate <{horizon})")
    return ok


def check_causality_template(df, **kw) -> bool:
    L = int(len(df) * 0.55)
    H = kw.get("H", 12)
    full = template_entries(df, **kw)
    trunc = template_entries(df.iloc[:L].reset_index(drop=True), **kw)
    horizon = L - H - 2
    f = {e["i"]: e["d"] for e in full if e["i"] < horizon}
    t = {e["i"]: e["d"] for e in trunc if e["i"] < horizon}
    ok = (f == t)
    print(f"  causalita' template: {'OK' if ok else 'VIOLATO'} "
          f"({len(f)} entries confrontate <{horizon})")
    return ok


# =========================================================================================
#  RUN
# =========================================================================================
def run_head_to_head(assets=SUBC_ASSETS, W=16, H=12, K=40, step=6, agree=0.62,
                     decide_step=4, dtw_prefilter=120):
    """Confronto HEAD-TO-HEAD delle metriche di forma a PARITA' di selettivita'.

    Tutte le metriche valutano le STESSE barre-decisione (decide_step) con lo STESSO
    W/H/K/agree: l'unica variabile e' la distanza. decide_step>1 serve a rendere DTW
    trattabile (pura Python ~9ms/query); applicato a tutte per equita'.
    """
    print("=" * 100)
    print(f"  FILONE 1 — ANALOG head-to-head metriche (W{W} H{H} K{K} step{step} "
          f"agree{agree} decide_step{decide_step}) | netto fee, OOS")
    print("=" * 100)
    results = {}
    for asset in assets:
        df = get_df(asset, "1h")
        print(f"\n  --- {asset} 1h (n={len(df)}) ---", flush=True)
        for dist in ["euclid", "corr", "cosine", "dtw"]:
            t0 = time.time()
            ents = analog_dist_entries(df, dist=dist, W=W, H=H, K=K, step=step, agree=agree,
                                       dtw_band=max(2, W // 5), dtw_prefilter=dtw_prefilter,
                                       decide_step=decide_step)
            dt = time.time() - t0
            res = evaluate(f"{dist:<7s}", ents, df)
            results[(asset, dist)] = res
            print(f"      ^ time={dt:>5.1f}s robust={'YES' if robust(res) else 'no '}", flush=True)
    return results


def run_templates(assets=SUBC_ASSETS, W=20, H=12, corr_min=0.85, dir_min=0.10):
    print("=" * 100)
    print(f"  FILONE 2 — TEMPLATE canonici (W{W} H{H} corr>={corr_min} dir>={dir_min}) | netto fee, OOS")
    print("=" * 100)
    results = {}
    for asset in assets:
        df = get_df(asset, "1h")
        print(f"\n  --- {asset} 1h (n={len(df)}) ---")
        for cm in [0.80, 0.85, 0.90]:
            ents = template_entries(df, W=W, H=H, corr_min=cm, dir_min=dir_min)
            res = evaluate(f"corr_min={cm}", ents, df)
            results[(asset, cm)] = res
            print(f"      ^ robust={'YES' if robust(res) else 'no '}")
    return results


def run_sweep():
    """Sweep largo (lento per via di DTW). Usa run_in_background."""
    print("=" * 100)
    print("  SWEEP LARGO — analog griglia W/H/K/step x metriche + template griglia")
    print("=" * 100)
    for W in [16, 20, 28]:
        for H in [8, 12, 24]:
            print(f"\n##### W={W} H={H} #####")
            run_head_to_head(W=W, H=H, K=40, step=6, agree=0.62)
    for W in [16, 20, 28]:
        for H in [8, 12, 24]:
            print(f"\n##### TEMPLATE W={W} H={H} #####")
            run_templates(W=W, H=H, corr_min=0.85, dir_min=0.10)


def run():
    np.random.seed(RNG_SEED)
    print("#" * 100)
    print("  SHAPE_TEMPLATE_RESEARCH — distanze di forma alternative + template canonici")
    print("#" * 100)
    # 1) check causalita' espliciti
    print("\n[CAUSALITA']")
    dfb = get_df("BTC", "1h")
    check_causality_analog(dfb, dist="euclid", W=20, H=12, K=40, step=6, min_lib=2000)
    check_causality_analog(dfb, dist="dtw", W=16, H=12, K=40, step=8, min_lib=2000,
                           dtw_band=3, decide_step=20)
    check_causality_template(dfb, W=20, H=12, corr_min=0.85)
    # 2) head-to-head metriche
    print()
    run_head_to_head()
    # 3) template
    print()
    run_templates()


if __name__ == "__main__":
    if "--sweep" in sys.argv:
        run_sweep()
    elif "--templates" in sys.argv:
        run_templates()
    elif "--h2h" in sys.argv:
        run_head_to_head()
    else:
        run()