#!/usr/bin/env python3 """Gegen-Squeeze-EXIT unter Volatilitäts-Regimen (2026-07-24, User-Frage „Markt ist gerade sehr volatil — hilft der Gegen-Ausstieg da wirklich nicht?"). Verschärfter Re-Test des 2026-07-22-Befunds (backtest_squeeze_exit.py = EMA-Entries): 1. ENTRIES = echte SQUEEZE-Ausbrüche (Live-Params _SQ_N/_SQ_MULT/_SQ_K aus core.wave_rec) — exakt das Auto-Squeeze-Szenario, nicht EMA-Crosses. 2. Exit-Varianten: BASE = Live-Exit (SL 2,0×ATR + Trailing 1,5 + Breakeven 1,3) MOD = Gegen-Squeeze-Ausbruch schließt SOFORT (am Bar-Close), immer MODL = wie MOD, aber nur wenn Position im MINUS ist („Verlust minimieren") MODL5 = nur wenn Verlust > 0,5×ATR (erst bei klarem Minus kappen) 3. VOLA-SPLIT: Trades je Hälfte nach ATR-Terzil am Entry (low/mid/HIGH) getrennt — beantwortet direkt „gilt der Befund auch im Hoch-Vola-Regime?". 4. Extra-Fenster: letzte ~30 Tage (8640 M5-Bars) = die aktuelle volatile Phase. Innerhalb eines Bars: erst SL/Trailing (intrabar H/L), dann Gegen-Squeeze (Close). Kosten = Bar-Spread/ATR am Entry. Verdict-Regel (vorab): eine MOD-Variante wird nur eingebaut, wenn sie in BEIDEN Hälften UND im High-Vol-Terzil ΣR ≥ BASE liefert. """ import sys import MetaTrader5 as mt5 from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K _MAXH = 288; _ATRMIN = 0.06; _COOL = 12 _RECENT_BARS = 8640 # ~30 Tage M5 def _atr_series(H, L, C, p=14): t = [0.0] for i in range(1, len(C)): t.append(max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1]))) return [(sum(t[max(1, i-p+1):i+1])/max(1, len(t[max(1, i-p+1):i+1]))) if i else None for i in range(len(C))] def _sq_dir(H, L, C, A): """Squeeze-Ausbruchs-Richtung je Bar (+1/-1/0), Live-Params — Box = _SQ_N abgeschlossene Bars VOR i, aktueller Close bricht ±_SQ_K×ATR aus der Box.""" out = [0]*len(C) for i in range(_SQ_N+1, len(C)): atr = A[i] if not atr or atr <= 0: continue hi = max(H[i-_SQ_N:i]); lo = min(L[i-_SQ_N:i]) if (hi-lo)/atr > _SQ_MULT: continue if C[i] >= hi + _SQ_K*atr: out[i] = 1 elif C[i] <= lo - _SQ_K*atr: out[i] = -1 return out def sim(entry, d, atr, H, L, C, j0, SQ, mode): """mode: base | mod | modl (nur Minus) | modl5 (nur Verlust>0,5×ATR).""" eff = entry - d*2.0*atr; hw = entry end = min(j0+_MAXH, len(C)-1); exit_px = C[end]; exit_j = end for j in range(j0, end+1): hj, lj = H[j], L[j] if (lj <= eff) if d > 0 else (hj >= eff): return (eff-entry)*d/atr, j if mode != 'base' and SQ[j] == -d: prof = (C[j]-entry)*d fire = (mode == 'mod' or (mode == 'modl' and prof < 0) or (mode == 'modl5' and prof < -0.5*atr)) if fire: return prof/atr, j hw = max(hw, hj) if d > 0 else min(hw, lj) prof = (C[j]-entry)*d if prof >= 0.3*atr: cand = hw - d*1.5*atr if prof >= 1.3*atr: cand = max(cand, entry) if d > 0 else min(cand, entry) eff = max(eff, cand) if d > 0 else min(eff, cand) return (exit_px-entry)*d/atr, exit_j def collect(H, L, C, A, SP, SQ, point, lo, hi): """Squeeze-ENTRIES in [lo,hi) → Liste (atr_at_entry, {mode: R_netto}).""" def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr out = [] i = max(lo, _SQ_N+15) while i < min(hi, len(C)-_MAXH-1): d = SQ[i] if d == 0: i += 1; continue atr = max(A[i], _ATRMIN); c = cost(i, atr) rec = {} xj = i+1 for m in ('base', 'mod', 'modl', 'modl5'): r, xj = sim(C[i], d, atr, H, L, C, i+1, SQ, m) rec[m] = r - c out.append((A[i], rec)) i = i + _COOL return out def st(Rs): if not Rs: 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) return dict(n=n, wr=100*w/n, oR=s/n, pf=(up/dn if dn > 0 else 9.99), sum=s, worst=min(Rs)) def block(title, trades): """Vergleichstabelle BASE vs Varianten für eine Trade-Menge.""" if not trades: print(f" {title}: (keine Trades)"); return print(f" {title} (n={len(trades)}):") base = st([r['base'] for _, r in trades]) print(f" {'BASE (Live-Exit)':<22} WR={base['wr']:>3.0f}% ØR={base['oR']:+.3f} " f"PF={base['pf']:>4.2f} ΣR={base['sum']:>+7.1f} Worst={base['worst']:+.2f}") for m, lbl in (('mod', 'MOD (immer)'), ('modl', 'MODL (im Minus)'), ('modl5', 'MODL5 (Minus>0,5ATR)')): s = st([r[m] for _, r in trades]) d = s['sum']-base['sum'] print(f" {lbl:<22} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} " f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.1f} Worst={s['worst']:+.2f}" f" Δ={d:+.1f}") def vol_split(trades): """Terzile nach ATR am Entry (innerhalb der Menge).""" if len(trades) < 9: return [] atrs = sorted(a for a, _ in trades) t1 = atrs[len(atrs)//3]; t2 = atrs[2*len(atrs)//3] lo = [(a, r) for a, r in trades if a <= t1] mid = [(a, r) for a, r in trades if t1 < a <= t2] hi = [(a, r) for a, r in trades if a > t2] return [(f"Vola NIEDRIG (ATR≤{t1:.3f})", lo), (f"Vola MITTEL", mid), (f"Vola HOCH (ATR>{t2:.3f})", hi)] 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() 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] A = _atr_series(H, L, C); SQ = _sq_dir(H, L, C, A) N = len(C); mid = N//2 print("="*88) print(f" Gegen-Squeeze-EXIT × Vola-Regime — {sym} M5 ({N} Bars) · ENTRIES=Squeeze-Ausbrüche") print(f" Live-Params Box={_SQ_N}·mult={_SQ_MULT}·k={_SQ_K} · Exit BASE=SL2,0/Trail1,5/BE1,3") print("="*88) for lbl, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N)): trades = collect(H, L, C, A, SP, SQ, point, lo, hi) print(f"\n{lbl}:") block("ALLE Trades", trades) for t, sub in vol_split(trades): block(t, sub) recent = collect(H, L, C, A, SP, SQ, point, max(0, N-_RECENT_BARS), N) print(f"\nAKTUELLE PHASE (letzte ~30 Tage):") block("ALLE Trades", recent) for t, sub in vol_split(recent): block(t, sub) print("\n Verdict-Regel (vorab): Variante nur einbauen, wenn sie in BEIDEN Hälften") print(" UND im Hoch-Vola-Terzil ΣR ≥ BASE liefert. Sonst bleibt der Live-Exit.") if __name__ == "__main__": main()