AH-Oil-Trader/backtest_auto_signal.py

#!/usr/bin/env python3
"""backtest_auto_signal.py — Auto-Entry auf die EMPFEHLUNG (B2, User-Wunsch
2026-07-30 „Trades bei Empfehlung automatisch eröffnen").

Kernfrage: Trägt ein AUTONOMER Entry auf dem Wave-Signal (mit den HEUTIGEN Gates)
netto in BEIDEN Halbjahren — und hält er gegen den einzigen validierten Auto-Entry
(Squeeze: ØR +0,14…+0,23, PF>1)?

Vorgeschichte (hohe Beweislast!): Der Wave-Auto-Trader lief live als Dry-Run und
war ein klarer VERLIERER (WR 29 % / PF 0,64 vs. manuell 56–76 %) → am 2026-07-23
komplett entfernt. Gegenbefund: die MANUELLEN „MIT Signal"-Trades des Users sind
profitabel (+118 €, WR 56 %) — aber das ist MENSCHLICHE AUSWAHL unter den Signalen,
nicht „alle nehmen". Genau diese Differenz misst dieses Skript.

Methodik (bewusst pessimistisch, wie backtest_breakout_squeeze.py):
  · Signal   = die ECHTE Logik `WaveRecommender._build` (kein Nachbau), inkl.
               Totband, Anti-Überdehnung, Konfidenz-Gate 55, Reversal-Zweig
  · HTF      = M30- UND H1-Trend aus den M5-Bars aggregiert und in `_build`
               gefüttert (`htf_trend`/`h1_trend`) → der live aktive M30-Gegen-
               Trend-Filter und der H1-Konfluenz-Bonus wirken mit
  · Entry    = am Bar-CLOSE des Signal-Bars, nur wenn FLAT (sequentiell, 1 Position)
  · Exit     = Live-Modell: SL 2,0×ATR · Trailing 1,5 ab 0,3 · Breakeven 1,3 ·
               Time-Stop 120 min (24 M5-Bars) wenn nie Trail-Start erreicht
  · Kosten   = echter Bar-Spread/ATR (Ø ~0,265×ATR — NICHT die alte 0,1-Annahme)
  · 2 Halbjahre (H1/H2) — eine Variante gilt nur als „trägt", wenn sie in BEIDEN
    positiv ist (Dead-Hour-Lektion)

NICHT simuliert (bewusst): Entry-Raum-Gate + ATR-Breakout-Bestätigung (beide
brauchen Live-State/pb_levels, s. CLAUDE.md — Backtests bleiben gate-frei) sowie
das EIA-Blackout-Gate. Beide würden Trades ENTFERNEN, die Ergebnisse hier sind
also eher optimistisch als zu streng.

Aufruf:  python backtest_auto_signal.py [n_bars]     (Default 40000 M5-Bars)
"""
from __future__ import annotations
import sys
from datetime import datetime, timezone, timedelta

import MetaTrader5 as mt5

from core.wave_rec import (WaveRecommender, _atr, _ema_last,
                           _EMA_FAST, _EMA_SLOW, _HTF_DEADBAND)

# ── Exit-Parameter = die LIVE-Werte (core/trailing.py) ────────────────────────
_SL_ATR    = 2.0      # Initial-SL (Band 1,8–2,2 → Ziel 2,0)
_TRAIL     = 1.5      # Trailing-Abstand ×ATR (M5-Mult)
_TRAIL_ON  = 0.3      # Trail-Start ab diesem Profit ×ATR
_BE_ON     = 1.3      # Breakeven-Floor ab diesem Profit ×ATR
_TIMESTOP  = 24       # Time-Stop: 120 min = 24 M5-Bars in Phase „Init"
_MAXH      = 288      # Sim-Horizont (24 h auf M5)
_ATRMIN    = 0.06     # ATR-Floor wie live (_ATR_MIN in trailing.py)
_NIGHT     = (0, 1, 2, 3, 4, 5, 6, 7)   # Berlin-Stunden = gemessene Kostenfalle
_ADV_BARS  = 3        # 15-Min-Regel: 3 M5-Bars = 15 min nach Entry prüfen
_HTF_N     = 6        # 6 M5-Bars = 1 M30-Bar
_H1_N      = 12       # 12 M30-Bars … (H1 = 12 M5-Bars)


class _NeutralTU:
    """TU ist aus der Empfehlung entfernt — im Test neutral (kein Einfluss)."""
    def snapshot(self):
        return {"intervals": {}}


def _atr_series(H, L, C, p=14):
    tr = [0.0]
    for i in range(1, len(C)):
        tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    out = [None] * len(C)
    run = 0.0
    for i in range(1, len(C)):
        run += tr[i]
        if i > p:
            run -= tr[i - p]
        out[i] = run / min(i, p)
    return out


