AH-Oil-Trader/backtest_metalabel.py

#!/usr/bin/env python3
"""META-LABELING auf dem Wellensignal (López de Prado) — 2026-08-05.

IDEE (aus der Web-Recherche): NICHT ein 23. Signal-Filter, sondern ein
**Zweitmodell**, das für ein bereits erzeugtes Signal die Wahrscheinlichkeit
schätzt, dass es GEWINNT — und nur die guten durchlässt. Das Primärmodell
(Richtung) bleibt unangetastet.

WARUM DAS HIER ETWAS ANDERES IST ALS DIE 22 VERWORFENEN EINGRIFFE:
Jene waren einzelne, handgewählte Bedingungen, je einzeln getestet („shorte nicht
gegen den Winkel", „meide Chop", …). Meta-Labeling lernt eine **kalibrierte
Wahrscheinlichkeit aus vielen Merkmalen gleichzeitig** und wird wie das
P(break)-Modell out-of-sample auf AUC UND Kalibrierung geprüft. Genau dieses
Vorgehen hat im Projekt schon einmal funktioniert (P(break), AUC 0,65 oos).

⚠ ERWARTUNGSMANAGEMENT (Hudson & Thames, wörtlich): „if the algorithm is bad then
meta-labeling would likely only reduce the downside." Das Wellensignal ist in H1
gemessen negativ. Realistisch macht ein Zweitmodell daraus einen kleineren
Verlierer oder Breakeven — keinen Gewinner. Genau das wird hier geprüft, nicht
behauptet.

AUFBAU
 * Primärsignal = der volle LIVE-Gate-Stack über die ECHTEN Methoden (`_build`,
   `_confirm_breakout`-Mechanik, `_room_gate`) — identisch zu
   `backtest_breakout_gated.py`, damit die Population die live gehandelte ist.
 * Label = hat der Trade mit dem kanonischen Exit + Echtkosten GEWONNEN (R > 0)?
 * Merkmale: nur zum Signalzeitpunkt bekannte Größen (kausal).
 * Fit auf einer Hälfte → gemessen auf der ANDEREN. **Beide Richtungen**
   (H1→H2 und H2→H1), sonst ist ein Einweg-Zufall nicht auszuschließen.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  Übernehmen nur, wenn eine Schwelle in **BEIDEN** Fit-Richtungen ΣR **und** ØR
  gegenüber „alle Signale nehmen" verbessert, UND die Nachbar-Schwellen mithalten.

Aufruf:  python backtest_metalabel.py [bars]
"""
import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import numpy as np
import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _EMA_FAST, _EMA_SLOW, _N_BARS,
                           _ANGLE_LR, _ANGLE_DEAD, _HTF_DEADBAND)

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_ATRMIN = 0.06
_COOL = 3
_TIMEOUT_BARS = 12
_PIV_K = 3
_PIV_LOOK = 300
_ATR_FLOOR_PB = 0.12
_K = 0.3                      # breakout_k wie live
_SCHWELLEN = (0.30, 0.35, 0.40, 0.45, 0.50, 0.55)

_MERKMALE = ["conf", "stretch", "htf", "h1", "kosten", "vola",
             "mom3", "mom6", "er", "nacht", "raum", "winkel"]


def _berlin(ts):
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN))


def _ema_series(v, p):
    k = 2 / (p + 1); out = [v[0]]
    for x in v[1:]:
        out.append(out[-1] + k * (x - out[-1]))
    return out


def _atr_series(H, L, C, p=14):
    tr = [0.0]
    for i in range(1, len(C)):
        tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    out = [None]
    for i in range(1, len(C)):
        seg = tr[max(1, i - p + 1):i + 1]
        out.append(sum(seg) / len(seg))
    return out


