#!/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()