PythagorasGoal/scripts/waste/W22_perpetual_only.py

"""S2-12: Strategie SOLO su perpetual — dati 100% reali.
Niente opzioni, niente IV stimata. Solo prezzo OHLCV.
Mix di approcci diversi da quelli già testati su main.

1. Intraday range breakout con filtro volatilità
2. Daily open range breakout (prima ora di trading)
3. RSI divergence (prezzo fa nuovo min/max, RSI no)
4. Close-to-close momentum filtrato da volatilità regime
5. Multi-timeframe confirmation (15m signal + 1h trend)

Test per-anno, onesto, con fee reali perpetual (0.05% taker × 2 = 0.1% roundtrip).
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from src.data.downloader import load_data

FEE_RT = 0.002  # 0.1% taker roundtrip
INITIAL = 1000
LEVERAGE = 3


def rsi(close, period=14):
    delta = np.diff(close)
    gain = np.where(delta > 0, delta, 0)
    loss = np.where(delta < 0, -delta, 0)
    result = np.full(len(close), 50.0)
    if len(gain) < period:
        return result
    ag = np.mean(gain[:period])
    al = np.mean(loss[:period])
    for i in range(period, len(delta)):
        ag = (ag * (period - 1) + gain[i]) / period
        al = (al * (period - 1) + loss[i]) / period
        if al == 0:
            result[i + 1] = 100
        else:
            result[i + 1] = 100 - 100 / (1 + ag / al)
    return result


def ema(arr, period):
    r = np.full(len(arr), np.nan)
    k = 2 / (period + 1)
    r[period - 1] = np.mean(arr[:period])
    for i in range(period, len(arr)):
        r[i] = arr[i] * k + r[i - 1] * (1 - k)
    return r


def run_all_perpetual(asset):
    print(f"\n{'#'*70}")
    print(f"  {asset} — STRATEGIE PERPETUAL (dati reali)")
    print(f"  Fee: {FEE_RT*100}% roundtrip, Leva: {LEVERAGE}x")
    print(f"{'#'*70}")

    df_1h = load_data(asset, "1h")
    df_15m = load_data(asset, "15m")
    c1h = df_1h["close"].values
    h1h = df_1h["high"].values
    l1h = df_1h["low"].values
    v1h = df_1h["volume"].values
    n1h = len(c1h)
    ts1h = pd.to_datetime(df_1h["timestamp"], unit="ms", utc=True)

    rsi_14 = rsi(c1h, 14)
    ema_20 = ema(c1h, 20)
    ema_50 = ema(c1h, 50)

    results = {}

    # ======================================================
    # STRAT 1: Daily Open Range Breakout
    # Prima ora (08-09 UTC) definisce il range. Breakout = entrata.
    # ======================================================
    for hold, stop_m in [(6, 1.0), (12, 1.5), (4, 0.8)]:
        name = f"ORB_h{hold}_s{stop_m}"
        capital = float(INITIAL)
        yearly = {}

        for i in range(50, n1h - hold):
            if ts1h.iloc[i].hour != 9:  # fine della prima ora
                continue
            day = ts1h.iloc[i].strftime("%Y-%m-%d")
            if day in yearly and len(yearly[day]) >= 1:
                continue

            range_high = h1h[i - 1]
            range_low = l1h[i - 1]
            range_size = range_high - range_low
            if range_size <= 0:
                continue

            # ATR per stop
            atr_14 = np.mean(h1h[max(0,i-14):i] - l1h[max(0,i-14):i])
            if atr_14 <= 0:
                continue

            # Breakout detection: la candela attuale rompe il range
            if c1h[i] > range_high:
                direction = "long"
            elif c1h[i] < range_low:
                direction = "short"
            else:
                continue

            entry = c1h[i]
            stop_dist = atr_14 * stop_m
            exit_price = c1h[min(i + hold, n1h - 1)]

            for j in range(i + 1, min(i + hold + 1, n1h)):
                if direction == "long":
                    if l1h[j] <= entry - stop_dist:
                        exit_price = entry - stop_dist
                        break
                    if h1h[j] >= entry + stop_dist * 2:
                        exit_price = entry + stop_dist * 2
                        break
                else:
                    if h1h[j] >= entry + stop_dist:
                        exit_price = entry + stop_dist
                        break
                    if l1h[j] <= entry - stop_dist * 2:
                        exit_price = entry - stop_dist * 2
                        break
                exit_price = c1h[j]

            trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry)
            net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE
            capital += capital * 0.15 * net
            capital = max(capital, 10)

            year = ts1h.iloc[i].year
            if year not in yearly:
                yearly[year] = []
            yearly[year].append(net > 0)
            if day not in yearly:
                yearly[day] = []

        if sum(len(v) for v in yearly.values() if isinstance(v, list) and all(isinstance(x, bool) for x in v)) > 30:
            all_wins = [w for v in yearly.values() if isinstance(v, list) for w in v if isinstance(w, bool)]
            acc = sum(all_wins) / len(all_wins) * 100
            ret = (capital - INITIAL) / INITIAL * 100
            results[name] = {"acc": acc, "ret": ret, "trades": len(all_wins), "capital": capital}

    # ======================================================
    # STRAT 2: RSI Divergence
    # Prezzo fa nuovo low, RSI no = bullish divergence → long
    # ======================================================
    for lookback, rsi_thr_low, rsi_thr_high, hold in [(20, 30, 70, 6), (14, 25, 75, 8), (10, 35, 65, 4)]:
        name = f"RSIdiv_lb{lookback}_h{hold}"
        capital = float(INITIAL)
        trades_list = []

        for i in range(max(50, lookback + 1), n1h - hold):
            day = ts1h.iloc[i].strftime("%Y-%m-%d")

            # Bullish divergence: price new low, RSI higher low
            price_new_low = c1h[i] < np.min(c1h[i - lookback : i])
            rsi_higher = rsi_14[i] > np.min(rsi_14[i - lookback : i]) and rsi_14[i] < rsi_thr_low

            # Bearish divergence: price new high, RSI lower high
            price_new_high = c1h[i] > np.max(c1h[i - lookback : i])
            rsi_lower = rsi_14[i] < np.max(rsi_14[i - lookback : i]) and rsi_14[i] > rsi_thr_high

            direction = None
            if price_new_low and rsi_higher:
                direction = "long"
            elif price_new_high and rsi_lower:
                direction = "short"

            if direction is None:
                continue

            entry = c1h[i]
            exit_price = c1h[min(i + hold, n1h - 1)]
            trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry)
            net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE
            capital += capital * 0.12 * net
            capital = max(capital, 10)
            trades_list.append({"year": ts1h.iloc[i].year, "win": trade_ret > 0})

        if len(trades_list) > 30:
            acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100
            ret = (capital - INITIAL) / INITIAL * 100
            results[name] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital}

    # ======================================================
    # STRAT 3: Momentum regime — trend following solo in low-vol regime
    # ======================================================
    for fast, slow, vol_w, vol_thr, hold in [
        (8, 21, 48, 0.8, 12),
        (5, 13, 24, 0.8, 6),
        (13, 34, 72, 0.7, 24),
        (8, 21, 48, 0.9, 8),
    ]:
        name = f"MomReg_f{fast}s{slow}_h{hold}"
        ema_f = ema(c1h, fast)
        ema_s = ema(c1h, slow)

        rv_short = np.full(n1h, np.nan)
        rv_long = np.full(n1h, np.nan)
        lr = np.diff(np.log(np.where(c1h == 0, 1e-10, c1h)))
        for idx in range(vol_w, len(lr)):
            rv_short[idx + 1] = np.std(lr[idx - min(12, vol_w) : idx])
            rv_long[idx + 1] = np.std(lr[idx - vol_w : idx])

        capital = float(INITIAL)
        trades_list = []
        daily_done = set()

        for i in range(max(60, slow + 1), n1h - hold):
            if np.isnan(ema_f[i]) or np.isnan(ema_s[i]) or np.isnan(rv_short[i]) or np.isnan(rv_long[i]):
                continue
            if rv_long[i] <= 0:
                continue

            day = ts1h.iloc[i].strftime("%Y-%m-%d")
            if day in daily_done:
                continue

            # Only trade in low-vol regime
            vol_ratio = rv_short[i] / rv_long[i]
            if vol_ratio > vol_thr:
                continue

            # EMA crossover signal
            cross_up = ema_f[i] > ema_s[i] and ema_f[i - 1] <= ema_s[i - 1]
            cross_down = ema_f[i] < ema_s[i] and ema_f[i - 1] >= ema_s[i - 1]

            if not (cross_up or cross_down):
                continue

            direction = "long" if cross_up else "short"
            entry = c1h[i]
            exit_price = c1h[min(i + hold, n1h - 1)]
            trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry)
            net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE
            capital += capital * 0.15 * net
            capital = max(capital, 10)
            trades_list.append({"year": ts1h.iloc[i].year, "win": trade_ret > 0})
            daily_done.add(day)

        if len(trades_list) > 30:
            acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100
            ret = (capital - INITIAL) / INITIAL * 100
            results[name] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital}

    # ======================================================
    # STRAT 4: Multi-TF confirmation (15m entry, 1h trend)
    # ======================================================
    c15 = df_15m["close"].values
    h15 = df_15m["high"].values
    l15 = df_15m["low"].values
    ts15 = df_15m["timestamp"].values
    n15 = len(c15)

    ema_1h_50 = ema(c1h, 50)
    rsi_15m = rsi(c15, 14)

    capital = float(INITIAL)
    trades_list = []
    daily_done = set()

    for i in range(100, n15 - 12):
        ts_dt = pd.Timestamp(ts15[i], unit="ms", tz="UTC")
        day = ts_dt.strftime("%Y-%m-%d")
        if day in daily_done:
            continue

        # 15m signal: RSI extreme
        if rsi_15m[i] > 35 and rsi_15m[i] < 65:
            continue

        # Find matching 1h candle
        h_idx = np.searchsorted(ts1h.values.astype("int64"), ts15[i]) - 1
        if h_idx < 50 or h_idx >= n1h or np.isnan(ema_1h_50[h_idx]):
            continue

        # 1h trend confirmation
        trend_up = c1h[h_idx] > ema_1h_50[h_idx]
        trend_down = c1h[h_idx] < ema_1h_50[h_idx]

        direction = None
        if rsi_15m[i] < 30 and trend_up:
            direction = "long"  # oversold in uptrend
        elif rsi_15m[i] > 70 and trend_down:
            direction = "short"  # overbought in downtrend

        if direction is None:
            continue

        entry = c15[i]
        hold_bars = 12  # 12 × 15m = 3h
        exit_price = c15[min(i + hold_bars, n15 - 1)]
        trade_ret = ((exit_price - entry) / entry if direction == "long" else (entry - exit_price) / entry)
        net = trade_ret * LEVERAGE - FEE_RT * LEVERAGE
        capital += capital * 0.12 * net
        capital = max(capital, 10)
        trades_list.append({"year": ts_dt.year, "win": trade_ret > 0})
        daily_done.add(day)

    if len(trades_list) > 30:
        acc = sum(1 for t in trades_list if t["win"]) / len(trades_list) * 100
        ret = (capital - INITIAL) / INITIAL * 100
        results["MultiTF_15m1h"] = {"acc": acc, "ret": ret, "trades": len(trades_list), "capital": capital}

    # === PRINT RESULTS ===
    print(f"\n  {'Strategy':<25s} {'Trades':>7s} {'Acc':>6s} {'Return':>10s} {'Capital':>10s}")
    print(f"  {'-'*60}")
    for name, r in sorted(results.items(), key=lambda x: x[1]["acc"], reverse=True):
        tag = "✅" if r["acc"] >= 60 and r["ret"] > 30 else ""
        print(f"  {name:<25s} {r['trades']:>7d} {r['acc']:>5.1f}% {r['ret']:>+9.1f}% €{r['capital']:>9,.0f} {tag}")


for asset in ["ETH", "BTC"]:
    run_all_perpetual(asset)