AH-Oil-Trader/backtest_squeeze_srclose2.py

#!/usr/bin/env python3
"""Kappt der S/R-Auto-Close die Squeeze-Laeufer? — NEU unter dem HEUTIGEN Stand.

ANLASS (2026-08-13). Der Pending-Umbau vom 05.08. hat gewirkt (7 von 8 Bot-Trades
fuellen per ruhender Order), der Squeeze verliert aber weiter. Die Signatur zeigt
auf den Exit, nicht mehr auf den Einstieg:
    Trefferquote 73 % · OeGewinn 13,62 EUR gegen OeVerlust 94,15 EUR (Faktor 7)
    Drei von sieben Pending-Trades per `sr_close` bei +0,26 / +2,88 / +0,28 EUR
    geschlossen, waehrend die Verlierer bis zum vollen Stop liefen.
Der Squeeze lebt gemessen von LAEUFERN (43 % Treffer, Ertrag im Tail).

⚠⚠ WARUM DIE ALTE MESSUNG NICHT MEHR BINDET. `backtest_squeeze_srclose.py`
(24.07.) fand auf Squeeze-Entries einen WASH (+0,190→+0,224 / +0,394→+0,377) —
aber mit dem **alten** P(break)-Modell und Schwelle **0,55/0,60**. Seit dem
31.07. laeuft das **nachtrainierte** Modell mit Schwelle **0,35**, und das
schliesst live **94-97 %** aller Level-Beruehrungen. Gemessen unter X, betrieben
unter Y — dasselbe Muster wie die acht dokumentierten Deployment-Drift-Faelle.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  Der S/R-Close bleibt fuer Squeeze-Trades an, wenn er in BEIDEN Haelften
  mindestens so gut ist wie OHNE ihn (Variante A). Sonst: Squeeze-Trades vom
  S/R-Close ausnehmen — oder, falls eine Mindestgewinn-Schwelle in BEIDEN
  Haelften A schlaegt, diese setzen.

AUFBAU — live-treu, mit den geteilten Kernen:
  · Einstiege ueber `core.squeeze_scan.scan(on_entry=...)` = DERSELBE Code wie
    der Live-Squeeze. ⚠ `on_entry` statt `exit_fn`, sonst bekommt man still die
    Kontrollgruppe ALLER Box-Ausbrueche (real: 519 statt 132).
  · Einstieg als ruhende Stop-Order am Level, **Fill bei Beruehrung** — das ist
    das Live-Verhalten seit dem 05.08.
  · Exit `core.exit_model.simulate` mit dem `stop_when`-Haken; damit wird die
    Phasen-Mechanik NICHT kopiert (die Kopiererei hat vier divergierende
    Exit-Modelle erzeugt).
  · Level ueber `core.sr_levels` (M30-Pivots, geclustert) = derselbe Code wie
    `_draw_levels`. P(break) ueber `core.engine._p_break` mit den Live-Gewichten,
    Merkmale trainingsgleich aus M5.
  · Sequentiell EIN Slot, Echtkosten aus dem Bar-Spread, 2 Halbjahre, Bootstrap-KI.

⚠ ZWEI BEWUSSTE NAEHERUNGEN, ehrlich benannt:
  (1) Der Haken schliesst zum BAR-CLOSE; live feuert der Check im 1-s-Takt.
      Dieselbe Naeherung wie bei der Flip-Close-Messung.
  (2) Live glaettet P(break) ueber ~60 s und hat eine Hysterese von ±5 Pp. Auf
      5-Minuten-Bars ist beides gegenstandslos — jeder Bar ist eine frische
      Auswertung. Die Schwellen-Naehe wird dadurch etwas haeufiger ueberschritten
      als live, also wirkt der S/R-Close hier eher ZU stark. Das faellt zu
      Gunsten der Nullhypothese aus (kein Eingriff).

Aufruf:  python backtest_squeeze_srclose2.py [bars]
"""
from __future__ import annotations

import random
import sys
from bisect import bisect_left, bisect_right

import MetaTrader5 as mt5

from core.engine import _p_break
from core.exit_model import LIVE, simulate
from core.sr_levels import halten, linien, naechste
from core.squeeze_scan import atr_series, scan

