AH-Oil-Trader/backtest_auto_signal_v3.py

#!/usr/bin/env python3
"""AUTO-SIGNAL (🎯 SIG) — vollständige Neumessung mit dem Stand vom 2026-08-05.

WARUM SCHON WIEDER: seit `backtest_auto_signal_norev.py` (04.08.) haben sich vier
Dinge geändert, von denen drei die Messgrundlage berühren:
  1. **Nacht-Fenster 0–7 → 3–6** (05.08., `auto_squeeze_night_hours` gilt auch für
     den Signal-Pfad). Die alte Messung sperrte 8 Stunden, live sind es noch 4.
  2. **Reversal aus** (04.08.) — war dort schon berücksichtigt.
  3. **Trail 1,0** im kanonischen Exit (31.07.) — dort schon drin.
  4. ⚠ **NEU und der eigentliche Anlass: die AUSFÜHRUNG.** Am 05.08. wurde für den
     Squeeze gemessen, dass sein gesamter Edge im Einstiegspreis liegt (am Level
     ØR +0,456, hinterhergelaufen −0,158). **Dieselbe Frage stellt sich für SIG** —
     denn `_confirm_breakout` hat ein LEVEL (`_pend["level"]` = Anker ± k×ATR), an
     dem eine ruhende Stop-Order liegen könnte, statt am Markt nachzukaufen. Alle
     bisherigen SIG-Messungen sind am Bar-Close eingestiegen.

GEMESSEN WIRD ENTLANG DREI ACHSEN
  * Einstieg: **am Bestätigungs-Level** vs. **am Bar-Close** (Ist-Zustand)
  * Nacht-Fenster: **3–6 (live)** vs. 0–7 (alt) vs. keins
  * Konfidenz-Schwelle: 55 / 65 / **75 (live)** / 85

METHODIK: voller Live-Gate-Stack über die ECHTEN Methoden (`_build` inkl. Totband,
M30-Filter, Anti-Überdehnung, Winkel, min_conf; `_confirm_breakout`-Mechanik mit
k=0,3; `_room_gate`), sequentielle 1-Slot-Sim wie live, kanonischer Exit
(`exit_model.LIVE`), Echtkosten aus dem Bar-Spread, 2 Halbjahre.
⚠ Live-Dedup nachgebildet: **1× je Signal-Episode** — re-armt erst bei WARTEN oder
Richtungswechsel (`_signal_last_sig`), nicht nach jedem Exit.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  SIG bleibt nur an, wenn eine Variante in **BEIDEN** Halbjahren ØR > 0 UND PF > 1
  liefert und die Nachbar-Konfidenzschwellen mithalten. Sonst: gemessen nicht
  tragfähig — die Entscheidung bleibt beim User, die Zahl nicht.

Aufruf:  python backtest_auto_signal_v3.py [bars]
"""
import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _EMA_FAST, _EMA_SLOW, _N_BARS,
                           _ANGLE_LR, _ANGLE_DEAD, _HTF_DEADBAND)

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_ATRMIN = 0.06
_TIMEOUT_BARS = 12          # `_breakout_timeout_s` 3600 s = 12 M5-Bars
_PIV_K, _PIV_LOOK = 3, 300
_ATR_FLOOR_PB = 0.12
_K = 0.3                    # breakout_k live
_NACHT_LIVE = (3, 4, 5, 6)
_NACHT_ALT = (0, 1, 2, 3, 4, 5, 6, 7)


def _berlin(ts):
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN))


def d_slip(sig):
    """Slippage wirkt IMMER gegen den Trade (LONG teurer, SHORT billiger)."""
    return 1 if sig == "LONG" else -1


def _ema_series(v, p):
    k = 2 / (p + 1); out = [v[0]]
    for x in v[1:]:
        out.append(out[-1] + k * (x - out[-1]))
    return out


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]
    for i in range(1, len(C)):
        seg = tr[max(1, i - p + 1):i + 1]
        out.append(sum(seg) / len(seg))
    return out


