AH-Oil-Trader/backtest_verworfene_v2.py

#!/usr/bin/env python3
"""Die verworfenen Signale — neu gemessen auf der EINEN Achse, die heute neu ist.

ANLASS (2026-08-06, User: „backteste alle verworfenen Signale").

⚠ WAS DAS HIER NICHT IST: ein erneuter Durchlauf aller Ablehnungen. Der lief am
2026-08-04 (`run_all_backtests.py`, 14 Klassen, alle blieben verworfen) und waere
nur eine Wiederholung — mit dem Mehrfachvergleichs-Problem obendrauf.

NEU ist genau eine Achse: der **Einstiegspreis**. Sie hat am 05.08. Squeeze und
Auto-Signal von negativ auf positiv gedreht. ALLE alten Ablehnungen wurden mit
Markt-/Bar-Close-Einstieg gemessen — sie koennten also unter einer Annahme
gefallen sein, die nicht mehr gilt.

⚠⚠ ABER MIT DER HEUTIGEN KORREKTUR. `backtest_momentum_v2.py` hat gezeigt, dass
„Einstieg am Level" zweideutig ist:
    (B) Fill, WO der Bar-CLOSE das Level bricht   → Look-ahead, ~+0,3…0,45 R zu gut
    (C) Fill bei der ERSTEN BERUEHRUNG            → das, was live passiert
Gemessen wird deshalb C. B laeuft nur als Diagnose mit, um den Aufschlag sichtbar
zu machen.

⚠⚠ UND NUR DORT, WO DIE ACHSE UEBERHAUPT GREIFEN KANN. Der Vorteil entsteht an
Levels, die **STILLSTEHEN** (Squeeze-Box: +0,124/+0,292) — nicht an mitwandernden
(Momentum: −0,04…−0,09). Geprueft werden deshalb die verworfenen Setups mit einem
stehenden Level und einer AUSBRUCHS-Richtung:
    ORB (EU/US) · PDH/PDL · Asian Range · Equal Highs/Lows
⚠ Bewusst NICHT dabei, weil die Achse dort nichts aendern kann:
  · Momentum, Inter-Market, Chop-Gate, Meta-Labeling, RSI/MACD/ADX — kein Level.
  · Reversal/Bounce, Kerzen-Fade, POC-Reversion, Fibonacci, Kanalrand — REVERSIONS-
    Einstiege. Dort fuellt eine ruhende Order gerade dann, wenn man falsch liegt
    (gemessen `backtest_candle_fade2.py`: Limit am Extrem macht es SCHLECHTER).
  · Volume Profile — Level steht zwar, aber das Profil muesste je Bar neu gebinnt
    werden; Laufzeit unverhaeltnismaessig. Offen.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL (vor dem ersten Lauf):
  (1) in Variante C: ØR > 0 UND PF > 1 in BEIDEN Haelften
  (2) mind. 40 Trades je Haelfte
  (3) muss die KONTROLLE (Squeeze, +0,124/+0,292) nicht schlagen, aber selbst
      beidhaelftig tragen
  (4) Mehrfachvergleich wird ausgewiesen: bei ~10 Tests sind ~0,5 Zufallstreffer
      zu erwarten — ein einzelnes ✅ ist damit schwach.

Aufruf:  python backtest_verworfene_v2.py [bars]
"""
import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import MetaTrader5 as mt5

import core.wave_rec as wr
from backtest_cost_gate import _atr_series, kennzahlen
from core.exit_model import LIVE, simulate

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_ATRMIN = 0.06
_MIN_N = 40


def _berlin(ts):
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN))


# ─────────────────────────────────────────────────────────────────────────
#  Setups: jedes liefert STEHENDE Level als (ab_bar, bis_bar, level, richtung)
#  richtung: +1 = Ausbruch nach oben ist LONG, −1 = nach unten ist SHORT
# ─────────────────────────────────────────────────────────────────────────
def orb(H, L, T, A, start_h, start_m, minuten):
    """Opening Range Breakout: Box = erste `minuten` nach dem Open, gilt bis 22:00."""
    out, tag, box = [], None, None
    for i in range(len(H)):
        b = _berlin(T[i])
        d = b.date()
        if d != tag:
            tag, box = d, None
        beginn = b.replace(hour=start_h, minute=start_m, second=0, microsecond=0)
        ende = beginn + timedelta(minutes=minuten)
        if box is None and b >= ende:
            js = [j for j in range(max(0, i - 40), i)
                  if beginn <= _berlin(T[j]) < ende]
            if js:
                hi, lo = max(H[j] for j in js), min(L[j] for j in js)
                bis = i
                while bis < len(H) - 2 and _berlin(T[bis]).date() == d \
                        and _berlin(T[bis]).hour < 22:
                    bis += 1
                out.append((i, bis, hi, +1))
                out.append((i, bis, lo, -1))
            box = True
    return out


