#!/usr/bin/env python3 """Marktstruktur-Kanal als SIGNAL — über MEHRERE Zeitebenen (2026-08-03). Anlass: User-Beobachtung im M5-Chart „der Kurs folgt dem Marktstrukturkanal und bricht die untere Linie nicht — können wir das nutzen?". Die vorhandene Messung (`backtest_structure.py`, 2026-07-20) lief ausschliesslich auf **M30** und fiel in beiden Hälften durch. M5 ist damit streng genommen eine UNGETESTETE Variante. Unterschiede zu `backtest_structure.py` — bewusst, und beide relevant: 1. **Kanonischer Exit** (`core/exit_model.simulate`, LIVE) statt der dortigen Exit-KOPIE. Jene nutzt Trail **1,5** und kennt weder Time-Stop noch TP; live läuft seit 2026-07-31 Trail **1,0** MIT Time-Stop. Für eine NEUE Messung ist der kanonische Kern richtig (die alte Datei bleibt unverändert — sie umzu- stellen wäre laut Migrations-Regel nur erlaubt, wenn die Zahlen gleich blieben). 2. **Zeitebene als Parameter**, plus zwei Lesarten des „M5-Kanals": · `bars` — gleiche BAR-Anzahl wie auf M30 (=schnellerer, kürzerer Kanal; das, was man im M5-Chart tatsächlich sieht) · `time` — gleiche ZEIT-Spanne (Fenster ×6 auf M5) = derselbe Kanal, nur feiner aufgelöst Beide zu testen ist Pflicht: kippt das Vorzeichen zwischen ihnen, ist es Rauschen (die ORB-/SMC-Lehre — Nachbar-Parameter müssen mithalten). Maßstab wie immer: beidhälftig positiv UND parameter-robust. Vergleich: Squeeze ØR +0,14…+0,23 bei PF>1. Aufruf: python backtest_structure_tf.py [bars] """ import sys import MetaTrader5 as mt5 from core.exit_model import LIVE, simulate from core.structure import _pivots, _classify, _channel _ATRMIN = 0.06 _COOL = 3 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]))) return [(sum(t[max(1, i - p + 1):i + 1]) / max(1, len(t[max(1, i - p + 1):i + 1]))) if i else None for i in range(len(C))] def st(Rs): if not Rs: return None n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs) up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0) return dict(n=n, wr=100 * w / n, oR=s / n, pf=(up / dn if dn > 0 else 9.99), sum=s) def line(lbl, s): if not s: return f" {lbl:<26} — (keine Trades)" return (f" {lbl:<26} n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} " f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}") def run(H, L, C, A, SP, lo, hi, rule, X, look, reg): """Sequentielle 1-Positions-Sim mit dem KANONISCHEN Live-Exit.""" def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr Rs = [] i = max(lo, look) prev_pos = None while i < hi: atr = A[i] if not atr or atr < _ATRMIN: i += 1; prev_pos = None; continue ch = _channel(C[max(0, i - reg + 1):i + 1], atr) if not ch: i += 1; prev_pos = None; continue pos, cdir = ch["pos"], ch["dir"] d = 0 if rule in ("A", "A2"): # frischer Eintritt ins untere (Aufwärtskanal) bzw. obere (Abwärts) Drittel if cdir == "up" and pos <= X and (prev_pos is None or prev_pos > X): d = 1 elif cdir == "down" and pos >= 1 - X and (prev_pos is None or prev_pos < 1 - X): d = -1 if d != 0 and rule == "A2": pl = _pivots(H[max(0, i - look):i + 1], L[max(0, i - look):i + 1], 3) lab = _classify(pl) recent = [x["type"] for x in lab[-4:]] ups = sum(1 for t in recent if t in ("HH", "HL")) dns = sum(1 for t in recent if t in ("LH", "LL")) strend = ("up" if ups >= 3 and ups > dns else "down" if dns >= 3 and dns > ups else "range") if (d > 0 and strend != "up") or (d < 0 and strend != "down"): d = 0 prev_pos = pos if d == 0: i += 1; continue entry = C[i] a = max(atr, _ATRMIN) r, xj = simulate(entry, d, a, H, L, C, i + 1, LIVE, timestop=True, use_tp=True, ret_bar=True) Rs.append(r - cost(i, a)) i = xj + _COOL prev_pos = None return Rs def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000 mt5.initialize() sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(c)), None) point = mt5.symbol_info(sym).point TFS = [("M5", mt5.TIMEFRAME_M5), ("M15", mt5.TIMEFRAME_M15), ("M30", mt5.TIMEFRAME_M30)] data = {} for lbl, tf in TFS: b = mt5.copy_rates_from_pos(sym, tf, 0, n) if b is not None and len(b) > 5000: data[lbl] = ([float(x["high"]) for x in b], [float(x["low"]) for x in b], [float(x["close"]) for x in b], [float(x["spread"]) * point for x in b]) mt5.shutdown() print("=" * 92) print(f" Kanal-Pullback über Zeitebenen — {sym}, kanonischer Live-Exit " f"(SL {LIVE.sl_atr}× · Trail {LIVE.mult}× · BE {LIVE.be}× · Time-Stop), Echtkosten") print(" Regel: im Aufwärtskanal am UNTEREN Drittel long / im Abwärtskanal am OBEREN short") print(" Maßstab: beidhälftig ØR>0 UND PF>1 UND Nachbar-Varianten halten mit") print("=" * 92) # (reg, look) je Lesart. M30-Referenz = 60/220 wie in backtest_structure.py. plans = [ ("M5 Kanal 60 Bars (=5 h, 'was man im M5-Chart sieht')", "M5", 60, 220), ("M5 Kanal 360 Bars (=30 h, zeitgleich zum M30-Kanal)", "M5", 360, 1320), ("M15 Kanal 120 Bars (=30 h, Nachbar-Zeitebene)", "M15", 120, 440), ("M30 Kanal 60 Bars (=30 h, die ALTE Messung)", "M30", 60, 220), ] for title, tfl, reg, look in plans: if tfl not in data: print(f"\n {title}: keine Daten"); continue H, L, C, SP = data[tfl] A = _atr_series(H, L, C) N = len(C); mid = N // 2 print(f"\n {title} [{N} Bars]") for rl, X, sub in (("A", 0.20, "Kanal-Pullback X=0.20"), ("A", 0.30, "Kanal-Pullback X=0.30"), ("A2", 0.25, "+Swing-Filter X=0.25")): s1 = st(run(H, L, C, A, SP, 0, mid, rl, X, look, reg)) s2 = st(run(H, L, C, A, SP, mid, N - LIVE.max_hold - 1, rl, X, look, reg)) print(f" · {sub}") print(line("H1 (alt)", s1)) print(line("H2 (neu)", s2)) ok = (s1 and s2 and s1["oR"] > 0 and s2["oR"] > 0 and s1["pf"] > 1 and s2["pf"] > 1) print(f" → {'ROBUST' if ok else 'fällt durch'}") print() if __name__ == "__main__": main()