PythagorasGoal/scripts/analysis/intrabar_test.py

"""Test ingresso intra-barra: rottura banda squeeze rilevata sul 5m vs close 15m.

Domanda: entrando sul 5m appena il prezzo rompe la banda di Bollinger dello
squeeze (bande dall'ultima barra 15m CHIUSA -> nessun look-ahead), si recupera
parte del movimento che l'ingresso al close della barra 15m si perde?

Confronto a parita' di EXIT (stesso wall-clock): l'unica differenza e' il prezzo
d'ingresso (5m anticipato vs close 15m ritardato). La differenza di rendimento e'
esattamente lo "scatto" del breakout catturato in piu'.
"""
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 src.live.signal_engine import keltner_ratio

OOS_START = "2023-11-20"
BB_W = 14
SQ_THR = 0.8
MIN_DUR = 5
LEV = 3.0
POS = 0.15
M15 = 15 * 60 * 1000
M5 = 5 * 60 * 1000


def build_15m_levels(df15: pd.DataFrame) -> pd.DataFrame:
    c = df15["close"].values
    h = df15["high"].values
    l = df15["low"].values
    n = len(c)
    kcr = keltner_ratio(c, h, l, BB_W)
    ma = np.full(n, np.nan)
    sd = np.full(n, np.nan)
    for t in range(BB_W, n):
        w = c[t - BB_W + 1 : t + 1]
        ma[t] = w.mean()
        sd[t] = w.std()
    upper = ma + 2 * sd
    lower = ma - 2 * sd

    # durata squeeze consecutiva e maturita'
    dur = np.zeros(n, dtype=int)
    run = 0
    for t in range(n):
        if not np.isnan(kcr[t]) and kcr[t] < SQ_THR:
            run += 1
        else:
            run = 0
        dur[t] = run
    mature = dur >= MIN_DUR

    return pd.DataFrame({
        "ts15": df15["timestamp"].values,
        "close_time15": df15["timestamp"].values + M15,
        "close15": c,
        "upper": upper,
        "lower": lower,
        "mature": mature,
    })


def run_asset(asset: str, hold_min: int, fee_rt: float) -> dict:
    df5 = load_data(asset, "5m").reset_index(drop=True)
    df15 = load_data(asset, "15m").reset_index(drop=True)
    lvl = build_15m_levels(df15)

    d5 = pd.DataFrame({
        "ts5": df5["timestamp"].values,
        "close_time5": df5["timestamp"].values + M5,
        "close5": df5["close"].values,
    })

    # banda armata: ultima barra 15m CHIUSA prima della chiusura del bar 5m
    armed = pd.merge_asof(
        d5.sort_values("close_time5"),
        lvl[["close_time15", "upper", "lower", "mature"]].sort_values("close_time15"),
        left_on="close_time5", right_on="close_time15", direction="backward",
    )
    # barra 15m CONTENENTE il bar 5m (per l'ingresso ritardato a close 15m)
    cont = pd.merge_asof(
        d5.sort_values("ts5"),
        lvl[["ts15", "close15", "close_time15"]].rename(
            columns={"close_time15": "cont_close_time"}).sort_values("ts15"),
        left_on="ts5", right_on="ts15", direction="backward",
    )

    m = armed.copy()
    m["cont_close"] = cont["close15"].values
    m["cont_close_time"] = cont["cont_close_time"].values

    oos_ms = int(pd.Timestamp(OOS_START, tz="UTC").timestamp() * 1000)
    close5 = m["close5"].values
    ct5 = m["close_time5"].values
    upper = m["upper"].values
    lower = m["lower"].values
    mature = m["mature"].values
    cont_close = m["cont_close"].values
    cont_ct = m["cont_close_time"].values
    n = len(m)

    cap_e = cap_l = 1000.0   # equity ingresso early(5m) e late(15m)
    peak_e = peak_l = 1000.0
    dd_e = dd_l = 0.0
    trades = win_e = win_l = 0
    thrust_sum = 0.0
    fee = fee_rt * LEV
    busy_until = -1

    for i in range(n):
        if ct5[i] < oos_ms or ct5[i] <= busy_until:
            continue
        if not mature[i] or np.isnan(upper[i]):
            continue
        if close5[i] > upper[i]:
            d = 1
        elif close5[i] < lower[i]:
            d = -1
        else:
            continue

        entry_e = close5[i]
        entry_l = cont_close[i]
        exit_time = cont_ct[i] + hold_min * 60 * 1000
        # primo close 5m al/oltre exit_time
        j = np.searchsorted(ct5, exit_time, side="left")
        if j >= n:
            break
        exit_p = close5[j]

        ret_e = ((exit_p - entry_e) / entry_e) * d * LEV - fee
        ret_l = ((exit_p - entry_l) / entry_l) * d * LEV - fee
        thrust_sum += (entry_l - entry_e) / entry_e * d * 100  # scatto % (no leva)

        cb_e, cb_l = cap_e, cap_l
        cap_e = max(cb_e + cb_e * POS * ret_e, 10.0)
        cap_l = max(cb_l + cb_l * POS * ret_l, 10.0)
        peak_e = max(peak_e, cap_e); dd_e = max(dd_e, (peak_e - cap_e) / peak_e)
        peak_l = max(peak_l, cap_l); dd_l = max(dd_l, (peak_l - cap_l) / peak_l)
        trades += 1
        win_e += ret_e > 0
        win_l += ret_l > 0
        busy_until = exit_time

    return {
        "trades": trades,
        "avg_thrust": thrust_sum / trades if trades else 0.0,
        "early_win": win_e / trades * 100 if trades else 0.0,
        "late_win": win_l / trades * 100 if trades else 0.0,
        "early_ret": (cap_e / 1000 - 1) * 100,
        "late_ret": (cap_l / 1000 - 1) * 100,
        "early_dd": dd_e * 100,
        "late_dd": dd_l * 100,
    }


def main():
    for fee_rt in (0.002, 0.001):
        print("=" * 104)
        print(f"  INGRESSO INTRA-BARRA 5m vs CLOSE 15m — OOS da {OOS_START} | leva={LEV:.0f}x "
              f"| fee={fee_rt*100:.2f}% RT")
        print("  EARLY = entra al close 5m che rompe la banda | LATE = entra al close della barra 15m | stesso exit")
        print("=" * 104)
        print(f"  {'Asset':>5s}{'Hold':>6s}{'Trd':>6s}{'Scatto%':>9s}"
              f"{'EARLY win%':>12s}{'EARLY ret%':>12s}{'LATE win%':>11s}{'LATE ret%':>11s}{'Δret%':>9s}")
        print("  " + "-" * 100)
        for asset in ["BTC", "ETH"]:
            for hold_min in (15, 30, 45):
                r = run_asset(asset, hold_min, fee_rt)
                print(f"  {asset:>5s}{hold_min:>5d}m{r['trades']:>6d}{r['avg_thrust']:>+9.3f}"
                      f"{r['early_win']:>12.1f}{r['early_ret']:>+12.1f}"
                      f"{r['late_win']:>11.1f}{r['late_ret']:>+11.1f}"
                      f"{r['early_ret']-r['late_ret']:>+9.1f}")
            print("  " + "-" * 100)
        print("  Scatto% = movimento medio (no leva) catturato tra rottura 5m e close 15m, nella direzione.")
        print("  Δret% = vantaggio dell'ingresso anticipato. Se ~0 o negativo, il 5m non aiuta.\n")


if __name__ == "__main__":
    main()