def tagesextrem(H, L, T):
    """PDH/PDL — Hoch/Tief des Vortages, gilt den ganzen Folgetag."""
    tage = {}
    for i in range(len(H)):
        d = _berlin(T[i]).date()
        r = tage.setdefault(d, [i, i, H[i], L[i]])
        r[1] = i; r[2] = max(r[2], H[i]); r[3] = min(r[3], L[i])
    folge = sorted(tage)
    out = []
    for k in range(1, len(folge)):
        _, _, hi, lo = tage[folge[k - 1]]
        a, b, _, _ = tage[folge[k]]
        out.append((a, b, hi, +1))
        out.append((a, b, lo, -1))
    return out


def asia(H, L, T):
    """Asian Range 00:00–08:00 Berlin, gilt 08:00 bis Tagesende."""
    out, tag, gesammelt = [], None, []
    for i in range(len(H)):
        b = _berlin(T[i])
        if b.date() != tag:
            tag, gesammelt, gesetzt = b.date(), [], False
        if b.hour < 8:
            gesammelt.append(i)
        elif gesammelt and not globals().get("_dummy"):
            if b.hour == 8 and b.minute < 5:
                hi = max(H[j] for j in gesammelt); lo = min(L[j] for j in gesammelt)
                bis = i
                while bis < len(H) - 2 and _berlin(T[bis]).date() == tag:
                    bis += 1
                out.append((i, bis, hi, +1))
                out.append((i, bis, lo, -1))
                gesammelt = []
    return out


def equal_hl(H, L, A, tol=0.15, gueltig=120):
    """Equal Highs/Lows aus M5-Pivots (k=3) — zwei Extrema innerhalb tol×ATR."""
    ph = [i for i in range(3, len(H) - 3)
          if H[i] == max(H[i - 3:i + 4])]
    pl = [i for i in range(3, len(L) - 3)
          if L[i] == min(L[i - 3:i + 4])]
    out = []
    for piv, arr, d in ((ph, H, +1), (pl, L, -1)):
        for a in range(1, len(piv)):
            i0, i1 = piv[a - 1], piv[a]
            if not (6 <= i1 - i0 <= 240):
                continue
            atr = A[i1]
            if not atr or atr < _ATRMIN:
                continue
            if abs(arr[i1] - arr[i0]) > tol * atr:
                continue
            lvl = (arr[i1] + arr[i0]) / 2.0
            out.append((i1 + 4, min(i1 + 4 + gueltig, len(H) - 2), lvl, d))
    return out


def squeeze_kontrolle(H, L, C, A):
    """Kontrolle: die Squeeze-Box als stehendes Level (muss +0,124/+0,292 treffen)."""
    out = []
    for i in range(wr._SQ_N + 2, len(C) - 2):
        atr = A[i - 1]
        if not atr or atr < _ATRMIN:
            continue
        hb = H[i - wr._SQ_N - 1:i - 1]; lb = L[i - wr._SQ_N - 1:i - 1]
        if not hb or not lb:
            continue
        hi, lo = max(hb), min(lb)
        if (hi - lo) / atr > wr._SQ_MULT:
            continue
        out.append((i, i, hi + wr._SQ_K * atr, +1))
        out.append((i, i, lo - wr._SQ_K * atr, -1))
    return out


