AH-Oil-Trader/backtest_pbreak_retrain.py

#!/usr/bin/env python3
"""P(break) NEU TRAINIEREN auf einer LIVE-SPIEGELNDEN Stichprobe (Weg b, 2026-07-31).

WARUM: `analyze_pbreak_live.py` + `backtest_pbreak_touches.py` haben belegt, dass das
Modell nicht verfallen ist, sondern live eine ANDERE Merkmalsverteilung bekommt:

    Merkmal   Live-Mittel   Training-µ   Differenz
    mom3          0,149       1,4523      −1,303
    mom6          0,181       1,6565      −1,476

Ursache: das ALTE Training fixiert das Level beim Entry (≥0,3×ATR entfernt) und wartet
auf den ersten Touch nach einem echten ANLAUF → mom3 ≈ 1,45×ATR (ein Schub). **Live
wählt `_draw_levels` das nächstgelegene Level jede Sekunde NEU** (mit Hysterese) — der
Kurs steht dann oft neben einem Level, das gerade erst zum nächsten wurde, ohne je
darauf zugelaufen zu sein → mom3 ≈ 0,15. Das Modell antwortet konsequent „kein Schub →
kein Durchbruch → 23 %", liegt damit immer unter der Schwelle 0,55, und das Gate war
seit Inbetriebnahme faktisch nie aktiv (91 % aller Berührungen geschlossen).

WAS DIESES SKRIPT ANDERS MACHT — die Stichprobe spiegelt den Live-Pfad:
  * Level-Auswahl **dynamisch mit Hysterese**, wie `engine._draw_levels`: die Linie
    bleibt stehen, bis der Kurs sie WIRKLICH durchbricht (±0,1×ATR); nur wenn sie
    >1,5×ATR entfernt ist UND ein deutlich näheres Level existiert, wird gewechselt.
  * Gesampelt wird an **JEDEM Bar** im 0,15×ATR-Band um dieses Level (= live
    `at_level`) — KEINE Anlauf-Bedingung.
  * Ausgang identisch zu live/Training: erreicht der Kurs binnen 12 M5-Bars zuerst
    `level ± 0,5×ATR` in Bruchrichtung (break) oder in Gegenrichtung (bounce);
    Timeout zählt als bounce.
  * Merkmale wie live: mom6/mom3 (M5, ×Richtung), wt (EMA-Richtung), dist
    (|level − entry|/ATR), plus optional die Touch-Zahl (User-Idee, im alten
    Backtest +0,010 AUC).

KONTROLLE (entscheidend): das ALTE Modell wird auf DERSELBEN live-spiegelnden
Stichprobe ausgewertet. Erwartung ~AUC 0,5 (= der live beobachtete Wert). Nur wenn
das neue Modell dort deutlich besser ist, ist der Umbau belegt — sonst hätte man
lediglich zwei Stichproben verglichen.

Urteil: neues Modell muss out-of-sample (Fit H1 → H2) sowohl TRENNEN (AUC) als auch
KALIBRIERT sein. Beides, sonst kein Einbau.
"""
import sys
from collections import deque

import numpy as np
import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD,
                           _REVERSAL_STRETCH, _STRETCH_MAX)

_MAXH = 200; _ATRMIN = 0.12
_PIV_K = 3; _BRK_W = 12; _BRK_ATR = 0.5
_M30 = 6; _PIV_LOOKBACK = 50
_ZONE = 0.15          # „am Level" (wie live `at_level`)
_CLUSTER = 0.5        # Level-Clusterung in ×ATR (wie `_draw_levels`)
_HYST = 0.1           # echter Durchbruch erst ab ±X×ATR (Hysterese)
_FAR = 1.5            # alte Linie gilt als „irrelevant weit" ab X×ATR
_TWIN = 6             # Touch-Fenster in M5-Bars (30 min)

