PythagorasGoal/Old/scripts/waste/W24_donchian.py

"""DC01 — Donchian Channel Breakout con filtri.

Trend-following classico: quando il prezzo rompe il massimo/minimo degli
ultimi N periodi, entra nella direzione del breakout.

Completamente diverso dallo squeeze (che usa Bollinger/Keltner).
Donchian cattura breakout di RANGE, non di VOLATILITÀ.

IN:
    - OHLCV DataFrame
    - Parametri: channel_period, volume_filter, atr_stop, trend_filter

OUT:
    - Signal al breakout del canale Donchian
    - BacktestResult

Logica:
    1. Donchian upper = max(high, N periodi), lower = min(low, N periodi)
    2. Close > upper → LONG (breakout rialzista)
    3. Close < lower → SHORT (breakout ribassista)
    4. Filtri: volume, trend EMA, ATR minimo
"""
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


class DonchianBreakout(Strategy):
    name = "DC01_donchian"
    description = "Donchian Channel breakout — trend-following su range"
    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)

        period = params.get("channel_period", 48)
        use_vol = params.get("vol_filter", False)
        use_trend = params.get("trend_filter", False)
        cooldown = params.get("cooldown", 6)

        # EMA per trend filter
        ema_50 = np.full(n, np.nan)
        k = 2 / 51
        ema_50[49] = np.mean(c[:50])
        for i in range(50, n):
            ema_50[i] = c[i] * k + ema_50[i - 1] * (1 - k)

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

        signals = []
        last_signal_idx = -cooldown

        for i in range(period + 1, n):
            if i - last_signal_idx < cooldown:
                continue

            upper = np.max(h[i - period:i])
            lower = np.min(l[i - period:i])

            direction = 0
            if c[i] > upper:
                direction = 1
            elif c[i] < lower:
                direction = -1

            if direction == 0:
                continue

            # Trend filter: breakout must align with EMA trend
            if use_trend and not np.isnan(ema_50[i]):
                if direction == 1 and c[i] < ema_50[i]:
                    continue
                if direction == -1 and c[i] > ema_50[i]:
                    continue

            # Volume filter
            if use_vol and not np.isnan(vol_ma[i]):
                if v[i] < vol_ma[i] * 1.3:
                    continue

            signals.append(Signal(
                idx=i, direction=direction, entry_price=c[i],
                metadata={"upper": float(upper), "lower": float(lower)},
            ))
            last_signal_idx = i

        return signals


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

    configs = [
        ("p=24", {"channel_period": 24}),
        ("p=48", {"channel_period": 48}),
        ("p=96", {"channel_period": 96}),
        ("p=48+trend", {"channel_period": 48, "trend_filter": True}),
        ("p=48+vol", {"channel_period": 48, "vol_filter": True}),
        ("p=48+t+v", {"channel_period": 48, "trend_filter": True, "vol_filter": True}),
        ("p=96+t+v", {"channel_period": 96, "trend_filter": True, "vol_filter": True}),
    ]

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

    all_results.sort(key=lambda r: r.accuracy, reverse=True)
    print(f"\n{'=' * 120}")
    print(f"  DC01 DONCHIAN BREAKOUT — TOP 20")
    print(f"{'=' * 120}")
    for r in all_results[:20]:
        r.print_summary()
    if all_results:
        all_results[0].print_yearly()