#!/usr/bin/env python3 """Kerzen-Anatomie als Vorhersage — trägt sie, und WANN? (User-Vorgabe 2026-07-31) Getestet werden die vier KONDITIONALEN Behauptungen des Users (nicht „Kerzenmuster funktionieren", sondern „sie funktionieren IM KONTEXT"): 1. Langer Docht zählt **am Level**, in der Range-Mitte ist er Noise. 2. Langer Körper **ohne Volumen** ist ein Fake-Out; mit 2× Ø-Volumen ist er echt. 3. Auf M15 sind Dochte Noise — erst H1 bestätigt. 4. Langer Körper GEGEN den Trend = Erschöpfung, MIT dem Trend = Fortsetzung. METHODE: reine INFORMATIONS-Messung (Forward-Return in ×ATR über N Bars), kein Exit/Sizing — die Frage ist, ob die Kerzenklasse überhaupt Richtungsinformation trägt. Erst wenn ja, lohnt ein Trade-Backtest mit Exit+Kosten. ⚠ Die Baseline ist entscheidend: H1 war ein Abwärts-, H2 ein Aufwärts-Regime. Ein „+0,10 ØR"-Bucket ist wertlos, wenn ALLE Bars in dieser Hälfte +0,10 liefern. Deshalb wird jeder Bucket **relativ zur Regime-Baseline derselben Hälfte** gelesen (Lehre aus `backtest_volume_profile.py`). Urteil „trägt" nur, wenn ein Bucket in BEIDEN Hälften relativ zur Baseline richtungs-konsistent UND deutlich abweicht. """ import sys from bisect import bisect_left from statistics import median import MetaTrader5 as mt5 _FWD = 12 # Forward-Fenster in M5-Bars (1 h) — wie beim P(break)-Training _ATRMIN = 0.06 # Live-Floor _M30 = 6 _PIV_K = 3 _PIV_LOOKBACK = 50 _NEAR_ATR = 0.5 # „am Level" = näher als X×ATR _VOL_N = 20 # Fenster für Ø-Volumen und Ø-Body 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 _ema(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 build_levels(H, L): """M30-Pivots k=3, kausal (Pivot erst 3 M30-Bars später bekannt) — dieselbe Quelle wie der Live-Bot seit 2026-07-30.""" from collections import deque 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 classify(o, h, l, c, body_ref): """Kerzenklasse + implizierte Richtung. body_ref = Median-Body der letzten _VOL_N Bars (Massstab für „lang"). Rückgabe: (klasse, richtung) — richtung 0 = keine implizierte Richtung. """ rng = h - l if rng <= 0: return None, 0 body = abs(c - o) up_w = h - max(o, c) lo_w = min(o, c) - l br = body / rng d = 1 if c > o else -1 if c < o else 0 long_body = body_ref > 0 and body >= 1.5 * body_ref if br >= 0.90: return "Marubozu", d if br < 0.10: return "Doji", 0 if br < 0.30 and up_w / rng > 0.30 and lo_w / rng > 0.30: return "SpinningTop", 0 if up_w / rng >= 0.50: return "LangerDochtOben", -1 # Ablehnung nach oben → Short impliziert if lo_w / rng >= 0.50: return "LangerDochtUnten", 1 if long_body and br >= 0.60: return "LangerKoerper", d return None, 0 def st(xs): if not xs: return " n=0" n = len(xs); m = sum(xs) / n w = sum(1 for x in xs if x > 0) return f"n={n:>6} ØR={m:+.3f} WR={100*w/n:>3.0f}%" def main(): n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000 mt5.initialize(); sym = None for cand in ("SpotCrude", "USOIL", "WTI", "XTIUSD"): if mt5.symbol_info(cand): sym = cand; break bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n) mt5.shutdown() O = [float(b["open"]) for b in bars]; H = [float(b["high"]) for b in bars] L = [float(b["low"]) for b in bars]; C = [float(b["close"]) for b in bars] V = [float(b["tick_volume"]) for b in bars] AT = _atr_series(H, L, C) EF = _ema(C, 12); ES = _ema(C, 50) LV = build_levels(H, L) N = len(C); mid = N // 2 halves = [("H1", 100, mid), ("H2", mid, N - _FWD - 1)] print("=" * 100) print(f" KERZEN-ANATOMIE als Vorhersage — {sym} M5 ({N} Bars), " f"Forward {_FWD} Bars = {_FWD*5} min") print(" Alle Werte in ×ATR. ⚠ Buckets IMMER gegen die Regime-Baseline derselben") print(" Hälfte lesen (H1/H2 hatten unterschiedliche Drift).") print("=" * 100) for lbl, a, b in halves: # ── Baseline: durchschnittlicher Forward-Return ALLER Bars, long-normiert base = [] for i in range(a, b): atr = AT[i] if not atr: continue base.append((C[i + _FWD] - C[i]) / max(atr, _ATRMIN)) bm = sum(base) / len(base) if base else 0.0 print(f"\n{'='*100}\n{lbl} — Regime-Baseline (alle Bars, LONG-Sicht): " f"ØR={bm:+.3f} (n={len(base)})\n{'='*100}") # Buckets sammeln by_cls: dict = {} by_cond: dict = {} dsum: dict = {} dcond: dict = {} for i in range(a, b): atr = AT[i] if not atr: continue atr = max(atr, _ATRMIN) body_ref = median([abs(C[k] - O[k]) for k in range(max(0, i - _VOL_N), i)]) \ if i > _VOL_N else 0.0 cls, d = classify(O[i], H[i], L[i], C[i], body_ref) if not cls: continue fwd = (C[i + _FWD] - C[i]) / atr # in IMPLIZIERTER Richtung; ohne Richtung (Doji/Spinning) long-normiert r = fwd * d if d else fwd # Richtungs-Bilanz mitführen: bei gerichteten Klassen ist die korrekte # NULL-Linie nicht die Baseline, sondern bm×E[d] — die Regime-Drift hebt # sich auf, wenn Long-/Short-Kerzen ausgewogen sind. Ohne das liest man # reine Drift als „Edge" (der Fehler, der `backtest_volume_profile.py` # fast gekippt hätte). by_cls.setdefault(cls, []).append(r) dsum.setdefault(cls, []).append(d) if d == 0: continue # ── Bedingung 1: am Level? lv = LV[min(i // _M30, len(LV) - 1)] near = False if lv: j = bisect_left(lv, C[i]) for k in (j - 1, j): if 0 <= k < len(lv) and abs(lv[k] - C[i]) <= _NEAR_ATR * atr: near = True # ── Bedingung 2: Volumen vref = sum(V[max(0, i - _VOL_N):i]) / max(1, len(V[max(0, i - _VOL_N):i])) hot = vref > 0 and V[i] >= 2.0 * vref # ── Bedingung 3: mit/gegen M30-Trend (EMA12 vs EMA50 auf M5 als Proxy) trend = 1 if EF[i] > ES[i] else -1 withtrend = (d == trend) key = "Docht" if cls.startswith("LangerDocht") else \ ("Koerper" if cls in ("LangerKoerper", "Marubozu") else None) if key: for cond in ("am Level" if near else "Range-Mitte", "Vol>=2x" if hot else "Vol normal", "mit Trend" if withtrend else "gegen Trend"): by_cond.setdefault((key, cond), []).append(r) dcond.setdefault((key, cond), []).append(d) def line(label, xs, ds): """Zeigt ØR, die DRIFT-NULL (bm×E[d]) und den Überschuss darüber.""" null = bm * (sum(ds) / len(ds)) if ds else 0.0 m = sum(xs) / len(xs) print(f" {label:<24} {st(xs)} Null={null:+.3f} Ueberschuss={m-null:+.3f}") print("\n -- Klassen (in implizierter Richtung; Doji/SpinningTop long-normiert) --") print(" 'Null' = reine Regime-Drift dieser Klasse; nur der UEBERSCHUSS ist Information.") for cls in ("LangerKoerper", "Marubozu", "LangerDochtOben", "LangerDochtUnten", "SpinningTop", "Doji"): xs = by_cls.get(cls) if xs: line(cls, xs, dsum.get(cls, [0] * len(xs))) print("\n -- Konditional (die 4 User-Behauptungen) --") for key in ("Docht", "Koerper"): for cond in ("am Level", "Range-Mitte", "Vol>=2x", "Vol normal", "mit Trend", "gegen Trend"): xs = by_cond.get((key, cond)) if xs: line(f"{key} {cond}", xs, dcond.get((key, cond), [0] * len(xs))) if __name__ == "__main__": main()