AH-Oil-Trader/backtest_pbreak_regime.py

#!/usr/bin/env python3
"""P(break): liegt der Live-Fehler am NIVEAU (Basisrate) oder an der RANGFOLGE?

ANLASS (2026-08-07). Die B4-Prüfung des am 31.07. nachtrainierten Modells zeigt
zwei getrennte Dinge:
  · KALIBRIERUNG messbar daneben — Training-Basisrate 37,8 %, live 53,2 %,
    stabil über vier volle Tage. Alle unteren Klassen „zu niedrig".
  · TRENNSCHÄRFE nicht entscheidbar — entkoppelt n=87, AUC 0,556,
    95-%-KI [0,434 … 0,674] (enthält Zufall UND Messlatte).
Die Vermutung lautet deshalb: das Modell RANKT brauchbar, aber sein NIVEAU ist an
eine Basisrate genagelt, die wandert. Ein drittes Fitten derselben Bauart würde
daran nichts ändern — es würde nur einen neuen festen Achsenabschnitt lernen.

⚠ VORFRAGE, die über alles entscheidet: schwankt die Basisrate im Backtest-Zeitraum
überhaupt? Ist sie dort konstant, lässt sich das Problem an diesen Daten NICHT
studieren, und jede Variante wäre in-sample-Kosmetik. Wird zuerst geprüft und bei
Bedarf ausdrücklich gesagt.

GEPRÜFT WERDEN DREI VARIANTEN, alle auf derselben live-spiegelnden Stichprobe:
  A) BASIS      — 4 Merkmale, fester Achsenabschnitt (der heutige Stand)
  B) +REGIME    — 5. Merkmal: ATR relativ zum eigenen gleitenden Median
                  ⚠ RELATIV, nicht absolut. Ein absoluter ATR-Schwellwert wäre
                  ein Selektions-Artefakt (Lehre aus `analyze_tageszeit.py` und
                  dem ORB-Fehlschlag) und würde über Regime hinweg nicht gelten.
  C) ROLL-ACHSE — Merkmale wie A, aber der Achsenabschnitt wird laufend an die
                  zuletzt BEOBACHTETE Basisrate angepasst.
                  ⚠ KAUSAL: ein Ausgang steht erst `_BRK_W` Bars später fest,
                  deshalb Nachlauf. Ohne den wäre es Look-ahead — genau der
                  Fehler, der am 06.08. Momentum und SIG scheinbar rettete.

⚠⚠ ENTSCHEIDUNGSREGEL, VORAB FESTGELEGT (sonst verschiebt sich hinterher die Latte):
Eine Variante gilt nur dann als besser, wenn sie in BEIDEN Fit-Richtungen
(H1→H2 UND H2→H1)
    (1) den Kalibrierungsfehler senkt (mittlerer |vorhergesagt − real| über die
        Klassen, gewichtet nach n), UND
    (2) die AUC NICHT verschlechtert (Toleranz 0,005), UND
    (3) bei den Nachbar-Parametern nicht kippt (Fenster/Nachlauf).
Sonst: kein Einbau. Zwei Richtungen, weil eine einzelne Richtung schon einmal in
die Irre führte (`backtest_metalabel.py`, 05.08.).
"""
import sys

import numpy as np
import MetaTrader5 as mt5

from backtest_pbreak_retrain import (_ATRMIN, _BRK_W, _N_BARS, _atr_series,
                                     _ema_series, auc, build_levels, collect,
                                     fit_logreg, old_model)
from core.wave_rec import _EMA_FAST, _EMA_SLOW

_REG_FENSTER = 288      # gleitender ATR-Median (~1 Tag auf M5) für das Regime
_ROLL_N = 400           # wie viele zurückliegende Ausgänge die Achse nachführt
_ROLL_LAG = _BRK_W      # Nachlauf: so lange braucht ein Ausgang, bis er feststeht


