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