AH-Oil-Trader/analyze_verdict_modules.py

#!/usr/bin/env python3
"""Welches Verdict-Modul sagt tatsächlich etwas voraus? (2026-08-05)

FRAGE (User): „Bewerte die Signale, die in die Gesamtempfehlung einfließen."
Die Tabelle `verdict_votes` wird seit 2026-07-24 genau dafür geschrieben — bisher
hat sie niemand ausgewertet.

⚠⚠ ZWEI FALLSTRICKE, beide hier adressiert:
 1. **Autokorrelation.** Die Tabelle schreibt ~1 Zeile/Minute. Ein Modul, das drei
    Stunden lang LONG sagt, erzeugt 180 Zeilen — das ist EINE Marktlage, nicht 180
    Beobachtungen. Wer roh zählt, hält 20 Episoden für 286 Datenpunkte (genau dieser
    Fehler ist bei der Copilot-Sichtung am 2026-08-02 aufgefallen). Deshalb werden
    aufeinanderfolgende gleiche Stimmen zu **EPISODEN** zusammengefasst und nur die
    ERSTE Zeile je Episode gewertet.
 2. **Kurzer Zeitraum.** 24.07.–05.08. = ~12 Tage, EIN Regime. Der Zwei-Stichproben-
    Split ist hier bestenfalls ein Konsistenz-Check, KEIN Robustheitsnachweis. Das
    Ergebnis ist eine SICHTUNG; die Reminder-Schwelle steht bewusst bei 25.000
    Verdicts (~3 Wochen), erreicht sind ~13.500.

METHODIK: Forward-Return in Stimmen-Richtung, ATR-normiert, aus MT5-M5-Bars
(NICHT aus `candles_m1` — das deckt nur ~14 Tage und liefert für ältere Zeitstempel
still denselben Randwert; dieser Fehler ist bei der News-Messung am 03.08. passiert).
Gegen die **Drift-Null** gelesen: in einem Aufwärts-Regime sieht jedes LONG-Modul
gut aus, deshalb wird der Mittelwert ALLER Bars als Nulllinie abgezogen.

Aufruf:  python analyze_verdict_modules.py [minuten]
"""
import bisect
import sqlite3
import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import MetaTrader5 as mt5

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_MODULE = [("wave", "Welle", 3.0), ("m30", "M30", 1.5), ("h1", "H1", 1.5),
           ("ki", "KI-Copilot", 1.0), ("elliott", "Elliott", 1.0),
           ("squeeze", "Squeeze", 2.0), ("pattern", "Muster", 0.25),
           ("orderbook", "Orderbuch", 0.5), ("liqtrend", "Liq-Trend", 0.5)]


def _berlin_ep(ts):
    """Bar-Zeitstempel → lokale Epoch (DST-korrekt, s. CLAUDE.md 05.08.)."""
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN).timestamp())


