#!/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 from bisect import bisect_left, bisect_right 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)} # ⚠ bisect statt linearer Filterung (2026-08-07). Die alte Fassung # scannte je Bar die KOMPLETTE Pivot-Liste = O(Bars x Pivots) und war # der Engpass des Projekts (80k-Lauf: ~9 min -> 8,9 s). PH/PL sind # sortiert, die Binaersuche liefert DIESELBE Menge — bewiesen in # tests/test_pivotfenster.py, nicht behauptet. _a1 = bisect_left(PH, i - _PIV_LOOK); _b1 = bisect_right(PH, i - _PIV_K) _a2 = bisect_left(PL, i - _PIV_LOOK); _b2 = bisect_right(PL, i - _PIV_K) w._pb_levels = {"ph": [C[j] for j in PH[_a1:_b1]], "pl": [C[j] for j in PL[_a2:_b2]]} 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()