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