# Altes Live-Modell (M30-Fit) — dient als KONTROLLE auf derselben Stichprobe
_OLD_MU = (1.6565, 1.4523, 0.9465, 1.9133)
_OLD_SD = (2.5291, 2.4177, 0.2250, 2.1257)
_OLD_W = (-0.3264, 0.4719, 1.4330, 0.0943, -0.1885)


def _ema_series(v, p):
    k = 2.0 / (p + 1); o = []; e = v[0]
    for i, x in enumerate(v):
        e = x if i == 0 else x * k + e * (1 - k)
        o.append(e)
    return o


def _atr_series(H, L, C, p=14):
    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])))
    out = []
    for i in range(len(C)):
        if not i:
            out.append(None); continue
        w = t[max(1, i - p + 1):i + 1]
        out.append(sum(w) / max(1, len(w)))
    return out


def build_levels(H, L):
    """M30-Pivots k=3, kausal, geclustert — wie `wave.pb_levels30` + `_draw_levels`."""
    nm = len(H) // _M30
    mh = [max(H[i * _M30:(i + 1) * _M30]) for i in range(nm)]
    ml = [min(L[i * _M30:(i + 1) * _M30]) for i in range(nm)]
    born = [[] for _ in range(nm)]
    for p in range(_PIV_K, nm - _PIV_K):
        if mh[p] == max(mh[p - _PIV_K:p + _PIV_K + 1]) and p + _PIV_K < nm:
            born[p + _PIV_K].append(mh[p])
        if ml[p] == min(ml[p - _PIV_K:p + _PIV_K + 1]) and p + _PIV_K < nm:
            born[p + _PIV_K].append(ml[p])
    lv = []; win = deque(); cur = []
    for m in range(nm):
        win.append(born[m]); cur.extend(born[m])
        while len(win) > _PIV_LOOKBACK:
            for x in win.popleft():
                try: cur.remove(x)
                except ValueError: pass
        lv.append(sorted(cur))
    return lv


def _cluster(levels, atr):
    """Nahe Pivots zu einer Linie zusammenfassen (~0,5×ATR) — wie `_draw_levels`."""
    if not levels:
        return []
    out = []; grp = [levels[0]]
    for x in levels[1:]:
        if x - grp[-1] <= _CLUSTER * atr:
            grp.append(x)
        else:
            out.append(sum(grp) / len(grp)); grp = [x]
    out.append(sum(grp) / len(grp))
    return out


def pick_level(cands, px, d, atr, prev):
    """Ziel-Level in Trade-Richtung MIT Hysterese (Nachbau von `_draw_levels`).

    d>0 → nächster Widerstand ÜBER px · d<0 → nächste Unterstützung UNTER px.
    `prev` = zuletzt gezeichnete Linie; sie bleibt stehen, bis der Kurs sie wirklich
    durchbricht (±_HYST×ATR) — außer sie ist >_FAR×ATR weg und es gibt eine deutlich
    nähere.
    """
    side = [x for x in cands if (x > px if d > 0 else x < px)]
    fresh = (min(side) if d > 0 else max(side)) if side else None
    if prev is not None:
        broken = (px > prev + _HYST * atr) if d > 0 else (px < prev - _HYST * atr)
        if not broken:
            far = abs(prev - px) > _FAR * atr
            if not (far and fresh is not None and abs(fresh - px) < abs(prev - px) - 0.5 * atr):
                return prev
    return fresh


def outcome(H, L, level, atr, j, d):
    """break/bounce wie live: level ± 0,5×ATR, Fenster 12 Bars, Timeout = bounce."""
    up = level + d * _BRK_ATR * atr
    dn = level - d * _BRK_ATR * atr
    for k in range(j, min(j + _BRK_W, len(H))):
        if (H[k] >= up) if d > 0 else (L[k] <= up):
            return 1
        if (L[k] <= dn) if d > 0 else (H[k] >= dn):
            return 0
    return 0


