AH-Oil-Trader/backtest_candle_fade2.py

#!/usr/bin/env python3
"""Klimax-Kerzen-Fade, NEUMESSUNG mit dem Stand vom 2026-08-05.

⚠⚠ HOHE BEWEISLAST — diese Klasse ist bereits ZWEIMAL durchgefallen:
  * `backtest_candles.py` (31.07.): die Lehrbuch-Lesart ist in beiden Hälften
    INVERTIERT — nach einer volumenstarken großen Kerze läuft der Kurs GEGEN sie
    (Überschuss −0,117/−0,204 bei Volumen ≥2×). Das ist der Informations-Befund.
  * `backtest_candle_fade.py` (31.07.): als TRADE fällt der Fade durch — benachbarte
    Volumen-Schwellen exakt gegenläufig (Vol≥2,0: −0,059/**+0,103**; Vol≥2,5:
    **+0,144**/−0,080) = Rauschmuster. 18. verworfener Signal-Eingriff.
Es wird hier NICHT neu gefischt. Gemessen wird EINE Achse, die im alten Lauf fehlte
und die am 05.08. Squeeze UND Signal von negativ auf positiv gedreht hat:

    **DER EINSTIEGSPREIS.**

Der alte Lauf steigt am **Bar-Close der Klimax-Kerze** ein (Market). Für einen Fade
ist das der denkbar schlechteste Moment — man verkauft, NACHDEM die Kerze schon
gelaufen ist. Die natürliche ruhende Order liegt am **Extrem der Kerze**: eine
Sell-Limit am Hoch (bzw. Buy-Limit am Tief) füllt nur, wenn die Klimax sich noch
etwas ausdehnt — besserer Preis UND zusätzlicher Filter in einem.

ZWEITE Änderung: **kanonischer Exit** (`exit_model.LIVE`, Trail **1,0**). Der alte
Lauf nutzte eine Exit-KOPIE mit Trail 1,5 — seit dem 31.07. nicht mehr der Live-Stand.

⚠ DIE KONTROLLGRUPPE BLEIBT PFLICHT („Fade OHNE Volumen-Bedingung"). Ohne sie wäre
nicht unterscheidbar, ob der Ertrag vom Volumen kommt oder ob schlicht JEDER Fade
einer großen Kerze funktioniert — genau der Kontrollfehler, der den Chartmuster-
„Edge" als Exit-Artefakt entlarvte, als er nachgeholt wurde.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL (vor dem ersten Lauf festgelegt):
  Der Fade gilt nur als tragfähig, wenn ALLE vier Bedingungen halten:
    (a) ØR netto > 0 UND PF > 1 in BEIDEN Halbjahren,
    (b) die Volumen-Variante schlägt die KONTROLLE in BEIDEN Halbjahren,
    (c) die benachbarten Volumen-Schwellen kippen das Vorzeichen NICHT,
    (d) n ≥ 100 je Hälfte (sonst ist es keine Stichprobe, sondern eine Anekdote).
  Sonst: bleibt verworfen, und die Kachel bleibt reine Anzeige.

Aufruf:  python backtest_candle_fade2.py [bars]
"""
import sys

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate

_ATRMIN = 0.06
_REF_N = 20            # Fenster für Median-Body und Ø-Volumen (wie core/candles.py)
_LONG_BR = 0.60        # Body/Range für „langer Körper"
_LONG_X = 1.5          # Body >= X × Median-Body
_WARTE = (3, 6, 12)    # Gültigkeit der Limit-Order in Bars
_VOL = (0.0, 1.5, 2.0, 2.5, 3.0)   # 0.0 = KONTROLLE (kein Volumen-Filter)


def _atr_series(H, L, C, p=14):
    tr = [0.0]
    for i in range(1, len(C)):
        tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    out = [None]
    for i in range(1, len(C)):
        seg = tr[max(1, i - p + 1):i + 1]
        out.append(sum(seg) / len(seg))
    return out