_PIV_K = 3
_LOOK = 50
_ATR5_FLOOR = 0.12
_AT_LEVEL = 0.15      # „am Level" wie live (_sr_close_hint)
_WARTE = 8            # Bars, die die ruhende Order liegt
# Live-Parameter des Squeeze (core/wave_rec.py)
_SQ_N, _SQ_K, _SQ_MULT = 10, 0.1, 2.5
# ⚠⚠ ATR-FLOOR: `core.squeeze_scan` hat als Default ATRMIN_STD = 0,12 — LIVE und
#   der dokumentierte Referenzlauf (`backtest_squeeze_touchfill.py`) nutzen aber
#   **0,06** (am 2026-07-xx gemessen gesenkt, `backtest_atrfloor.py`: in BEIDEN
#   Haelften besser). Der erste Lauf hier nahm den Modul-Default und bekam damit
#   eine voellig andere Population — H1 127 gegen H2 483 Einstiege, weil der
#   hohe Floor die Niedrig-Vola-Haelfte wegfiltert. Dieselbe Fehlerklasse wie
#   das fehlende `set_entry_room(0.6)` am 05.08.: **ein aufgerufenes Gate ist
#   nicht dasselbe wie ein aktives Gate.**
_ATRMIN = 0.06


def agg(H, L, C, step):
    hh, hl, hc = [], [], []
    for s in range(0, len(C) - step + 1, step):
        hh.append(max(H[s:s + step])); hl.append(min(L[s:s + step]))
        hc.append(C[s + step - 1])
    return hh, hl, hc


def pivots_m30(H, L, C):
    """ALLE M30-Pivots EINMAL (Index + Preis) — O(log n) statt quadratisch."""
    hh, hl, _ = agg(H, L, C, 6)
    ph_i, ph_v, pl_i, pl_v = [], [], [], []
    for x in range(_PIV_K, len(hh) - _PIV_K):
        if hh[x] == max(hh[x - _PIV_K:x + _PIV_K + 1]):
            ph_i.append(x); ph_v.append(round(hh[x], 3))
        if hl[x] == min(hl[x - _PIV_K:x + _PIV_K + 1]):
            pl_i.append(x); pl_v.append(round(hl[x], 3))
    return (ph_i, ph_v), (pl_i, pl_v)


def level_serie(H, L, C, A5):
    """(res, sup) je M5-Bar — DERSELBE Weg wie `_draw_levels`, inkl. Hysterese."""
    (ph_i, ph_v), (pl_i, pl_v) = pivots_m30(H, L, C)
    out = [(None, None)] * len(C)
    pr = ps = None
    for i in range(len(C)):
        a = A5[i]
        if not a:
            continue
        a5f = max(a, _ATR5_FLOOR)
        j = i // 6 - 1
        if j < _PIV_K + 5:
            continue
        a1, b1 = bisect_left(ph_i, j - _LOOK), bisect_right(ph_i, j - _PIV_K)
        a2, b2 = bisect_left(pl_i, j - _LOOK), bisect_right(pl_i, j - _PIV_K)
        ph, pl = ph_v[a1:b1], pl_v[a2:b2]
        if not ph and not pl:
            continue
        r, s = naechste(linien(ph, pl, a5f), C[i])
        r = halten(r, pr, C[i], a5f, True)
        s = halten(s, ps, C[i], a5f, False)
        pr, ps = r, s
        out[i] = (r, s)
    return out


def merkmale(C, A5):
    """mom6 / mom3 / ema_diff je Bar — trainingsgleich aus M5."""
    def ema(v, p):
        k = 2 / (p + 1); o = [v[0]]
        for x in v[1:]:
            o.append(o[-1] + k * (x - o[-1]))
        return o
    ef, es = ema(C, 12), ema(C, 50)
    m6, m3, ed = [0.0] * len(C), [0.0] * len(C), [0.0] * len(C)
    for i in range(6, len(C)):
        a5f = max(A5[i] or _ATR5_FLOOR, _ATR5_FLOOR)
        m6[i] = (C[i] - C[i - 6]) / a5f
        m3[i] = (C[i] - C[i - 3]) / a5f
        ed[i] = (ef[i] - es[i]) / a5f
    return m6, m3, ed


def entries(H, L, C, A, SP, point, lo, hi):
    """Squeeze-Ausbrueche ueber den GETEILTEN Kern einsammeln."""
    got = []
    scan(H, L, C, A, SP, point, lo, hi, _SQ_MULT, box_n=_SQ_N, k=_SQ_K,
         atr_min=_ATRMIN,
         exit_fn=lambda e, d, a, h, l, c, j: 0.0,
         on_entry=lambda j, d, lvl, a, komp: (
             got.append({"j": j, "d": d, "lvl": lvl, "atr": a}) if komp else None))
    return got