def live_model(feat):
    """Das HEUTE laufende Modell — Gewichte direkt aus `core/engine.py`, damit die
    Kontrolle nicht von einer Kopie abhängt (Deployment-Drift wäre sonst genau
    hier möglich: gemessen mit anderen Zahlen als betrieben)."""
    from core.engine import _PB_MU, _PB_SD, _PB_W
    z = _PB_W[0]
    for k in range(4):
        z += _PB_W[k + 1] * (feat[k] - _PB_MU[k]) / _PB_SD[k]
    return 1 / (1 + np.exp(-z))


def kalib_fehler(p, y, kanten=(0.0, 0.2, 0.35, 0.5, 0.65, 1.01)):
    """Mittlerer |vorhergesagt − real| über die Klassen, nach n gewichtet.
    Das ist die Größe, die live danebenlag — nicht die AUC."""
    ges = 0.0
    n_ges = 0
    for a, b in zip(kanten, kanten[1:]):
        m = (p >= a) & (p < b)
        k = int(m.sum())
        if k < 20:
            continue
        ges += k * abs(p[m].mean() - y[m].mean())
        n_ges += k
    return ges / n_ges if n_ges else float("nan")


def zeig(titel, p, y):
    print(f"    {titel:<34} AUC {auc(p, y):.3f}   Ø-Vorhersage {100*p.mean():>5.1f} % "
          f"vs. real {100*y.mean():>5.1f} %   Kalibr.-Fehler {100*kalib_fehler(p, y):>5.2f} Pp")


def regime_reihe(AT):
    """ATR relativ zum eigenen gleitenden Median — regime-normiert, damit der Wert
    über verschiedene Vola-Niveaus dasselbe bedeutet. Kausal (nur Vergangenheit)."""
    out = [1.0] * len(AT)
    puffer = []
    for i, a in enumerate(AT):
        if a:
            puffer.append(a)
            if len(puffer) > _REG_FENSTER:
                puffer.pop(0)
            med = float(np.median(puffer))
            out[i] = (a / med) if med > 0 else 1.0
    return out


