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`

Aufruf:  python analyze_divergence.py [tage]      (Default 7)
"""
import re
import sqlite3
import sys
from datetime import datetime, timedelta

DB = "oil_widget_history.db"
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)


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("  B) WELCHES GATE BLOCKT? (block_reason, seit 2026-07-31)")
    print("=" * 84)
    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:
        print("  keine TF-Wechsel im Zeitraum  OK"); 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)
    c.execute("""SELECT AVG(p_break), AVG(CASE WHEN outcome='break' THEN 100.0 ELSE 0 END),
                        COUNT(*) FROM pbreak_predictions
                 WHERE ts>=? AND outcome IS NOT NULL""", (t0,))
    pm, rm, n = c.fetchone()
    if not n or n < 30:
        print(f"  nur {n or 0} ausgewertete Vorhersagen — zu wenig"); return
    gap = (rm or 0) - (pm or 0)
    print(f"  n={n}   Ø-Vorhersage {pm:.1f} %   echte Rate {rm:.1f} %   "
          f"Δ {gap:+.1f} Pp   {_flag(abs(gap) <= 8)}")
    if abs(gap) > 8:
        print("  ⚠ Das Modell ist systematisch verschoben — es bewertet live eine")
        print("    andere Situation als beim Training (Deployment-Drift).")


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():
    days = int(sys.argv[1]) if len(sys.argv) > 1 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)
    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()