# ─────────────────────────────────────────────────────────────────────────
def lauf(levels, H, L, C, SP, A, lo, hi, modus):
    """modus: 'markt' = Bar-Close · 'close' = Level, aber nur bei Close-Bruch
       · 'touch' = Level, Fill bei erster Beruehrung (REALITAET)."""
    ereignisse = []
    for ab, bis, lvl, d in levels:
        for i in range(max(ab, 1), min(bis + 1, len(C) - 2)):
            atr = A[i - 1]
            if not atr or atr < _ATRMIN:
                continue
            beruehrt = (H[i] >= lvl) if d > 0 else (L[i] <= lvl)
            if not beruehrt:
                continue
            gebrochen = (C[i] > lvl) if d > 0 else (C[i] < lvl)
            if modus != "touch" and not gebrochen:
                break          # Level verbraucht (Docht) — kein Signal
            entry = C[i] if modus == "markt" else lvl
            ereignisse.append((i, d, entry, atr, SP[i]))
            break              # ein Fill je Level
    ereignisse.sort()
    rs, frei_ab = [], lo
    for i, d, entry, atr, sp in ereignisse:
        if not (lo <= i < hi) or i < frei_ab:
            continue
        r, xb = simulate(entry, d, atr, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        rs.append(r - (sp if sp > 0 else 0.0225) / atr)
        frei_ab = xb + 1
    return rs


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    if not mt5.initialize():
        print("MT5 nicht erreichbar"); return 1
    sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
                if mt5.symbol_info(c)), None)
    b = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    H = [float(x["high"]) for x in b]; L = [float(x["low"]) for x in b]
    C = [float(x["close"]) for x in b]; T = [int(x["time"]) for x in b]
    SP = [float(x["spread"]) * point for x in b]
    A = _atr_series(H, L, C)
    m = len(C) // 2

    setups = [
        ("KONTROLLE Squeeze", squeeze_kontrolle(H, L, C, A)),
        ("ORB EU 15 min", orb(H, L, T, A, 10, 0, 15)),
        ("ORB EU 30 min", orb(H, L, T, A, 10, 0, 30)),
        ("ORB US 15 min", orb(H, L, T, A, 16, 30, 15)),
        ("ORB US 30 min", orb(H, L, T, A, 16, 30, 30)),
        ("PDH / PDL Bruch", tagesextrem(H, L, T)),
        ("Asian Range Bruch", asia(H, L, T)),
        ("Equal H/L Bruch", equal_hl(H, L, A)),
    ]

    print("=" * 104)
    print(f"  VERWORFENE SETUPS — neu auf der Ausfuehrungs-Achse ({sym}, {len(C)} M5-Bars)")
    print("=" * 104)
    print("  Geprueft wird NUR, was ein STEHENDES Level + Ausbruchsrichtung hat —")
    print("  dort kann die Achse ueberhaupt wirken (Squeeze ja, Momentum nein).")
    print("  Massgeblich ist Spalte C. B steht nur da, um den Look-ahead zu zeigen.")
    print()
    print(f"  {'Setup':<22} {'A) Markt H1/H2':^22} {'B) Close-Bruch H1/H2':^24} "
          f"{'C) BERUEHRUNG H1/H2':^24} n  Urteil")
    print("  " + "-" * 100)

    treffer = []
    for name, lv in setups:
        z = f"  {name:<22}"
        erg = {}
        for modus in ("markt", "close", "touch"):
            k1 = kennzahlen(lauf(lv, H, L, C, SP, A, 0, m, modus))
            k2 = kennzahlen(lauf(lv, H, L, C, SP, A, m, len(C), modus))
            erg[modus] = (k1, k2)
            z += (f" {k1['oer']:>+7.3f}/{k2['oer']:>+7.3f} " if k1 and k2
                  else f" {'—':^16} ")
        c1, c2 = erg["touch"]
        if not c1 or not c2:
            print(z + "  —"); continue
        ok = (c1["oer"] > 0 and c2["oer"] > 0 and c1["pf"] > 1 and c2["pf"] > 1
              and c1["n"] >= _MIN_N and c2["n"] >= _MIN_N)
        z += f" {c1['n']:>4}  {'✅' if ok else '❌'}"
        if c1["n"] < _MIN_N or c2["n"] < _MIN_N:
            z += " (duenn)"
        print(z)
        if ok and not name.startswith("KONTROLLE"):
            treffer.append(name)

    print()
    print("=" * 104)
    print(f"  ERGEBNIS: {len(treffer)} von {len(setups)-1} verworfenen Setups tragen "
          f"auch mit realistischem Fill")
    if treffer:
        print(f"  → {', '.join(treffer)}")
        print("  ⚠ Vor jeder Konsequenz: Nachbar-Parameter pruefen und gegen die")
        print("    Kontrolle halten. Ein einzelnes ✅ bei ~7 Tests ist erwartbar.")
    else:
        print("  → Die Ablehnungen bleiben bestehen. Die Ausfuehrungs-Achse rettet")
        print("    keines dieser Setups.")
    print()
    print("  ⚠ Spalte B gegen C gelesen = der Look-ahead-Aufschlag. Ist er gross,")
    print("    war das Setup nie so gut, wie ein Level-Backtest suggeriert.")
    return 0


if __name__ == "__main__":
    sys.exit(main())