AH-Oil-Trader/backtest_level_tf.py

#!/usr/bin/env python3
"""backtest_level_tf.py — Auf WELCHER Timeframe sollen die S/R-Level gebildet werden?

User-Beobachtung 2026-07-30: „die S/R-Linien liegen oft sehr nahe beieinander".
Stimmt — `_draw_levels` nutzt rohe **M5**-Pivots (k=3), und M5 erzeugt viele kleine
Zacken. Der User wählte Option 2: Level-TF umstellen UND das P(break)-Modell auf der
neuen TF nachtrainieren/prüfen (statt nur die Chart-Anzeige zu ändern).

⚠ Warum das NICHT nur Kosmetik ist: dieselben Level speisen den **P(break)-Auto-Close**
(`engine._sr_close_hint`). Das Modell (`_PB_W`) ist auf M5-Pivots trainiert (AUC 0,71).
Mit M15-/M30-Leveln wäre es nicht mehr kalibriert → erst messen, dann umstellen.

Gemessen wird je Level-TF (M5 / M15 / M30), Trades bleiben M5-basiert:
  1. **Level-ABSTÄNDE** — wie weit ist das Gegenlevel typisch weg (×ATR)? Beantwortet
     die Ausgangsbeobachtung mit Zahlen.
  2. **P(break)-Modell NEU trainiert** (H1) und **out-of-sample geprüft** (H2):
     AUC + Kalibrierung. Ein Level-Satz taugt nur, wenn Durchbrüche vorhersagbar bleiben.
  3. **R-Ertrag** der Close-Regel (P<Schwelle → am Level schließen) gegen die
     Trailing-Baseline, in BEIDEN Hälften, mit Echtkosten.

Umgestellt wird nur, wenn eine höhere TF in AUC **und** R-Ertrag mindestens
gleichwertig ist — sonst bleibt M5 (dann ggf. nur die Chart-Anzeige ändern).

Aufruf:  python backtest_level_tf.py [n_bars]
"""
from __future__ import annotations
import sys

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
_SL_ATR = 2.0; _TRAILON = 0.3; _MULT = 1.5; _BE = 1.3
_LOCK_START = 3.5; _LOCK_SCALE = 0.6; _LOCK_MIN = 1.2; _TP_INIT = 3.5
_PIV_K = 3; _LOOKBACK = 300
_BRK_W = 12; _BRK_ATR = 0.5          # Bruch-Definition (wie im Live-Modell)
_TFS = (("M5", 1), ("M15", 3), ("M30", 6))   # (Label, M5-Bars je HTF-Bar)


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 = [None] * len(C); run = 0.0
    for i in range(1, len(C)):
        run += t[i]
        if i > p: run -= t[i - p]
        out[i] = run / min(i, p)
    return out


def sim_run(entry, d, atr, H, L, C, j0):
    """Live-Exit (Trailing + Phasen-Ratsche) — identisch zu backtest_srclose_prob.py."""
    sl = entry - d * _SL_ATR * atr; tp = entry + d * _TP_INIT * atr
    hw = entry; rank = 0
    end = min(j0 + _MAXH, len(C) - 1)
    for j in range(j0, end + 1):
        hi, lo = H[j], L[j]
        if (lo <= sl) if d > 0 else (hi >= sl): return (sl - entry) * d / atr
        if (hi >= tp) if d > 0 else (lo <= tp): return (tp - entry) * d / atr
        hw = max(hw, hi) if d > 0 else min(hw, lo)
        profit = (hw - entry) * d
        ph = 0 if profit < _TRAILON * atr else (1 if profit < _LOCK_START * atr else 2)
        if ph < rank: ph = rank
        rank = ph
        if ph == 1:
            cand = hw - d * _MULT * atr
            if profit >= _BE * atr: cand = max(cand, entry) if d > 0 else min(cand, entry)
            sl = max(sl, cand) if d > 0 else min(sl, cand)
        elif ph == 2:
            tm = max(_LOCK_MIN, _MULT * _LOCK_SCALE); cand = hw - d * tm * atr
            cand = max(cand, entry) if d > 0 else min(cand, entry)
            sl = max(sl, cand) if d > 0 else min(sl, cand)
    return (C[end] - entry) * d / atr


def pivots_for_tf(H, L, step):
    """Pivots auf der aggregierten TF, zurückgegeben als (m5_index_bestätigt, preis).

    LOOK-AHEAD-frei: ein HTF-Pivot bei Index j ist erst bekannt, wenn k Bars DANACH
    abgeschlossen sind → verfügbar ab M5-Index (j+k+1)·step."""
    if step == 1:
        ph = [(i + _PIV_K + 1, H[i]) for i in range(_PIV_K, len(H) - _PIV_K)
              if H[i] == max(H[i - _PIV_K:i + _PIV_K + 1])]
        pl = [(i + _PIV_K + 1, L[i]) for i in range(_PIV_K, len(L) - _PIV_K)
              if L[i] == min(L[i - _PIV_K:i + _PIV_K + 1])]
        return ph, pl
    hh, ll = [], []
    for s in range(0, len(H) - step + 1, step):
        hh.append(max(H[s:s + step])); ll.append(min(L[s:s + step]))
    ph, pl = [], []
    for j in range(_PIV_K, len(hh) - _PIV_K):
        avail = (j + _PIV_K + 1) * step
        if hh[j] == max(hh[j - _PIV_K:j + _PIV_K + 1]): ph.append((avail, hh[j]))
        if ll[j] == min(ll[j - _PIV_K:j + _PIV_K + 1]): pl.append((avail, ll[j]))
    return ph, pl


