PythagorasGoal/scripts/waste/W25_squeeze_retest.py

"""SB01 — Squeeze Breakout con Retest.

Il problema di SQ01/SQ02: entri al breakout, ma molti breakout sono fakeout.
Soluzione: aspetta il RETEST. Dopo il breakout, il prezzo spesso torna a
testare il livello di breakout prima di continuare.

Più selettivo di SQ02 → meno trade ma più accurati.
Anti-overfitting: meccanismo strutturale (retest è fenomeno di mercato reale).

IN:
    - OHLCV DataFrame
    - Parametri: bb_window, sq_threshold, retest_window (quante barre aspettare
                 il retest), retest_tolerance (quanto può tornare indietro)

OUT:
    - Signal al retest confermato (non al breakout iniziale)
    - BacktestResult

Logica:
    1. Rileva squeeze release (come SQ01)
    2. NON entrare subito — segna direzione e livello di breakout
    3. Nelle N barre successive, aspetta che il prezzo torni verso il livello
    4. Se il prezzo torna nel range di tolleranza e poi rimbalza → ENTRA
    5. Se il prezzo non torna → skip (momentum troppo forte, entry persa)
    6. Se il prezzo sfonda il livello → fakeout confermato, skip
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from src.strategies.base import Strategy, Signal
from src.strategies.indicators import keltner_ratio, detect_squeezes


class SqueezeBreakoutRetest(Strategy):
    name = "SB01_squeeze_retest"
    description = "Squeeze breakout con retest — entra solo dopo pullback confermato"
    default_assets = ["BTC", "ETH"]
    default_timeframes = ["15m", "1h"]
    fee_rt = 0.002

    def generate_signals(self, df, ts, **params):
        c = df["close"].values
        h = df["high"].values
        l = df["low"].values
        v = df["volume"].values
        n = len(c)

        bb_w = params.get("bb_window", 14)
        sq_thr = params.get("sq_threshold", 0.8)
        retest_window = params.get("retest_window", 8)
        retest_tol = params.get("retest_tolerance", 0.5)
        use_vol = params.get("vol_filter", False)

        kcr = keltner_ratio(c, h, l, bb_w)
        events = detect_squeezes(c, h, l, kcr, sq_thr)

        vol_ma = np.full(n, np.nan)
        for i in range(20, n):
            vol_ma[i] = np.mean(v[i - 20:i])

        signals = []

        for ev in events:
            brk_idx = ev["idx"]
            if brk_idx + retest_window + 3 >= n or brk_idx < 1:
                continue

            # Direzione breakout
            first_ret = (c[brk_idx] - c[brk_idx - 1]) / c[brk_idx - 1]
            if abs(first_ret) < 0.001:
                continue

            direction = 1 if first_ret > 0 else -1
            breakout_level = c[brk_idx - 1]
            breakout_move = abs(first_ret)

            # Aspetta retest nelle prossime N barre
            retest_found = False
            retest_idx = -1

            for j in range(brk_idx + 1, min(brk_idx + retest_window + 1, n)):
                if direction == 1:
                    # Long: il prezzo deve tornare GIÙ verso breakout_level
                    pullback = (h[brk_idx] - l[j]) / (h[brk_idx] - breakout_level) if h[brk_idx] > breakout_level else 0
                    if pullback >= retest_tol:
                        # Tornato abbastanza — ora deve rimbalzare
                        if c[j] > breakout_level:
                            retest_found = True
                            retest_idx = j
                            break
                        elif c[j] < breakout_level * 0.998:
                            # Sfondato sotto → fakeout
                            break
                else:
                    # Short: il prezzo deve tornare SU verso breakout_level
                    pullback = (h[j] - l[brk_idx]) / (breakout_level - l[brk_idx]) if breakout_level > l[brk_idx] else 0
                    if pullback >= retest_tol:
                        if c[j] < breakout_level:
                            retest_found = True
                            retest_idx = j
                            break
                        elif c[j] > breakout_level * 1.002:
                            break

            if not retest_found or retest_idx < 0:
                continue

            # Volume filter al retest
            if use_vol and not np.isnan(vol_ma[retest_idx]):
                if v[retest_idx] < vol_ma[retest_idx] * 0.8:
                    continue

            signals.append(Signal(
                idx=retest_idx, direction=direction,
                entry_price=c[retest_idx],
                metadata={
                    "breakout_idx": brk_idx,
                    "retest_bars": retest_idx - brk_idx,
                    "breakout_move": round(breakout_move * 100, 3),
                },
            ))

        return signals


if __name__ == "__main__":
    strategy = SqueezeBreakoutRetest()

    configs = [
        ("rt8 tol50%", {"retest_window": 8, "retest_tolerance": 0.5}),
        ("rt6 tol50%", {"retest_window": 6, "retest_tolerance": 0.5}),
        ("rt10 tol50%", {"retest_window": 10, "retest_tolerance": 0.5}),
        ("rt8 tol30%", {"retest_window": 8, "retest_tolerance": 0.3}),
        ("rt8 tol70%", {"retest_window": 8, "retest_tolerance": 0.7}),
        ("rt8 tol50%+vol", {"retest_window": 8, "retest_tolerance": 0.5, "vol_filter": True}),
        ("rt6 tol30%", {"retest_window": 6, "retest_tolerance": 0.3}),
        ("rt12 tol50%", {"retest_window": 12, "retest_tolerance": 0.5}),
    ]

    all_results = []
    for label, params in configs:
        for asset in ["BTC", "ETH"]:
            for tf in ["15m", "1h"]:
                for hold in [3, 6]:
                    r = strategy.backtest(asset, tf, hold=hold, **params)
                    if r and r.trades >= 30:
                        r.strategy_name = f"SB01 {label} h={hold}"
                        all_results.append(r)

    all_results.sort(key=lambda r: r.accuracy, reverse=True)
    print(f"\n{'=' * 130}")
    print(f"  SB01 SQUEEZE BREAKOUT RETEST — TOP 25")
    print(f"{'=' * 130}")
    for r in all_results[:25]:
        r.print_summary()
    if all_results:
        all_results[0].print_yearly()

    # Confronto con benchmark
    print(f"\n  BENCHMARK SQ02: 79.7% acc, 1250 trades, DD 6.5%, 9/9 anni")