#!/usr/bin/env python3 """Divergenz-Wächter: läuft der Bot unter den Bedingungen, unter denen gemessen wurde? Stufe 2 der Deployment-Drift-Behebung (2026-07-31). WARUM ES DAS GIBT: An EINEM Tag wurden drei Fälle gefunden, in denen eine Komponente unter anderen Bedingungen BETRIEBEN als VALIDIERT wurde: 1. P(break) — trainiert auf Anlauf zu FIXIERTEM Level, live dynamisch gewähltes Level → AUC 0,715 fiel auf 0,368. 2. `breakout_k=0,3` — validiert auf FESTER Zeitebene, live wechselt sie 33×/Tag und löschte dabei den Bestätigungs-Anker → 93 % WARTEN statt ~43 %. 3. Trailing-`mult` — validiert bei 1,5, live 1,5–3,0 je nach TF → H1 ΣR −249 statt −1367, Worst −1,50 statt −2,50. Keiner davon wäre durch MEHR Backtesting gefunden worden — die Backtests waren korrekt. Gefehlt hat die Kontrolle, ob der Betrieb ihren Annahmen entspricht. Track B schützt gegen OVERFITTING (Edge existiert gar nicht). Dieses Skript schützt gegen DEPLOYMENT-DRIFT (Edge existiert, läuft aber unter anderen Bedingungen). GEPRÜFT WIRD: A) Signal-Verteilung live vs. Backtest-Erwartung (`backtest_dist.py`: ~43 % WARTEN) B) WELCHES Gate blockt — Anteile je `block_reason` (seit 2026-07-31 geloggt) C) TF-Wechsel-Rate (die Bedingung, unter der `breakout_k` validiert wurde, ist „feste TF" — je mehr Wechsel, desto weiter weg) D) Merkmals-Drift der eingebetteten Modelle: Live-Mittel gegen `_PB_MU`/`_PB_SD` E) GLEICHES FENSTER — Backtest auf genau dem Live-Zeitraum, gegen das Konfidenzintervall der ENTKOPPELTEN Live-Rate (nur mit --backtest) Aufruf: python analyze_divergence.py [tage] [--backtest] (Default 7) """ import re import sqlite3 import sys from datetime import datetime, timedelta from core.stichprobe import anteil_ci, entkoppeln, schluessel_level_stunde DB = "oil_widget_history.db" _BROKER_OFF = 10800 # Brokerzeit = UTC+3 (candles/copy_rates sind Broker-Wallclock) LOG = "oil_widget.log" # Erwartungswerte aus den Backtests — die Messlatte # ⚠⚠ KORRIGIERT 2026-08-11 (Review-Durchgang 2): hier standen **43,0 %** aus # `backtest_dist.py`. Dieser Massstab ist seit dem 2026-08-05 ausdrücklich # UNGÜLTIG — er misst `_build` ALLEIN, ohne Breakout-Bestätigung und ohne # Entry-Raum-Gate, also ohne die beiden groessten Live-Blocker. Mit dem vollen # Live-Stack liegt die Erwartung bei **80,5 %** (`backtest_dist_gated.py`: # entry_room 41,4 % · breakout_pending 17,0 % · stretch 13,8 % · deadband # 7,4 % · min_conf 0,9 %). # ⚠ Wirkung des Fehlers: der Waechter meldete DAUERHAFT „+50,8 Pp Abweichung" — # und ein Waechter, der immer warnt, wird ignoriert. Genau diese Lehre steht # seit dem 02.08. im Epochen-Block darunter; sie galt fuer die Modelle, nicht # fuer diese Konstante. _EXP_WAIT = 80.5 # backtest_dist_gated.py (2026-08-05), VOLLER Live-Stack _EXP_WAIT_TOL = 15.0 # ab dieser Abweichung wird geflaggt (Pp) # ── EPOCHEN ─────────────────────────────────────────────────────────────────── # ⚠ Ergänzt 2026-08-02 (Review-Durchgang 2). Der Wächter rechnete über ein starres # Zeitfenster und wusste NICHT, ab wann ein Modell bzw. eine Telemetrie gilt. # Dadurch meldete er zwei FEHLALARME, die beide nur Artefakte des Fensters waren: # · D0 mischte Vorhersagen des ALTEN und des am 31.07. nachtrainierten # P(break)-Modells (Alarm „Δ +15 Pp" stammte zu 100 % aus dem alten; das neue # hatte n=0, weil der Markt seit Fr 22:55 zu ist). # · B zeigte „block_reason: None 66,8 %" — die Spalte existiert aber erst seit # dem 31.07., davor ist NULL schlicht „nicht erhoben", kein blinder Fleck. # Ein Wächter, der nach jedem Nachtraining wochenlang Alarm schlägt, wird # ignoriert — und dann fängt er den echten Fall auch nicht mehr. # Beim Ändern eines Modells/einer Telemetrie hier das Datum mitziehen. _EPOCHS = { # P(break) auf live-spiegelnder Stichprobe neu gefittet (neue _PB_MU/_SD/_W) "pbreak_model": "2026-07-31", # `recommendations.block_reason` eingeführt (DB-Migration) "block_reason": "2026-07-31", } def _epoch_ts(name: str) -> float: return datetime.fromisoformat(_EPOCHS[name]).timestamp() def _clamp(t0: float, epoch: str) -> tuple[float, bool]: """→ (effektiver Startzeitpunkt, wurde beschnitten?). Sorgt dafür, dass eine Auswertung nie über den Beginn ihrer eigenen Epoche hinausreicht.""" e = _epoch_ts(epoch) return (max(t0, e), e > t0) def _flag(ok: bool) -> str: return "OK " if ok else "⚠ " def sec_a(c, t0): print("\n" + "=" * 84) print(" A) SIGNAL-VERTEILUNG — läuft der Bot so oft wie gemessen?") print("=" * 84) c.execute("SELECT signal, COUNT(*) FROM recommendations WHERE timestamp>=? " "GROUP BY signal", (t0,)) d = dict(c.fetchall()); tot = sum(d.values()) if not tot: print(" keine Daten"); return w = 100.0 * d.get("WARTEN", 0) / tot dev = w - _EXP_WAIT print(f" n={tot} WARTEN {w:.1f} % LONG {100*d.get('LONG',0)/tot:.1f} % " f"SHORT {100*d.get('SHORT',0)/tot:.1f} %") print(f" Backtest-Erwartung: {_EXP_WAIT:.0f} % WARTEN → Abweichung {dev:+.1f} Pp " f"{_flag(abs(dev) <= _EXP_WAIT_TOL)}") if dev > _EXP_WAIT_TOL: print(" ⚠ Der Bot wartet deutlich öfter als gemessen — ein Live-Gate blockt") print(" mehr, als der Backtest kennt. Siehe Abschnitt B.") def sec_b(c, t0): print("\n" + "=" * 84) print(f" B) WELCHES GATE BLOCKT? (block_reason, seit {_EPOCHS['block_reason']})") print("=" * 84) # ⚠ Fenster auf die Epoche beschneiden: vor der Migration ist block_reason # NULL, weil die Spalte nicht existierte — nicht, weil kein Gate gegriffen # hätte. Ungeschnitten meldete der Report „None 66,8 %" und legte damit einen # blinden Fleck nahe, den es nicht gibt (nachgeprüft: seit 01.08. 100 % Abdeckung). t0, cut = _clamp(t0, "block_reason") if cut: print(f" (Fenster auf den Beginn der Telemetrie gekürzt — davor wurde der " f"Grund nicht erhoben)") try: c.execute("SELECT block_reason, COUNT(*) FROM recommendations " "WHERE timestamp>=? AND signal='WARTEN' GROUP BY block_reason " "ORDER BY COUNT(*) DESC", (t0,)) rows = c.fetchall() except sqlite3.OperationalError: print(" Spalte block_reason fehlt (ältere DB)"); return tot = sum(n for _, n in rows) if not tot: print(" noch keine Daten — das Logging läuft erst seit dem Neustart am 31.07.") return # Welche Gates kennt der Backtest? (nur diese dürfen den Vergleich prägen) # ⚠⚠ `entry_room` KORRIGIERT 2026-08-11 (Review-Durchgang 2): stand auf # False = „live-only". Das stimmt seit dem 2026-08-04/05 nicht mehr — # `backtest_breakout_gated.py` und `backtest_dist_gated.py` rufen das # ECHTE `_room_gate` (mit `set_entry_room(0.6)`) und modellieren es damit. # Wirkung: der grösste Blocker (58,9 %) wurde als „nicht gedeckt" gezählt, # wodurch die Zeile „Nur-Live-Gates blocken X %" massiv zu hoch auswies # (63,9 % statt real ~5,0 %) — und damit einen Deployment-Drift behauptete, # den es an dieser Stelle nicht gibt. im_backtest = {"deadband": True, "htf_counter": True, "stretch": True, "min_conf": True, "breakout_pending": True, "entry_room": True, "dead_hour": False, "eia": False, "no_data": False, "stale": False, None: None} print(f" {'Gate':<20}{'n':>8}{'Anteil':>9} im Backtest modelliert?") for br, n in rows: mb = im_backtest.get(br) mark = "ja" if mb else ("NEIN — live-only" if mb is False else "—") print(f" {str(br):<20}{n:>8}{100*n/tot:>8.1f} % {mark}") live_only = sum(n for br, n in rows if im_backtest.get(br) is False) print(f"\n Nur-Live-Gates blocken {100*live_only/tot:.1f} % aller WARTEN-Fälle.") print(" ⚠ Dieser Anteil ist per Definition NICHT durch einen Backtest gedeckt.") def sec_c(t0): print("\n" + "=" * 84) print(" C) TF-WECHSEL — `breakout_k` wurde auf FESTER Zeitebene validiert") print("=" * 84) try: sw = [] for ln in open(LOG, encoding="utf-8", errors="ignore"): m = re.search(r"^(\S+ \S+).*Wellen-Timeframe → (\w+)", ln) if m: try: t = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S") except ValueError: continue if t.timestamp() >= t0: sw.append((t, m.group(2))) except FileNotFoundError: print(" kein Log gefunden"); return if not sw: # ⚠ Kein „OK" mehr (2026-08-02): 0 Wechsel heißt bei geschlossenem Markt # schlicht „keine Daten". Das als Entwarnung zu drucken, wäre genau die # Sorte stiller Fehlschluss, die dieser Report aufdecken soll. print(" 0 TF-Wechsel gefunden — das ist KEINE Entwarnung, sondern") print(" möglicherweise fehlende Evidenz (Markt zu / Log rotiert / Bot aus).") print(" Aussagekräftig ist dieser Abschnitt nur über Handelszeiten.") return days = max(1.0, (datetime.now().timestamp() - t0) / 86400) gaps = sorted((sw[i][0] - sw[i - 1][0]).total_seconds() for i in range(1, len(sw))) med = gaps[len(gaps) // 2] / 60 if gaps else 0 print(f" {len(sw)} Wechsel in {days:.1f} Tagen = {len(sw)/days:.1f}/Tag") if gaps: print(f" Median-Abstand {med:.0f} min {_flag(med >= 15)}") if med < 15: print(" ⚠ Wechselt schneller, als ein Signal seine Breakout-Bestätigung") print(" abschließen kann → prüfe `tf_min_dwell_s`.") def sec_d0(c, t0): """Der SCHÄRFSTE Test: sagt das Modell im Mittel das voraus, was eintritt? ⚠ Diese Prüfung ersetzt NICHT die Merkmals-Drift unten, sie ist ihr überlegen — und das ist eine Lehre aus dem Bau dieses Skripts: mit den ALTEN Modellwerten lag `mom3` nur **0,50σ** unter µ und wäre an einer 1σ-Schwelle NICHT aufgefallen. Entscheidend ist nicht die Abweichung des Merkmals, sondern ihre WIRKUNG nach Gewichtung (mom3 hatte Gewicht 1,43 → 0,50σ wurden zu −0,71 im Logit). Der Vergleich „Ø-Vorhersage vs. echte Rate" fängt das direkt ab: er hätte 23 % vs 41 % sofort gezeigt. """ print("\n" + "=" * 84) print(" D0) MODELL-KALIBRIERUNG — sagt es voraus, was eintritt?") print("=" * 84) # ⚠ NUR Vorhersagen des AKTUELLEN Modells auswerten (Epochen-Schnitt, 2026-08-02). # Ungeschnitten mischte diese Sektion die Vorhersagen des alten und des am # 31.07. nachtrainierten Modells und meldete dessen Fehlkalibrierung als # aktuellen Alarm — obwohl das neue Modell noch gar keine ausgewertete # Vorhersage hatte. Nach einem Nachtraining hätte der Wächter so wochenlang # falsch Alarm geschlagen, bis die Altdaten aus dem Fenster gerollt wären. t0, cut = _clamp(t0, "pbreak_model") if cut: print(f" (nur Vorhersagen seit dem Nachtraining am {_EPOCHS['pbreak_model']})") # ⚠⚠ ENTKOPPELN, BEVOR IRGENDEINE RATE BERICHTET WIRD (Fix 2026-08-07). # Vorher stand hier ein schlichtes `AVG(...)` über ALLE Rohzeilen. Die Zeilen # sind aber NICHT unabhängig — dieselbe Position pendelt vielfach um dasselbe # Level. Real am 07.08.: roh 922 Zeilen mit 52,6 % Bruchrate, entkoppelt 93 # Zeilen mit 45,2 %. Der Wächter hätte auf der Rohzahl Alarm geschlagen # (Δ +10,8 Pp > 8), obwohl der Backtest DESSELBEN Zeitfensters bei 43,4 % lag # und damit im 95-%-Intervall der entkoppelten Rate [35,5 … 54,8] liegt. # Genau diese Verwechslung hat an dem Tag eine Untersuchung mit elf # widerlegten Hypothesen ausgelöst. rows = c.execute("""SELECT ts, level, p_break, outcome FROM pbreak_predictions WHERE ts>=? AND outcome IS NOT NULL ORDER BY ts""", (t0,)).fetchall() roh = len(rows) ent = entkoppeln(rows, schluessel_level_stunde(1, 0)) # Tupel: ts=0, level=1 n = len(ent) if n < 30: print(f" nur {n} ENTKOPPELTE Vorhersagen seit dem Nachtraining " f"(roh {roh}) — zu wenig.") print(f" (Das ist KEIN Entwarnungssignal, sondern fehlende Evidenz. " f"Fälligkeit: Messung `pbreak_accuracy_v2`, ≥350 ENTKOPPELTE.)") return # ⚠ `p_break` steht in der DB bereits in PROZENT (0–100) — nicht nochmal # skalieren. Der erste Entwurf tat es und meldete „Ø-Vorhersage 3763,9 %". pm = sum(float(r[2] or 0) for r in ent) / n treffer = [1 if r[3] == "break" else 0 for r in ent] rate, lo, hi = anteil_ci(treffer, n) rm = 100 * rate gap = rm - pm # ⚠ Geflaggt wird nur, wenn die Ø-Vorhersage AUSSERHALB des Intervalls der # echten Rate liegt — ein Punktabstand allein ist auf dieser Datenlage keine # Aussage. Die alte 8-Pp-Schwelle ohne Intervall war genau der Fehler. drin = lo is not None and lo <= pm / 100 <= hi print(f" n={n} entkoppelt (roh {roh}) Ø-Vorhersage {pm:.1f} % " f"echte Rate {rm:.1f} % Δ {gap:+.1f} Pp") if lo is not None: print(f" 95-%-KI der echten Rate: [{100*lo:.1f} … {100*hi:.1f}] % " f"{_flag(bool(drin))}") if not drin: print(" ⚠ Die Ø-Vorhersage liegt AUSSERHALB des Intervalls — das Modell") print(" bewertet live eine andere Situation als beim Training") print(" (Deployment-Drift). Vor einer Reparatur: den Backtest auf") print(" DASSELBE Zeitfenster einschränken (s. Abschnitt E).") def sec_e(c, t0): """Der Vergleich, der heute gefehlt hat: der Backtest auf DEMSELBEN Fenster. ⚠⚠ WARUM (2026-08-07). Eine Live-Rate gegen einen Backtest-Durchschnitt aus 208 Tagen zu halten, ist kein Vergleich — es mischt den Regime-Unterschied in die Aussage. Real gemessen: die Live-Bruchrate lag bei 52,7 %, der Langfrist-Backtest bei 38,6 %. Auf DIESELBEN sechs Tage eingeschränkt liefert der Backtest **43,4 %**, und entkoppelt liegt live bei **45,2 %** — kein belegbarer Unterschied. Von 14,1 Pp „Lücke" waren **34 % Zeitraum, 53 % Clusterung, der Rest Rauschen**. ⚠ Braucht MT5 und ~1–2 min, deshalb nur mit `--backtest`. """ print("\n" + "=" * 84) print(" E) GLEICHES FENSTER — Backtest auf genau dem Live-Zeitraum") print("=" * 84) rows = c.execute("""SELECT ts, level, outcome FROM pbreak_predictions WHERE ts>=? AND outcome IS NOT NULL ORDER BY ts""", (t0,)).fetchall() ent = entkoppeln(rows, schluessel_level_stunde(1, 0)) # Tupel: ts=0, level=1 if len(ent) < 30: print(f" nur {len(ent)} entkoppelte Live-Vorhersagen — zu wenig.") return rate, lo, hi = anteil_ci([1 if r[2] == "break" else 0 for r in ent], len(ent)) t_von, t_bis = rows[0][0], rows[-1][0] try: import MetaTrader5 as mt5 import numpy as np from backtest_pbreak_retrain import (_atr_series, _ema_series, build_levels, collect, _N_BARS) from core.wave_rec import _EMA_FAST, _EMA_SLOW mt5.initialize() sym = next((s for s in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(s)), None) bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, 20000) mt5.shutdown() T = [int(b["time"]) - _BROKER_OFF 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] import bisect a = max(bisect.bisect_left(T, t_von), _N_BARS) b = bisect.bisect_right(T, t_bis) if b - a < 200: print(" M5-Daten decken das Live-Fenster nicht ab."); return EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW) AT = _atr_series(H, L, C) s = collect(a, b, H, L, C, EF, ES, AT, build_levels(H, L)) bt = float(np.mean([e["y"] for e in s])) if s else float("nan") except Exception as e: # pragma: no cover print(f" Backtest nicht möglich ({e})"); return print(f" Fenster: {datetime.fromtimestamp(t_von):%d.%m.} bis " f"{datetime.fromtimestamp(t_bis):%d.%m.}") print(f" LIVE entkoppelt n={len(ent):<5} Bruchrate {100*rate:.1f} %" + (f" 95-%-KI [{100*lo:.1f} … {100*hi:.1f}] %" if lo is not None else "")) print(f" BACKTEST gleiches Fenster n={len(s):<5} Bruchrate {100*bt:.1f} %") if lo is not None: drin = lo <= bt <= hi print(f" {_flag(bool(drin))} " + ("Backtest liegt im Intervall — kein belegbarer Unterschied." if drin else "Backtest liegt AUSSERHALB — echte Divergenz, nachgehen.")) def sec_d(c, t0): print("\n" + "=" * 84) print(" D) MERKMALS-DRIFT der eingebetteten Modelle") print("=" * 84) try: from core.engine import _PB_MU, _PB_SD except Exception as e: print(f" Modell nicht ladbar: {e}"); return # Live-Merkmale liegen (noch) nicht in der DB → aus candles_m1 rekonstruieren c.execute("SELECT MAX(time) FROM candles_m1") last = c.fetchone()[0] if not last: print(" keine M1-Kerzen — Rekonstruktion nicht möglich"); return import time as _t off = round((last - _t.time()) / 3600) * 3600 c.execute("""SELECT ts,direction,level,price_at_pred,atr FROM pbreak_predictions WHERE ts>=? AND atr IS NOT NULL""", (t0,)) rows = c.fetchall() if len(rows) < 30: print(f" nur {len(rows)} Vorhersagen im Zeitraum — zu wenig"); return def close_at(ts): c.execute("SELECT c FROM candles_m1 WHERE time<=? ORDER BY time DESC LIMIT 1", (ts,)) r = c.fetchone(); return r[0] if r else None m3, m6 = [], [] for ts, d, lvl, px, atr in rows: b = ts + off c0, c3, c6 = close_at(b), close_at(b - 15 * 60), close_at(b - 30 * 60) if None in (c0, c3, c6): continue dd = 1 if d == "LONG" else -1 m3.append((c0 - c3) * dd / atr); m6.append((c0 - c6) * dd / atr) if len(m3) < 30: print(" zu wenige rekonstruierbare Merkmale"); return print(f" n={len(m3)} (P(break)-Modell, Merkmale aus candles_m1 rekonstruiert)") print(f" {'Merkmal':<8}{'Live-µ':>10}{'Modell-µ':>11}{'Δ in SD':>10}") for nm, v, k in (("mom6", m6, 0), ("mom3", m3, 1)): mean = sum(v) / len(v) dsd = (mean - _PB_MU[k]) / _PB_SD[k] print(f" {nm:<8}{mean:>10.3f}{_PB_MU[k]:>11.3f}{dsd:>9.2f}σ {_flag(abs(dsd) <= 0.5)}") print("\n ⚠ Schwelle bewusst 0,5σ (nicht 1σ): der reale P(break)-Fehler lag bei") print(" mom3 nur 0,50σ unter µ — bei Gewicht 1,43 wurden daraus aber −0,71 im") print(" Logit, was 23 % statt 41 % vorhersagte. Merkmals-Drift allein") print(" unterschätzt den Schaden; D0 (Kalibrierung) ist der schärfere Test.") def main(): args = [a for a in sys.argv[1:] if not a.startswith("--")] mit_bt = "--backtest" in sys.argv days = int(args[0]) if args else 7 t0 = (datetime.now() - timedelta(days=days)).timestamp() db = sqlite3.connect(DB); c = db.cursor() print("=" * 84) print(f" DIVERGENZ-WÄCHTER — läuft der Bot unter den validierten Bedingungen?") print(f" Zeitraum: letzte {days} Tage") print("=" * 84) sec_a(c, t0) sec_b(c, t0) sec_c(t0) sec_d0(c, t0) if mit_bt: sec_e(c, _clamp(t0, "pbreak_model")[0]) else: print() print(" (Abschnitt E — Backtest auf gleichem Fenster — mit `--backtest`)") sec_d(c, t0) db.close() print("\n" + "=" * 84) print(" Track B schützt gegen Overfitting. DIESER Report schützt gegen") print(" Deployment-Drift: gemessen unter X, betrieben unter Y.") print("=" * 84) if __name__ == "__main__": main()