diff --git a/backtest_candles.py b/backtest_candles.py new file mode 100644 index 0000000..ba4451b --- /dev/null +++ b/backtest_candles.py @@ -0,0 +1,233 @@ +#!/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()