AH-Oil-Trader/backtest_squeeze_entry.py

#!/usr/bin/env python3
"""Wie teuer ist der SPÄTE Squeeze-Einstieg — und hilft eine Abstands-Bremse?

ANLASS (2026-08-05, User): „Der Auto-Trade wurde am höchsten Punkt einer
aufsteigenden Kerze geschlossen und ein neuer Long eröffnet. Aus meiner Sicht ist
das nicht klug." Realfall 08:33:20 — Ausbruchs-Level **75,757**, Einstieg
**75,822** = **0,28×ATR darüber**, oberhalb des Hochs (75,808) der laufenden
M5-Kerze; die Kerze schloss dann bei 75,693.

⚠ DER VERDACHT IST EIN DEPLOYMENT-DRIFT-KANDIDAT, kein Bauchgefühl:
`backtest_breakout_squeeze.py` hat den Squeeze mit Einstieg **AM LEVEL** validiert
(+0,14…+0,23 ØR). Live läuft `_check_auto_squeeze` alle ~5 s und schickt eine
MARKET-Order — der Kurs ist dann schon ein Stück über dem Level. Gemessen unter X,
betrieben unter Y: dieselbe Klasse wie die fünf dokumentierten Fälle.

GEMESSEN WIRD DESHALB
  A) Einstieg AM LEVEL          (die Backtest-Annahme)
  B) Einstieg am BAR-CLOSE      (realistisch: so weit ist der Kurs gelaufen, wenn
                                 `_squeeze_one` den Ausbruch das erste Mal sieht)
  C) B + Abstands-Bremse: Ausbruch überspringen, wenn der Kurs bereits mehr als
     X×ATR über/unter dem Level steht („zu spät, der Zug ist weg")
  D) B + Bremse, aber die übersprungenen Ausbrüche NACHTRÄGLICH prüfen — steigt
     man nur später ein oder gar nicht? (Hier: gar nicht, wie live.)

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  Eine Bremse X wird nur empfohlen, wenn sie B in **ΣR und ØR in BEIDEN Hälften**
  schlägt UND die beiden Nachbar-Schwellen ebenfalls besser als B sind.

Sequentiell mit EINEM Positions-Slot (wie live), kanonischer Exit, Echtkosten.

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

import MetaTrader5 as mt5

from backtest_cost_gate import _atr_series, _berlin, kandidaten, kennzahlen
from core.exit_model import LIVE, simulate

_SCHWELLEN = (0.10, 0.15, 0.20, 0.30, 0.40)


def lauf(K, H, L, C, lo, hi, am_level: bool, max_abstand: float | None):
    """Sequentiell, EIN Slot. `am_level`=True → Backtest-Annahme, sonst Bar-Close.
    `max_abstand` in ×ATR → Ausbruch überspringen, wenn der Kurs schon weiter ist."""
    rs, abst = [], []
    frei_ab = lo
    for k in K:
        i = k["i"]
        if not (lo <= i < hi) or i < frei_ab:
            continue
        d, lvl, atr = k["d"], k["lvl"], k["atr"]
        weg = (C[i] - lvl) * d / atr          # wie weit ist der Kurs schon drüber?
        if max_abstand is not None and weg > max_abstand:
            continue                          # zu spät → gar nicht einsteigen
        entry = lvl if am_level else C[i]
        r, xb = simulate(entry, d, atr, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        # Kosten immer auf den ATR bezogen; beim Bar-Close kommt der bereits
        # gelaufene Weg als echter Nachteil über den schlechteren `entry` rein.
        rs.append(r - k["kosten"])
        abst.append(weg)
        frei_ab = xb + 1
    return rs, abst


def lauf_limit(K, H, L, C, lo, hi, warte: int, puffer: float = 0.0):
    """Variante D — LIMIT am Ausbruchs-Level statt Market.

    Die Order liegt `warte` Bars lang am Level (+`puffer`×ATR in Trade-Richtung
    zurückversetzt = etwas günstiger). Kommt der Kurs zurück → Fill zum Limit-Preis;
    kommt er nicht zurück → **kein Trade** (der Zug ist weg, das ist der Preis).
    Damit wird Variante A überhaupt erst erreichbar: dort wird der Level-Preis
    einfach unterstellt, obwohl der Kurs ihn schon verlassen hat."""
    rs = []
    frei_ab = lo
    verpasst = 0
    for k in K:
        i = k["i"]
        if not (lo <= i < hi) or i < frei_ab:
            continue
        d, atr = k["d"], k["atr"]
        limit = k["lvl"] - d * puffer * atr
        fill = None
        for j in range(i + 1, min(i + 1 + warte, len(C) - 1)):
            # LONG: Limit unter dem Kurs → Fill, wenn das Bar-Tief es erreicht
            if (L[j] <= limit) if d > 0 else (H[j] >= limit):
                fill = j
                break
        if fill is None:
            verpasst += 1
            continue
        r, xb = simulate(limit, d, atr, H, L, C, fill + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        rs.append(r - k["kosten"])
        frei_ab = xb + 1
    return rs, verpasst


def zeile(lbl, k1, k2, extra=""):
    def f(k):
        if not k:
            return f"{'—':>34}"
        return (f"{k['n']:>5} {k['wr']:>5.1f}% {k['oer']:>+7.3f} "
                f"{k['pf']:>5.2f} {k['sr']:>+8.1f}")
    print(f"  {lbl:<30}{f(k1)}  |{f(k2)}  {extra}")


def main():
    n_bars = 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_bars)
    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]; T = [int(b["time"]) for b in bars]
    SP = [float(b["spread"]) * point for b in bars]
    A = _atr_series(H, L, C)
    N = len(C); mid = N // 2

    print("=" * 104)
    print(f"  SQUEEZE-EINSTIEG: am Level vs. hinterhergelaufen — {sym} M5, {N} Bars")
    print(f"  {_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y} · 2 Halbjahre · "
          f"sequentiell (EIN Slot) · kanonischer Exit · Echtkosten")
    print("=" * 104)
    K = kandidaten(H, L, C, T, SP, A)

    # Wie weit läuft der Kurs typischerweise, bis der Ausbruch sichtbar ist?
    weg = sorted((C[k["i"]] - k["lvl"]) * k["d"] / k["atr"] for k in K)
    print(f"  Abstand Kurs↔Level beim ERSTEN Sichtbarwerden des Ausbruchs (×ATR):")
    print(f"    Median {weg[len(weg)//2]:.2f} · 75 % {weg[3*len(weg)//4]:.2f} · "
          f"90 % {weg[int(0.9*len(weg))]:.2f} · Max {weg[-1]:.2f}   (n={len(weg)})")
    print(f"    Der Realfall vom 05.08. lag bei 0,28×ATR — "
          f"{100.0*sum(1 for w in weg if w <= 0.28)/len(weg):.0f} % der Fälle sind näher dran.")
    print()

    print(f"  {'Variante':<30}{'H1  n    WR      OeR    PF     SigmaR':>34}"
          f"  |{'H2  n    WR      OeR    PF     SigmaR':>34}")
    print("  " + "-" * 100)
    erg = {}
    for lbl, am_level, mx in (("A  Einstieg AM LEVEL", True, None),
                              ("B  Einstieg am Bar-Close", False, None)):
        r1, _ = lauf(K, H, L, C, 0, mid, am_level, mx)
        r2, _ = lauf(K, H, L, C, mid, N, am_level, mx)
        erg[lbl] = (kennzahlen(r1), kennzahlen(r2))
        zeile(lbl, *erg[lbl])
    print("  " + "-" * 100)
    for x in _SCHWELLEN:
        lbl = f"C  Bremse: max {x:.2f}xATR"
        r1, _ = lauf(K, H, L, C, 0, mid, False, x)
        r2, _ = lauf(K, H, L, C, mid, N, False, x)
        erg[lbl] = (kennzahlen(r1), kennzahlen(r2))
        zeile(lbl, *erg[lbl])

    print("  " + "-" * 100)
    for warte, puf in ((3, 0.0), (6, 0.0), (12, 0.0), (6, 0.1), (12, 0.1)):
        lbl = f"D  Limit am Level {warte}B" + (f" +{puf:.1f}" if puf else "")
        r1, v1 = lauf_limit(K, H, L, C, 0, mid, warte, puf)
        r2, v2 = lauf_limit(K, H, L, C, mid, N, warte, puf)
        erg[lbl] = (kennzahlen(r1), kennzahlen(r2))
        zeile(lbl, *erg[lbl], extra=f"verpasst {v1}/{v2}")

    a1, a2 = erg["A  Einstieg AM LEVEL"]
    b1, b2 = erg["B  Einstieg am Bar-Close"]
    print()
    print("=" * 104)
    print("  BEFUND 1 — was kostet das Hinterherlaufen? (A gegen B)")
    print("=" * 104)
    print(f"    OeR   H1 {a1['oer']:+.3f} -> {b1['oer']:+.3f} ({b1['oer']-a1['oer']:+.3f})"
          f"      H2 {a2['oer']:+.3f} -> {b2['oer']:+.3f} ({b2['oer']-a2['oer']:+.3f})")
    print(f"    SigmaR H1 {a1['sr']:+.1f} -> {b1['sr']:+.1f} ({b1['sr']-a1['sr']:+.1f})"
          f"      H2 {a2['sr']:+.1f} -> {b2['sr']:+.1f} ({b2['sr']-a2['sr']:+.1f})")
    print("    ⚠ B ist der LIVE-Fall. Weicht es stark ab, gilt die validierte")
    print("      Squeeze-Kennzahl (+0,14…+0,23 OeR) fuer den Live-Betrieb NICHT.")

    print()
    print("=" * 104)
    print("  BEFUND 2 — hilft eine Abstands-Bremse? (Regel VOR dem Lauf fixiert:")
    print("             SigmaR UND OeR in BEIDEN Haelften > B, plus beide Nachbarn)")
    print("=" * 104)

    def schlaegt(x):
        k1, k2 = erg[f"C  Bremse: max {x:.2f}xATR"]
        return (k1 and k2 and k1["sr"] > b1["sr"] and k2["sr"] > b2["sr"]
                and k1["oer"] > b1["oer"] and k2["oer"] > b2["oer"])

    treffer = []
    for idx, x in enumerate(_SCHWELLEN):
        nb = [_SCHWELLEN[j] for j in (idx - 1, idx + 1) if 0 <= j < len(_SCHWELLEN)]
        selbst, nb_ok = schlaegt(x), all(schlaegt(m) for m in nb)
        if selbst and nb_ok:
            treffer.append(x)
        print(f"    X = {x:.2f}xATR   "
              + ("BESTEHT" if (selbst and nb_ok) else
                 "schlaegt B, Nachbarn NICHT" if selbst else "faellt durch"))
    print()
    if treffer:
        print(f"    => Kandidat(en): {', '.join(f'{x:.2f}' for x in treffer)}")
    else:
        print("    => KEINE Schwelle besteht. Eine Abstands-Bremse ist nicht belegt —")
        print("       uebersprungene Ausbrueche waeren im Schnitt NICHT schlechter.")

    print()
    print("=" * 104)
    print("  BEFUND 3 — schlaegt eine LIMIT-Order am Level den Market-Einstieg?")
    print("             (Regel: SigmaR UND OeR in BEIDEN Haelften > B, Nachbarn ebenso)")
    print("=" * 104)
    d_treffer = []
    d_keys = [k for k in erg if k.startswith("D ")]
    for k in d_keys:
        k1, k2 = erg[k]
        gut = (k1 and k2 and k1["sr"] > b1["sr"] and k2["sr"] > b2["sr"]
               and k1["oer"] > b1["oer"] and k2["oer"] > b2["oer"])
        if gut:
            d_treffer.append(k)
        print(f"    {k:<30} {'BESSER als Market' if gut else 'faellt durch'}")
    print()
    if d_treffer:
        print(f"    => {len(d_treffer)} von {len(d_keys)} Limit-Varianten schlagen den")
        print("       Market-Einstieg in beiden Haelften.")
    else:
        print("    => Auch die Limit-Order rettet es nicht.")
    print()


if __name__ == "__main__":
    main()