def collect(a, b, H, L, C, EF, ES, AT, SP, PH, PL):
    """Wellen-Events sammeln; Level aus der übergebenen Pivot-Liste (TF-spezifisch)."""
    TH = _REVERSAL_STRETCH; out = []
    ph_i = 0; pl_i = 0
    ph_live: list[tuple[int, float]] = []; pl_live: list[tuple[int, float]] = []
    start = max(a, _N_BARS, _LOOKBACK)
    # Pivots, die vor `start` schon bekannt sind, vorladen
    while ph_i < len(PH) and PH[ph_i][0] <= start: ph_live.append(PH[ph_i]); ph_i += 1
    while pl_i < len(PL) and PL[pl_i][0] <= start: pl_live.append(PL[pl_i]); pl_i += 1
    for i in range(start, min(b, len(C) - _MAXH - 1)):
        while ph_i < len(PH) and PH[ph_i][0] <= i: ph_live.append(PH[ph_i]); ph_i += 1
        while pl_i < len(PL) and PL[pl_i][0] <= i: pl_live.append(PL[pl_i]); pl_i += 1
        atr = AT[i]
        if not atr or atr <= 0: 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: continue
        entry = C[i]; cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
        R_run = sim_run(entry, d, atr, H, L, C, i + 1) - cost
        rec = {"R_run": R_run, "touched": False}
        # nur Pivots aus dem Lookback-Fenster verwenden
        lo_idx = i - _LOOKBACK
        if d > 0:
            cands = [p for (ix, p) in ph_live if ix >= lo_idx and p > entry + 0.3 * atr]
            level = min(cands) if cands else None
        else:
            cands = [p for (ix, p) in pl_live if ix >= lo_idx and p < entry - 0.3 * atr]
            level = max(cands) if cands else None
        if level is not None:
            rec["dist_atr"] = abs(level - entry) / atr
            jt = None
            for j in range(i + 1, min(i + _MAXH, len(C) - _BRK_W)):
                if ((H[j] >= level) if d > 0 else (L[j] <= level)): jt = j; break
                if ((entry - L[j]) if d > 0 else (H[j] - entry)) >= 2.0 * atr: break
            if jt is not None:
                up = level + d * _BRK_ATR * atr; dn = level - d * _BRK_ATR * atr
                brk = False
                for j in range(jt, min(jt + _BRK_W, len(H))):
                    if (H[j] >= up) if d > 0 else (L[j] <= up): brk = True; break
                    if (L[j] <= dn) if d > 0 else (H[j] >= dn): brk = False; break
                rec.update(touched=True, brk=1.0 if brk else 0.0,
                           R_close=(level - entry) * d / atr - cost,
                           feat=[(C[jt] - C[max(0, jt - 6)]) * d / atr,
                                 (C[jt] - C[max(0, jt - 3)]) * d / atr,
                                 1.0 if (EF[jt] - ES[jt]) * d > 0 else 0.0,
                                 abs(level - entry) / atr])
        out.append(rec)
    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(y, p):
    """ROC-AUC über Rangstatistik (Mann-Whitney)."""
    y = np.asarray(y); p = np.asarray(p)
    pos = p[y == 1]; neg = p[y == 0]
    if len(pos) == 0 or len(neg) == 0: return float("nan")
    order = np.argsort(np.concatenate([pos, neg]))
    ranks = np.empty(len(order), float); ranks[order] = np.arange(1, len(order) + 1)
    return (ranks[:len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))


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
    si = mt5.symbol_info(sym); point = si.point
    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]; SP = [float(b["spread"]) * point for b in bars]
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW); AT = _atr_series(H, L, C)
    mid = len(C) // 2

    print("=" * 96)
    print(f"  LEVEL-TIMEFRAME-VERGLEICH — {sym} ({len(C)} M5-Bars, 2 Halbjahre)")
    print(f"  Trades bleiben M5; nur die S/R-LEVEL-Quelle variiert. Echtkosten, Live-Exit.")
    print(f"  Aktuell live: M5-Pivots (k={_PIV_K}) — Modell _PB_W darauf trainiert (AUC 0,71)")
    print("=" * 96)

    summary = {}
    for lbl, step in _TFS:
        PH, PL = pivots_for_tf(H, L, step)
        ev1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, SP, PH, PL)
        ev2 = collect(mid, len(C), H, L, C, EF, ES, AT, SP, PH, PL)
        t1 = [e for e in ev1 if e["touched"]]; t2 = [e for e in ev2 if e["touched"]]
        if len(t1) < 200 or len(t2) < 200:
            print(f"\n### {lbl}: zu wenige Touches ({len(t1)}/{len(t2)}) — übersprungen")
            continue
        X1 = np.array([e["feat"] for e in t1]); y1 = np.array([e["brk"] for e in t1])
        mu = X1.mean(0); sd = X1.std(0) + 1e-9
        w = fit_logreg((X1 - mu) / sd, y1)

        def pb(e):
            x = (np.array(e["feat"]) - mu) / sd
            return 1 / (1 + np.exp(-(w @ np.concatenate([[1.0], x]))))

        for e in t1 + t2: e["_p"] = pb(e)
        a2 = auc([e["brk"] for e in t2], [e["_p"] for e in t2])
        a1 = auc([e["brk"] for e in t1], [e["_p"] for e in t1])
        d1 = [e["dist_atr"] for e in ev1 if "dist_atr" in e]
        d2 = [e["dist_atr"] for e in ev2 if "dist_atr" in e]
        near1 = 100 * sum(1 for x in d1 if x < 1.0) / max(1, len(d1))
        near2 = 100 * sum(1 for x in d2 if x < 1.0) / max(1, len(d2))
        print(f"\n### LEVEL-TF {lbl}  (Pivots auf {lbl}, Trades M5)")
        print(f"  Level-ABSTAND (×ATR):   H1 Median {np.median(d1):.2f} · H2 {np.median(d2):.2f}"
              f"   |  <1×ATR entfernt: H1 {near1:.0f}% · H2 {near2:.0f}%")
        print(f"  Touches:                H1 {len(t1)} · H2 {len(t2)}"
              f"   |  Basisrate P(break): H1 {np.mean(y1):.0%} · H2 {np.mean([e['brk'] for e in t2]):.0%}")
        print(f"  Modell (auf H1 trainiert): AUC in-sample H1 {a1:.3f} · "
              f"**out-of-sample H2 {a2:.3f}**")
        print(f"    Koeff: mom6={w[1]:+.2f} mom3={w[2]:+.2f} wt={w[3]:+.2f} dist={w[4]:+.2f}")
        # Kalibrierung out-of-sample (H2)
        print(f"  Kalibrierung H2:  ", end="")
        for lo, hi in ((0.0, 0.3), (0.3, 0.5), (0.5, 0.7), (0.7, 1.01)):
            g = [e for e in t2 if lo <= e["_p"] < hi]
            if len(g) >= 20:
                print(f"[{lo:.1f}-{hi:.1f}) pred {np.mean([e['_p'] for e in g]):.0%}"
                      f"/real {np.mean([e['brk'] for e in g]):.0%} (n={len(g)})  ", end="")
        print()
        # R-Ertrag der Close-Regel
        res = {}
        for hl, ev, ts in (("H1", ev1, t1), ("H2", ev2, t2)):
            base = sum(e["R_run"] for e in ev)
            best = None
            row = []
            for X in (0.45, 0.50, 0.55, 0.60, 0.70):
                tot = (sum((e["R_close"] if e["_p"] < X else e["R_run"]) for e in ts)
                       + sum(e["R_run"] for e in ev if not e["touched"]))
                row.append((X, tot - base))
                if best is None or tot - base > best[1]: best = (X, tot - base)
            res[hl] = (base, row, best)
            print(f"  R-Ertrag {hl}: Baseline(nur Trailing) ΣR={base:+.0f}  →  "
                  + "  ".join(f"P<{X:.2f}: {dR:+.0f}" for X, dR in row))
        summary[lbl] = {"auc_h2": a2, "d_med": (np.median(d1) + np.median(d2)) / 2,
                        "near": (near1 + near2) / 2,
                        "best55": (dict(res["H1"][1])[0.55], dict(res["H2"][1])[0.55]),
                        "touch": (len(t1), len(t2))}

    print("\n" + "=" * 96)
    print("  URTEIL — umstellen nur, wenn eine höhere TF in AUC UND R-Ertrag mithält:")
    print(f"  {'TF':<6}{'AUC(H2)':>9}{'Level-Abstand':>15}{'<1xATR':>9}"
          f"{'ΔR bei P<0,55 (H1/H2)':>26}")
    for lbl in [t[0] for t in _TFS]:
        if lbl not in summary: continue
        s = summary[lbl]
        print(f"  {lbl:<6}{s['auc_h2']:>9.3f}{s['d_med']:>13.2f}xATR{s['near']:>8.0f}%"
              f"{s['best55'][0]:>+13.0f}{s['best55'][1]:>+13.0f}")
    if "M5" in summary:
        base = summary["M5"]
        for lbl in ("M15", "M30"):
            if lbl not in summary: continue
            s = summary[lbl]
            ok_auc = s["auc_h2"] >= base["auc_h2"] - 0.02
            ok_r = (s["best55"][0] >= base["best55"][0] * 0.9
                    and s["best55"][1] >= base["best55"][1] * 0.9)
            print(f"  → {lbl}: AUC {'ok' if ok_auc else 'SCHLECHTER'} · "
                  f"R-Ertrag {'ok' if ok_r else 'SCHLECHTER'} · "
                  f"{'UMSTELLEN vertretbar' if (ok_auc and ok_r) else 'NICHT umstellen'}")


if __name__ == "__main__":
    main()