AH-Oil-Trader/backtest_squeeze_srclose.py

#!/usr/bin/env python3
"""Prüft die User-Beobachtung: Auto-Squeeze-Trades werden vom S/R-Auto-Close am
nächsten Level gekappt. Vergleich auf 2 Halbjahren, NUR für Squeeze-Entries
(Box<=2,5×ATR, k=0,1 — Live-Params):

  Exit A (Baseline, live-validiert): SL 2×ATR + Trailing 1,5 + BE 1,3.
  Exit B (aktuelles Live-Verhalten): wie A, ABER schließt am nächsten Gegen-Level
          (M5-Pivot in Trade-Richtung), sobald Kurs dort (≤0,15×ATR) UND
          P(break) < 0,60 (kalibriertes Modell aus core.engine, identisch live).

Frage: Ist B in BEIDEN Hälften schlechter als A? Dann kappt der S/R-Close den
Squeeze-Runner-Edge → Squeeze-Trades vom S/R-Close ausnehmen.
"""
import sys
import MetaTrader5 as mt5
from core.engine import _p_break

_MAXH = 288; _ATRMIN = 0.12
_N = 12; _W = 24; _COOL = 12; _K = 0.1; _SQ_MULT = 2.5
_PIV_K = 3; _LOOKBACK = 300; _SR_TOUCH = 0.15; _PBRK = 0.60
_EMA_FAST = 12; _EMA_SLOW = 50


def _ema(v, p):
    k = 2.0/(p+1); o = []; e = v[0]
    for i, x in enumerate(v): e = x if i == 0 else x*k+e*(1-k); o.append(e)
    return o
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])))
    out = [None]
    for i in range(1, len(C)):
        seg = t[max(1, i-p+1):i+1]; out.append(sum(seg)/len(seg))
    return out

def _opp_level(entry, d, atr, H, L, i):
    """Nächstes Gegen-/Ziel-Level (M5-Pivot) in Trade-Richtung, wie live _draw_levels."""
    phis, plos = [], []
    for j in range(max(_PIV_K, i-_LOOKBACK), i-_PIV_K):
        if H[j] == max(H[j-_PIV_K:j+_PIV_K+1]): phis.append(H[j])
        if L[j] == min(L[j-_PIV_K:j+_PIV_K+1]): plos.append(L[j])
    if d > 0:
        c = [p for p in phis if p > entry+0.3*atr]; return min(c) if c else None
    c = [p for p in plos if p < entry-0.3*atr]; return max(c) if c else None

def sim(entry, d, atr, H, L, C, EF, ES, j0, level=None,
        sl_atr=2.0, trail=1.5, trail_on=0.3, be_on=1.3, srclose=False):
    eff = entry - d*sl_atr*atr; hw = entry
    end = min(j0+_MAXH, len(C)-1); exit_px = C[end]
    for j in range(j0, end+1):
        hi, lo = H[j], L[j]
        if (lo <= eff) if d > 0 else (hi >= eff): exit_px = eff; break
        # S/R-Close: Kurs am Gegen-Level (Touch) + P(break)<Schwelle + im Plus → close
        if srclose and level is not None:
            gap = (level - C[j])*d
            if 0 <= gap <= _SR_TOUCH*atr and (C[j]-entry)*d > 0:
                mom6 = (C[j]-C[max(0, j-6)])*d/atr; mom3 = (C[j]-C[max(0, j-3)])*d/atr
                wt = 1.0 if (EF[j]-ES[j])*d > 0 else 0.0; dist = abs(level-entry)/atr
                if _p_break(mom6, mom3, wt, dist) < _PBRK:
                    exit_px = level; break
        hw = max(hw, hi) if d > 0 else min(hw, lo)
        prof = (C[j]-entry)*d
        if prof >= trail_on*atr:
            cand = hw - d*trail*atr
            if prof >= be_on*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

def rep(name, Rs):
    if not Rs: print(f"  {name:<34} -"); return
    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)
    pf = up/dn if dn > 0 else 9.99
    print(f"  {name:<34} Trades={n:>4} Treffer={100*w/n:>3.0f}% ØR={s/n:+.3f} PF={pf:.2f} ΣR={s:+.0f}")

def scan(H, L, C, A, EF, ES, SP, point, lo_i, hi_i):
    def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
    RsA, RsB = [], []
    i = max(lo_i, _N+15, _LOOKBACK)
    while i < min(hi_i, len(C)-_MAXH-1):
        atr = A[i]
        if not atr or atr < _ATRMIN: i += 1; continue
        boxHi = max(H[i-_N:i]); boxLo = min(L[i-_N:i])
        if (boxHi-boxLo) > _SQ_MULT*atr: i += 1; continue   # nur Squeeze
        hit = None
        for j in range(i, min(i+_W, len(C)-_MAXH-1)):
            up = boxHi+_K*atr; dn = boxLo-_K*atr
            if H[j] >= up: hit = (j, 1, up); break
            if L[j] <= dn: hit = (j, -1, dn); break
        if hit is None: i += 1; continue
        j, d, lvl = hit; c = cost(j, atr)
        level = _opp_level(lvl, d, atr, H, L, j)
        RsA.append(sim(lvl, d, atr, H, L, C, EF, ES, j+1, srclose=False) - c)
        RsB.append(sim(lvl, d, atr, H, L, C, EF, ES, j+1, level=level, srclose=True) - c)
        i = j + _COOL
    return RsA, RsB

def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    mt5.initialize(); sym = None
    for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
        if mt5.symbol_info(c): sym = c; break
    bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n+_MAXH+30)
    si = mt5.symbol_info(sym); point = si.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_FAST); ES = _ema(C, _EMA_SLOW)
    N = len(C); mid = N//2
    print("="*90)
    print(f"  Squeeze-Trades: Trailing-Exit (A) vs +S/R-Close (B) — {sym} M5 ({N} Bars)")
    print(f"  Box<={_SQ_MULT}×ATR · Ausbruch {_K}×ATR · S/R-Close bei P(break)<{_PBRK}")
    print("="*90)
    for label, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N)):
        print(f"\n{label}:")
        RsA, RsB = scan(H, L, C, A, EF, ES, SP, point, lo, hi)
        rep("A: nur Trailing (validiert)", RsA)
        rep("B: + S/R-Close (aktuell live)", RsB)
        if RsA and RsB:
            dR = sum(RsB)-sum(RsA)
            print(f"     → Δ(B−A) ΣR = {dR:+.0f}   (negativ = S/R-Close KOSTET beim Squeeze)")
    print("\n  S/R-Close beim Squeeze rausnehmen, wenn B in BEIDEN Hälften ΣR/ØR < A.")

if __name__ == "__main__":
    main()