def main():
    horizont = int(sys.argv[1]) if len(sys.argv) > 1 else 60      # Minuten

    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, 20000)
    mt5.shutdown()
    EP = [_berlin_ep(b["time"]) for b in bars]
    C = [float(b["close"]) for b in bars]
    H = [float(b["high"]) for b in bars]; L = [float(b["low"]) for b in bars]
    tr = [0.0] + [max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1]))
                  for i in range(1, len(C))]
    A = [None] + [sum(tr[max(1, i - 13):i + 1]) / len(tr[max(1, i - 13):i + 1])
                  for i in range(1, len(C))]
    schritt = max(1, horizont // 5)

    def fwd(ep):
        """Forward-Return in ×ATR, oder None wenn nicht sauber abgedeckt."""
        j = bisect.bisect_left(EP, ep)
        if j <= 20 or j + schritt >= len(C):
            return None
        if abs(EP[j] - ep) > 900:            # >15 min Lücke → keine Deckung
            return None
        a = A[j]
        if not a or a <= 0:
            return None
        return (C[j + schritt] - C[j]) / a

    db = sqlite3.connect("oil_widget_history.db")
    db.row_factory = sqlite3.Row
    rows = db.execute("SELECT * FROM verdict_votes ORDER BY ts").fetchall()
    db.close()
    if len(rows) < 500:
        print("Zu wenige Zeilen."); return

    # Drift-Null: Mittelwert ALLER abgedeckten Bars im Zeitraum
    alle = [x for x in (fwd(r["ts"]) for r in rows) if x is not None]
    drift = sum(alle) / len(alle) if alle else 0.0
    t0, t1 = rows[0]["ts"], rows[-1]["ts"]
    mitte = (t0 + t1) / 2

    print("=" * 92)
    print(f"  VERDICT-MODULE — sagen sie etwas voraus?  Horizont {horizont} min")
    print(f"  {datetime.fromtimestamp(t0, _BERLIN):%d.%m.%Y} bis "
          f"{datetime.fromtimestamp(t1, _BERLIN):%d.%m.%Y} · {len(rows)} Zeilen · "
          f"Drift-Null {drift:+.4f}×ATR")
    print("  ⚠ ~12 Tage = EIN Regime. Sichtung, kein Urteil (Reminder-Schwelle: 25.000).")
    print("=" * 92)
    print(f"  {'Modul':<13}{'Gew':>5}{'Episoden':>9}{'stumm':>7}"
          f"{'Übersch. H1':>13}{'Übersch. H2':>13}{'Treffer':>9}  Konsistenz")
    print("  " + "-" * 90)

    for feld, name, gew in _MODULE:
        # Episoden bilden: aufeinanderfolgende GLEICHE Stimme = eine Beobachtung
        # ⚠ Stimme UND Gewicht prüfen. `verdict_votes` speichert beides getrennt
        # (`elliott` / `elliott_w`); eine Stimme mit Gewicht 0 zählt im Verdict NICHT.
        # Erster Entwurf wertete nur die Stimme — Elliott sah damit aus wie ein Modul,
        # das permanent mitredet (stumm 1 %), obwohl es real fast nie Gewicht hat.
        wfeld = feld + "_w"
        hat_w = wfeld in rows[0].keys()
        eps, letzte, stumm = [], None, 0
        for r in rows:
            v = r[feld]
            v = 0 if v is None else int(v)
            if hat_w and not (r[wfeld] or 0) > 0:
                v = 0                      # Stimme ohne Gewicht = keine Stimme
            if v == 0:
                stumm += 1
                letzte = 0
                continue
            if v != letzte:
                eps.append((r["ts"], v))
            letzte = v
        werte = [(ts, v, fwd(ts)) for ts, v in eps]
        werte = [(ts, v, f) for ts, v, f in werte if f is not None]
        if len(werte) < 8:
            print(f"  {name:<13}{gew:>5.2f}{len(werte):>9}"
                  f"{100.0*stumm/len(rows):>6.0f}%{'—':>13}{'—':>13}{'—':>9}  "
                  f"zu wenige Episoden")
            continue
        h1 = [(v * f - v * drift) for ts, v, f in werte if ts < mitte]
        h2 = [(v * f - v * drift) for ts, v, f in werte if ts >= mitte]
        alle_u = [(v * f - v * drift) for ts, v, f in werte]
        tr_q = 100.0 * sum(1 for x in alle_u if x > 0) / len(alle_u)
        m1 = sum(h1) / len(h1) if h1 else float("nan")
        m2 = sum(h2) / len(h2) if h2 else float("nan")
        kons = ("beide +" if (m1 > 0 and m2 > 0) else
                "beide −" if (m1 < 0 and m2 < 0) else "KIPPT")
        if len(h1) < 5 or len(h2) < 5:
            kons = "n zu klein"
        print(f"  {name:<13}{gew:>5.2f}{len(werte):>9}{100.0*stumm/len(rows):>6.0f}%"
              f"{m1:>+13.4f}{m2:>+13.4f}{tr_q:>8.0f}%  {kons}")

    # ── Wieviel bewegt jedes Modul die Nadel WIRKLICH? ────────────────────
    # Gewicht allein sagt wenig: ein Modul mit Gewicht 2,0, das zu 81 % schweigt,
    # bewegt weniger als eines mit 1,0, das immer mitredet. Gemessen wird deshalb
    # der ANTEIL am Gesamtgewicht, gemittelt ueber alle Zeilen.
    print()
    print("=" * 92)
    print("  REALER EINFLUSS auf die Bias-Nadel (Anteil am Gesamtgewicht, Ø ueber alle Zeilen)")
    print("=" * 92)
    summe = {n: 0.0 for _, n, _ in _MODULE}
    ges = 0.0
    for r in rows:
        w_row = {}
        for feld, name, gew in _MODULE:
            wfeld = feld + "_w"
            v = r[feld]
            v = 0 if v is None else int(v)
            w = (r[wfeld] or 0.0) if wfeld in r.keys() else (gew if v != 0 else 0.0)
            w_row[name] = w if v != 0 else 0.0
        s = sum(w_row.values())
        if s <= 0:
            continue
        ges += 1
        for name, w in w_row.items():
            summe[name] += w / s
    for _, name, gew in _MODULE:
        a = 100.0 * summe[name] / max(1, ges)
        bar = "█" * int(round(a / 2))
        print(f"    {name:<13}{a:>6.1f} %  {bar}")
    print(f"    (n={int(ges)} Zeilen mit mindestens einer Stimme)")

    print()
    print("  LESART: Ueberschuss = Forward-Return in Stimmen-Richtung MINUS der")
    print("  allgemeinen Drift. >0 heisst: das Modul trifft besser als ein Muenzwurf,")
    print("  der einfach dem Markt folgt. 'stumm' = Anteil der Zeilen ohne Stimme")
    print("  (Gewicht 0) - ein Modul, das fast immer schweigt, bewegt wenig.")
    print()


if __name__ == "__main__":
    main()