#!/usr/bin/env python3 """backtest_consolidation.py — KONSOLIDIERUNGS-SETUPS messen (B2, User-Vorgabe 2026-07-30 „lerne für die Gesamtempfehlung die Konsolidierungen"). Drei vom User beschriebene Setups, alle auf DERSELBEN Box: A) **False Breakout / Liquidity Sweep** — Ausbruch kehrt sofort zurück → Einstieg in die GEGENRICHTUNG (Stops der Ausbruchs-Trader wurden abgeräumt). B) **Distribution (bearish)** — Range NACH einem Aufwärtstrend: Smart Money verteilt in die Range hinein, danach Bruch nach UNTEN → Short. C) **Accumulation (bullish)** — Range NACH einem Abwärtstrend: leises Einsammeln, danach Bruch nach OBEN → Long. (+) **Retest-Variante** zu B/C: Einstieg erst, wenn der Kurs zum gebrochenen Level zurückkommt, statt direkt beim Bruch. ⚠ Die eigentliche NEUE Frage bei B/C: Ist die **Ausbruchs-RICHTUNG aus dem Trend VOR der Box vorhersagbar?** Der validierte Squeeze handelt den Ausbruch richtungs-neutral (er nimmt, was kommt). B/C behaupten das Gegenteil der naiven Trendfolge — Range nach Aufwärtstrend soll nach UNTEN brechen. Das ist deskriptiv sofort prüfbar und wird hier zuerst gemessen (Basisraten), bevor überhaupt ein R-Ertrag gerechnet wird. Setup laut User: · Preis bricht scheinbar aus der Konsolidierung aus (über Resistance / unter Support) · kehrt sofort zurück → Stops der Ausbruchs-Trader sind abgeräumt (Liquidity Sweep) · **Close zurück IN der Range** = Einstieg in die GEGENRICHTUNG · WTI-Tipp: gehäuft vor News (EIA/API) und am Session-Open ⚠ ABGRENZUNG — warum das trotz zweier heute verworfener Sweep-Tests neu ist: `backtest_liquidity_sweep.py` (2026-07-30) hat den Sweep an **Equal Highs/Lows** gemessen (zwei ~gleiche Swing-Pivots) → durchgefallen. HIER ist die Zone eine **echte Konsolidierungs-BOX** (enge Spanne über N Bars, wie beim validierten Squeeze) — andere Mechanik, eigene Messung. Der Bezug zum Squeeze ist der eigentliche Reiz: der **Ausbruch** aus der Box ist validiert (ØR +0,14…+0,23), trifft aber nur ~43 %. Die anderen ~57 % sind genau diese Fehlausbrüche. Wenn die Gegenrichtung nach dem Rückfall trägt, wäre das die **komplementäre Hälfte eines bereits belegten Setups** — die stärkste Ausgangslage, die eine neue Idee in diesem Projekt haben kann. Methodik (wie alle Tests hier): M5, sequentielle 1-Positions-Sim, Live-Exit (SL 2,0×ATR · Trail 1,5 ab 0,3 · BE 1,3 · Time-Stop 120 min), Kosten = Bar-Spread/ATR, 2 Halbjahre. Zusätzlich der Setup-eigene Exit (Stop hinter dem Fake-Break-Docht, Ziel = Gegenseite der Range) — genau wie im Referenzbild. Achsen: Rückkehr-Fenster (1/2/3 Bars) · Ausbruchstiefe · Exit-Variante · Session. Maßstab: Squeeze ØR +0,14…+0,23 · PF>1 in BEIDEN Hälften. Aufruf: python backtest_false_breakout.py [n_bars] """ from __future__ import annotations import sys from datetime import datetime, timezone import MetaTrader5 as mt5 # ── Konsolidierungs-Box: dieselbe Definition wie der validierte Squeeze ────── _BOX_N = 10 # Box = Spanne der letzten N abgeschlossenen M5-Bars _BOX_MULT = 2.5 # Box ≤ x×ATR = Konsolidierung _BRK_K = 0.1 # Ausbruch ab k×ATR über/unter die Box-Grenze _MAX_OVER = 1.0 # mehr als x×ATR jenseits der Box = echter Ausbruch, kein Fake _COOL = 12 # Exit (Live-Modell) _SL_ATR, _TRAIL, _TRAIL_ON, _BE_ON, _TIMESTOP, _MAXH = 2.0, 1.5, 0.3, 1.3, 24, 288 _ATRMIN = 0.06 # Sessions in Berlin-Stunden (Broker UTC+3 → Berlin = Broker − 1 h) _SESS = {"EIA/US-Open (14-17)": (14, 15, 16), "EU-Open (8-10)": (8, 9, 10), "beide": (8, 9, 10, 14, 15, 16)} # Vor-Trend (fuer Distribution/Accumulation): Kursaenderung ueber die N Bars VOR der # Box, in ATR. Ab _PRE_MIN gilt der Trend als "klar". _PRE_N = 40 _PRE_MIN = 1.0 _RETEST_W = 24 # Fenster fuer den Retest-Einstieg (Bars nach dem Bruch) _RETEST_TOL = 0.15 # Retest gilt als erreicht, wenn Kurs bis x*ATR ans Level zurueck def _atr_series(H, L, C, p=14): tr = [0.0] for i in range(1, len(C)): tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1]))) out = [None] * len(C); run = 0.0 for i in range(1, len(C)): run += tr[i] if i > p: run -= tr[i - p] out[i] = run / min(i, p) return out def _sim_live(entry, d, atr, H, L, C, j0): eff = entry - d * _SL_ATR * atr; hw = entry; started = False end = min(j0 + _MAXH, len(C) - 1); px = C[end] for j in range(j0, end + 1): if (L[j] <= eff) if d > 0 else (H[j] >= eff): px = eff; break hw = max(hw, H[j]) if d > 0 else min(hw, L[j]) prof = (C[j] - entry) * d if prof >= _TRAIL_ON * atr: started = True cand = hw - d * _TRAIL * atr if prof >= _BE_ON * atr: cand = max(cand, entry) if d > 0 else min(cand, entry) eff = max(eff, cand) if d > 0 else min(eff, cand) if not started and (j - j0) >= _TIMESTOP: px = C[j]; break return (px - entry) * d / atr def _sim_setup(entry, d, atr, stop, target, H, L, C, j0): """Setup-eigener Exit: Stop hinter dem Fake-Break-Docht, Ziel = Gegenseite der Range (die Lehrbuch-Variante aus dem Referenzbild).""" end = min(j0 + _MAXH, len(C) - 1); px = C[end] for j in range(j0, end + 1): hit_stop = (L[j] <= stop) if d > 0 else (H[j] >= stop) hit_tgt = (H[j] >= target) if d > 0 else (L[j] <= target) if hit_stop: px = stop; break # konservativ: Stop vor Ziel if hit_tgt: px = target; break return (px - entry) * d / atr def pre_trend(C, i, atr): """Trend VOR der Box: +1 auf, -1 ab, 0 unklar (Kursaenderung/ATR ueber _PRE_N).""" a = i - _BOX_N - _PRE_N if a < 0: return 0 d = (C[i - _BOX_N] - C[a]) / atr return 1 if d >= _PRE_MIN else -1 if d <= -_PRE_MIN else 0 def scan_dist_acc(H, L, C, SP, A, lo_i, hi_i, want_pre, retest, stats=None, flip=False, any_pre=False): """Distribution (want_pre=+1: Box nach Aufwaertstrend -> Bruch nach UNTEN) bzw. Accumulation (want_pre=-1: Box nach Abwaertstrend -> Bruch nach OBEN). `retest`: Einstieg erst beim Ruecklauf ans gebrochene Level. `stats`: [boxen, brueche_erwartete_richtung, brueche_gegenrichtung].""" Rs = [] def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr i = max(lo_i, _BOX_N + _PRE_N + 20) hi_i = min(hi_i, len(C) - _MAXH - 2) while i < hi_i: atr = A[i] if not atr or atr < _ATRMIN: i += 1; continue boxHi = max(H[i - _BOX_N:i]); boxLo = min(L[i - _BOX_N:i]) if (boxHi - boxLo) > _BOX_MULT * atr: i += 1; continue if not any_pre and pre_trend(C, i, atr) != want_pre: i += 1; continue up = boxHi + _BRK_K * atr; dn = boxLo - _BRK_K * atr d = 0; lvl = None if H[i] >= up: d, lvl = 1, up elif L[i] <= dn: d, lvl = -1, dn if d == 0: i += 1; continue # erwartete Richtung: Distribution (+1 Vor-Trend) -> runter, Accumulation -> hoch want_dir = (want_pre if flip else -want_pre) if stats is not None: stats[0] += 1 stats[1 if d == want_dir else 2] += 1 if d != want_dir: i += _COOL; continue j0 = i + 1; entry = lvl if retest: # auf Ruecklauf ans gebrochene Level warten (altes Support = neuer Widerstand) hit = None for j in range(i + 1, min(i + _RETEST_W, len(C) - _MAXH - 2)): back = (L[j] <= lvl + _RETEST_TOL * atr) if d > 0 else (H[j] >= lvl - _RETEST_TOL * atr) if back: hit = j; break if hit is None: i += _COOL; continue j0 = hit + 1; entry = lvl Rs.append(_sim_live(entry, d, atr, H, L, C, j0) - cost(i, atr)) i += _COOL return Rs def _rep(name, Rs): if not Rs or len(Rs) < 10: print(f" {name:<34} n={len(Rs):>4} (zu wenige)"); 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) pf = up / dn if dn > 0 else 9.99 print(f" {name:<34} n={n:>4} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} " f"PF={pf:.2f} ΣR={s:+.0f} worst={min(Rs):+.1f}") return s / n def scan(H, L, C, SP, A, BH, lo_i, hi_i, back_bars, setup_exit, hours=None, stats=None): """Ein Durchlauf. `back_bars` = innerhalb wie vieler Bars der Close zurück in die Box muss. `stats` sammelt (ausbrueche, davon_fake) für die Quote.""" Rs = [] def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr i = max(lo_i, _BOX_N + 20) hi_i = min(hi_i, len(C) - _MAXH - 2) while i < hi_i: atr = A[i] if not atr or atr < _ATRMIN: i += 1; continue boxHi = max(H[i - _BOX_N:i]); boxLo = min(L[i - _BOX_N:i]) if (boxHi - boxLo) > _BOX_MULT * atr: # keine Konsolidierung i += 1; continue up = boxHi + _BRK_K * atr; dn = boxLo - _BRK_K * atr # Ausbruchs-Bar suchen d0 = 0 if H[i] >= up: d0 = 1 elif L[i] <= dn: d0 = -1 if d0 == 0: i += 1; continue if stats is not None: stats[0] += 1 # zu tief jenseits der Box = echter Ausbruch, kein Fake over = (H[i] - boxHi) if d0 > 0 else (boxLo - L[i]) if over > _MAX_OVER * atr: i += _COOL; continue # FAKE-OUT: Close muss innerhalb `back_bars` zurück IN die Box back_at = None for j in range(i, min(i + back_bars, len(C) - _MAXH - 2)): if (C[j] < boxHi) if d0 > 0 else (C[j] > boxLo): back_at = j; break if back_at is None: i += _COOL; continue if stats is not None: stats[1] += 1 if hours is not None: hb = (datetime.fromtimestamp(int(BH[back_at]), timezone.utc).hour - 1) % 24 if hb not in hours: i += _COOL; continue d = -d0 # GEGENRICHTUNG des Fehlausbruchs entry = C[back_at] wick = max(H[i:back_at + 1]) if d0 > 0 else min(L[i:back_at + 1]) stop = wick + (0.05 * atr if d < 0 else -0.05 * atr) target = boxLo if d < 0 else boxHi # Gegenseite der Range R = (_sim_setup(entry, d, atr, stop, target, H, L, C, back_at + 1) if setup_exit else _sim_live(entry, d, atr, H, L, C, back_at + 1)) - cost(back_at, atr) Rs.append(R) i = back_at + _COOL return Rs def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000 mt5.initialize(); sym = None for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"): if mt5.symbol_info(c): sym = c; break bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n) si = mt5.symbol_info(sym); point = si.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] BH = [int(b["time"]) for b in bars] A = _atr_series(H, L, C) N = len(C); mid = N // 2 print("=" * 94) print(f" FALSE BREAKOUT / Liquidity Sweep aus der KONSOLIDIERUNG — {sym} M5 ({N} Bars)") print(f" Box={_BOX_N} Bars ≤{_BOX_MULT}×ATR · Ausbruch {_BRK_K}×ATR · Fake wenn Close " f"zurück in die Box") print(f" Einstieg = GEGENRICHTUNG · Maßstab: Squeeze ØR +0,14…+0,23 · PF>1 beidseitig") print("=" * 94) # Wie oft ist ein Box-Ausbruch überhaupt ein Fehlausbruch? st = [0, 0] scan(H, L, C, SP, A, BH, _BOX_N + 20, N, 3, False, None, st) if st[0]: print(f"\n Basisrate: {st[1]} von {st[0]} Box-Ausbrüchen kehren binnen 3 Bars " f"zurück = {100*st[1]/st[0]:.0f} % Fehlausbrüche") verdict = {} for back in (1, 2, 3): print(f"\n### Rückkehr binnen {back} Bar(s)") for exl, se in (("Live-Exit", False), ("Setup-Exit (Ziel=Gegenseite)", True)): print(f" --- {exl} ---") for hl, lo, hi in (("H1", _BOX_N + 20, mid), ("H2", mid, N)): Rs = scan(H, L, C, SP, A, BH, lo, hi, back, se) verdict.setdefault((back, exl), []).append(_rep(f"[{hl}]", Rs)) # ── B/C: Distribution & Accumulation ──────────────────────────────────── print("\n" + "=" * 94) print("### B/C — DISTRIBUTION & ACCUMULATION: ist die Ausbruchsrichtung aus dem") print(f" Vor-Trend vorhersagbar? (Vor-Trend = Kursaenderung ueber {_PRE_N} Bars, " f">= {_PRE_MIN}xATR)") for lbl, want in (("Distribution (Box nach AUFwaertstrend -> erwartet: Bruch RUNTER)", 1), ("Accumulation (Box nach ABwaertstrend -> erwartet: Bruch HOCH)", -1)): st = [0, 0, 0] scan_dist_acc(H, L, C, SP, A, _BOX_N + _PRE_N + 20, N, want, False, st) print(f"\n {lbl}") if st[0]: print(f" Basisrate: {st[1]} erwartete vs {st[2]} Gegenrichtung " f"= {100*st[1]/max(1,st[0]):.0f} % treffen die Erwartung " f"(Zufall waere ~50 %)") for retest, rl in ((False, "Einstieg beim Bruch"), (True, "Einstieg beim RETEST")): print(f" --- {rl} ---") for hl, lo, hi in (("H1", _BOX_N + _PRE_N + 20, mid), ("H2", mid, N)): Rs = scan_dist_acc(H, L, C, SP, A, lo, hi, want, retest) key = ("dist" if want == 1 else "acc", rl) verdict.setdefault(key, []).append(_rep(f"[{hl}]", Rs)) print(f"\n### SESSION-Filter (Rückkehr ≤2 Bars, Live-Exit)") for sess, hours in _SESS.items(): print(f" --- {sess} ---") for hl, lo, hi in (("H1", _BOX_N + 20, mid), ("H2", mid, N)): Rs = scan(H, L, C, SP, A, BH, lo, hi, 2, False, hours) verdict.setdefault(("sess", sess), []).append(_rep(f"[{hl}]", Rs)) print("\n" + "=" * 94) print(" URTEIL (Kandidat für die Gesamtempfehlung nur, wenn ØR in BEIDEN Hälften") print(" > 0 UND ≥ +0,14 = Squeeze-Band):") any_ok = False for key, vals in verdict.items(): if len(vals) != 2 or any(v is None for v in vals): print(f" {str(key):<44} unvollständig"); continue ok = all(v > 0 for v in vals); band = all(v >= 0.14 for v in vals) mark = "✅ TRÄGT" if (ok and band) else ("⚠ positiv, unter Band" if ok else "❌ fällt durch") if ok and band: any_ok = True if key[0] == "sess": lbl = f"Session {key[1]}" elif key[0] == "dist": lbl = f"Distribution · {key[1]}" elif key[0] == "acc": lbl = f"Accumulation · {key[1]}" else: lbl = f"FalseBreak Rückkehr ≤{key[0]} · {key[1]}" print(f" {lbl:<44} H1 {vals[0]:+.3f} · H2 {vals[1]:+.3f} {mark}") print("\n " + ("→ Kandidat: Parameter-Robustheit prüfen, dann als Verdict-Modul " "(Gewicht analog Squeeze)." if any_ok else "→ KEINE Variante trägt beidhälftig im Squeeze-Band → NICHT in die " "Gesamtempfehlung.")) if __name__ == "__main__": main()