AH-Oil-Trader/backtest_adverse15_squeeze.py

#!/usr/bin/env python3
"""Schadet die 15-Minuten-Regel den SQUEEZE-Trades? (Anlass 2026-07-31)

BEOBACHTUNG: Von den drei heute per `adverse15` geschlossenen Bot-Trades hat die
Regel bei EINEM gerettet (Auto-Signal, danach −32 € weiter dagegen) und bei den
beiden SQUEEZE-Trades zu früh gekappt (danach +19,28 bzw. +17,60 € möglich).

VERDACHT — wieder ein Kalibrierungs-Bruch: die Schwelle `adverse_15min_atr=0.5`
wurde in `backtest_auto_signal.py` auf **Wave-Signal-Entries** gemessen und wird
live auf **ALLE** Bot-Trades angewandt, also auch auf den Squeeze. Der Squeeze ist
aber runner-abhängig (nur 43 % Treffer, der Ertrag steckt im Tail) — eine Regel,
die nach 15 Minuten kappt, trifft genau den Tail.

GEMESSEN: Squeeze-Entries (Regel exakt wie live), Exit aus `core/exit_model.py`,
Echtkosten, 2 Halbjahre; ohne Regel vs. mit Regel bei mehreren Schwellen.
Der Adverse-Check läuft über den `stop_when`-Hook — die Phasen-Mechanik wird nicht
kopiert.
"""
import sys

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.wave_rec import _N_BARS

_ATRMIN = 0.06
_BARS_15MIN = 3          # 15 min = 3 M5-Bars


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 = []
    for i in range(len(C)):
        if not i:
            out.append(None); continue
        w = t[max(1, i - p + 1):i + 1]
        out.append(sum(w) / max(1, len(w)))
    return out


def squeeze_signals(H, L, C, AT, lo, hi):
    """Regel exakt wie live (`wave._squeeze_one`), Einstieg AM Ausbruchslevel."""
    from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K
    _W = 24; _COOL = 12
    out = []
    i = max(lo, _N_BARS, _SQ_N + 15)
    while i < min(hi, len(C) - LIVE.max_hold - 1):
        atr = AT[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        bh = max(H[i - _SQ_N:i]); bl = min(L[i - _SQ_N:i])
        if (bh - bl) > _SQ_MULT * atr:
            i += 1; continue
        hit = None
        for j in range(i, min(i + _W, len(C) - LIVE.max_hold - 1)):
            up = bh + _SQ_K * atr; dn = bl - _SQ_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
        out.append((j, d, atr, lvl))
        i = j + _COOL
    return out


def run(sig, H, L, C, SP, adverse=None):
    """`adverse` = Schwelle in ×ATR (None = Regel aus)."""
    Rs = []
    for (j, d, atr, lvl) in sig:
        cost = (SP[j] if SP[j] > 0 else 0.0225) / atr
        hook = None
        if adverse is not None:
            j0 = j + 1

            def hook(k, prof_atr, _j0=j0, _x=adverse):
                # 1× genau beim 15-Min-Zeitpunkt prüfen (wie live)
                return (k - _j0) == _BARS_15MIN - 1 and prof_atr <= -_x
        Rs.append(simulate(lvl, d, atr, H, L, C, j + 1, LIVE, stop_when=hook) - cost)
    return Rs


def st(Rs):
    if not Rs:
        return "  —"
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    g = sum(x for x in Rs if x > 0); l = -sum(x for x in Rs if x < 0)
    return (f"n={n:>4} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} PF={(g/l if l > 0 else 9.99):>4.2f} "
            f"ΣR={s:>+6.0f}")


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)
    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]
    AT = _atr_series(H, L, C)
    mid = len(C) // 2
    halves = [("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C))]

    print("=" * 92)
    print(f"  15-MIN-REGEL auf SQUEEZE-Entries — {sym} M5 ({len(C)} Bars), Echtkosten")
    print(f"  Exit core/exit_model.py (Trail {LIVE.mult}) · live: adverse_15min_atr=0.5")
    print("  ⚠ Die 0,5 wurden auf WAVE-Signalen gemessen, nicht auf dem Squeeze.")
    print("=" * 92)
    SIG = {lbl: squeeze_signals(H, L, C, AT, a, b) for lbl, a, b in halves}
    for lbl, _, _ in halves:
        print(f"  {lbl}: {len(SIG[lbl])} Squeeze-Entries")

    base = {}
    for name, adv in (("OHNE Regel", None), ("0,3×ATR", 0.3),
                      ("0,5×ATR  ← live", 0.5), ("0,8×ATR", 0.8), ("1,2×ATR", 1.2)):
        print(f"\n{name}")
        for lbl, _, _ in halves:
            Rs = run(SIG[lbl], H, L, C, SP, adverse=adv)
            s = sum(Rs)
            if adv is None:
                base[lbl] = s
            d = f"   Δ ohne Regel {s - base.get(lbl, 0):+.0f}" if adv is not None else ""
            print(f"   {lbl}  {st(Rs)}{d}")


if __name__ == "__main__":
    main()