def klimax(O, H, L, C, V, i):
    """→ Fade-Richtung (+1 LONG / −1 SHORT) und Volumen-Vielfaches, sonst None."""
    rng = H[i] - L[i]
    body = abs(C[i] - O[i])
    if rng <= 0 or body <= 0:
        return None
    if body / rng < _LONG_BR:
        return None
    ref = sorted(abs(C[j] - O[j]) for j in range(i - _REF_N, i))
    med = ref[len(ref) // 2] if ref else 0.0
    if med <= 0 or body < _LONG_X * med:
        return None
    vs = [V[j] for j in range(i - _REF_N, i) if V[j] > 0]
    vx = (V[i] / (sum(vs) / len(vs))) if vs else 0.0
    # FADE = gegen die Körperrichtung
    return (-1 if C[i] > O[i] else 1), vx


def lauf(O, H, L, C, V, A, SP, lo, hi, vol_x, am_extrem, warte=6):
    """Sequentiell, EIN Positions-Slot (wie live). `am_extrem=False` → Market am
    Bar-Close (alte Annahme); True → ruhende Limit-Order am Kerzen-Extrem."""
    Rs = []
    frei_ab = lo
    verpasst = 0
    for i in range(max(lo, _REF_N + 2), hi - 2):
        if i < frei_ab:
            continue
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        k = klimax(O, H, L, C, V, i)
        if not k:
            continue
        d, vx = k
        if vol_x > 0 and vx < vol_x:
            continue
        kosten = (SP[i] if SP[i] > 0 else 0.0225) / atr
        if not am_extrem:
            entry, j0 = C[i], i + 1
        else:
            # SHORT-Fade → Sell-Limit am HOCH; LONG-Fade → Buy-Limit am TIEF.
            # Fill nur, wenn der Kurs das Extrem in `warte` Bars nochmal erreicht.
            limit = H[i] if d < 0 else L[i]
            j0 = None
            for j in range(i + 1, min(i + 1 + warte, len(C) - 1)):
                if (H[j] >= limit) if d < 0 else (L[j] <= limit):
                    j0 = j + 1
                    break
            if j0 is None:
                verpasst += 1
                continue
            entry = limit
        r, xb = simulate(entry, d, atr, H, L, C, j0, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        Rs.append(r - kosten)
        frei_ab = xb + 1
    return Rs, verpasst


def kenn(rs):
    if not rs:
        return None
    n = len(rs); s = sum(rs)
    g = sum(x for x in rs if x > 0); v = -sum(x for x in rs if x < 0)
    return {"n": n, "wr": 100 * sum(1 for x in rs if x > 0) / n,
            "oer": s / n, "sr": s, "pf": (g / v) if v > 0 else float("inf"),
            "worst": min(rs)}


def zeile(lbl, k1, k2, extra=""):
    def f(k):
        if not k:
            return f"{'—':>36}"
        return (f"{k['n']:>5} {k['wr']:>5.1f}% {k['oer']:>+7.3f} "
                f"{k['pf']:>5.2f} {k['sr']:>+8.1f}")
    gut = bool(k1 and k2 and k1["oer"] > 0 and k2["oer"] > 0
               and k1["pf"] > 1 and k2["pf"] > 1 and k1["n"] >= 100 and k2["n"] >= 100)
    print(f"  {lbl:<30}{f(k1)}  |{f(k2)}  {extra}"
          + ("  <<< (a)+(d) ok" if gut else ""))
    return gut


def main():
    n_bars = 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)
    bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n_bars)
    point = mt5.symbol_info(sym).point
    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]
    SP = [float(b["spread"]) * point for b in bars]
    A = _atr_series(H, L, C)
    N = len(C); mid = N // 2

    print("=" * 104)
    print(f"  KLIMAX-KERZEN-FADE — Neumessung · {sym} M5, {N} Bars · 2 Halbjahre")
    print(f"  sequentiell (EIN Slot) · kanonischer Exit (Trail {LIVE.mult}) · Echtkosten")
    print("  ⚠ Klasse bereits 2× verworfen — hier wird NUR der Einstiegspreis neu geprüft.")
    print("=" * 104)
    print(f"  {'Variante':<30}{'H1  n    WR      OeR    PF     SigmaR':>36}"
          f"  |{'H2  n    WR      OeR    PF     SigmaR':>36}")

    erg = {}
    for am_extrem, titel in ((False, "A) MARKET am Bar-Close (alte Annahme)"),
                             (True, "B) LIMIT am Kerzen-Extrem (6 Bars gueltig)")):
        print("  " + "-" * 102)
        print(f"  {titel}")
        for vx in _VOL:
            lbl = ("   KONTROLLE (ohne Volumen)" if vx == 0
                   else f"   Volumen >= {vx:.1f}x")
            r1, v1 = lauf(O, H, L, C, V, A, SP, 0, mid, vx, am_extrem)
            r2, v2 = lauf(O, H, L, C, V, A, SP, mid, N, vx, am_extrem)
            erg[(am_extrem, vx)] = (kenn(r1), kenn(r2))
            zeile(lbl, kenn(r1), kenn(r2),
                  extra=(f"verpasst {v1}/{v2}" if am_extrem else ""))

    # Wartefenster der Limit-Order variieren (Robustheit)
    print("  " + "-" * 102)
    print("  C) LIMIT — Gueltigkeitsdauer variiert (Volumen >= 2,0x)")
    for wt in _WARTE:
        r1, v1 = lauf(O, H, L, C, V, A, SP, 0, mid, 2.0, True, wt)
        r2, v2 = lauf(O, H, L, C, V, A, SP, mid, N, 2.0, True, wt)
        zeile(f"   {wt} Bars gueltig", kenn(r1), kenn(r2),
              extra=f"verpasst {v1}/{v2}")

    # ── Urteil nach der VORAB fixierten Regel ─────────────────────────────
    print()
    print("=" * 104)
    print("  URTEIL (Regel VOR dem Lauf fixiert)")
    print("=" * 104)
    for am_extrem, titel in ((False, "A) Market"), (True, "B) Limit am Extrem")):
        ko1, ko2 = erg[(am_extrem, 0.0)]
        treffer = []
        for idx, vx in enumerate([v for v in _VOL if v > 0]):
            k1, k2 = erg[(am_extrem, vx)]
            a = bool(k1 and k2 and k1["oer"] > 0 and k2["oer"] > 0
                     and k1["pf"] > 1 and k2["pf"] > 1)
            d_ = bool(k1 and k2 and k1["n"] >= 100 and k2["n"] >= 100)
            b = bool(k1 and k2 and ko1 and ko2
                     and k1["oer"] > ko1["oer"] and k2["oer"] > ko2["oer"])
            nb = [v for v in _VOL if v > 0 and abs(v - vx) <= 0.51 and v != vx]
            c = all((erg[(am_extrem, m)][0] or {}).get("oer", -1) > 0
                    and (erg[(am_extrem, m)][1] or {}).get("oer", -1) > 0 for m in nb)
            if a and b and c and d_:
                treffer.append(vx)
            print(f"    {titel:<22} Vol>={vx:.1f}x   "
                  f"(a){'ok ' if a else 'nein'} (b){'ok ' if b else 'nein'} "
                  f"(c){'ok ' if c else 'nein'} (d){'ok ' if d_ else 'nein'}")
        if treffer:
            print(f"    => {titel}: BESTEHT bei {treffer}")
        else:
            print(f"    => {titel}: faellt durch")
    print()
    print("  ⚠ Auch ein Bestehen waere KEIN Einbau-Auftrag: die Klasse ist 2× verworfen,")
    print("    ein dritter Lauf ist ein Mehrfachvergleich. Erst live-klein beobachten.")
    print()


if __name__ == "__main__":
    main()