AH-Oil-Trader/backtest_squeeze_exit3.py

#!/usr/bin/env python3
"""Gegen-Squeeze als EXIT — dritte Messung, jetzt mit dem kanonischen Exit.

ANLASS (User, 2026-08-05): Trade T=49423144 (SELL 3,49 @ 75,012) lief in 14 min
1,135 $ ins Minus und endete am Initial-SL bei **−343 €**. Wunsch: „der Breakout
hätte den Trade viel früher schließen sollen."

⚠ WAS AN DEM TAG WIRKLICH PASSIERT IST (rekonstruiert):
  * Beim Einstieg war der Squeeze **active SHORT** — er stand also FÜR den Trade,
    nicht dagegen. Die Empfehlung war WARTEN (`block=entry_room`), der Einstieg
    manuell.
  * Der Gegen-Ausbruch (armed LONG @ 75,729 auf dem 17:05-Bar) wurde erst
    ~3–5 min SPÄTER aktiv. Ein Close dort hätte bei ≈75,73 stattgefunden:
    (75,73 − 75,012) × 3,49 Lots × 87,6 ≈ **−219 €** statt −343 € → ~124 € gespart.
  Das ist EIN Fall. Die Idee ist im Projekt bereits ZWEIMAL gemessen und verworfen
  (`backtest_squeeze_exit.py` 22.07., `backtest_squeeze_exit_vol.py` 24.07.).

WARUM TROTZDEM EINE DRITTE MESSUNG: beide Vorläufer stammen von VOR der
Exit-Vereinheitlichung am 31.07. und ruhen damit auf einem Exit-Modell, das dem
Live-System nicht entspricht (Trail 1,5 statt 1,0, teils ohne Time-Stop). Genau
diese Konstellation hat heute schon mehrere alte Schlüsse gekippt.

MESSUNG: kanonischer Exit (`exit_model.simulate`) + der dokumentierte
**`stop_when`-Hook** — dadurch muss die Phasen-Mechanik NICHT kopiert werden
(genau diese Kopiererei hat im Projekt vier divergierende Exit-Modelle erzeugt).
Zwei Populationen (Squeeze-Ausbrüche, gegatete Wellensignale), feste Entry-Listen,
Echtkosten, 2 Halbjahre.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  Einbauen nur, wenn eine Variante ΣR **und** ØR in BEIDEN Halbjahren und auf
  BEIDEN Populationen verbessert. Sonst bleibt es beim Nein — dann zum dritten Mal.

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

import MetaTrader5 as mt5

import core.wave_rec as wr
from backtest_cost_gate import _atr_series, kandidaten
from backtest_trail_be import _ema_series, _htf_sign, wellen_entries
from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _EMA_FAST, _EMA_SLOW, _N_BARS,
                           _ANGLE_LR, _ANGLE_DEAD)

_ATRMIN = 0.06
_PIV_K, _PIV_LOOK = 3, 300


def squeeze_richtung(H, L, C, A):
    """Je Bar: +1 wenn ein LONG-Ausbruch aktiv ist, −1 bei SHORT, sonst 0.
    Nutzt die ECHTE `_squeeze_one` — kein Nachbau."""
    out = [0] * len(C)
    for i in range(wr._SQ_N + 2, len(C)):
        a = A[i]
        if not a or a < _ATRMIN:
            continue
        w0 = i - wr._SQ_N - 2
        st = WaveRecommender._squeeze_one(H[w0:i + 1], L[w0:i + 1], C[w0:i + 1], a)
        if st.get("state") == "active":
            out[i] = 1 if st.get("dir") == "LONG" else -1
    return out


def lauf(K, H, L, C, SQ, lo, hi, modus, min_verlust=0.0):
    """modus: 'aus' | 'immer' | 'nur_minus' | 'eng_sl'.

    ⚠ `eng_sl` ist die User-Idee vom 05.08.: beim Gegen-Ausbruch NICHT sofort
    schliessen, sondern den Stop ENG ans aktuelle Niveau ziehen. Der Unterschied
    ist inhaltlich: die Sofort-Variante fiel gemessen durch, WEIL Gegen-Ausbrueche
    oft Fehlausbrueche sind und die Position zurueckprallt. Ein enger Stop steigt
    nur aus, wenn die Bewegung WEITERLAEUFT — genau der Einwand.
    Modelliert ueber den Hook: Preis beim ersten Gegen-Ausbruch merken, schliessen
    sobald es `min_verlust` xATR WEITER dagegen geht.
    ⚠ Bar-Aufloesung: ein enger Stop wird von dieser Simulation systematisch
    UEBERSCHAETZT (s. analyze_trail_resolution.py, 05.08.) — die Zahlen unten sind
    also die optimistische Obergrenze."""
    rs = []
    frei = lo
    ausgeloest = 0
    for k in K:
        if not (lo <= k["i"] < hi) or k["i"] < frei:
            continue
        d = k["d"]

        traf = []
        ausl = []          # Profit-Stand beim ERSTEN Gegen-Ausbruch

        def stop_when(j, profit, _d=d, _t=traf, _a=ausl):
            if modus == "aus":
                return False
            if modus == "eng_sl":
                if not _a:
                    if SQ[j] != -_d:
                        return False
                    _a.append(profit)          # Stop ab hier eng nachziehen
                    return False
                if profit <= _a[0] - min_verlust:
                    _t.append(j); return True
                return False
            if SQ[j] != -_d:                  # kein GEGEN-Ausbruch
                return False
            if modus == "nur_minus" and profit > -min_verlust:
                return False
            _t.append(j)
            return True

        r, xb = simulate(k["lvl"], d, k["atr"], H, L, C, k["i"] + 1, LIVE,
                         timestop=True, use_tp=True,
                         stop_when=(None if modus == "aus" else stop_when),
                         ret_bar=True)
        rs.append(r - k["kosten"])
        ausgeloest += 1 if traf else 0
        frei = xb + 1
    return rs, ausgeloest


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


def zeile(lbl, k1, k2, b1, b2):
    def f(k):
        if not k:
            return f"{'—':>36}"
        return (f"{k['n']:>5} {k['wr']:>5.1f}% {k['oer']:>+7.3f} "
                f"{k['pf']:>5.2f} {k['sr']:>+8.1f}")
    gut = bool(k1 and k2 and b1 and b2 and k1["oer"] > b1["oer"]
               and k2["oer"] > b2["oer"] and k1["sr"] > b1["sr"] and k2["sr"] > b2["sr"])
    print(f"    {lbl:<26}{f(k1)}  |{f(k2)}" + ("   BESSER" if gut else ""))
    return gut


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)
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
    AD = [0.0] * len(C)
    for i in range(_N_BARS, len(C)):
        AD[i] = calc_trend_angle(C[i - _N_BARS:i + 1], _ANGLE_LR) - 90.0
        if abs(AD[i]) < _ANGLE_DEAD:
            AD[i] = 0.0
    HTF = [_htf_sign(C, i, 6, A[i] or 0.1) for i in range(len(C))]
    H1S = [_htf_sign(C, i, 12, A[i] or 0.1) for i in range(len(C))]
    PH = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
    PL = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]

    w = WaveRecommender(lambda: None)
    w.set_reversal_enabled(False)
    w.set_entry_room(0.6)
    w.set_dead_hours("")
    mid = len(C) // 2

    print("=" * 100)
    print(f"  GEGEN-SQUEEZE ALS EXIT — 3. Messung · {sym} M5, {len(C)} Bars")
    print(f"  kanonischer Exit (Trail {LIVE.mult}) + stop_when-Hook · Echtkosten · 2 Halbjahre")
    print("  ⚠ Idee bereits 2x verworfen (22.07. / 24.07.) — beide auf einem NICHT-Live-Exit.")
    print("=" * 100)
    SQ = squeeze_richtung(H, L, C, A)
    print(f"  Bars mit aktivem Ausbruch: {sum(1 for x in SQ if x)} von {len(SQ)}")

    pops = [("SQUEEZE-Entries", kandidaten(H, L, C, T, SP, A)),
            ("WELLE (gegated)",
             wellen_entries(w, H, L, C, A, SP, EF, ES, AD, HTF, H1S, PH, PL))]
    treffer = {}
    for name, K in pops:
        print(f"\n  {name} — {len(K)} Entries")
        print(f"    {'Variante':<26}{'H1  n    WR      OeR    PF     SigmaR':>36}"
              f"  |{'H2  n    WR      OeR    PF     SigmaR':>36}")
        b1 = kenn(lauf(K, H, L, C, SQ, 0, mid, "aus")[0])
        b2 = kenn(lauf(K, H, L, C, SQ, mid, len(C), "aus")[0])
        zeile("BASIS (kein Gegen-Exit)", b1, b2, None, None)
        for modus, mv, lbl in (("immer", 0.0, "sofort schliessen"),
                               ("nur_minus", 0.0, "sofort, nur im Minus"),
                               ("nur_minus", 0.5, "sofort, Verlust > 0,5xATR"),
                               ("eng_sl", 0.1, "ENGER SL 0,1xATR"),
                               ("eng_sl", 0.2, "ENGER SL 0,2xATR"),
                               ("eng_sl", 0.3, "ENGER SL 0,3xATR"),
                               ("eng_sl", 0.5, "ENGER SL 0,5xATR")):
            r1, a1 = lauf(K, H, L, C, SQ, 0, mid, modus, mv)
            r2, a2 = lauf(K, H, L, C, SQ, mid, len(C), modus, mv)
            k1, k2 = kenn(r1), kenn(r2)
            if zeile(lbl + f" [{a1}/{a2}x]", k1, k2, b1, b2):
                treffer.setdefault(lbl, set()).add(name)

    print("\n" + "=" * 100)
    print("  URTEIL (Regel VOR dem Lauf fixiert: OeR UND SigmaR besser, in BEIDEN")
    print("          Haelften UND auf BEIDEN Populationen)")
    print("=" * 100)
    voll = [k for k, v in treffer.items() if len(v) == len(pops)]
    if voll:
        print(f"    => BESTEHT: {voll}")
    else:
        print("    => KEINE Variante besteht auf beiden Populationen.")
        print("       Der Gegen-Squeeze taugt als ENTRY (validiert), nicht als EXIT —")
        print("       zum DRITTEN Mal, jetzt auch mit dem kanonischen Exit.")
    print()


if __name__ == "__main__":
    main()