AH-Oil-Trader/backtest_consolidation.py

#!/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()