AH-Oil-Trader/analyze_verdict_calibration.py

#!/usr/bin/env python3
"""Kalibrierung der Gesamtempfehlung (2026-07-24, Idee 2 „Verdict ans eigene
Erfolgsrezept halten"): Ist die angezeigte Konfidenz (conf_pct) PRÄDIKTIV — sagt
conf=80 mehr Erfolg voraus als conf=55? Und: sind die unbelegten Verdict-Stimmen
(KI-Copilot ai_sentiment, news_score) prädiktiv?

Methodik wie beim P(break)-Modell (dem kalibrierten Erfolgsfall):
  - `recommendations` (LONG/SHORT) aus der DB, Sampling ≥15 min Abstand
    (Autokorrelation dämpfen — die Empfehlung ändert sich minütlich kaum).
  - Forward-Return = Signalrichtung × (Close[t+H] − Close[t0]) / ATR(M5, t0)
    für Horizonte 1 h und 2 h (M5-Bars von MT5; rec.timestamp = lokale Epoch,
    Bar-Zeit = Broker-Zeit → −3 h).
  - Bins über conf_pct → n / WR(fwd>0) / Ø-fwdR je Bin, in ZWEI Zeitraum-Hälften
    (Sign-Stabilität wie immer).
  - Copilot: ai_sentiment (LONG/bullish=+1, SHORT/bearish=−1) → Ø-fwdR in
    Copilot-Richtung (unabhängig vom Wave-Signal!). News: Vorzeichen news_score.
Kalibriert = monoton steigende Ø-fwdR über die conf-Bins, in BEIDEN Hälften.
"""
import sqlite3, datetime as dt
import MetaTrader5 as mt5

_BROKER_OFF = 3*3600
_SAMPLE_S   = 15*60          # Mindestabstand zwischen zwei gewerteten Empfehlungen
_H1, _H2    = 12, 24         # Forward-Horizonte in M5-Bars (1 h / 2 h)
_ATR_P      = 14
_BINS       = [(0,45),(45,55),(55,65),(65,75),(75,85),(85,101)]


def _atr_series(H, L, C, p=_ATR_P):
    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])))
    return [(sum(t[max(1,i-p+1):i+1])/max(1,len(t[max(1,i-p+1):i+1]))) if i else None for i in range(len(C))]


def _dirnum(s):
    if not s: return 0
    s = str(s).strip().upper()
    if s in ("LONG","BULLISH"): return 1
    if s in ("SHORT","BEARISH"): return -1
    return 0


def line(lbl, rows, key):
    v=[r[key] for r in rows]
    if not v: return f"    {lbl:<22} —"
    n=len(v); wr=100*sum(1 for x in v if x>0)/n
    return f"    {lbl:<22} n={n:>5}  WR={wr:>3.0f}%  ØR={sum(v)/n:+.3f}"


def main():
    mt5.initialize()
    sym = next((c for c in ("SpotCrude","USOIL","WTI","XTIUSD") if mt5.symbol_info(c)), None)
    bars = None
    for req in (100000, 80000, 60000, 40000):
        bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, req)
        if bars is not None and len(bars) > 2000: break
    mt5.shutdown()
    H=[float(b["high"]) for b in bars]; L=[float(b["low"]) for b in bars]
    C=[float(b["close"]) for b in bars]
    T=[int(b["time"])-_BROKER_OFF for b in bars]     # → echte UTC-Epoch
    A=_atr_series(H,L,C)
    t2i = {t:i for i,t in enumerate(T)}

    con = sqlite3.connect("oil_widget_history.db"); con.row_factory=sqlite3.Row
    recs = con.execute(
        "SELECT timestamp, signal, conf_pct, ai_sentiment, news_score "
        "FROM recommendations WHERE signal IN ('LONG','SHORT') ORDER BY timestamp"
    ).fetchall()

    ev=[]; last_ts=0
    for r in recs:
        ts=r["timestamp"]
        if ts-last_ts < _SAMPLE_S: continue
        bar_t = (ts//300)*300                     # auf M5-Raster runden
        i = t2i.get(bar_t)
        if i is None or i+_H2 >= len(C) or not A[i]: continue
        last_ts=ts
        d = 1 if r["signal"]=="LONG" else -1
        atr=max(A[i],0.06)
        ev.append(dict(ts=ts, conf=r["conf_pct"] or 0, d=d,
                       ai=_dirnum(r["ai_sentiment"]),
                       news=(r["news_score"] if r["news_score"] is not None else None),
                       f1=(C[i+_H1]-C[i])*d/atr, f2=(C[i+_H2]-C[i])*d/atr))
    if not ev:
        print("Keine matchbaren Empfehlungen (History-Überlappung prüfen)."); return
    mid_ts = ev[len(ev)//2]["ts"]
    halves=[("H1 (alt)",[e for e in ev if e["ts"]<mid_ts]),
            ("H2 (neu)",[e for e in ev if e["ts"]>=mid_ts])]
    def span(rows):
        return (f"{dt.datetime.fromtimestamp(rows[0]['ts']):%d.%m.%y}–"
                f"{dt.datetime.fromtimestamp(rows[-1]['ts']):%d.%m.%y}")
    print("="*84)
    print(f"  Verdict-Kalibrierung — {sym}, {len(ev)} gesampelte Signale (≥15 min Abstand)")
    print(f"  fwd-Return in Signalrichtung, ×ATR(M5) · Horizonte 1 h/2 h")
    print("="*84)

    for lbl, rows in halves:
        print(f"\n{lbl} ({span(rows)}, n={len(rows)}):")
        print("  conf_pct-Bins (Horizont 2 h) — kalibriert = ØR steigt monoton:")
        for lo,hi in _BINS:
            sub=[e for e in rows if lo<=e["conf"]<hi]
            print(line(f"conf {lo}–{hi-1}", sub, "f2"))
        print("  KI-Copilot (eigene Richtung, unabhängig vom Wave-Signal):")
        for tag,dd in (("Copilot LONG",1),("Copilot SHORT",-1)):
            sub=[dict(f2=e["f2"]*e["d"]*dd) for e in rows if e["ai"]==dd]
            print(line(tag, sub, "f2"))
        print("  News-Score (Richtung des Scores, |score|≥0.3):")
        for tag,dd in (("News bullisch",1),("News bärisch",-1)):
            sub=[dict(f2=e["f2"]*e["d"]*dd) for e in rows
                 if e["news"] is not None and (e["news"]>=0.3 if dd>0 else e["news"]<=-0.3)]
            print(line(tag, sub, "f2"))

    print("\n  Lesart: conf kalibriert → höhere Bins klar bessere ØR in BEIDEN Hälften.")
    print("  Copilot/News prädiktiv → 'LONG'-Zeile positiv UND 'SHORT'-Zeile positiv")
    print("  (jeweils fwd in der EIGENEN Richtung gemessen).")


if __name__ == "__main__":
    main()