AH-Oil-Trader/backtest_pbreak_touches.py

#!/usr/bin/env python3
"""P(break) mit TOUCH-ZAHL als 5. Merkmal — trägt sie über 2 echte Halbjahre?

User-Idee 2026-07-31: „berücksichtige, ob der Kurs das erste, zweite oder dritte Mal
in einer Zeitspanne den S/R trifft — je öfter, desto wahrscheinlicher der Durchbruch."

AUF LIVE-DATEN BEREITS BESTÄTIGT (`analyze_pbreak_live.py`, 2275 Vorhersagen, beide
Hälften): Bruchrate nach n-ter Berührung im 30-min-Fenster
    1.  38,3 / 41,7 %      3.  43,2 / 45,1 %
    2.  40,8 / 45,6 %      4.  45,1 / 47,4 %   ← Maximum, +6/+4 Pp über Basis
    5.+ 37,8 / 41,8 %      ← FÄLLT ZURÜCK unter die Basis (größte Gruppe, n≈1400)
Also ein **umgekehrtes U**: 2–4 Berührungen machen das Level mürbe, ab 5 in 30 min
ist es eine Range-Begrenzung und HÄLT. Das aktuelle Modell sieht davon nichts (sagt
über alle Berührungszahlen konstant ~23 % voraus).

⚠ Die Live-Bestätigung ist schwach (beide „Hälften" nur 4 Tage auseinander, Buckets
n=82–139) — DESHALB dieser Test über 80k M5-Bars / 2 echte Halbjahre.
⚠ `backtest_srbreak.py` fand die Touch-Zahl früher „nicht robust" — allerdings ohne
Fenster-Definition und ohne die 5+-Umkehr.

WAS HIER ANDERS IST ALS IN `backtest_srclose_prob.py`:
  * **M30-Level** (wie live seit 2026-07-30), nicht die alten M5-Pivots.
  * Direkter Vergleich **4 Merkmale (Live-Modell) vs. 5 Merkmale (+Touch-Zahl)** —
    ohne diesen Kontrast wäre nicht unterscheidbar, ob eine bessere AUC vom neuen
    Merkmal kommt oder von der Neuanpassung.
  * Urteil: das 5-Merkmal-Modell muss in BEIDEN Hälften besser sein (AUC UND R-Ertrag).

Die Touch-Zahl wird als **umgekehrtes U** kodiert (zwei Dummies statt eines linearen
Terms): `t_mid` = 1 bei 2–4 Berührungen, `t_many` = 1 ab 5. Ein linearer Term würde
die gemessene Umkehr systematisch verfehlen.
"""
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; _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; _BRK_W = 12; _BRK_ATR = 0.5
_M30 = 6; _PIV_LOOKBACK = 50            # M30-Level wie live
_ZONE = 0.15                            # „am Level" = |Kurs−Level| <= X×ATR (wie live)
_TWIN = 6                               # Touch-Fenster in M5-Bars (6 = 30 min, wie live)


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 (Pivot erst 3 M30-Bars später bekannt) — wie live."""
    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 touch_count(H, L, level, atr, jt, win):
    """Wievielte Berührung dieses Levels innerhalb der letzten `win` Bars?

    Gezählt werden EPISODEN (Eintritte in die Zone |Kurs−Level| <= _ZONE×ATR nach
    einem Bar außerhalb) — nicht Bars, sonst zählt ein langes Verweilen als viele
    Berührungen. Die aktuelle Berührung bei jt zählt mit, Ergebnis also >= 1.
    """
    z = _ZONE * atr
    n = 0; inside_prev = False
    for j in range(max(0, jt - win), jt + 1):
        inside = (L[j] <= level + z) and (H[j] >= level - z)
        if inside and not inside_prev:
            n += 1
        inside_prev = inside
    return max(1, n)


def sim_run(entry, d, atr, H, L, C, j0):
    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 collect(a, b, H, L, C, EF, ES, AT, SP, LV):
    TH = _REVERSAL_STRETCH; out = []
    for i in range(max(a, _N_BARS, 320), min(b, len(C) - _MAXH - 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}
        lv = LV[min(i // _M30, len(LV) - 1)]
        level = None
        if lv:
            if d > 0:
                cands = [p for p in lv if p > entry + 0.3 * atr]
                level = min(cands) if cands else None
            else:
                cands = [p for p in lv if p < entry - 0.3 * atr]
                level = max(cands) if cands else None
        if level is not None:
            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
                tc = touch_count(H, L, level, atr, jt, _TWIN)
                base4 = [(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]
                rec.update(touched=True, brk=1.0 if brk else 0.0, tc=tc,
                           R_close=(level - entry) * d / atr - cost,
                           feat4=base4,
                           # umgekehrtes U: 2–4 Berührungen vs. 5+
                           feat5=base4 + [1.0 if 2 <= tc <= 4 else 0.0,
                                          1.0 if tc >= 5 else 0.0])
        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(ps, ys):
    order = np.argsort(ps); ys = np.asarray(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 main():
    global _TWIN
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    if len(sys.argv) > 2:
        _TWIN = int(sys.argv[2])
    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)
    LV = build_levels(H, L)
    mid = len(C) // 2
    ev1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, SP, LV)
    ev2 = collect(mid, len(C), H, L, C, EF, ES, AT, SP, LV)
    t1 = [e for e in ev1 if e["touched"]]; t2 = [e for e in ev2 if e["touched"]]

    print("=" * 98)
    print(f"  P(break) + TOUCH-ZAHL — {sym} M5 ({len(C)} Bars), M30-Level wie live")
    print(f"  Touch-Fenster {_TWIN} Bars = {_TWIN*5} min · Zone {_ZONE}×ATR")
    print(f"  Fit auf H1 ({len(t1)} Touches), geprüft auf H2 ({len(t2)} Touches)")
    print("=" * 98)

    y1 = np.array([e["brk"] for e in t1]); y2 = np.array([e["brk"] for e in t2])
    print(f"\n  Basisrate Durchbruch:  H1 {100*y1.mean():.1f} %   H2 {100*y2.mean():.1f} %")

    # ── Rohbefund: Bruchrate je Berührungszahl, BEIDE Hälften
    print(f"\n  Rohbefund — Bruchrate je Berührungszahl (User-Hypothese):")
    print(f"    {'Berührung':<12}{'H1 n':>7}{'H1 Bruch':>11}{'H2 n':>7}{'H2 Bruch':>11}")
    for b in (1, 2, 3, 4, 5):
        lab = f"{b}." if b < 5 else "5.+"
        a1 = [e["brk"] for e in t1 if (e["tc"] == b if b < 5 else e["tc"] >= 5)]
        a2 = [e["brk"] for e in t2 if (e["tc"] == b if b < 5 else e["tc"] >= 5)]
        if len(a1) < 20 or len(a2) < 20:
            print(f"    {lab:<12}{len(a1):>7}{'-':>11}{len(a2):>7}{'-':>11}"); continue
        print(f"    {lab:<12}{len(a1):>7}{100*np.mean(a1):>10.1f}%"
              f"{len(a2):>7}{100*np.mean(a2):>10.1f}%")

    # ── Modellvergleich 4 vs 5 Merkmale
    results = {}
    for tag, key in (("4 Merkmale (Live-Modell)", "feat4"),
                     ("5 Merkmale (+Touch-Zahl)", "feat5")):
        X1 = np.array([e[key] for e in t1]); X2 = np.array([e[key] for e in t2])
        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)))
        results[key] = (mu, sd, w, p1, p2)
        print(f"\n  {tag}")
        print(f"    AUC  in-sample H1 {auc(p1, y1):.3f}   out-of-sample H2 {auc(p2, y2):.3f}")
        if key == "feat5":
            print(f"    Gewichte Touch: 2–4 Ber. {w[5]:+.3f} · 5+ Ber. {w[6]:+.3f}")
        print(f"    Kalibrierung H2:  {'Bucket':<12}{'n':>6}{'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(p2, y2) if lo <= p < hi]
            if len(sel) < 40: continue
            print(f"                      {lo:.2f}–{hi:<7.2f}{len(sel):>6}"
                  f"{100*np.mean([p for p, _ in sel]):>9.1f}%"
                  f"{100*np.mean([y for _, y in sel]):>7.1f}%")

    # ── R-Ertrag der Close-Regel, beide Modelle, beide Hälften
    print(f"\n{'='*98}\n  R-ERTRAG der Close-Regel (Echtkosten, Baseline = immer laufen lassen)\n{'='*98}")
    for lbl, ev, ts, idx in (("H1", ev1, t1, 3), ("H2", ev2, t2, 4)):
        base = sum(e["R_run"] for e in ev)
        untouched = sum(e["R_run"] for e in ev if not e["touched"])
        print(f"\n  {lbl}: {len(ev)} Wellen · {len(ts)} Touches · Baseline ΣR={base:+.0f}")
        print(f"    {'Schwelle':<10}{'4 Merkmale':>14}{'5 Merkmale':>14}{'Differenz':>12}")
        for X in (0.35, 0.45, 0.55, 0.60):
            row = []
            for key in ("feat4", "feat5"):
                mu, sd, w, p1, p2 = results[key]
                ps = p1 if lbl == "H1" else p2
                tot = sum((e["R_close"] if p < X else e["R_run"])
                          for e, p in zip(ts, ps)) + untouched
                row.append(tot)
            print(f"    P≥{X:.2f}    {row[0]:>+14.0f}{row[1]:>+14.0f}{row[1]-row[0]:>+12.0f}")


if __name__ == "__main__":
    main()