AH-Oil-Trader/analyze_divergence.py

#!/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
_EXP_WAIT = 43.0          # backtest_dist.py: ~57 % LONG/SHORT, ~43 % WARTEN
_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)
    im_backtest = {"deadband": True, "htf_counter": True, "stretch": True,
                   "min_conf": True, "breakout_pending": True,
                   "entry_room": False, "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()