#!/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 2026-08-04 w.set_dead_hours("") 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()