AH-Oil-Trader/backtest_squeeze_exit.py

#!/usr/bin/env python3
"""Squeeze-als-EXIT (Variante B, 2026-07-22, User-Idee „Gegen-Position beim Ausbruch
schneller schließen"). Reine EXIT-Regel — KEIN Re-Entry (anders als der verworfene
Stop-&-Reverse): Halte ich eine Position und feuert ein **validierter Squeeze-Ausbruch
in GEGENrichtung**, schließe ich sofort (statt bis zum SL/Trailing zu laufen).

Test: gleiche Entries (frischer EMA12/50-Cross M5), zwei Exits vergleichen:
  BASE = Live-Exit (SL 2,0×ATR + Trailing 1,5 + BE 1,3).
  MOD  = BASE, aber Gegen-Squeeze-Ausbruch schließt die Position sofort (am Bar-Close).
  MOD-Verlust = wie MOD, aber nur wenn die Position im MINUS ist (Verlust begrenzen).
Squeeze mit den LIVE-Params (_SQ_N/_SQ_MULT/_SQ_K). Kosten = Bar-Spread/ATR am Entry.
Innerhalb eines Bars: erst SL/Trailing (intrabar H/L), dann Squeeze (Close) — der
Squeeze-Exit greift also nur, wenn der SL im Bar NICHT schon lief.

Verdict: MOD baut nur ein, wenn ΣR in BEIDEN Hälften ≥ BASE (verbessert den Exit,
kostet keine der Hälften). Sonst raus (wie der Reverse).
"""
import sys
import MetaTrader5 as mt5
from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K

_MAXH = 288; _ATRMIN = 0.06; _COOL = 6
_EMA_F = 12; _EMA_S = 50; _DEAD = 0.15


def _ema(C, p):
    k = 2.0 / (p + 1); e = C[0]; out = [e]
    for x in C[1:]:
        e += k * (x - e); out.append(e)
    return out


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 LONG / −1 SHORT / 0), Live-Params —
    Box = _SQ_N Bars VOR i (abgeschlossen), aktueller Close bricht ±_SQ_K×ATR."""
    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:          # keine Kompression
            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' (Gegen-Squeeze schließt) | 'modl' (nur im Minus)."""
    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):        # SL/Trailing (intrabar)
            return (eff-entry)*d/atr, j
        if mode != 'base' and SQ[j] == -d:               # Gegen-Squeeze am Close
            if mode == 'mod' or (C[j]-entry)*d < 0:      # 'modl': nur wenn im Minus
                return (C[j]-entry)*d/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 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)


def line(lbl, s):
    if not s: return f"    {lbl:<14} —"
    return (f"    {lbl:<14} n={s['n']:>4} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
            f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+6.0f}")


def run(H, L, C, A, SP, EF, ES, SQ, lo, hi, mode):
    def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
    Rs = []; i = max(lo, _EMA_S + 2)
    sig_prev = 0
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        diff = EF[i] - ES[i]; dead = _DEAD * atr
        sig = 1 if diff > dead else -1 if diff < -dead else 0
        if sig != 0 and sig != sig_prev:                 # frischer Cross
            r, xj = sim(C[i], sig, max(atr, _ATRMIN), H, L, C, i+1, SQ, mode)
            Rs.append(r - cost(i, max(atr, _ATRMIN)))
            sig_prev = sig
            i = xj + _COOL; continue
        if sig != 0: sig_prev = sig
        i += 1
    return Rs


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); EF = _ema(C, _EMA_F); ES = _ema(C, _EMA_S)
    SQ = _sq_dir(H, L, C, A); N = len(C); mid = N//2

    print("="*80)
    print(f"  Squeeze-als-EXIT (Variante B) — {sym} M5 ({N} Bars, EMA-Entries, Echtkosten)")
    print(f"  BASE=Live-Exit · MOD=Gegen-Squeeze schließt · MODL=nur im Minus. 2 Halbjahre.")
    print("="*80)
    for lbl, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N-_MAXH-1)):
        base = st(run(H, L, C, A, SP, EF, ES, SQ, lo, hi, 'base'))
        mod  = st(run(H, L, C, A, SP, EF, ES, SQ, lo, hi, 'mod'))
        modl = st(run(H, L, C, A, SP, EF, ES, SQ, lo, hi, 'modl'))
        print(f"\n  {lbl}:")
        print(line("BASE", base))
        print(line("MOD (immer)", mod))
        print(line("MODL (Minus)", modl))
        if base and mod and modl:
            print(f"    Δ ΣR:  MOD {mod['sum']-base['sum']:+.0f}   MODL {modl['sum']-base['sum']:+.0f}"
                  f"   (>0 = besser als BASE)")
    print(f"\n  Verdict: einbauen nur, wenn eine MOD-Variante in BEIDEN Hälften ΣR ≥ BASE.")


if __name__ == "__main__":
    main()