#!/usr/bin/env python3 """Backtest der visuellen Chartmuster (Track B, Gate VOR Signal-Einfluss). Nutzt die ECHTE Erkennungslogik aus `core/patterns.py` (`_pivots` + dieselben Toleranzen) → misst genau das, was die Dashboard-Karte zeigt. Ereignisbasiert: 1. Pivots auf der vollen M30-History (mit Bestätigungs-Lag k, KEIN Look-ahead). 2. Jedes Muster-Vorkommen (Doppeltop/-boden, SKS/invers, Tasse/invers, Dreiecke) an seinem FORMATIONS-Bar (letzter Pivot + k) mit Trigger/Ziel erfassen. 3. Einstieg beim ersten Trigger-Bruch DANACH (in Musterrichtung, Fenster 48 Bars, sonst verfällt das Muster = kein Trade). 4. Exit = Live-Modell (SL 2,0×ATR + Trailing 1,5 + BE 1,3), Echtkosten (Bar- Spread/ATR). Sequentiell (1 Position) → keine überlappenden Doppel-Trades. 5. Auswertung je Muster-Typ, 2 Halbjahre: n / WR / ØR / PF / Ziel-Trefferquote. Einbau in die Empfehlung NUR, wenn ein Typ in BEIDEN Hälften ØR netto > ~+0,1×ATR UND PF>1 zeigt (Squeeze-Maßstab). Erwartung niedrig — Muster-Klasse 2× verworfen. """ import sys import MetaTrader5 as mt5 from core.patterns import (_pivots, _PIVOT_K, _MIN_HEIGHT_ATR, _TOL_ATR, _NECK_TOL_ATR) _MAXH = 288; _ATRMIN = 0.06; _BREAK_WIN = 48; _SL_ATR = 2.0; _TRAIL = 1.5 _TRAIL_ON = 0.3; _BE = 1.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 _sim(a, b, atr): return max(0.0, 1.0 - abs(a-b)/(_TOL_ATR*atr)) if atr else 0.0 def scan(P, A): """→ Liste Muster-Events (form_bar, typ, dir, trigger, extreme). Nutzt dieselben Bedingungen wie core.patterns._detect, aber über ALLE Pivot-Fenster der History.""" ev = [] K = _PIVOT_K for a in range(len(P)-2): (ia, pa, ka), (ib, pb, kb), (ic, pc, kc) = P[a], P[a+1], P[a+2] fb = ic + K # Formations-Bar (letzter Pivot bestätigt) if fb >= len(A) or not A[fb] or A[fb] <= 0: continue atr = max(A[fb], _ATRMIN) def ok_h(ext, trig): return abs(ext-trig) >= _MIN_HEIGHT_ATR*atr # Doppeltop H,L,H if ka == "H" and kb == "L" and kc == "H": if _sim(pa, pc, atr) > 0.35 and pb < min(pa, pc) and ok_h(max(pa, pc), pb): ev.append((fb, "double_top", -1, pb, max(pa, pc))) # Doppelboden L,H,L if ka == "L" and kb == "H" and kc == "L": if _sim(pa, pc, atr) > 0.35 and pb > max(pa, pc) and ok_h(min(pa, pc), pb): ev.append((fb, "double_bottom", +1, pb, min(pa, pc))) # (Inverse) Tasse+Henkel — Rand-Extrema ~gleich, breit (>=10 Bars) span = ic - ia if ka == "L" and kb == "H" and kc == "L" and span >= 10: if _sim(pa, pc, atr) > 0.4 and pb > max(pa, pc) and ok_h(min(pa, pc), pb): ev.append((fb, "inv_cup", -1, min(pa, pc), pb)) if ka == "H" and kb == "L" and kc == "H" and span >= 10: if _sim(pa, pc, atr) > 0.4 and pb < min(pa, pc) and ok_h(max(pa, pc), pb): ev.append((fb, "cup", +1, max(pa, pc), pb)) # Kopf-Schulter / inverse SKS (5 Pivots) for a in range(len(P)-4): w = P[a:a+5]; kinds = [x[2] for x in w]; pr = [x[1] for x in w] fb = w[4][0] + K if fb >= len(A) or not A[fb] or A[fb] <= 0: continue atr = max(A[fb], _ATRMIN) if kinds == ["H", "L", "H", "L", "H"] and pr[2] > pr[0] and pr[2] > pr[4]: neck = (pr[1]+pr[3])/2 if _sim(pr[0], pr[4], atr)*_sim(pr[1], pr[3], atr) > 0.25 \ and abs(pr[1]-pr[3]) < _NECK_TOL_ATR*atr and abs(pr[2]-neck) >= _MIN_HEIGHT_ATR*atr: ev.append((fb, "hs", -1, neck, pr[2])) if kinds == ["L", "H", "L", "H", "L"] and pr[2] < pr[0] and pr[2] < pr[4]: neck = (pr[1]+pr[3])/2 if _sim(pr[0], pr[4], atr)*_sim(pr[1], pr[3], atr) > 0.25 \ and abs(pr[1]-pr[3]) < _NECK_TOL_ATR*atr and abs(pr[2]-neck) >= _MIN_HEIGHT_ATR*atr: ev.append((fb, "inv_hs", +1, neck, pr[2])) # Dreiecke (letzte 2 Hochs + 2 Tiefs bis zu jedem Pivot) for a in range(3, len(P)): sub = P[:a+1] hs = [(x[0], x[1]) for x in sub if x[2] == "H"][-2:] ls = [(x[0], x[1]) for x in sub if x[2] == "L"][-2:] if len(hs) < 2 or len(ls) < 2: continue fb = P[a][0] + K if fb >= len(A) or not A[fb] or A[fb] <= 0: continue atr = max(A[fb], _ATRMIN) (_, h1), (_, h2) = hs; (_, l1), (_, l2) = ls wide = abs(max(h1, h2)-min(l1, l2)) if wide < _MIN_HEIGHT_ATR*atr: continue flat_h = abs(h2-h1) <= _NECK_TOL_ATR*atr; flat_l = abs(l2-l1) <= _NECK_TOL_ATR*atr rising_l = l2 > l1+0.4*atr; falling_h = h2 < h1-0.4*atr if flat_h and rising_l: ev.append((fb, "tri_asc", +1, max(h1, h2), min(l1, l2))) elif flat_l and falling_h: ev.append((fb, "tri_desc", -1, min(l1, l2), max(h1, h2))) ev.sort() return ev def sim(entry, d, atr, H, L, C, j0, target): """Live-Exit (SL 2×ATR + Trailing) → (R, ziel_getroffen bool).""" eff = entry - d*_SL_ATR*atr; hw = entry end = min(j0+_MAXH, len(C)-1); exit_px = C[end]; hit = False for j in range(j0, end+1): hi, lo = H[j], L[j] if (target is not None) and ((hi >= target) if d > 0 else (lo <= target)): hit = True if (lo <= eff) if d > 0 else (hi >= eff): exit_px = eff; break hw = max(hw, hi) if d > 0 else min(hw, lo) prof = (C[j]-entry)*d if prof >= _TRAIL_ON*atr: cand = hw - d*_TRAIL*atr if prof >= _BE*atr: cand = max(cand, entry) if d > 0 else min(cand, entry) eff = max(eff, cand) if d > 0 else min(eff, cand) return (exit_px-entry)*d/atr, hit def rep(name, rows): if not rows: print(f" {name:<26} —"); return R = [r[0] for r in rows]; n = len(R); w = sum(1 for x in R if x > 0) up = sum(x for x in R if x > 0); dn = -sum(x for x in R if x < 0) pf = up/dn if dn > 0 else 9.99; hit = 100*sum(1 for r in rows if r[1])/n print(f" {name:<26} n={n:>4} WR={100*w/n:>3.0f}% ØR={sum(R)/n:+.3f} " f"PF={pf:>4.2f} ΣR={sum(R):>+6.0f} Ziel={hit:>3.0f}%") def run(H, L, C, SP, A, point, lo, hi): """Sequentielle Sim über Events in [lo,hi). → dict typ→[(R,hit)].""" P = _pivots(H, L, _PIVOT_K) ev = [e for e in scan(P, A) if lo <= e[0] < hi] def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr res = {}; next_free = lo for fb, typ, d, trig, extreme in ev: if fb < next_free: # überlappt laufenden Trade → skip continue entry_j = None for j in range(fb+1, min(fb+1+_BREAK_WIN, len(C)-_MAXH-1)): if (L[j] <= trig) if d < 0 else (H[j] >= trig): entry_j = j; break if entry_j is None: continue atr = max(A[entry_j] or 0, _ATRMIN) h = abs(extreme-trig); target = trig - d*(-h) if False else (trig - h if d < 0 else trig + h) r, hit = sim(trig, d, atr, H, L, C, entry_j+1, target) r -= cost(entry_j, atr) res.setdefault(typ, []).append((r, hit)) next_free = None # Exit-Bar approximieren: bis zum SL/MAXH; einfach _MAXH als Cooldown-Deckel next_free = entry_j + 6 # kleiner Cooldown gegen Cluster return res def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 120000 mt5.initialize() sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(c)), None) bars = None for req in (n, 100000, 80000, 60000, 40000): bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, req) if bars is not None and len(bars) > 3000: break si = mt5.symbol_info(sym); point = si.point; 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] A = _atr_series(H, L, C); N = len(C); mid = N//2 print("="*90) print(f" Chartmuster-Backtest — {sym} M30 ({N} Bars) · Exit=live(SL2/Trail1,5/BE1,3) · Echtkosten") print(f" Einbau NUR, wenn ein Typ in BEIDEN Hälften ØR>~+0,1 UND PF>1 (Muster-Klasse 2× verworfen)") print("="*90) types = ["double_top", "double_bottom", "hs", "inv_hs", "cup", "inv_cup", "tri_asc", "tri_desc"] label = {"double_top": "Doppeltop", "double_bottom": "Doppelboden", "hs": "Kopf-Schulter", "inv_hs": "Inverse SKS", "cup": "Tasse+Henkel", "inv_cup": "Inv. Tasse+Henkel", "tri_asc": "Aufst. Dreieck", "tri_desc": "Abst. Dreieck"} for lbl, a, b in (("H1 (alt)", _PIVOT_K, mid), ("H2 (neu)", mid, N-_MAXH-1)): res = run(H, L, C, SP, A, point, a, b) allrows = [r for rows in res.values() for r in rows] print(f"\n{lbl}:") rep("ALLE Muster", allrows) for t in types: rep(label[t], res.get(t, [])) print("\n ØR/PF je Typ in BEIDEN Hälften vergleichen. Ziel% = wie oft die Measured-Move-") print(" Marke vor dem SL berührt wurde (Muster-Logik-Check, nicht der R-Ertrag).") if __name__ == "__main__": main()