AH-Oil-Trader/backtest_candle_fade.py

#!/usr/bin/env python3
"""Fade der Klimax-Kerze: langer Körper + hohes Volumen GEGEN die Körperrichtung.

Folgetest zu `backtest_candles.py` (2026-07-31). Dort ergab die reine INFORMATIONS-
Messung (Forward-Return 60 min) den einzigen robusten Lead des Laufs:

    Langer Körper           Überschuss  -0,059 / -0,076   (beide Hälften)
    ... + Volumen >= 2x Ø   Überschuss  -0,117 / -0,204   (monoton verstärkt)

d. h. nach einer volumenstarken großen Kerze läuft der Kurs GEGEN sie — das Gegenteil
der Lehrbuch-Lesart („2× Volumen = echt"). Brutto +0,117/+0,204 beim Fade, real
kosten aber ~0,265×ATR → unter der Schwelle. OFFEN blieb, ob ein **Live-Exit** mehr
holt als das starre 60-min-Fenster (bei den Chartmustern machte genau das den
Unterschied zwischen negativ und +0,43).

DESHALB HIER: echter Trade-Sim mit dem LIVE-Exit (Initial-SL 2,0×ATR · Trail 1,5 ab
0,3 · Breakeven 1,3 · Lock ab 3,5 · Time-Stop 120 min) und ECHTKOSTEN (Bar-Spread/ATR).

⚠ ENTSCHEIDEND — die KONTROLLGRUPPE: „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 + Trailing funktioniert (der Kontrollfehler, der in
`backtest_patterns.py` unentdeckt blieb und den Muster-„Edge" als Exit-Artefakt
entlarvte, als er nachgeholt wurde).

Urteil „trägt" nur, wenn (a) in BEIDEN Hälften ØR netto > 0, (b) die Volumen-Variante
die Kontrolle in BEIDEN Hälften schlägt, (c) benachbarte Volumen-Schwellen nicht das
Vorzeichen kippen.
"""
import sys
from statistics import median

import MetaTrader5 as mt5

_MAXH = 200          # max. Haltedauer in Bars
_ATRMIN = 0.06       # Live-Floor
_VOL_N = 20          # Fenster für Ø-Volumen / Median-Body
_COOL = 12           # Cooldown nach einem Trade (keine Überlappung)
# Live-Exit
_SL = 2.0; _TRAIL = 1.5; _TRAIL_START = 0.3; _BE = 1.3
_LOCK_START = 3.5; _LOCK_SCALE = 0.6; _LOCK_MIN = 1.2
_TIMESTOP = 24       # 120 min = 24 M5-Bars, greift nur in Phase „Init"


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 sim(entry, d, atr, H, L, C, j0):
    """LIVE-Exit-Modell (identisch zu backtest_trailing_sr.py)."""
    sl = entry - d * _SL * atr
    hw = entry; rank = 0
    end = min(j0 + _MAXH, len(C) - 1)
    for j in range(j0, end + 1):
        hi, lo = H[j], L[j]
        if (lo <= sl) if d > 0 else (hi >= sl):
            return (sl - entry) * d / atr
        hw = max(hw, hi) if d > 0 else min(hw, lo)
        profit = (hw - entry) * d
        ph = 0 if profit < _TRAIL_START * atr else (1 if profit < _LOCK_START * atr else 2)
        if ph < rank:
            ph = rank
        rank = ph
        # Time-Stop: nach 120 min noch in Phase „Init" (nie 0,3×ATR im Plus)
        if rank == 0 and (j - j0) >= _TIMESTOP:
            return (C[j] - entry) * d / atr
        if ph == 0:
            continue
        m = _TRAIL if ph == 1 else max(_LOCK_MIN, _TRAIL * _LOCK_SCALE)
        cand = hw - d * m * atr
        if profit >= _BE * atr or ph == 2:
            cand = max(cand, entry) if d > 0 else min(cand, entry)
        sl = max(sl, cand) if d > 0 else min(sl, cand)
    return (C[end] - entry) * d / atr


def st(Rs):
    if not Rs:
        return "Trades=   0"
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
    pf = up / dn if dn > 0 else 9.99
    return (f"Trades={n:>5} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} PF={pf:>4.2f} "
            f"Worst={min(Rs):+.2f} ΣR={s:+.0f}")


def scan(O, H, L, C, V, A, SP, EF, ES, lo_i, hi_i, vol_x, body_x=1.5,
         br_min=0.60, only_with_trend=False):
    """vol_x = 0 → KONTROLLE (kein Volumen-Filter)."""
    Rs = []
    i = max(lo_i, _VOL_N + 2)
    while i < min(hi_i, len(C) - _MAXH - 2):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        rng = H[i] - L[i]
        body = abs(C[i] - O[i])
        if rng <= 0 or body / rng < br_min:
            i += 1; continue
        bref = median([abs(C[k] - O[k]) for k in range(i - _VOL_N, i)])
        if bref <= 0 or body < body_x * bref:
            i += 1; continue
        if vol_x > 0:
            vref = sum(V[i - _VOL_N:i]) / _VOL_N
            if vref <= 0 or V[i] < vol_x * vref:
                i += 1; continue
        d_body = 1 if C[i] > O[i] else -1
        if only_with_trend:
            trend = 1 if EF[i] > ES[i] else -1
            if d_body != trend:            # nur Klimax IM Trend
                i += 1; continue
        d = -d_body                        # FADE: gegen die Körperrichtung
        entry = C[i]                       # Einstieg am Close der Signal-Kerze
        cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
        Rs.append(sim(entry, d, atr, H, L, C, i + 1) - cost)
        i += _COOL
    return Rs


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)
    si = mt5.symbol_info(sym); point = si.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); EF = _ema(C, 12); ES = _ema(C, 50)
    N = len(C); mid = N // 2
    halves = [("H1", 100, mid), ("H2", mid, N)]

    print("=" * 104)
    print(f"  KLIMAX-FADE — {sym} M5 ({N} Bars) · Live-Exit (SL 2,0 · Trail 1,5 ab 0,3 · "
          f"BE 1,3 · Time-Stop 120') · Echtkosten")
    print("  Einstieg GEGEN einen langen Körper (Body/Range>=0,60, Body>=1,5x Median)")
    print("=" * 104)

    variants = [
        ("KONTROLLE: Fade ohne Volumen-Filter", dict(vol_x=0)),
        ("Volumen >= 1,5x Ø",                   dict(vol_x=1.5)),
        ("Volumen >= 2,0x Ø",                   dict(vol_x=2.0)),
        ("Volumen >= 2,5x Ø",                   dict(vol_x=2.5)),
        ("Volumen >= 3,0x Ø",                   dict(vol_x=3.0)),
        ("Vol >= 2,0x + nur IM Trend",          dict(vol_x=2.0, only_with_trend=True)),
        ("Vol >= 2,0x + Marubozu (BR>=0,90)",   dict(vol_x=2.0, br_min=0.90)),
    ]
    for name, kw in variants:
        print(f"\n{name}")
        for lbl, a, b in halves:
            Rs = scan(O, H, L, C, V, A, SP, EF, ES, a, b, **kw)
            print(f"   {lbl}  {st(Rs)}")


if __name__ == "__main__":
    main()