def collect(a, b, H, L, C, EF, ES, AT, LV, REG=None):
    """Live-spiegelnde Stichprobe: jeder Bar im 0,15×ATR-Band um das dyn. Level.

    ⚠ `REG` ist ADDITIV (2026-08-07, für `backtest_pbreak_regime.py`): wird eine
    Regime-Reihe übergeben, bekommt jede Zeile zusätzlich `"reg"`. OHNE das
    Argument ist die Ausgabe **bitgenau wie vorher** — die dokumentierten Zahlen
    dieses Skripts dürfen sich nicht verschieben (Migrations-Regel)."""
    TH = _REVERSAL_STRETCH
    out = []
    i = max(a, _N_BARS, 320)
    while i < min(b, len(C) - _MAXH - 1):
        atr = AT[i]
        if not atr or atr <= 0:
            i += 1; continue
        atr = max(atr, _ATRMIN); es = ES[i]; ef = EF[i]
        stretch = (C[i] - es) / atr
        ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR); ad = ang - 90.0
        d = 0
        if stretch <= -TH and ad >= _ANGLE_DEAD: d = 1
        elif stretch >= TH and ad <= -_ANGLE_DEAD: d = -1
        elif abs(stretch) < _STRETCH_MAX: d = 1 if ef > es else -1 if ef < es else 0
        if not d:
            i += 1; continue
        entry = C[i]
        # ── Position „läuft" bis _MAXH Bars; dabei jeden Bar prüfen (wie `_pos_loop`)
        prev_lvl = None; hist = deque()
        for j in range(i + 1, min(i + _MAXH, len(C) - _BRK_W)):
            aj = AT[j]
            if not aj:
                continue
            aj = max(aj, _ATRMIN)
            lv = LV[min(j // _M30, len(LV) - 1)]
            cl = _cluster(lv, aj)
            level = pick_level(cl, C[j], d, aj, prev_lvl)
            prev_lvl = level
            if level is None:
                continue
            at_level = abs(C[j] - level) <= _ZONE * aj
            hist.append((j, at_level, level))
            while hist and hist[0][0] < j - _TWIN:
                hist.popleft()
            if not at_level:
                continue
            # Touch-Zahl: Eintritte in die Zone desselben Levels im Fenster
            tc = 0; prev_in = False
            for _jj, inz, lvl2 in hist:
                inside = inz and abs(lvl2 - level) < 1e-9
                if inside and not prev_in:
                    tc += 1
                prev_in = inside
            tc = max(1, tc)
            y = outcome(H, L, level, aj, j, d)
            feat = [(C[j] - C[max(0, j - 6)]) * d / aj,
                    (C[j] - C[max(0, j - 3)]) * d / aj,
                    1.0 if (EF[j] - ES[j]) * d > 0 else 0.0,
                    abs(level - entry) / aj]
            zeile = {"feat": feat, "y": y, "tc": tc,
                     "feat5": feat + [1.0 if 2 <= tc <= 4 else 0.0,
                                      1.0 if tc >= 5 else 0.0]}
            if REG is not None:
                zeile["reg"] = REG[min(j, len(REG) - 1)]
            out.append(zeile)
        i += 12          # nächste „Position" versetzt starten (weniger Überlappung)
    return out


def fit_logreg(X, y, iters=3000, lr=0.3):
    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(ps, ys):
    ps = np.asarray(ps); ys = np.asarray(ys)
    order = np.argsort(ps); ys = ys[order]
    pos = ys.sum(); neg = len(ys) - pos
    if not pos or not neg:
        return 0.5
    ranks = np.arange(1, len(ys) + 1)
    return float((ranks[ys == 1].sum() - pos * (pos + 1) / 2) / (pos * neg))


def calib(ps, ys, label):
    print(f"    Kalibrierung {label}: {'Bucket':<12}{'n':>7}{'vorherg.':>10}{'real':>8}")
    for lo, hi in ((0, .2), (.2, .35), (.35, .5), (.5, .65), (.65, 1.01)):
        sel = [(p, y) for p, y in zip(ps, ys) if lo <= p < hi]
        if len(sel) < 40:
            continue
        print(f"                         {lo:.2f}–{hi:<7.2f}{len(sel):>7}"
              f"{100*np.mean([p for p, _ in sel]):>9.1f}%"
              f"{100*np.mean([y for _, y in sel]):>7.1f}%")


def old_model(feat):
    z = _OLD_W[0]
    for k in range(4):
        z += _OLD_W[k + 1] * (feat[k] - _OLD_MU[k]) / _OLD_SD[k]
    return 1 / (1 + np.exp(-max(-30, min(30, z))))


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    mt5.initialize(); sym = None
    for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
        if mt5.symbol_info(c): sym = c; break
    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)
    mid = len(C) // 2
    s1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, LV)
    s2 = collect(mid, len(C), H, L, C, EF, ES, AT, LV)

    print("=" * 100)
    print(f"  P(break) NEU TRAINIERT auf live-spiegelnder Stichprobe — {sym} M5 ({len(C)} Bars)")
    print(f"  Level dynamisch mit Hysterese · gesampelt an JEDEM Bar im {_ZONE}×ATR-Band")
    print(f"  H1 n={len(s1)}   H2 n={len(s2)}")
    print("=" * 100)
    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)
    print(f"\n  Basisrate Durchbruch:  H1 {100*y1.mean():.1f} %   H2 {100*y2.mean():.1f} %")

    # Merkmalsverteilung — spiegelt sie wirklich die Live-Werte?
    F1 = np.array([e["feat"] for e in s1])
    print(f"\n  Merkmalsverteilung (Kontrolle gegen die Live-Rekonstruktion):")
    print(f"    {'Merkmal':<8}{'hier µ':>10}{'ALT-Training µ':>16}{'Live gemessen':>15}")
    for k, nm, old, live in ((0, "mom6", 1.6565, 0.181), (1, "mom3", 1.4523, 0.149)):
        print(f"    {nm:<8}{F1[:, k].mean():>10.3f}{old:>16.3f}{live:>15.3f}")

    # ── KONTROLLE: altes Modell auf DIESER Stichprobe
    p_old2 = np.array([old_model(e["feat"]) for e in s2])
    print(f"\n  KONTROLLE — ALTES Modell auf dieser Stichprobe (H2):")
    print(f"    AUC {auc(p_old2, y2):.3f}   Ø-Vorhersage {100*p_old2.mean():.1f} % "
          f"vs. real {100*y2.mean():.1f} %")
    print(f"    (live beobachtet: AUC 0,539 · 23 % vs 41 % — passt es zusammen?)")

    # ── NEUES Modell, 4 und 6 Merkmale
    for tag, key in (("NEU 4 Merkmale", "feat"), ("NEU 6 Merkmale (+Touch-Zahl)", "feat5")):
        X1 = np.array([e[key] for e in s1]); X2 = np.array([e[key] for e in s2])
        mu = X1.mean(0); sd = X1.std(0) + 1e-9
        w = fit_logreg((X1 - mu) / sd, y1)
        p2 = 1 / (1 + np.exp(-(np.hstack([np.ones((len(X2), 1)), (X2 - mu) / sd]) @ w)))
        p1 = 1 / (1 + np.exp(-(np.hstack([np.ones((len(X1), 1)), (X1 - mu) / sd]) @ w)))
        print(f"\n  {tag}")
        print(f"    AUC  in-sample H1 {auc(p1, y1):.3f}   out-of-sample H2 {auc(p2, y2):.3f}")
        calib(p2, y2, "H2")
        if key == "feat":
            print(f"\n    → zum Einbetten (falls dieses Modell gewinnt):")
            print(f"    _PB_MU = ({', '.join(f'{x:.4f}' for x in mu)})")
            print(f"    _PB_SD = ({', '.join(f'{x:.4f}' for x in sd)})")
            print(f"    _PB_W  = ({', '.join(f'{x:.4f}' for x in w)})")


if __name__ == "__main__":
    main()