AH-Oil-Trader/backtest_candles.py

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