#!/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() # ── TEIL 2: welche Nachtstunden sind WIRKLICH tot? ───────────────────── print("=" * 100) print(" TEIL 2 — WELCHE Nachtstunden tragen den Block? (je Stunde, sequentiell)") print("=" * 100) print(f" {'Std':>3} {'Kosten':>7} | {'n H1':>5} {'OeR H1':>8} {'SigmaR H1':>10}" f" | {'n H2':>5} {'OeR H2':>8} {'SigmaR H2':>10} Urteil") print(" " + "-" * 92) beide_neg = [] for s in range(24): k1 = kennzahlen(sequentiell(K, H, L, C, lambda e, s=s: e["stunde"] == s, 0, mid)) k2 = kennzahlen(sequentiell(K, H, L, C, lambda e, s=s: e["stunde"] == s, mid, N)) ks_ = [e["kosten"] for e in E if e["stunde"] == s] kmed = sorted(ks_)[len(ks_) // 2] if ks_ else float("nan") if not k1 or not k2: continue neg = k1["sr"] < 0 and k2["sr"] < 0 if neg: beide_neg.append(s) print(f" {s:>3} {kmed:>7.3f} | {k1['n']:>5} {k1['oer']:>+8.3f} {k1['sr']:>+10.1f}" f" | {k2['n']:>5} {k2['oer']:>+8.3f} {k2['sr']:>+10.1f} " f"{'beide NEGATIV' if neg else ''}") print() print(f" In BEIDEN Haelften negativ: {beide_neg or 'keine'}") print(" ⚠ Diese Liste NICHT direkt als Gate verwenden — 24 Stunden = 24 Tests,") print(" das waere in-sample-zirkulaer (dokumentierter Fehler, backtest_realcosts).") print(" Deshalb unten nur ZUSAMMENHAENGENDE Fenster: ein Fenster hat 1 Freiheits-") print(" grad statt 24 und laesst sich nicht auf das Rauschen zurechtlegen.") print() # ── Fenster-Sweep, Regel vorab fixiert ──────────────────────────────── print(" ZUSAMMENHAENGENDE NACHT-FENSTER (sequentiell, EIN Slot)") print(" REGEL (vor dem Lauf fixiert): das Fenster muss das heutige 0-7 im SigmaR") print(" in BEIDEN Haelften schlagen. Bei mehreren Treffern gewinnt das LAENGSTE") print(" — der Beaufsichtigungs-Grund hat Wert, der in keiner Zahl steht.") print(f" {'Fenster':<18}{'H1 n WR OeR PF SigmaR':>32}" f" |{'H2 n WR OeR PF SigmaR':>32}") print(" " + "-" * 96) fenster = [("kein Gate", ())] for a in range(0, 8): for b in range(a, 8): fenster.append((f"{a}-{b} Uhr", tuple(range(a, b + 1)))) b1, b2 = ergebnis["ohne Gate (Basis)"] zeilen = [] for lbl, hh in fenster: if not hh: continue k1 = kennzahlen(sequentiell(K, H, L, C, lambda e, hh=hh: e["stunde"] not in hh, 0, mid)) k2 = kennzahlen(sequentiell(K, H, L, C, lambda e, hh=hh: e["stunde"] not in hh, mid, N)) if not k1 or not k2: continue p1 = k1["sr"] - b1["sr"]; p2 = k2["sr"] - b2["sr"] # Preis (negativ) nblk = sum(1 for e in E if e["stunde"] in hh) zeilen.append({"lbl": lbl, "hh": hh, "h": len(hh), "p1": p1, "p2": p2, "n": nblk, "je": (p1 + p2) / len(hh)}) print(" PREIS DER BEAUFSICHTIGUNG — was kostet ein Fenster gegenueber KEINEM Gate?") print(" ⚠ Keine Stunde ist in beiden Haelften negativ. Es gibt also keine") print(" 'toten Stunden' zu streichen — jedes Fenster KOSTET. Die Frage ist") print(" nur, wieviel Schlaf-Schutz man wie teuer einkauft. Das ist eine") print(" PREIS-Minimierung fuer eine Praeferenz, KEINE Edge-Behauptung —") print(" deshalb ist die Auswahl hier auch nicht overfitting-gefaehrdet.") print(f" {'Fenster':<12}{'Std':>4}{'geblockt':>10}{'Preis H1':>10}{'Preis H2':>10}" f"{'SigmaR/Std':>12}") print(" " + "-" * 60) for z in sorted(zeilen, key=lambda z: -z["je"])[:12]: print(f" {z['lbl']:<12}{z['h']:>4}{z['n']:>10}{z['p1']:>+10.1f}" f"{z['p2']:>+10.1f}{z['je']:>+12.1f}") heute = next(z for z in zeilen if z["lbl"] == "0-7 Uhr") print(f" {'... heute:':<12}{heute['h']:>4}{heute['n']:>10}{heute['p1']:>+10.1f}" f"{heute['p2']:>+10.1f}{heute['je']:>+12.1f} <- 0-7 Uhr") print() # Auswahl: guenstigstes Fenster, das noch >=4 Stunden Schlaf-Schutz gibt. # 4 h ist gesetzt, nicht gemessen — darunter waere der Schutzzweck hinfaellig. kand = [z for z in zeilen if z["h"] >= 4] best = max(kand, key=lambda z: z["je"]) print(f" => EMPFEHLUNG (>=4 h Schutz, guenstigstes Fenster): **{best['lbl']}**") print(f" Preis H1 {best['p1']:+.1f} / H2 {best['p2']:+.1f} R " f"gegen heute {heute['p1']:+.1f} / {heute['p2']:+.1f}") print(f" = {100*abs(best['p1'])/abs(heute['p1']):.0f} % / " f"{100*abs(best['p2'])/abs(heute['p2']):.0f} % der heutigen Kosten," f" bei {best['h']} statt 8 Stunden") print(f" freigegeben: {[s for s in range(8) if s not in best['hh']]} Uhr") print() if __name__ == "__main__": main()