def _ema_series(vals, period):
    k = 2.0 / (period + 1)
    out = [vals[0]]
    for v in vals[1:]:
        out.append(v * k + out[-1] * (1.0 - k))
    return out


def _htf_signs(H, L, C, step):
    """Aggregiert die M5-Bars zu einer höheren TF (step M5-Bars je HTF-Bar) und
    liefert je M5-Index das Trend-Vorzeichen (+1/-1/0, Totband _HTF_DEADBAND×ATR)
    — LOOK-AHEAD-frei: an Index i zählen nur HTF-Bars, die vor i abgeschlossen sind."""
    # HTF-Bars bilden
    hc, hh, hl = [], [], []
    for s in range(0, len(C) - step + 1, step):
        seg_h = H[s:s + step]; seg_l = L[s:s + step]
        hh.append(max(seg_h)); hl.append(min(seg_l)); hc.append(C[s + step - 1])
    if len(hc) < _EMA_SLOW + 5:
        return [0] * len(C)
    ef = _ema_series(hc, _EMA_FAST); es = _ema_series(hc, _EMA_SLOW)
    hatr = _atr_series(hh, hl, hc)
    signs_htf = []
    for j in range(len(hc)):
        a = hatr[j]
        if j < _EMA_SLOW + 2 or not a:
            signs_htf.append(0); continue
        d = ef[j] - es[j]
        signs_htf.append(0 if abs(d) < _HTF_DEADBAND * a else (1 if d > 0 else -1))
    # auf M5-Index abbilden: bei i ist der letzte ABGESCHLOSSENE HTF-Bar (i//step)-1
    out = [0] * len(C)
    for i in range(len(C)):
        j = i // step - 1
        out[i] = signs_htf[j] if 0 <= j < len(signs_htf) else 0
    return out


def _sim(entry, d, atr, H, L, C, j0, adv15=None):
    """Live-Exit-Modell. Rückgabe R = Profit/ATR (brutto, ohne Kosten).
    Phasen-Ratsche wie live: Time-Stop greift nur, solange nie Trail-Start erreicht.

    `adv15` = 15-Minuten-Regel (User-Wunsch 2026-07-30): steht der Trade nach
    _ADV_BARS (3 M5-Bars = 15 min) mehr als adv15×ATR im Minus → schließen.
    None = aus. adv15=0.0 bedeutet „schließen wenn überhaupt im Minus"."""
    eff = entry - d * _SL_ATR * atr
    hw = entry
    trail_started = False
    end = min(j0 + _MAXH, len(C) - 1)
    exit_px = C[end]
    for j in range(j0, end + 1):
        hi, lo = H[j], L[j]
        # SL/Trail-Stop zuerst (konservativ: Stop schlägt Weiterlaufen im selben Bar)
        if (lo <= eff) if d > 0 else (hi >= eff):
            exit_px = eff; break
        hw = max(hw, hi) if d > 0 else min(hw, lo)
        prof = (C[j] - entry) * d
        # 15-Minuten-Regel: EINMALIGE Prüfung genau beim Bar _ADV_BARS nach Entry.
        # Läuft der Trade dann gegen mich → raus (Einstieg war falsch).
        if adv15 is not None and (j - j0) == _ADV_BARS - 1:
            if prof <= -adv15 * atr:
                exit_px = C[j]; break
        if prof >= _TRAIL_ON * atr:
            trail_started = True
            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)
        # Time-Stop: nach 120 min noch in „Init" (nie Trail-Start) → raus
        if not trail_started and (j - j0) >= _TIMESTOP:
            exit_px = C[j]; break
    return (exit_px - entry) * d / atr


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


