AH-Oil-Trader/backtest_breakout_gated.py

#!/usr/bin/env python3
"""`breakout_k` mit dem VOLLEN Live-Gate-Stack (2026-08-04).

Offener Punkt aus der Vollprüfung: `backtest_breakout_canon.py` zeigte, dass
k=0,5/0,8/1,0 den Live-Wert 0,3 in beiden Hälften schlagen — aber dort lief das
**ungegatete** EMA-Signal. Live stehen mehrere Gates davor, und die verschieben
die Population erheblich. Diese Messung schließt die Lücke.

Modelliert (alles über die ECHTEN Methoden, kein Nachbau):
  · `_build`  → Totband · HTF-Gegen-Trend (M30) · Anti-Überdehnung · Winkel-
                Konfidenz · Mindest-Konfidenz 55 · Reversal (per Schalter)
  · Breakout-Bestätigung → Mechanik 1:1 aus `_confirm_breakout` übernommen,
                aber **bar-basiert** statt Wall-Clock (s. u.)
  · `_room_gate` → mit echten M5-Pivots (`_pb_levels`) und `_pb_feats`

⚠ ZWEI bewusste Abweichungen, beide unvermeidbar und hier benannt:
 1. **Timeout bar-basiert.** `_confirm_breakout` misst den Timeout mit
    `time.time()` gegen `_breakout_timeout_s`=3600 s. In einem Bar-Loop vergeht
    keine Wall-Clock-Zeit — der Timeout würde NIE feuern. 3600 s sind auf M5
    exakt **12 Bars**; genau so wird hier gezählt. (Das ist zugleich ein Hinweis
    für künftige Backtests: jede Zeitmessung über `time.time()` im Live-Code ist
    in einer Bar-Simulation stumm.)
 2. **`hour=None`.** Würde man die Bar-Stunde übergeben, liefe die EIA-Prüfung in
    `_build` über `datetime.now(_BERLIN)` — also die ECHTE Uhrzeit des Laufs, für
    JEDEN Bar gleich. Ein Lauf am Mittwoch 15:30–16:30 würde damit alle Bars
    blocken. `dead_hours` ist live ohnehin leer; es fehlt also nur das
    EIA-Fenster (ein seltenes Ereignis-Gate).

Exit: `core/exit_model.LIVE` · Echtkosten (Bar-Spread/ATR) · 2 Halbjahre.
⚠ Regel VORAB: ein anderes k wird nur empfohlen, wenn es in BEIDEN Hälften mehr
ΣR liefert UND die Nachbarwerte mithalten.

Aufruf:  python backtest_breakout_gated.py [bars]
"""
import sys

import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _ANGLE_LR, _EMA_FAST, _EMA_SLOW,
                           _HTF_DEADBAND, _ema_series)

_ATRMIN = 0.06
_COOL = 3
_TIMEOUT_BARS = 12          # = _breakout_timeout_s 3600 s auf M5
_PIV_K = 3                  # Pivot-Fenster wie in `refresh_market`
_PIV_LOOK = 300             # Pivot-Rückblick wie live
_ATR_FLOOR_PB = 0.12        # `_pb_feats`-Floor wie live


class _TU:
    def snapshot(self):
        return {}


def _atr_series(H, L, C, p=14):
    t = [0.0]
    for i in range(1, len(C)):
        t.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    return [(sum(t[max(1, i - p + 1):i + 1]) / max(1, len(t[max(1, i - p + 1):i + 1])))
            if i else None for i in range(len(C))]


def _htf_sign(C, i, n, atr):
    """M30-/H1-Trend aus M5 aggregiert: EMA12/50 auf jedem n-ten Close."""
    lo = max(0, i - n * (_EMA_SLOW + 5))
    sub = C[lo:i + 1:n]
    if len(sub) < _EMA_SLOW + 2 or not atr:
        return 0
    ef = _ema_series(sub, _EMA_FAST)[-1]; es = _ema_series(sub, _EMA_SLOW)[-1]
    d = (ef - es) / atr
    return 0 if abs(d) < _HTF_DEADBAND else (1 if d > 0 else -1)


def st(Rs):
    if not Rs:
        return None
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
    return dict(n=n, wr=100 * w / n, oR=s / n, pf=(up / dn if dn > 0 else 9.99), sum=s)