def roll_achse(p_roh, y, n_fenster=_ROLL_N, lag=_ROLL_LAG):
    """Verschiebt den Achsenabschnitt im Logit-Raum so, dass die mittlere
    Vorhersage der zuletzt BEOBACHTETEN Basisrate folgt.

    ⚠ Kausal: für Zeile i zählen nur Ausgänge bis i-lag. Vor dem ersten vollen
    Fenster bleibt die Vorhersage unverändert (kein Raten aus der Zukunft)."""
    p = np.clip(p_roh, 1e-6, 1 - 1e-6)
    logit = np.log(p / (1 - p))
    out = np.array(p_roh, dtype=float)
    for i in range(len(p)):
        a = i - lag - n_fenster
        b = i - lag
        if a < 0 or b <= a:
            continue
        beob = y[a:b].mean()
        erw = p_roh[a:b].mean()
        if not (0 < beob < 1) or not (0 < erw < 1):
            continue
        # Differenz der Logits der MITTELWERTE = Niveau-Korrektur
        delta = np.log(beob / (1 - beob)) - np.log(erw / (1 - erw))
        out[i] = 1 / (1 + np.exp(-(logit[i] + delta)))
    return out


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    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, n)
    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]
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
    AT = _atr_series(H, L, C); LV = build_levels(H, L)
    REG = regime_reihe(AT)
    mid = len(C) // 2

    print("=" * 104)
    print(f"  P(break) — NIVEAU oder RANGFOLGE?   {sym} M5, {len(C)} Bars")
    print("=" * 104)

    s1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, LV, REG)
    s2 = collect(mid, len(C), H, L, C, EF, ES, AT, LV, REG)
    if len(s1) < 500 or len(s2) < 500:
        print("zu wenige Stichproben"); return

    y1 = np.array([e["y"] for e in s1], float)
    y2 = np.array([e["y"] for e in s2], float)
    X1 = np.array([e["feat"] for e in s1]); X2 = np.array([e["feat"] for e in s2])

    # ── VORFRAGE: wandert die Basisrate in diesen Daten ueberhaupt? ────────
    print(f"\n  H1 n={len(s1)}  Basisrate {100*y1.mean():.1f} %")
    print(f"  H2 n={len(s2)}  Basisrate {100*y2.mean():.1f} %")
    print(f"  Live gemessen (ab 01.08.): 53,2 %   ·   ALT-Training: 37,8 %")
    print("\n  Basisrate in Zehnteln des Zeitraums (wandert sie?):")
    alle_y = np.concatenate([y1, y2])
    blk = len(alle_y) // 10
    raten = [100 * alle_y[i*blk:(i+1)*blk].mean() for i in range(10)]
    print("    " + " ".join(f"{r:5.1f}" for r in raten))
    print(f"    Spanne {min(raten):.1f} … {max(raten):.1f} %  "
          f"(Schwankung {max(raten)-min(raten):.1f} Pp)")
    if max(raten) - min(raten) < 8:
        print("    ⚠ Die Basisrate ist in diesen Daten weitgehend STABIL — der live")
        print("      beobachtete Drift (37,8 → 53,2 %) lässt sich hier nur begrenzt")
        print("      nachstellen. Die Ergebnisse unten sind entsprechend zu lesen.")

    # ── die drei Varianten, in BEIDEN Fit-Richtungen ───────────────────────
    print("\n" + "=" * 104)
    print("  VARIANTEN (vorab fixierte Regel: beide Richtungen besser kalibriert,")
    print("             AUC nicht schlechter, Nachbarn kippen nicht)")
    print("=" * 104)

    for name, (Xa, ya, Xb, yb, sa, sb) in (
            ("FIT H1 → TEST H2", (X1, y1, X2, y2, s1, s2)),
            ("FIT H2 → TEST H1", (X2, y2, X1, y1, s2, s1))):
        print(f"\n  {name}")
        w = fit_logreg(Xa, ya)
        pb = 1 / (1 + np.exp(-(np.hstack([np.ones((len(Xb), 1)), Xb]) @ w)))
        zeig("A) BASIS (fester Achsenabschnitt)", pb, yb)

        # B) +Regime als 5. Merkmal
        ra = np.array([[e["reg"]] for e in sa]); rb = np.array([[e["reg"]] for e in sb])
        w5 = fit_logreg(np.hstack([Xa, ra]), ya)
        p5 = 1 / (1 + np.exp(-(np.hstack([np.ones((len(Xb), 1)), Xb, rb]) @ w5)))
        zeig("B) + Regime (ATR/Median)", p5, yb)

        # C) rollierender Achsenabschnitt auf A
        pc = roll_achse(pb, yb)
        zeig(f"C) ROLL-ACHSE (n={_ROLL_N}, Nachlauf {_ROLL_LAG})", pc, yb)

        # Nachbarn von C
        for nf in (200, 800):
            zeig(f"   C-Nachbar n={nf}", roll_achse(pb, yb, n_fenster=nf), yb)

        # ── ZWEI Kontrollen, und sie sind NICHT dasselbe ──────────────────
        # ⚠ Beim ersten Lauf hatte ich `old_model` als „Modell wie live"
        # beschriftet — falsch. `old_model` sind die Gewichte VOR dem
        # Nachtrainieren (31.07.); live laufen seitdem `_PB_*` aus `engine.py`.
        p_old = np.array([old_model(e["feat"]) for e in sb])
        zeig("KONTROLLE 1: Modell VOR dem 31.07.", p_old, yb)
        p_live = np.array([live_model(e["feat"]) for e in sb])
        zeig("KONTROLLE 2: Modell wie HEUTE live", p_live, yb)


if __name__ == "__main__":
    main()