def scan(bars, H, L, C, SP, A, HTF, H1, EF, ES, w, lo_i, hi_i, min_conf, skip_night, adv15=None):
    """Sequentielle 1-Positions-Simulation: nimmt das ERSTE Signal, wenn flat."""
    Rs = []
    i = max(lo_i, _EMA_SLOW + 20)
    hi_i = min(hi_i, len(C) - _MAXH - 2)
    while i < hi_i:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        if skip_night:
            # Broker-Zeit ist UTC+3, Berlin (Sommer) UTC+2 → Berlin = Broker − 1 h
            hb = (datetime.fromtimestamp(int(bars[i]["time"]), timezone.utc).hour - 1) % 24
            if hb in _NIGHT:
                i += 1; continue
        ef, es = EF[i], ES[i]
        rec, _snap = w._build(ef, es, C[i], atr, "M5", 5,
                              htf_trend=HTF[i], h1_trend=H1[i])
        sig = rec.get("signal")
        if sig not in ("LONG", "SHORT") or (rec.get("conf_pct") or 0) < min_conf:
            i += 1; continue
        d = 1 if sig == "LONG" else -1
        entry = C[i]
        cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
        R = _sim(entry, d, atr, H, L, C, i + 1, adv15) - cost
        Rs.append(R)
        # nach dem Trade weiterspringen: der Exit fand irgendwo in [i+1, i+_MAXH]
        # statt — konservativ mit dem Time-Stop-Fenster weiterrücken (kein Overlap)
        i += _TIMESTOP
    return Rs


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 40000
    if not mt5.initialize():
        print(f"MT5-Init fehlgeschlagen: {mt5.last_error()}"); sys.exit(1)
    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 + 60)
    si = mt5.symbol_info(sym); point = si.point
    mt5.shutdown()
    if bars is None or len(bars) < 5000:
        print("Zu wenige Bars."); sys.exit(1)

    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)
    HTF = _htf_signs(H, L, C, _HTF_N)          # M30-Trend je M5-Bar
    H1 = _htf_signs(H, L, C, _HTF_N * 2)       # H1-Trend je M5-Bar (12 M5-Bars)

    w = WaveRecommender(_NeutralTU(), mt5.TIMEFRAME_M5)
    # EMA-Serien vorab (identisch zur Live-Rekursion, nur ohne Fenster-Neuaufbau)
    EF = _ema_series(C, _EMA_FAST)
    ES = _ema_series(C, _EMA_SLOW)

    N = len(C); mid = N // 2
    print("=" * 92)
    print(f"  AUTO-ENTRY auf die EMPFEHLUNG — {sym} M5 ({N} Bars) · echte _build-Logik")
    print(f"  Exit=live(SL2,0/Trail1,5/BE1,3/Time-Stop120min) · Kosten=Bar-Spread/ATR")
    print(f"  Referenz (validierter Auto-Entry Squeeze): ØR +0,14…+0,23 · PF>1")
    print("=" * 92)

    verdict = {}
    # 15-Minuten-Regel (User-Wunsch): welche Schwelle ist die beste? Basis =
    # Konf>=55 + Nacht-Sperre, damit nur die Regel selbst variiert.
    ADV = [(None, "OHNE 15-Min-Regel (Basis)"), (0.0, "15min raus wenn im Minus"),
           (0.15, "15min raus wenn < -0,15xATR"), (0.3, "15min raus wenn < -0,3xATR"),
           (0.5, "15min raus wenn < -0,5xATR"), (0.8, "15min raus wenn < -0,8xATR")]
    for label, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N)):
        print(f"\n{label} — 15-MIN-REGEL (Basis Konf>=55 + Nacht-Sperre):")
        for a, alab in ADV:
            Rs = scan(bars, H, L, C, SP, A, HTF, H1, EF, ES, w, lo, hi, 55, True, a)
            verdict.setdefault(("adv", a), []).append(_rep(alab, Rs))
        print(f"\n{label} — Konfidenz-Achse (ohne 15-Min-Regel):")
        for mc in (55, 65, 70, 75):
            for night in (False, True):
                tag = f"Konf>={mc}" + (" · Nacht-Sperre" if night else "")
                Rs = scan(bars, H, L, C, SP, A, HTF, H1, EF, ES, w, lo, hi, mc, night)
                oR = _rep(tag, Rs)
                verdict.setdefault((mc, night), []).append(oR)

    print("\n" + "=" * 92)
    print("  URTEIL (trägt nur, wenn ØR in BEIDEN Hälften > 0 UND im Squeeze-Band):")
    any_ok = False
    for key, vals in sorted(verdict.items(), key=lambda kv: str(kv[0])):
        mc, night = key
        if len(vals) != 2 or any(v is None for v in vals):
            continue
        ok = all(v > 0 for v in vals)
        band = all(v >= 0.14 for v in vals)
        mark = "✅ TRÄGT" if (ok and band) else ("⚠ positiv, unter Squeeze-Band" if ok else "❌ fällt durch")
        if ok and band:
            any_ok = True
        if mc == "adv":
            tag = "15Min-Regel " + (str(night) if night is not None else "AUS")
        else:
            tag = f"Konf>={mc}" + (" +Nacht-Sperre" if night else "")
        print(f"  {tag:<28} H1 ØR {vals[0]:+.3f} · H2 ØR {vals[1]:+.3f}   {mark}")
    print("\n  " + ("→ mindestens eine Variante trägt: Einbau vertretbar (klein starten, B4-Monitor)."
                    if any_ok else
                    "→ KEINE Variante trägt beidhälftig im Squeeze-Band → NICHT einbauen."))


if __name__ == "__main__":
    main()