def run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL, lo, hi, k):
    """Ein Durchlauf mit dem vollen Gate-Stack für ein bestimmtes k."""
    def cost(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    Rs = []
    pend = None                       # spiegelt `_pend` aus `_confirm_breakout`
    i = max(lo, _PIV_LOOK + _EMA_SLOW + 5)
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        # Raum-Gate mit echten, KAUSALEN M5-Pivots füttern (ein Pivot bei j ist
        # erst ab j+_PIV_K bekannt — sonst schaut das Gate in die Zukunft).
        w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
        w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
                        "pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
        snap = {}
        rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
                          htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
        sig = rec.get("signal")

        # ── Breakout-Bestätigung: Mechanik 1:1 aus `_confirm_breakout`,
        #    Timeout bar-basiert statt Wall-Clock (s. Modulkopf).
        if k > 0 and sig in ("LONG", "SHORT"):
            d = 1 if sig == "LONG" else -1
            if pend is None or pend["dir"] != d:
                pend = {"dir": d, "level": C[i] + d * k * atr,
                        "invalid": C[i] - d * k * atr, "confirmed": False, "bar": i}
            if not pend["confirmed"]:
                broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
                against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
                if broke:
                    pend["confirmed"] = True
                else:
                    if against or (i - pend["bar"]) > _TIMEOUT_BARS:
                        pend = {"dir": d, "level": C[i] + d * k * atr,
                                "invalid": C[i] - d * k * atr,
                                "confirmed": False, "bar": i}
                    i += 1; continue          # noch nicht bestätigt → WARTEN
        elif sig == "WARTEN":
            pend = None

        if sig not in ("LONG", "SHORT"):
            i += 1; continue
        rec = w._room_gate(rec, C[i])         # Entry-Raum-Gate (echte Methode)
        if rec.get("signal") not in ("LONG", "SHORT"):
            i += 1; continue

        d = 1 if rec["signal"] == "LONG" else -1
        a = max(atr, _ATRMIN)
        r, xj = simulate(C[i], d, a, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        Rs.append(r - cost(i, a))
        pend = None
        i = xj + _COOL
    return Rs


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    mt5.initialize()
    sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
                if mt5.symbol_info(c)), None)
    bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    H = [float(b["high"]) for b in bars]; L = [float(b["low"]) for b in bars]
    C = [float(b["close"]) for b in bars]; SP = [float(b["spread"]) * point for b in bars]
    A = _atr_series(H, L, C)
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
    N = len(C); mid = N // 2

    print("Vorberechnung (Winkel, HTF-Trends, Pivots) ...", flush=True)
    AD = [90.0] * N
    for i in range(_ANGLE_LR + 2, N):
        AD[i] = calc_trend_angle(C[i - (_ANGLE_LR + 1):i + 1], _ANGLE_LR)
    HTF = [0] * N; H1 = [0] * N
    for i in range(N):
        if A[i]:
            HTF[i] = _htf_sign(C, i, 6, A[i])       # M30 = 6 × M5
            H1[i] = _htf_sign(C, i, 12, A[i])       # H1  = 12 × M5
    PH = [j for j in range(_PIV_K, N - _PIV_K)
          if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
    PL = [j for j in range(_PIV_K, N - _PIV_K)
          if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]

    w = WaveRecommender(_TU(), mt5.TIMEFRAME_M5)
    w.set_reversal_enabled(False)          # Live-Stand seit 04.08.
    w.set_entry_room(0.6)                  # Live-Wert
    w.set_dead_hours([])                   # live leer

    print("=" * 94)
    print(f"  breakout_k mit VOLLEM Gate-Stack — {sym} M5 ({N} Bars, 2 Halbjahre)")
    print(f"  Gates: Totband · M30-Filter · Anti-Überdehnung · Winkel · min_conf 55 ·")
    print(f"         Entry-Raum 0,6×ATR (echte M5-Pivots) · Reversal AUS")
    print(f"  Exit: SL {LIVE.sl_atr}× · Trail {LIVE.mult}× · BE {LIVE.be}× · Time-Stop · TP")
    print("=" * 94)
    erg = {}
    for k in (0.0, 0.2, 0.3, 0.5, 0.8, 1.0):
        w.set_breakout_k(k)
        s1 = st(run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL, 0, mid, k))
        s2 = st(run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL,
                    mid, N - LIVE.max_hold - 1, k))
        erg[k] = (s1, s2)
        tag = "  ← LIVE" if abs(k - 0.3) < 1e-9 else ""
        print(f"\n  k = {k:.1f}{tag}")
        for lbl, s in (("H1 (alt)", s1), ("H2 (neu)", s2)):
            print(f"    {lbl}  " + (f"n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
                                    f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}"
                                    if s else "— keine Trades"))
    live = erg.get(0.3)
    if live and live[0] and live[1]:
        print("\n" + "=" * 94)
        print("  ΣR-Differenz gegen den Live-Wert k=0,3:")
        for k, (s1, s2) in sorted(erg.items()):
            if abs(k - 0.3) < 1e-9 or not (s1 and s2):
                continue
            beide = s1["sum"] > live[0]["sum"] and s2["sum"] > live[1]["sum"]
            print(f"    k={k:.1f}:  H1 {s1['sum'] - live[0]['sum']:+7.0f}   "
                  f"H2 {s2['sum'] - live[1]['sum']:+7.0f}"
                  + ("   << in BEIDEN besser" if beide else ""))
    print()


if __name__ == "__main__":
    main()