AH-Oil-Trader/backtest_cost_gate.py

#!/usr/bin/env python3
"""Kosten-Gate statt Stunden-Gate für den Auto-Squeeze (2026-08-05).

ANLASS (B1 — Hypothese + Hebel-Check): Am 05.08. blockte `auto_squeeze_skip_night`
fünf Squeeze-Ausbrüche, darunter den Einstieg direkt vor einem −1,8-$-Rutsch. Beim
Nachrechnen fiel auf: die REALEN Kosten lagen in dieser Nacht bei 0,11–0,22×ATR —
also klar unter der 0,32er-Schwelle, mit der die Nachtsperre begründet ist
(`backtest_realcosts.py`: Spread ÷ Nacht-ATR = 0,32–0,50).

Die Nachtsperre ist also ein **Proxy**: sie sperrt die Stunde, in der die Kosten
ÜBLICHERWEISE hoch sind. Ein Kosten-Gate würde die Größe messen, um die es
eigentlich geht. Hebel = Gate/Filter → laut CLAUDE.md „klein, hohe Beweislast".

FRAGE: Schlägt „blocke wenn Spread/ATR > X" das Stunden-Gate 0–7 — in BEIDEN
Halbjahren, und halten die Nachbar-Schwellen mit?

VORAB FIXIERTE ENTSCHEIDUNGSREGEL (vor dem ersten Lauf festgelegt):
  Ein Kosten-Gate ersetzt das Stunden-Gate nur, wenn
    (a) ΣR **und** ØR in BEIDEN Hälften ≥ Stunden-Gate, UND
    (b) die beiden NACHBAR-Schwellen ebenfalls ≥ Stunden-Gate liegen
        (sonst ist es Parameter-Zufall — die Lehre aus ORB und den SMC-Sweeps).
  Sonst: keine Änderung.

METHODIK
 * Population = Auto-Squeeze-Ausbrüche über die ECHTE Live-Methode
   `WaveRecommender._squeeze_one` (kein Nachbau — sonst misst man wieder etwas
   anderes als betrieben wird, Deployment-Drift).
 * **GETEILTE, FESTE Entry-Liste.** Alle Gates sind Teilmengen DERSELBEN Liste.
   Eine sequentielle Sim je Variante würde jeder Variante eine andere Trade-Folge
   geben und die Zahlen unvergleichbar machen — genau der Methodik-Fehler, der in
   `backtest_exit_combo.py` dokumentiert ist.
 * Exit = `core/exit_model.simulate()` mit `LIVE` (timestop + TP = die einzige
   live-treue Variante). Kosten = echter Bar-Spread ÷ ATR.
 * 2 Halbjahre, Urteil nur bei beidhälftiger Robustheit.

⚠ ABGRENZUNG: Gemessen wird das Gate für den **Auto-Squeeze** (`auto_squeeze_
skip_night`, live AN). Das zweite „Dead-Hour"-Gate (`[trading] dead_hours` für das
Wellensignal) ist live LEER und betrifft eine andere Population.

⚠ Der zweite Grund für die Nachtsperre — „kein Trade soll unbeaufsichtigt feuern,
während ich schlafe" — ist eine Präferenz und wird von diesem Skript NICHT berührt.
Selbst ein gewinnendes Kosten-Gate beantwortet sie nicht.

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

import MetaTrader5 as mt5

import core.wave_rec as wr
from core.exit_model import LIVE, simulate
from core.wave_rec import WaveRecommender

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_NACHT = (0, 1, 2, 3, 4, 5, 6, 7)      # _SQUEEZE_NIGHT aus engine.py
_COOL = 12          # Bars Sperre nach einem Entry (Überlappung dämpfen) — für ALLE
                    # Varianten identisch, damit die Entry-Liste geteilt bleibt
_ATRMIN = 0.06      # Live-Floor (core/trailing.py)
_SCHWELLEN = (0.15, 0.20, 0.25, 0.30, 0.35, 0.40, 0.50)


def _berlin(ts):
    """MT5-Bar-Zeitstempel → echte Berliner Zeit.

    ⚠ `fromtimestamp(ts, UTC+3)` wäre um 3 h daneben: MT5 liefert dort bereits die
    Broker-Wallclock, keinen echten UTC-Epoch (Fehler vom 04.08., Fix am 05.08.)."""
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN))


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 entries(H, L, C, T, SP, A):
    """Die GETEILTE Entry-Liste: jeder Squeeze-Ausbruch genau einmal.

    Dedup wie live (`_squeeze_last_sig`): 1× je Ausbruch über (dir, level); erst
    wenn der Squeeze sich klärt (state != 'active'), ist der nächste scharf.
    """
    out = []
    last_sig = None
    sperre = -1
    for i in range(wr._SQ_N + 2, len(C) - 2):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        # Nur das benötigte Fenster übergeben: `_squeeze_one` liest ausschliesslich
        # `[-_SQ_N-1:-1]` und `[-1]`. `H[:i+1]` wäre bitgleich, kopiert aber je Bar
        # die ganze Historie → O(n²) und bei 80k Bars unbrauchbar. (Gegengeprüft:
        # identische Ergebnisse auf 8.000 Bars.)
        w0 = i - wr._SQ_N - 2
        st = WaveRecommender._squeeze_one(H[w0:i + 1], L[w0:i + 1], C[w0:i + 1], atr)
        if st.get("state") != "active":
            last_sig = None                     # re-armt, sobald sich der Squeeze klärt
            continue
        sig = (st["dir"], st["level"])
        if sig == last_sig:
            continue                            # derselbe Ausbruch, nicht doppelt
        last_sig = sig
        if i < sperre:
            continue                            # Überlappungs-Cooldown
        sperre = i + _COOL
        d = 1 if st["dir"] == "LONG" else -1
        lvl = float(st["level"])
        kosten = (SP[i] if SP[i] > 0 else 0.0225) / atr
        R = simulate(lvl, d, atr, H, L, C, i + 1, LIVE,
                     timestop=True, use_tp=True) - kosten
        out.append({"i": i, "dir": st["dir"], "R": R, "kosten": kosten,
                    "stunde": _berlin(T[i]).hour, "atr": atr})
    return out


def kandidaten(H, L, C, T, SP, A):
    """Rohe Ausbruchs-Kandidaten OHNE Cooldown — Grundlage der sequentiellen Sim.

    Dort regelt die Belegung des einen Positions-Slots die Überlappung von selbst,
    ein künstlicher Cooldown wäre doppelt gemoppelt."""
    out = []
    last_sig = None
    for i in range(wr._SQ_N + 2, len(C) - 2):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        w0 = i - wr._SQ_N - 2
        st = WaveRecommender._squeeze_one(H[w0:i + 1], L[w0:i + 1], C[w0:i + 1], atr)
        if st.get("state") != "active":
            last_sig = None
            continue
        sig = (st["dir"], st["level"])
        if sig == last_sig:
            continue
        last_sig = sig
        out.append({"i": i, "d": 1 if st["dir"] == "LONG" else -1,
                    "lvl": float(st["level"]), "atr": atr,
                    "kosten": (SP[i] if SP[i] > 0 else 0.0225) / atr,
                    "stunde": _berlin(T[i]).hour})
    return out


def sequentiell(K, H, L, C, pred, lo, hi):
    """EIN Positions-Slot, wie live: ein Ausbruch wird nur gehandelt, wenn gerade
    kein Trade läuft. Genau das entscheidet, ob ein Entry-Gate dem KONTO nützt —
    ein geblockter Nacht-Trade gibt den Slot für den nächsten Tag-Trade frei."""
    rs = []
    frei_ab = lo
    for k in K:
        if not (lo <= k["i"] < hi) or k["i"] < frei_ab or not pred(k):
            continue
        r, xb = simulate(k["lvl"], k["d"], k["atr"], H, L, C, k["i"] + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        rs.append(r - k["kosten"])
        frei_ab = xb + 1
    return rs


def kennzahlen(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.0 * 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"{'—':>32}"
        return (f"{k['n']:>5} {k['wr']:>5.1f}% {k['oer']:>+7.3f} "
                f"{k['pf']:>5.2f} {k['sr']:>+8.1f}")
    print(f"  {lbl:<26}{f(k1)}  |{f(k2)}")


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]
    A = _atr_series(H, L, C)
    N = len(C); mid = N // 2

    print("=" * 100)
    print(f"  KOSTEN-GATE statt STUNDEN-GATE — Auto-Squeeze, {sym} M5, {N} Bars")
    print(f"  {_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y} · 2 Halbjahre")
    print(f"  Exit: kanonisch (SL {LIVE.sl_atr} · Trail {LIVE.mult} · BE {LIVE.be} · "
          f"Time-Stop {LIVE.timestop_bars} Bars · TP {LIVE.tp_init}) · Echtkosten")
    print("=" * 100)

    E = entries(H, L, C, T, SP, A)
    e1 = [e for e in E if e["i"] < mid]
    e2 = [e for e in E if e["i"] >= mid]
    print(f"  Squeeze-Ausbrueche gesamt: {len(E)}   (H1 {len(e1)} · H2 {len(e2)})")
    ks = sorted(e["kosten"] for e in E)
    print(f"  Kostenverteilung Spread/ATR: Median {ks[len(ks)//2]:.3f} · "
          f"25% {ks[len(ks)//4]:.3f} · 75% {ks[3*len(ks)//4]:.3f}")
    nacht = [e for e in E if e["stunde"] in _NACHT]
    print(f"  davon in 0-7 Uhr: {len(nacht)} ({100.0*len(nacht)/max(1,len(E)):.0f} %)"
          f" · deren Median-Kosten {sorted(e['kosten'] for e in nacht)[len(nacht)//2]:.3f}")
    # ── MECHANISMUS: gilt die Nacht-Kostenfalle fuer SQUEEZE-Bars ueberhaupt? ──
    # `backtest_realcosts.py` misst 0,32-0,50xATR fuer die Nacht — das ist der
    # DURCHSCHNITTLICHE Nacht-Bar. Der Squeeze feuert aber nur beim Ausbruch, also
    # wenn der ATR gerade anzieht. Wenn diese Zahlen auseinanderlaufen, ist die
    # Nachtsperre auf der falschen Population begruendet.
    def stat(v):
        if not v:
            return "      —"
        s = sorted(v)
        return f"Med {s[len(s)//2]:.3f} / Ø {sum(v)/len(v):.3f} (n={len(v)})"

    def bars_kosten(nacht, floor):
        return [SP[i] / A[i] for i in range(len(C))
                if A[i] and SP[i] > 0 and (A[i] >= _ATRMIN if floor else True)
                and ((_berlin(T[i]).hour in _NACHT) == nacht)]

    print("  MECHANISMUS — Kosten (Spread/ATR):")
    print("    ⚠ MIT ATR-Floor 0,06 gerechnet ist die Aussage verzerrt: gerade die")
    print("      teuersten Nacht-Bars (winziger ATR) fallen raus. Deshalb beides.")
    for floor, lbl in ((True, "mit Floor 0,06"), (False, "OHNE Floor    ")):
        print(f"    ALLE Bars  {lbl}   Nacht {stat(bars_kosten(True, floor))}"
              f"   Tag {stat(bars_kosten(False, floor))}")
    sq_n = [e["kosten"] for e in E if e["stunde"] in _NACHT]
    sq_t = [e["kosten"] for e in E if e["stunde"] not in _NACHT]
    print(f"    nur SQUEEZE-Ausbrueche       Nacht {stat(sq_n)}   Tag {stat(sq_t)}")
    print("    -> Der Squeeze feuert beim Ausbruch, also bei ANZIEHENDEM ATR. Liegt")
    print("       seine Nacht-Zeile deutlich unter der aller Nacht-Bars, ist die")
    print("       Nacht-Kostenfalle fuer DIESE Population nicht einschlaegig.")
    print()

    # ── Was BLEIBT je Gate (die gehandelten Trades) ────────────────────────
    print("  GEHANDELT (was das Gate durchlaesst)")
    print(f"  {'Gate':<26}{'H1  n    WR      OeR    PF      SigmaR':>32}"
          f"  |{'H2  n    WR      OeR    PF      SigmaR':>32}")
    print("  " + "-" * 96)
    varianten = [("ohne Gate (Basis)", lambda e: True),
                 ("STUNDEN-GATE 0-7 (live)", lambda e: e["stunde"] not in _NACHT)]
    for x in _SCHWELLEN:
        varianten.append((f"KOSTEN-GATE > {x:.2f}xATR", lambda e, x=x: e["kosten"] <= x))
    varianten.append(("KOMBI Stunde + 0,30", lambda e: e["stunde"] not in _NACHT
                      and e["kosten"] <= 0.30))
    ergebnis = {}
    for lbl, pred in varianten:
        k1 = kennzahlen([e["R"] for e in e1 if pred(e)])
        k2 = kennzahlen([e["R"] for e in e2 if pred(e)])
        ergebnis[lbl] = (k1, k2)
        zeile(lbl, k1, k2)

    # ── Was das Gate WEGWIRFT — der eigentliche Test ───────────────────────
    print()
    print("  GEBLOCKT (war der weggeworfene Teil wirklich schlecht?)")
    print("  " + "-" * 96)
    for lbl, pred in varianten[1:]:
        k1 = kennzahlen([e["R"] for e in e1 if not pred(e)])
        k2 = kennzahlen([e["R"] for e in e2 if not pred(e)])
        zeile(lbl, k1, k2)

    # ── DER ENTSCHEIDENDE TEST: EIN Positions-Slot, sequentiell ────────────
    # Oben ist jeder Ausbruch unabhaengig bewertet — das ueberschaetzt, was ein
    # Konto mit EINER Position einsammeln kann (Trades ueberlappen). Beim Vergleich
    # von ENTRY-Gates ist die Abweichung der Trade-Folge kein Artefakt, sondern der
    # Kern der Frage: ein geblockter Nacht-Trade gibt den Slot frei.
    print()
    print("=" * 100)
    print("  SEQUENTIELL — EIN Positions-Slot (das ist die Konto-relevante Zahl)")
    print("=" * 100)
    K = kandidaten(H, L, C, T, SP, A)
    print(f"  rohe Ausbruchs-Kandidaten: {len(K)}")
    print(f"  {'Gate':<26}{'H1  n    WR      OeR    PF      SigmaR':>32}"
          f"  |{'H2  n    WR      OeR    PF      SigmaR':>32}")
    print("  " + "-" * 96)
    seq = {}
    for lbl, pred in varianten:
        k1 = kennzahlen(sequentiell(K, H, L, C, pred, 0, mid))
        k2 = kennzahlen(sequentiell(K, H, L, C, pred, mid, N))
        seq[lbl] = (k1, k2)
        zeile(lbl, k1, k2)

    # ── Urteil nach der VORAB fixierten Regel ──────────────────────────────
    print()
    print("=" * 100)
    print("  URTEIL (Regel VOR dem Lauf fixiert: SigmaR UND OeR in BEIDEN Haelften")
    print("          >= Stunden-Gate, UND beide Nachbar-Schwellen ebenfalls)")
    print("  ⚠ Gewertet wird die SEQUENTIELLE Tabelle — sie bildet den einen")
    print("    Positions-Slot ab, den der Bot real hat.")
    print("=" * 100)
    ergebnis = seq
    sh1, sh2 = ergebnis["STUNDEN-GATE 0-7 (live)"]

    def schlaegt(lbl):
        k1, k2 = ergebnis.get(lbl, (None, None))
        if not k1 or not k2 or not sh1 or not sh2:
            return False
        return (k1["sr"] >= sh1["sr"] and k2["sr"] >= sh2["sr"]
                and k1["oer"] >= sh1["oer"] and k2["oer"] >= sh2["oer"])

    besteht = []
    for idx, x in enumerate(_SCHWELLEN):
        lbl = f"KOSTEN-GATE > {x:.2f}xATR"
        selbst = schlaegt(lbl)
        nachbarn = [f"KOSTEN-GATE > {_SCHWELLEN[j]:.2f}xATR"
                    for j in (idx - 1, idx + 1) if 0 <= j < len(_SCHWELLEN)]
        nb_ok = all(schlaegt(m) for m in nachbarn)
        mark = "BESTEHT" if (selbst and nb_ok) else \
               ("schlaegt Stunden-Gate, Nachbarn NICHT" if selbst else "faellt durch")
        if selbst and nb_ok:
            besteht.append(x)
        print(f"  X = {x:.2f}xATR   {mark}")
    print()
    if besteht:
        print(f"  => Kandidat(en): {', '.join(f'{x:.2f}' for x in besteht)}")
    else:
        print("  => KEINE Schwelle erfuellt die Regel als ERSATZ.")

    # ── Zusatzbefund: taugt das Stunden-Gate ueberhaupt? ───────────────────
    # Die Regel oben vergleicht Kosten-Gate GEGEN Stunden-Gate. Sie beantwortet
    # NICHT, ob das Stunden-Gate selbst traegt. Das steht in derselben Tabelle.
    b1, b2 = ergebnis["ohne Gate (Basis)"]
    print()
    print("  ZUSATZBEFUND — traegt das Stunden-Gate selbst? (sequentiell, SigmaR)")
    print(f"    ohne Gate    H1 {b1['sr']:+8.1f}   H2 {b2['sr']:+8.1f}")
    print(f"    Stunden-Gate H1 {sh1['sr']:+8.1f}   H2 {sh2['sr']:+8.1f}")
    print(f"    Differenz    H1 {sh1['sr']-b1['sr']:+8.1f}   H2 {sh2['sr']-b2['sr']:+8.1f}"
          f"   {'-> Gate KOSTET in beiden Haelften' if (sh1['sr'] < b1['sr'] and sh2['sr'] < b2['sr']) else ''}")
    print()
    print("  ⚠ NICHT modelliert: Slippage. Nachts ist das Buch duenner, und der")
    print("    gemessene Tail liegt bei 0,75xATR ueber den Stop hinaus. Das trifft")
    print("    Nacht-Trades selektiv und ist die groesste Bedrohung dieses Befunds.")
    print("  ⚠ Der Beaufsichtigungs-Grund der Nachtsperre („kein Trade soll feuern,")
    print("    waehrend ich schlafe\") ist eine Praeferenz und wird hier NICHT beruehrt.")
    print()


if __name__ == "__main__":
    main()