def lauf(ein, H, L, C, SP, A5, LV, M6, M3, ED, schwelle, min_gain=0.0):
    """Sequentiell EIN Slot. `schwelle=None` → ohne S/R-Close."""
    Rs = []
    n_close = 0          # wie oft der S/R-Close wirklich gefeuert hat
    frei_ab = -1
    grenze = len(C) - LIVE.max_hold - 2
    for e in ein:
        i, d, lvl = e["j"], e["d"], e["lvl"]
        if i < frei_ab or i >= grenze:
            continue
        a5f = max(A5[i] or _ATR5_FLOOR, _ATR5_FLOOR)
        fill = None
        for k in range(i + 1, min(i + 1 + _WARTE, grenze)):
            if L[k] <= lvl <= H[k]:
                fill = k; break
        if fill is None:
            continue

        # ── (1) natuerlicher Exit ohne S/R-Close ─────────────────────────
        R, xb = simulate(lvl, d, a5f, H, L, C, fill + 1, LIVE, ret_bar=True)

        # ── (2) S/R-Close davor? ─────────────────────────────────────────
        # ⚠⚠ NICHT ueber `stop_when`. Der Haken schliesst zum BAR-CLOSE, und
        #   die Bedingung „am Level" ist ein Fenster von 0,15xATR ≈ 0,018 $ —
        #   dass ein 5-Minuten-Close genau darin landet, ist selten. Ein
        #   Probelauf zeigte deshalb Δ ≈ 0,000, waehrend der Close live
        #   94-97 % aller Beruehrungen schliesst. Live prueft den LAUFENDEN
        #   Kurs im Sekundentakt — das Analogon ist die Bar-SPANNE.
        # ⚠ Ausstieg zum ERSTEN Moment, in dem die Bedingung feuern kann
        #   (Level minus Fensterbreite), nicht zum Level selbst. Das ist das
        #   konservative Ende und entspricht dem 1-s-Loop.
        # ⚠ Strikt VOR `xb`: faellt beides in denselben Bar, hat der SL
        #   Vorrang — dieselbe pessimistische Konvention wie in `simulate`.
        if schwelle is not None:
            for j in range(fill + 1, xb):
                if profit_bedingung(j, d, lvl, a5f, min_gain, LV, C, M6, M3, ED,
                                    H, L, schwelle):
                    ziel = LV[j][0] if d > 0 else LV[j][1]
                    R = (ziel - d * _AT_LEVEL * a5f - lvl) * d / a5f
                    xb = j
                    n_close += 1
                    break
        Rs.append(R - SP[i] / a5f)
        frei_ab = xb + 1
    return Rs, n_close


def profit_bedingung(j, d, entry, a5f, min_gain, LV, C, M6, M3, ED, H, L, schwelle):
    """Feuert der S/R-Auto-Close in Bar `j`? — die Live-Bedingungen der Reihe nach."""
    ziel = LV[j][0] if d > 0 else LV[j][1]        # Ziel-Seite wie _sr_close_hint
    if ziel is None:
        return False
    # „am Level": der Kurs kam im Bar bis auf 0,15xATR heran (Bar-SPANNE, s. o.)
    nah = (H[j] >= ziel - _AT_LEVEL * a5f) if d > 0 else (L[j] <= ziel + _AT_LEVEL * a5f)
    if not nah:
        return False
    aus = ziel - d * _AT_LEVEL * a5f              # Preis im Moment des Ausloesens
    gewinn = (aus - entry) * d / a5f
    if gewinn <= 0 or gewinn < min_gain:          # live: nur im PLUS
        return False
    gap = abs(ziel - C[j])
    wt = 1.0 if ED[j] * d > 0 else 0.0
    return _p_break(M6[j] * d, M3[j] * d, wt, gap / a5f) < schwelle


def ci(Rs, n=3000, seed=42):
    if len(Rs) < 20:
        return None, None
    r = random.Random(seed); m = len(Rs)
    b = sorted(sum(Rs[r.randrange(m)] for _ in range(m)) / m for _ in range(n))
    return b[int(.025 * n)], b[int(.975 * n)]