def _htf_sign(C, i, n, atr):
    if i < n * _EMA_SLOW:
        return 0
    seg = C[i - n * _EMA_SLOW:i + 1:n]
    if len(seg) < _EMA_SLOW + 2:
        return 0
    ef = _ema_series(seg, _EMA_FAST)[-1]; es = _ema_series(seg, _EMA_SLOW)[-1]
    dead = _HTF_DEADBAND * atr * (n ** 0.5)
    return 1 if ef - es > dead else (-1 if es - ef > dead else 0)


def lauf(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1, PH, PL, lo, hi,
         min_conf, nacht, am_level, slip=0.0, fill="close"):
    """Ein Durchlauf: sequentiell, EIN Positions-Slot, Live-Dedup.

    `fill` (2026-08-06 ergaenzt, Default = bisheriges Verhalten, bitgenau):
      "close" — die Bestaetigung gilt, wenn der Bar-CLOSE das Level bricht.
      "touch" — die ruhende Stop-Order fuellt bei der ERSTEN BERUEHRUNG, auch
                wenn der Bar wieder zurueckfaellt. Das ist, was live passiert.
    ⚠ Der Unterschied ist kein Detail: beim Momentum betrug er 0,45 R und drehte
    +0,40 auf −0,06 (`backtest_momentum_v2.py`), beim Squeeze 0,32 R
    (`backtest_squeeze_touchfill.py`). "close" konditioniert auf den Bar-Schluss,
    den eine liegende Order nicht kennen kann.
    """
    def kosten(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    Rs, abst = [], []
    pend = None
    letztes_sig = None            # spiegelt `_signal_last_sig`
    frei_ab = lo
    i = max(lo, _PIV_LOOK + _EMA_SLOW + 5)
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
        w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
                        "pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
        rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
                          htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
        sig = rec.get("signal")

        # ── Breakout-Bestätigung (Mechanik 1:1 aus `_confirm_breakout`) ──────
        # ⚠⚠ `frisch` = die Bestätigung ist GERADE auf diesem Bar eingetreten.
        # Nur dann ist ein Einstieg AM LEVEL überhaupt erreichbar: eine ruhende
        # Stop-Order füllt in genau diesem Moment — oder gar nicht. Ein erster
        # Entwurf liess `pend["level"]` über weitere Bars stehen und wäre bei
        # verzögertem Einstieg (Slot belegt / Konfidenz zu niedrig) zu einem Preis
        # eingestiegen, den der Markt längst verlassen hatte. Das ist kein
        # Look-ahead, aber unerreichbar — und hat die Zahlen sichtbar verbogen
        # (Bruch zwischen Konf 65 und 75). Deshalb gilt `frisch` für BEIDE
        # Varianten, damit A und B dieselbe Einstiegs-Population haben.
        lvl_best = None
        frisch = False
        if _K > 0 and sig in ("LONG", "SHORT"):
            d = 1 if sig == "LONG" else -1
            if pend is None or pend["dir"] != d:
                pend = {"dir": d, "level": C[i] + d * _K * atr,
                        "invalid": C[i] - d * _K * atr, "confirmed": False, "bar": i}
            if not pend["confirmed"]:
                if fill == "touch":
                    # Beruehrung genuegt — die Order liegt dort und wird ausgeloest.
                    broke = ((H[i] >= pend["level"]) if d > 0
                             else (L[i] <= pend["level"]))
                else:
                    broke = ((C[i] >= pend["level"]) if d > 0
                             else (C[i] <= pend["level"]))
                against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
                if broke:
                    pend["confirmed"] = True
                    frisch = True
                else:
                    if against or (i - pend["bar"]) > _TIMEOUT_BARS:
                        pend = {"dir": d, "level": C[i] + d * _K * atr,
                                "invalid": C[i] - d * _K * atr,
                                "confirmed": False, "bar": i}
                    i += 1; continue
            lvl_best = pend["level"]
        elif sig == "WARTEN":
            pend = None
            letztes_sig = None            # Dedup re-armt bei WARTEN (wie live)
        if sig not in ("LONG", "SHORT"):
            i += 1; continue
        rec = w._room_gate(rec, C[i])
        if rec.get("signal") not in ("LONG", "SHORT"):
            i += 1; continue

        sig = rec["signal"]
        if _K > 0 and not frisch:
            i += 1; continue          # nur im Moment der Bestätigung (s. oben)
        if (rec.get("conf_pct") or 0) < min_conf:
            i += 1; continue                       # Dedup NICHT setzen (wie live)
        if nacht and _berlin(T[i]).hour in nacht:
            i += 1; continue                       # Nacht-Sperre, kein Dedup
        if sig == letztes_sig:
            i += 1; continue                       # dieselbe Episode schon gehandelt
        letztes_sig = sig
        if i < frei_ab:
            i += 1; continue                       # Slot belegt → Episode verfällt

        d = 1 if sig == "LONG" else -1
        a = max(atr, _ATRMIN)
        # `slip` = unterstellte Fill-Slippage der ausgeloesten Stop-Order in ×ATR.
        # Eine Stop-Order fuellt zum Markt, sobald sie triggert — der Idealpreis ist
        # die Obergrenze, nicht die Erwartung. Ohne diesen Test waere die Empfehlung
        # eine Behauptung (gleiche Pruefung wie beim Squeeze, Variante E).
        entry = ((lvl_best + d_slip(sig) * slip * atr)
                 if (am_level and lvl_best is not None) else C[i])
        abst.append((C[i] - (lvl_best if lvl_best is not None else C[i])) * d / a)
        r, xj = simulate(entry, d, a, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        Rs.append(r - kosten(i, a))
        pend = None
        frei_ab = xj + 1
        i += 1
    return Rs, abst


def kenn(rs):
    if not rs:
        return None
    n = len(rs); s = sum(rs)
    g = sum(x for x in rs if x > 0); v = -sum(x for x in rs if x < 0)
    return {"n": n, "wr": 100 * sum(1 for x in rs if x > 0) / n,
            "oer": s / n, "sr": s, "pf": (g / v) if v > 0 else float("inf")}


def zeile(lbl, k1, k2):
    def f(k):
        if not k:
            return f"{'—':>34}"
        return (f"{k['n']:>5} {k['wr']:>5.1f}% {k['oer']:>+7.3f} "
                f"{k['pf']:>5.2f} {k['sr']:>+8.1f}")
    beide = bool(k1 and k2 and k1["oer"] > 0 and k2["oer"] > 0
                 and k1["pf"] > 1 and k2["pf"] > 1)
    print(f"  {lbl:<34}{f(k1)}  |{f(k2)}" + ("   <<< BESTEHT" if beide else ""))
    return beide


def main():
    n_bars = 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_bars)
    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]; T = [int(b["time"]) for b in bars]
    SP = [float(b["spread"]) * point for b in bars]
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
    A = _atr_series(H, L, C)
    AD = [0.0] * len(C)
    for i in range(_N_BARS, len(C)):
        AD[i] = calc_trend_angle(C[i - _N_BARS:i + 1], _ANGLE_LR) - 90.0
        if abs(AD[i]) < _ANGLE_DEAD:
            AD[i] = 0.0
    HTF = [_htf_sign(C, i, 6, A[i] or 0.1) for i in range(len(C))]
    H1S = [_htf_sign(C, i, 12, A[i] or 0.1) for i in range(len(C))]
    PH = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
    PL = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]

    w = WaveRecommender(lambda: None)
    w.set_reversal_enabled(False)          # Live-Stand seit 04.08.
    # ⚠⚠ OHNE DIESE ZEILE IST DAS ENTRY-RAUM-GATE AUS: `_entry_room_atr`
    # ist im Konstruktor 0.0, und `_room_gate` gibt dann `rec` unveraendert
    # zurueck — der Aufruf sieht aus wie ein Gate, ist aber ein No-op.
    # Genau das ist mir am 05.08. in drei Skripten passiert (live blockt das
    # Gate 35 % aller Empfehlungen, im Backtest 0 %).
    w.set_entry_room(0.6)                  # Live-Wert
    w.set_dead_hours("")                   # live leer
    mid = len(C) // 2

    print("=" * 106)
    print(f"  AUTO-SIGNAL (🎯 SIG) — Neumessung Stand 05.08.2026 · {sym} M5, {len(C)} Bars")
    print(f"  {_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y} · 2 Halbjahre · "
          f"sequentiell (EIN Slot) · Live-Gate-Stack · kanonischer Exit · Echtkosten")
    print(f"  Exit: SL {LIVE.sl_atr} · Trail {LIVE.mult} · BE {LIVE.be} · "
          f"Time-Stop {LIVE.timestop_bars} Bars · TP {LIVE.tp_init}")
    print("=" * 106)

    def paar(min_conf, nacht, am_level, slip=0.0):
        r1, a1 = lauf(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
                      0, mid, min_conf, nacht, am_level, slip)
        r2, a2 = lauf(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
                      mid, len(C), min_conf, nacht, am_level, slip)
        return kenn(r1), kenn(r2), (a1 + a2)

    print(f"  {'Variante':<34}{'H1  n    WR      OeR    PF     SigmaR':>34}"
          f"  |{'H2  n    WR      OeR    PF     SigmaR':>34}")
    print("  " + "-" * 104)
    print("  A) EINSTIEG AM BAR-CLOSE  (Ist-Zustand: Market-Order)")
    besteht = []
    for mc in (55, 65, 75, 85):
        k1, k2, ab = paar(mc, _NACHT_LIVE, False)
        if zeile(f"   Konf>={mc} · Nacht 3-6 (live)", k1, k2):
            besteht.append(("close", mc))
    print("  " + "-" * 104)
    print("  B) EINSTIEG AM BESTAETIGUNGS-LEVEL  (ruhende Stop-Order, wie beim Squeeze)")
    for mc in (55, 65, 75, 85):
        k1, k2, ab = paar(mc, _NACHT_LIVE, True)
        if zeile(f"   Konf>={mc} · Nacht 3-6 (live)", k1, k2):
            besteht.append(("level", mc))
    print("  " + "-" * 104)
    print("  C) NACHT-FENSTER im Vergleich (Konf>=75, Einstieg am Level)")
    for lbl, nn in (("3-6 (live)", _NACHT_LIVE), ("0-7 (alt)", _NACHT_ALT),
                    ("keine Sperre", ())):
        k1, k2, _ = paar(75, nn, True)
        zeile(f"   Nacht {lbl}", k1, k2)

    print("  " + "-" * 104)
    print("  D) HAERTETEST — Fill-Slippage der Stop-Order (Konf>=65, Nacht 3-6)")
    for sl in (0.00, 0.05, 0.10, 0.20, 0.31):
        k1, k2, _ = paar(65, _NACHT_LIVE, True, sl)
        zeile(f"   Level +{sl:.2f}xATR"
              + ("  (= Ist-Zustand Market)" if sl >= 0.30 else ""), k1, k2)

    # Wie weit laeuft der Kurs ueber das Bestaetigungs-Level?
    _, _, ab = paar(75, _NACHT_LIVE, True)
    if ab:
        s = sorted(ab)
        print()
        print(f"  Abstand Bar-Close ↔ Bestaetigungs-Level (×ATR): Median "
              f"{s[len(s)//2]:+.3f} · Ø {sum(s)/len(s):+.3f} · Max {s[-1]:+.3f}  (n={len(s)})")
        print("  = so viel guenstiger waere eine ruhende Stop-Order am Level.")

    print()
    print("=" * 106)
    print("  URTEIL (Regel VOR dem Lauf fixiert: OeR > 0 UND PF > 1 in BEIDEN")
    print("          Halbjahren, Nachbar-Konfidenzschwellen muessen mithalten)")
    print("=" * 106)
    if besteht:
        print(f"  Einzeln bestanden: {besteht}")
        print("  ⚠ Nachbar-Pruefung von Hand vornehmen (s. Tabelle) — eine einzelne")
        print("    bestehende Schwelle zwischen zwei durchgefallenen ist Rauschen.")
    else:
        print("  => KEINE Variante besteht. Der Auto-Signal-Pfad ist auch mit den")
        print("     Aenderungen vom 05.08. nicht tragfaehig.")
    print()


if __name__ == "__main__":
    main()