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, minutes=15):
    """`adverse` = Schwelle in ×ATR (None = Regel aus), `minutes` = Prüfzeitpunkt."""
    bars = max(1, minutes // 5)
    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, _b=bars):
                # 1× genau zum Prüfzeitpunkt (wie live: einmal je Ticket)
                return (k - _j0) == _b - 1 and prof_atr <= -_x
        Rs.append(simulate(lvl, d, atr, H, L, C, j + 1, LIVE, stop_when=hook) - cost)
    return Rs


def wave_signals(H, L, C, AT, lo, hi, step=6):
    """WAVE-Signal-Entries — die Population, auf der die Regel kalibriert WURDE
    und auf der sie live noch gilt (Auto-Signal-Pfad)."""
    from core.analysis import calc_trend_angle
    from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _ANGLE_LR, _ANGLE_DEAD,
                               _REVERSAL_STRETCH, _STRETCH_MAX)
    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
    EF = ema(C, _EMA_FAST); ES = ema(C, _EMA_SLOW)
    out = []
    for i in range(max(lo, _N_BARS), min(hi, len(C) - LIVE.max_hold - 1), step):
        atr = AT[i]
        if not atr or atr < _ATRMIN:
            continue
        stretch = (C[i] - ES[i]) / atr
        ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR) - 90.0
        d = 0
        if stretch <= -_REVERSAL_STRETCH and ang >= _ANGLE_DEAD: d = 1
        elif stretch >= _REVERSAL_STRETCH and ang <= -_ANGLE_DEAD: d = -1
        elif abs(stretch) < _STRETCH_MAX: d = 1 if EF[i] > ES[i] else -1 if EF[i] < ES[i] else 0
        if d:
            out.append((i, d, atr, C[i]))
    return out


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")

    WAV = {lbl: wave_signals(H, L, C, AT, a, b) for lbl, a, b in halves}
    for lbl, _, _ in halves:
        print(f"  {lbl}: {len(WAV[lbl])} Wave-Signal-Entries")

    # ZEIT-Achse mitmessen (User-Frage 2026-07-31 „vielleicht die auto close auf
    # 30 min setzen?"): nicht nur WIE WEIT gegen die Richtung, sondern auch WANN
    # geprüft wird. Beide Populationen, weil die Regel auf der einen kalibriert
    # wurde und auf der anderen (bis heute) angewandt.
    for pop, SETS in (("SQUEEZE-Entries", SIG), ("WAVE-SIGNAL-Entries", WAV)):
        print(f"\n{'=' * 92}\n  {pop}\n{'=' * 92}")
        base = {}
        for lbl, _, _ in halves:
            Rs0 = run(SETS[lbl], H, L, C, SP, adverse=None)
            base[lbl] = sum(Rs0)
            print(f"   {lbl}  OHNE Regel   {st(Rs0)}")
        print(f"\n   {'Zeitpunkt':<12}{'Schwelle':<12}"
              + "".join(f"{l[:2]:>11}" for l, _, _ in halves) + "   (Δ gegen OHNE Regel)")
        for mins in (15, 30, 45, 60):
            for adv in (0.5, 0.8):
                cells = []
                for lbl, _, _ in halves:
                    tot = sum(run(SETS[lbl], H, L, C, SP, adverse=adv, minutes=mins))
                    cells.append(f"{tot - base[lbl]:>+11.0f}")
                mark = "   ← live" if (mins == 15 and adv == 0.5) else ""
                print(f"   {mins:>4} min    {adv:>5.1f}×ATR " + "".join(cells) + mark)


if __name__ == "__main__":
    main()