AH-Oil-Trader/backtest_breakout_canon.py

#!/usr/bin/env python3
"""`breakout_k` mit dem KANONISCHEN Exit nachgerechnet (2026-08-04).

Offener Punkt aus CLAUDE.md: „**`breakout_k=0,3` ruht weiter auf dem
`LEGACY_BE10`-Exit** (Breakeven 1,0 statt live 1,3). Der Legacy-Recheck hat EIA
und Dead-Hours nachgerechnet, `breakout_k` nicht. Kandidat für den nächsten
Recheck." — Das wiegt schwer, weil es ein AKTIVES Gate ist: die erste
`block_reason`-Auswertung zeigte `breakout_pending` als Ursache von **63 %**
aller WARTEN-Zustände.

Aufbau: echte `_confirm_breakout`-Mechanik (stateful `_pend`, Neuverankerung bei
Richtungswechsel/Timeout) auf dem EMA12/50-Signal, Entry erst nach Bestätigung.
Exit = `core/exit_model.LIVE` (SL 2,0 · Trail 1,0 · BE 1,3 · Time-Stop · TP),
Echtkosten = Bar-Spread/ATR, sequentielle 1-Positions-Sim, 2 Halbjahre.

⚠ VORAB festgelegte Regel (gegen Mehrfachvergleich): ein anderer k-Wert wird nur
dann empfohlen, wenn er in BEIDEN Hälften mehr ΣR liefert als der Live-Wert 0,3
UND seine Nachbarwerte mithalten. Ein einzelner Ausreißer zählt nicht.

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

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.wave_rec import _EMA_FAST, _EMA_SLOW, _TREND_DEADBAND, _ema_series

_ATRMIN = 0.06
_COOL = 3
_PEND_TIMEOUT = 24          # Bars, bis eine unbestätigte Anwartschaft verfällt


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 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(H, L, C, A, SP, EF, ES, lo, hi, k):
    """k = geforderte Bewegung in ATR, bevor das Signal gehandelt wird. k=0 → aus."""
    def cost(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    Rs = []
    pend = None            # {"dir": ±1, "level": float, "bar": int}
    i = max(lo, _EMA_SLOW + 5)
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        sep = (EF[i] - ES[i]) / atr
        d = 0 if abs(sep) < _TREND_DEADBAND else (1 if sep > 0 else -1)
        if d == 0:
            pend = None; i += 1; continue

        if k <= 0:                       # Bestätigung aus → sofort handeln
            trigger = True
        else:
            if pend is None or pend["dir"] != d:
                pend = {"dir": d, "level": C[i], "bar": i}
                i += 1; continue
            if i - pend["bar"] > _PEND_TIMEOUT:       # Timeout → neu verankern
                pend = {"dir": d, "level": C[i], "bar": i}
                i += 1; continue
            fortschritt = (C[i] - pend["level"]) * d / atr
            if fortschritt <= -k:                    # gegen die Richtung → neu
                pend = {"dir": d, "level": C[i], "bar": i}
                i += 1; continue
            trigger = fortschritt >= k
            if not trigger:
                i += 1; continue

        r, xj = simulate(C[i], d, max(atr, _ATRMIN), H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        Rs.append(r - cost(i, max(atr, _ATRMIN)))
        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("=" * 92)
    print(f"  breakout_k — KANONISCHER Exit — {sym} M5 ({N} Bars, 2 Halbjahre, Echtkosten)")
    print(f"  Exit: SL {LIVE.sl_atr}× · Trail {LIVE.mult}× · BE {LIVE.be}× · Time-Stop · TP")
    print(f"  ⚠ Die Validierung von k=0,3 stammt aus `backtest_breakout.py` mit BE 1,0")
    print(f"     (LEGACY_BE10) — hier erstmals mit dem Live-Exit gegengerechnet.")
    print("=" * 92)
    erg = {}
    for k in (0.0, 0.2, 0.3, 0.5, 0.8, 1.0):
        s1 = st(run(H, L, C, A, SP, EF, ES, 0, mid, k))
        s2 = st(run(H, L, C, A, SP, EF, ES, 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)):
            if s:
                print(f"    {lbl}  n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
                      f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}")
            else:
                print(f"    {lbl}  — keine Trades")
    live = erg.get(0.3)
    print("\n" + "=" * 92)
    print("  Vergleich gegen den Live-Wert k=0,3 (ΣR-Differenz je Hälfte):")
    for k, (s1, s2) in sorted(erg.items()):
        if abs(k - 0.3) < 1e-9 or not (s1 and s2 and live[0] and live[1]):
            continue
        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 (s1['sum'] > live[0]['sum']
                                              and s2['sum'] > live[1]['sum']) else ""))
    print("\n  Regel (vorab): Wechsel nur, wenn ein k in BEIDEN Hälften besser ist")
    print("  UND seine Nachbarn mithalten. Ein einzelner Ausreißer zählt nicht.")
    print()


if __name__ == "__main__":
    main()