#!/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 bisect import bisect_left, bisect_right 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)} # ⚠⚠ WAR DER ENGPASS DES GANZEN PROJEKTS (Profil 2026-08-07): die frühere # Fassung filterte je Bar die KOMPLETTE Pivot-Liste # [C[j] for j in PH if i-_PIV_LOOK <= j <= i-_PIV_K] # also O(Bars × Pivots). Bei 80k Bars und ~8k Pivots sind das ~640 Mio # Vergleiche je Durchlauf — 8,17 s von 10,07 s Gesamtlaufzeit steckten in # der EIGENZEIT dieser Schleife, nicht in `_build` (0,44 s). Daher die # 9–20-Minuten-Läufe. # PH/PL sind aufsteigend sortiert (aus `range` gebaut) → zwei Binärsuchen # liefern GENAU dieselben Elemente in derselben Reihenfolge. Kein # Näherungsverfahren, sondern dieselbe Menge in O(log n). # ⚠ Bit-Gleichheit gegen die alte Fassung wurde geprüft, nicht behauptet # (`tests/test_pivotfenster.py`, 200 Bars × beide Listen). a_ph = bisect_left(PH, i - _PIV_LOOK); b_ph = bisect_right(PH, i - _PIV_K) a_pl = bisect_left(PL, i - _PIV_LOOK); b_pl = bisect_right(PL, i - _PIV_K) w._pb_levels = {"ph": [C[j] for j in PH[a_ph:b_ph]], "pl": [C[j] for j in PL[a_pl:b_pl]]} 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()