def kenn(Rs):
    if not Rs:
        return None
    g = sum(x for x in Rs if x > 0); v = -sum(x for x in Rs if x < 0)
    u, o = ci(Rs)
    return {"n": len(Rs), "oer": sum(Rs) / len(Rs), "sr": sum(Rs),
            "wr": 100 * sum(1 for x in Rs if x > 0) / len(Rs),
            "pf": (g / v if v else float("inf")), "u": u, "o": o}


def zeile(name, k, basis=None):
    if not k:
        return f"  {name:<38}  —"
    kib = "      —      " if k["u"] is None else f"[{k['u']:+.3f} … {k['o']:+.3f}]"
    dlt = "" if basis is None else f"  Δ {k['oer'] - basis['oer']:+.3f}"
    cl = "" if k.get("cl") is None else f"  Close {k['cl']:>3} ({k['clq']:>4.1f}%)"
    return (f"  {name:<38} n={k['n']:>4} WR {k['wr']:>4.1f}% OeR {k['oer']:>+.3f} "
            f"PF {k['pf']:>5.2f} SR {k['sr']:>+7.1f} {kib}{dlt}{cl}")


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()
    if bars is None:
        print("  keine Bars"); return 1
    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]
    A5 = atr_series(H, L, C)
    LV = level_serie(H, L, C, A5)
    M6, M3, ED = merkmale(C, A5)
    mid = len(C) // 2
    e1 = entries(H, L, C, A5, SP, point, 60, mid)
    e2 = entries(H, L, C, A5, SP, point, mid, len(C))
    print(f"\n  {sym} · {len(C)} M5-Bars · Squeeze-Entries H1 {len(e1)} · H2 {len(e2)}")
    print(f"  Einstieg ruhend am Level (Fill bei Beruehrung) · kanonischer Exit · "
          f"Echtkosten · EIN Slot")
    print(f"  Regel vorab: der S/R-Close bleibt an, wenn er in BEIDEN Haelften "
          f"mindestens so gut ist wie OHNE ihn.\n")

    def rechne(schw, mg=0.0):
        r1, c1 = lauf(e1, H, L, C, SP, A5, LV, M6, M3, ED, schw, mg)
        r2, c2 = lauf(e2, H, L, C, SP, A5, LV, M6, M3, ED, schw, mg)
        k1, k2 = kenn(r1), kenn(r2)
        for k, c in ((k1, c1), (k2, c2)):
            if k:
                k["cl"] = c
                k["clq"] = 100 * c / k["n"] if k["n"] else 0
        return k1, k2

    b1, b2 = rechne(None)
    print("  ── BASIS: OHNE S/R-Close (nur kanonischer Exit) ───────────────")
    print(zeile("A) ohne S/R-Close        H1", b1))
    print(zeile("A) ohne S/R-Close        H2", b2))

    print("\n  ── MIT S/R-Close, Schwellen-Sweep (live = 0,35) ───────────────")
    for s in (0.30, 0.35, 0.45, 0.55):
        k1, k2 = rechne(s)
        mark = "  ← LIVE" if abs(s - 0.35) < 1e-9 else ""
        print(zeile(f"B) p_break < {s:.2f}        H1{mark}", k1, b1))
        print(zeile(f"B) p_break < {s:.2f}        H2{mark}", k2, b2))

    print("\n  ── ABHILFE-VARIANTE: Close erst ab Mindestgewinn (bei 0,35) ───")
    for mg in (0.5, 1.0, 1.5):
        k1, k2 = rechne(0.35, mg)
        print(zeile(f"C) 0,35 + erst ab {mg:.1f}xATR  H1", k1, b1))
        print(zeile(f"C) 0,35 + erst ab {mg:.1f}xATR  H2", k2, b2))

    l1, l2 = rechne(0.35)
    print("\n  ── URTEIL ─────────────────────────────────────────────────────")
    ok = bool(l1 and l2 and b1 and b2
              and l1["oer"] >= b1["oer"] and l2["oer"] >= b2["oer"])
    print(f"    S/R-Close (0,35) mindestens so gut wie ohne, in BEIDEN: "
          f"{'✅ ja — anlassen' if ok else '❌ nein'}")
    if not ok:
        print(f"      H1 {l1['oer']:+.3f} gegen {b1['oer']:+.3f}   "
              f"H2 {l2['oer']:+.3f} gegen {b2['oer']:+.3f}")
        print("      ➜ Squeeze-Trades vom S/R-Close ausnehmen, ODER eine "
              "Mindestgewinn-Schwelle, die BEIDE Haelften schlaegt (Tabelle C).")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())