#!/usr/bin/env python3 """Legacy-Befunde mit dem ECHTEN Exit gegenprüfen (2026-07-31, Stufe-3-Folgearbeit). BEFUND BEIM STUFE-3-REFACTOR: sechs Backtests nutzen einen **stark vereinfachten Exit** (`_TPTRAIL=0.5`, KEIN Breakeven, KEIN Lock, KEIN Time-Stop, `_MAXH=240`): `backtest_hourly`, `_hourly_split`, `_bounce`, `_events`, `_deadhour`, `_chopgate`. Auf ihnen ruhen aber Befunde, die live **echte Gates** steuern: * **EIA-Blackout** (Mi 15:30–16:30) — AKTIV, blockt Trades (`backtest_events.py`) * **Dead-Hours** (0–7, 12, 16) — aktuell aus, aber jederzeit reaktivierbar (`backtest_hourly_split.py`) Wenn diese Befunde nur unter dem vereinfachten Exit gelten, gaten wir live auf einer Grundlage, die es nicht mehr gibt. Genau das ist Deployment-Drift. DIESES SKRIPT rechnet beide Aussagen mit `core/exit_model.simulate()` neu — dem Exit, den das Live-System wirklich fährt (Phasen, Breakeven 1,3, Time-Stop, `max_hold` 200) — plus ECHTKOSTEN (Bar-Spread/ATR) und dem 2-Hälften-Protokoll. ⚠ Es geht NICHT darum, die alten Zahlen zu reproduzieren (das kann es nicht — anderer Exit), sondern um die FRAGE: hält die SCHLUSSFOLGERUNG? Also: ist das EIA-Fenster auch mit dem echten Exit in beiden Hälften schlechter als der Rest, und sind die Dead-Hours es auch? """ import sys from datetime import datetime, timezone, timedelta import MetaTrader5 as mt5 from core.analysis import calc_trend_angle from core.exit_model import LIVE, simulate from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD, _REVERSAL_STRETCH, _STRETCH_MAX) _ATRMIN = 0.12 NL = chr(10) _BROKER_OFF_H = 3 # Brokerzeit = UTC+3 _BERLIN_OFF_H = 2 # Sommerzeit (die History liegt im Sommerhalbjahr) def _ema_series(v, p): k = 2.0 / (p + 1); o = []; e = v[0] for i, x in enumerate(v): e = x if i == 0 else x * k + e * (1 - k) o.append(e) return o def _atr_series(H, L, C, p=14): t = [0.0] for i in range(1, len(C)): t.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1]))) out = [] for i in range(len(C)): if not i: out.append(None); continue w = t[max(1, i - p + 1):i + 1] out.append(sum(w) / max(1, len(w))) return out def _berlin(ts_broker: int) -> datetime: """Broker-Epoch (UTC+3) → Berliner Wanduhr.""" return datetime.fromtimestamp(ts_broker, timezone.utc) \ - timedelta(hours=_BROKER_OFF_H) + timedelta(hours=_BERLIN_OFF_H) def _er(C, i, n=14): """Efficiency Ratio: Netto-Weg / Summe der Einzelschritte. 1 = glatter Trend, ~0 = Chop. Grundlage des verworfenen ER-Chop-Gates (`backtest_chopgate.py`).""" if i - n < 1: return None net = abs(C[i] - C[i - n]) path = sum(abs(C[k] - C[k - 1]) for k in range(i - n + 1, i + 1)) return (net / path) if path > 0 else 0.0 def trades(T, H, L, C, EF, ES, AT, SP, lo, hi, step=2): """→ [(berlin_dt, R_netto, is_reversal, er)] mit dem ECHTEN Exit + Echtkosten.""" TH = _REVERSAL_STRETCH; out = [] for i in range(max(lo, _N_BARS), min(hi, len(C) - LIVE.max_hold - 1), step): atr = AT[i] if not atr or atr <= 0: continue atr = max(atr, _ATRMIN); es = ES[i]; ef = EF[i] stretch = (C[i] - es) / atr ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR); ad = ang - 90.0 d = 0; rev = False if stretch <= -TH and ad >= _ANGLE_DEAD: d, rev = 1, True elif stretch >= TH and ad <= -_ANGLE_DEAD: d, rev = -1, True elif abs(stretch) < _STRETCH_MAX: d = 1 if ef > es else -1 if ef < es else 0 if not d: continue cost = (SP[i] if SP[i] > 0 else 0.0225) / atr R = simulate(C[i], d, atr, H, L, C, i + 1, LIVE) - cost out.append((_berlin(int(T[i])), R, rev, _er(C, i))) return out def stat(rs): if not rs: return "n=0" n = len(rs); s = sum(rs) g = sum(x for x in rs if x > 0); l = -sum(x for x in rs if x < 0) return (f"n={n:>6} ØR={s/n:+.3f} PF={(g/l if l > 0 else 9.99):>4.2f} ΣR={s:+.0f}") def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000 mt5.initialize(); 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) si = mt5.symbol_info(sym); point = si.point mt5.shutdown() T = [int(b["time"]) for b in bars] 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] EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW); AT = _atr_series(H, L, C) mid = len(C) // 2 print("=" * 90) print(f" LEGACY-RECHECK mit dem ECHTEN Exit — {sym} M5 ({len(C)} Bars)") print(f" Exit: core/exit_model.py (trail {LIVE.mult} · BE {LIVE.be} · " f"Time-Stop {LIVE.timestop_bars}) · Echtkosten") print(" Alt gemessen wurde mit einem Exit OHNE Breakeven/Lock/Time-Stop.") print("=" * 90) halves = [("H1", _N_BARS, mid), ("H2", mid, len(C))] data = {lbl: trades(T, H, L, C, EF, ES, AT, SP, a, b) for lbl, a, b in halves} for lbl, _, _ in halves: print(f" {lbl}: {len(data[lbl])} Trades") # ── 1) EIA-Blackout: Mi 15:30–16:30 Berlin (LIVE AKTIVES GATE) print("\n" + "=" * 90) print(" 1) EIA-BLACKOUT (Mi 15:30–16:30 Berlin) — blockt live echte Trades") print("=" * 90) for lbl, _, _ in halves: inw = [r for t, r, _rv, _e in data[lbl] if t.weekday() == 2 and (15, 30) <= (t.hour, t.minute) < (16, 30)] out = [r for t, r, _rv, _e in data[lbl] if not (t.weekday() == 2 and (15, 30) <= (t.hour, t.minute) < (16, 30))] print(f" {lbl} im Fenster : {stat(inw)}") print(f" {lbl} ausserhalb : {stat(out)}") if inw and out: delta = sum(inw) / len(inw) - sum(out) / len(out) print(f" → Differenz ØR {delta:+.3f} " f"{'Fenster schlechter (Gate bestätigt)' if delta < 0 else '⚠ Fenster BESSER — Gate nicht gedeckt'}") # ── 2) Dead-Hours: Netto-R je Berliner Stunde print("\n" + "=" * 90) print(" 2) STUNDEN-PROFIL (Dead-Hours-Grundlage) — Netto-ØR je Berliner Stunde") print("=" * 90) print(f" {'Std':<5}{'H1 n':>7}{'H1 ØR':>9}{'H2 n':>7}{'H2 ØR':>9} Urteil") robust_neg, robust_pos = [], [] for h in range(24): a = [r for t, r, _rv, _e in data["H1"] if t.hour == h] b = [r for t, r, _rv, _e in data["H2"] if t.hour == h] if len(a) < 30 or len(b) < 30: continue ma, mb = sum(a) / len(a), sum(b) / len(b) verdict = "" if ma < 0 and mb < 0: verdict = "beidseitig NEGATIV"; robust_neg.append(h) elif ma > 0 and mb > 0: verdict = "beidseitig positiv"; robust_pos.append(h) print(f" {h:<5}{len(a):>7}{ma:>+9.3f}{len(b):>7}{mb:>+9.3f} {verdict}") print(f"\n Robust NEGATIV: {robust_neg or '—'}") print(f" Robust POSITIV: {robust_pos or '—'}") print(" ⚠ Alt (vereinfachter Exit) waren 0–7 sowie 12 und 16 die Kandidaten.") print(" Deckt sich das? Falls nicht, ruht das Dead-Hours-Gate auf dem alten Exit.") # ── 3) BOUNCE / Reversal — AKTIVES Setup (WAVE_REV_*) print(NL + "=" * 90) print(" 3) BOUNCE / REVERSAL — aktives Setup, alt gemessen ØR +0,185 · PF 1,35") print("=" * 90) for lbl, _, _ in halves: rev = [r for _t, r, rv, _e in data[lbl] if rv] trd = [r for _t, r, rv, _e in data[lbl] if not rv] print(f" {lbl} Reversal : {stat(rev)}") print(f" {lbl} Trend : {stat(trd)}") rr = [sum(1 for _t, _r, rv, _e in data[l] if rv) for l in ("H1", "H2")] print(f"{NL} ⚠ Alt: ØR +0,185 / PF 1,35 / Treffer 70 % (vereinfachter Exit, ohne Kosten).") print(f" Reversal-Anteil hier: {rr[0]} (H1) / {rr[1]} (H2) Trades.") print(" Urteil: traegt das Setup mit dem ECHTEN Exit noch, und schlaegt es Trend?") # ── 4) ER-CHOP-GATE — verworfen; haelt die Ablehnung? print(NL + "=" * 90) print(" 4) ER-CHOP-GATE (verworfen) — trennt Efficiency Ratio mit echtem Exit?") print("=" * 90) print(f" {'ER-Band':<14}{'H1 n':>7}{'H1 ØR':>9}{'H2 n':>7}{'H2 ØR':>9} Urteil") bands = [(0.0, 0.15), (0.15, 0.30), (0.30, 0.50), (0.50, 1.01)] for lo_e, hi_e in bands: a = [r for _t, r, _rv, e in data["H1"] if e is not None and lo_e <= e < hi_e] b = [r for _t, r, _rv, e in data["H2"] if e is not None and lo_e <= e < hi_e] if len(a) < 50 or len(b) < 50: continue ma, mb = sum(a) / len(a), sum(b) / len(b) v = "konsistent" if (ma - mb) * 0 == 0 and ((ma < 0) == (mb < 0)) else "kippt" print(f" {lo_e:.2f}–{hi_e:<9.2f}{len(a):>7}{ma:>+9.3f}{len(b):>7}{mb:>+9.3f} {v}") print(NL + " ⚠ Alt: Chop-Signale NICHT robust schlechter (H1 Trend leicht besser,") print(" H2 umgekehrt) -> Gate verworfen. Bestaetigt sich das mit echtem Exit?") if __name__ == "__main__": main()