def _htf_sign(C, i, n, atr):
    """M30/H1-Trend aus M5 aggregiert (wie in `backtest_breakout_gated.py`)."""
    if i < n * _EMA_SLOW:
        return 0
    seg = C[i - n * _EMA_SLOW:i + 1:n]
    if len(seg) < _EMA_SLOW + 2:
        return 0
    ef = _ema_series(seg, _EMA_FAST)[-1]; es = _ema_series(seg, _EMA_SLOW)[-1]
    dead = _HTF_DEADBAND * atr * (n ** 0.5)
    return 1 if ef - es > dead else (-1 if es - ef > dead else 0)


def sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1, PH, PL, lo, hi, vol_med):
    """Alle live-gegateten Signale mit Merkmalen UND Ergebnis."""
    def kosten(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    out = []
    pend = None
    i = max(lo, _PIV_LOOK + _EMA_SLOW + 5)
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
        w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
                        "pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
        rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
                          htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
        sig = rec.get("signal")
        if _K > 0 and sig in ("LONG", "SHORT"):
            d = 1 if sig == "LONG" else -1
            if pend is None or pend["dir"] != d:
                pend = {"dir": d, "level": C[i] + d * _K * atr,
                        "invalid": C[i] - d * _K * atr, "confirmed": False, "bar": i}
            if not pend["confirmed"]:
                broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
                against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
                if broke:
                    pend["confirmed"] = True
                else:
                    if against or (i - pend["bar"]) > _TIMEOUT_BARS:
                        pend = {"dir": d, "level": C[i] + d * _K * atr,
                                "invalid": C[i] - d * _K * atr,
                                "confirmed": False, "bar": i}
                    i += 1; continue
        elif sig == "WARTEN":
            pend = None
        if sig not in ("LONG", "SHORT"):
            i += 1; continue
        rec = w._room_gate(rec, C[i])
        if rec.get("signal") not in ("LONG", "SHORT"):
            i += 1; continue

        d = 1 if rec["signal"] == "LONG" else -1
        a = max(atr, _ATRMIN)
        # ── Merkmale, ALLE zum Signalzeitpunkt bekannt ──────────────────────
        seg = [abs(C[j] - C[j - 1]) for j in range(max(1, i - 11), i + 1)]
        er = (abs(C[i] - C[max(0, i - 12)]) / sum(seg)) if sum(seg) > 0 else 0.0
        # Abstand zum nächsten GEGENlevel (dieselbe Quelle wie das Raum-Gate)
        gg = [p for p in (w._pb_levels["ph"] if d > 0 else w._pb_levels["pl"])
              if ((p > C[i]) if d > 0 else (p < C[i]))]
        raum = (min(abs(p - C[i]) for p in gg) / a) if gg else 5.0
        f = [rec.get("conf_pct", 0) / 100.0,
             abs(C[i] - ES[i]) / a,
             float(HTF[i] * d),
             float(H1[i] * d),
             kosten(i, a),
             a / vol_med,
             (C[i] - C[max(0, i - 3)]) * d / a,
             (C[i] - C[max(0, i - 6)]) * d / a,
             er,
             1.0 if _berlin(T[i]).hour in (0, 1, 2, 3, 4, 5, 6, 7) else 0.0,
             min(raum, 5.0),
             AD[i] * d]
        r, xj = simulate(C[i], d, a, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        rn = r - kosten(i, a)
        out.append({"f": f, "R": rn, "win": 1.0 if rn > 0 else 0.0})
        pend = None
        i = xj + _COOL
    return out


def fit_logreg(X, y, iters=4000, lr=0.3):
    """Dieselbe schlichte Gradientenmethode wie beim P(break)-Modell —
    bewusst kein neues Verfahren, damit die Ergebnisse vergleichbar bleiben."""
    n, m = X.shape
    Xb = np.hstack([np.ones((n, 1)), X]); w = np.zeros(m + 1)
    for _ in range(iters):
        p = 1 / (1 + np.exp(-(Xb @ w)))
        w -= lr * (Xb.T @ (p - y)) / n
    return w


def auc(y, p):
    y = np.asarray(y); p = np.asarray(p)
    pos, neg = p[y == 1], p[y == 0]
    if len(pos) == 0 or len(neg) == 0:
        return float("nan")
    r = np.argsort(np.argsort(np.concatenate([pos, neg]))) + 1
    return (r[:len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))


def kenn(rs):
    if not rs:
        return None
    n = len(rs); s = sum(rs)
    g = sum(x for x in rs if x > 0); v = -sum(x for x in rs if x < 0)
    return {"n": n, "wr": 100 * sum(1 for x in rs if x > 0) / n,
            "oer": s / n, "sr": s, "pf": (g / v) if v > 0 else float("inf")}


def bewerte(train, test, lbl):
    Xt = np.array([e["f"] for e in train]); yt = np.array([e["win"] for e in train])
    mu = Xt.mean(0); sd = Xt.std(0) + 1e-9
    w = fit_logreg((Xt - mu) / sd, yt)
    Xs = (np.array([e["f"] for e in test]) - mu) / sd
    p = 1 / (1 + np.exp(-(np.hstack([np.ones((len(Xs), 1)), Xs]) @ w)))
    ys = np.array([e["win"] for e in test]); R = [e["R"] for e in test]
    print(f"\n  {lbl}  (Fit n={len(train)} · Test n={len(test)})")
    print(f"    AUC out-of-sample: {auc(ys, p):.3f}   "
          f"(0,5 = Muenzwurf; P(break) erreicht 0,65)")
    print(f"    Kalibrierung: Ø-Vorhersage {p.mean()*100:.1f} % vs. echte "
          f"Gewinnrate {ys.mean()*100:.1f} %")
    print(f"    {'Bucket':<12}{'n':>6}{'vorhergesagt':>14}{'real':>8}")
    for a_, b_ in ((0, .35), (.35, .45), (.45, .55), (.55, 1.01)):
        m = (p >= a_) & (p < b_)
        if m.sum() >= 20:
            print(f"    {f'{a_:.2f}-{b_:.2f}':<12}{int(m.sum()):>6}"
                  f"{p[m].mean()*100:>13.1f} %{ys[m].mean()*100:>7.1f} %")
    basis = kenn(R)
    print(f"    {'Schwelle':<12}{'n':>6}{'WR':>7}{'OeR':>9}{'PF':>7}{'SigmaR':>9}")
    print(f"    {'alle nehmen':<12}{basis['n']:>6}{basis['wr']:>6.1f}%"
          f"{basis['oer']:>+9.3f}{basis['pf']:>7.2f}{basis['sr']:>+9.1f}")
    erg = {}
    for t in _SCHWELLEN:
        rs = [R[i] for i in range(len(R)) if p[i] >= t]
        k = kenn(rs)
        erg[t] = k
        if k:
            print(f"    {f'p >= {t:.2f}':<12}{k['n']:>6}{k['wr']:>6.1f}%"
                  f"{k['oer']:>+9.3f}{k['pf']:>7.2f}{k['sr']:>+9.1f}"
                  + ("   besser" if (k['sr'] > basis['sr'] and k['oer'] > basis['oer'])
                     else ""))
    return basis, erg, dict(zip(_MERKMALE, w[1:]))


def main():
    n_bars = 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_bars)
    point = mt5.symbol_info(sym).point
    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"]) for b in bars]
    SP = [float(b["spread"]) * point for b in bars]
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
    A = _atr_series(H, L, C)
    AD = [0.0] * len(C)
    for i in range(_N_BARS, len(C)):
        AD[i] = calc_trend_angle(C[i - _N_BARS:i + 1], _ANGLE_LR) - 90.0
        if abs(AD[i]) < _ANGLE_DEAD:
            AD[i] = 0.0
    HTF = [_htf_sign(C, i, 6, A[i] or 0.1) for i in range(len(C))]
    H1S = [_htf_sign(C, i, 12, A[i] or 0.1) for i in range(len(C))]
    PH = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
    PL = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]
    vol_med = sorted(x for x in A if x)[len(A) // 2]

    w = WaveRecommender(lambda: None)
    w.set_reversal_enabled(False)          # Live-Stand seit 04.08.
    # ⚠⚠ OHNE DIESE ZEILE IST DAS ENTRY-RAUM-GATE AUS: `_entry_room_atr`
    # ist im Konstruktor 0.0, und `_room_gate` gibt dann `rec` unveraendert
    # zurueck — der Aufruf sieht aus wie ein Gate, ist aber ein No-op.
    # Genau das ist mir am 05.08. in drei Skripten passiert (live blockt das
    # Gate 35 % aller Empfehlungen, im Backtest 0 %).
    w.set_entry_room(0.6)                  # Live-Wert
    w.set_dead_hours("")                   # live leer
    mid = len(C) // 2

    print("=" * 92)
    print(f"  META-LABELING auf dem Wellensignal — {sym} M5, {len(C)} Bars")
    print(f"  {_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y} · voller Live-Gate-Stack"
          f" · kanonischer Exit · Echtkosten")
    print("=" * 92)
    e1 = sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
                0, mid, vol_med)
    e2 = sammle(w, H, L, C, T, A, SP, EF, ES, AD, HTF, H1S, PH, PL,
                mid, len(C), vol_med)
    print(f"  Signale: H1 {len(e1)} · H2 {len(e2)}")
    if len(e1) < 100 or len(e2) < 100:
        print("  Zu wenige Signale fuer eine Aussage."); return

    b1, g1, w1 = bewerte(e1, e2, "FIT auf H1  ->  TEST auf H2")
    b2, g2, w2 = bewerte(e2, e1, "FIT auf H2  ->  TEST auf H1")

    print("\n" + "=" * 92)
    print("  URTEIL (Regel VOR dem Lauf fixiert: in BEIDEN Fit-Richtungen SigmaR UND")
    print("          OeR besser als „alle nehmen\", plus Nachbar-Schwellen)")
    print("=" * 92)

    def gut(t, g, b):
        k = g.get(t)
        return bool(k) and k["sr"] > b["sr"] and k["oer"] > b["oer"]

    treffer = []
    for idx, t in enumerate(_SCHWELLEN):
        selbst = gut(t, g1, b1) and gut(t, g2, b2)
        nb = [_SCHWELLEN[j] for j in (idx - 1, idx + 1) if 0 <= j < len(_SCHWELLEN)]
        nb_ok = all(gut(m, g1, b1) and gut(m, g2, b2) for m in nb)
        if selbst and nb_ok:
            treffer.append(t)
        print(f"    p >= {t:.2f}   " + ("BESTEHT" if (selbst and nb_ok) else
                                        "beidseitig besser, Nachbarn NICHT" if selbst
                                        else "faellt durch"))
    print()
    if treffer:
        print(f"  => Kandidat(en): {', '.join(f'{t:.2f}' for t in treffer)}")
        print("     Naechster Schritt waere ein Live-Einbau als ANZEIGE (nicht als Gate),")
        print("     analog zum P(break)-Modell — erst kalibriert beobachten, dann schalten.")
    else:
        print("  => KEINE Schwelle besteht. Meta-Labeling traegt auf dieser Population")
        print("     nicht — passt zu Hudson & Thames: ein schwaches Primaersignal laesst")
        print("     sich auch mit einem Zweitmodell nicht retten.")
    print("\n  Merkmalsgewichte (standardisiert, Fit H1 | Fit H2):")
    for m in _MERKMALE:
        print(f"    {m:<10}{w1[m]:>+8.3f} | {w2[m]:>+8.3f}"
              + ("   (Vorzeichen kippt)" if w1[m] * w2[m] < 0 else ""))
    print()


if __name__ == "__main__":
    main()