AH-Oil-Trader/backtest_patterns_v2.py

#!/usr/bin/env python3
"""Die NEUEN Handbuch-Muster messen (User-Vorgabe 2026-07-31 „messe die neuen Muster").

Ergänzt wurden am selben Tag: **Dreifach-Top/-Boden · Flagge · Wimpel · Rechteck ·
Steigender/Fallender Keil**. Sie stehen in `core/patterns.py` auf `measured=False`
und tragen die Verdict-Stimme NICHT — dieses Skript prüft, ob sich das ändern darf.

⚠ ZWEI UNTERSCHIEDE ZU `backtest_patterns.py` (bewusst):
 1. **Der ECHTE Detektor wird benutzt** (`core.patterns._detect`), nicht eine
    Nachbildung. Das alte Skript hat die Bedingungen in `scan()` NACHGEBAUT — misst
    also möglicherweise etwas anderes als das, was live läuft. Nach den fünf
    Deployment-Drift-Fällen dieses Tages ist das keine akzeptable Grundlage mehr.
 2. **Exit aus `core/exit_model.py`** statt einer eigenen Kopie (das alte Skript war
    ein FÜNFTES Exit-Modell: Trail fest 1,5 · `_MAXH` 288 · kein Time-Stop).

KONTROLLGRUPPE (entscheidend): derselbe Einstieg/Exit auf einem **generischen
Swing-Bruch** — Bruch des letzten Swing-Hochs/-Tiefs ohne jedes Muster. Nur wenn ein
Muster die Kontrolle in BEIDEN Hälften schlägt, trägt das MUSTER etwas bei und nicht
bloss „Breakout + Trailing" (die Lehre aus dem 30.07.-Kontrolltest).

Zusätzlich: **Ziel-Trefferquote** — wie oft wird der Measured Move vor dem SL
erreicht? Das ist der ehrliche Muster-Check (alt: nur 13–38 %).
"""
import sys
from collections import defaultdict

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.patterns import _atr, _pivots, _detect, _PIVOT_K, _MEASURED

_BREAK_WIN = 48      # so viele Bars auf den Trigger-Bruch warten
_ATRMIN = 0.06
_NEW = ("triple_top", "triple_bottom", "flag", "pennant", "wedge_rise", "wedge_fall")


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


def scan(H, L, C, A, SP, lo, hi, step=3):
    """Läuft die History kausal ab und sammelt (typ, entry_bar, dir, trigger, target).

    Nutzt den ECHTEN Detektor auf dem jeweils bis `i` bekannten Fenster.
    """
    found = []
    seen = set()
    for i in range(max(lo, 120), min(hi, len(C) - LIVE.max_hold - 1), step):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        piv = _pivots(H[:i + 1], L[:i + 1], _PIVOT_K)
        if len(piv) < 3:
            continue
        try:
            pats = _detect(piv, C[i], atr, i)
        except Exception:
            continue
        for m in pats:
            if m.get("status") != "forming" or m.get("dir") not in ("bull", "bear"):
                continue
            key = (m["type"], round(m["trigger"], 2), m["dir"])
            if key in seen:
                continue
            seen.add(key)
            d = 1 if m["dir"] == "bull" else -1
            trig = m["trigger"]
            # auf den Trigger-Bruch warten
            for j in range(i + 1, min(i + 1 + _BREAK_WIN, len(C) - LIVE.max_hold - 1)):
                broke = (H[j] >= trig) if d > 0 else (L[j] <= trig)
                if broke:
                    found.append((m["type"], j, d, trig, m.get("target"), max(A[j] or 0, _ATRMIN)))
                    break
    return found


def control(H, L, C, A, lo, hi, step=3):
    """KONTROLLE: Bruch des letzten Swing-Hochs/-Tiefs, ohne jedes Muster."""
    out = []
    seen = set()
    for i in range(max(lo, 120), min(hi, len(C) - LIVE.max_hold - 1), step):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        piv = _pivots(H[:i + 1], L[:i + 1], _PIVOT_K)
        if len(piv) < 2:
            continue
        for d, kind in ((1, "H"), (-1, "L")):
            last = next((p for p in reversed(piv) if p[2] == kind), None)
            if not last:
                continue
            trig = last[1]
            key = (kind, round(trig, 2))
            if key in seen:
                continue
            seen.add(key)
            for j in range(i + 1, min(i + 1 + _BREAK_WIN, len(C) - LIVE.max_hold - 1)):
                broke = (H[j] >= trig) if d > 0 else (L[j] <= trig)
                if broke:
                    out.append(("control", j, d, trig, None, max(A[j] or 0, _ATRMIN)))
                    break
    return out


def evaluate(rows, H, L, C, SP):
    """→ {typ: [(R_netto, ziel_getroffen)]}"""
    res = defaultdict(list)
    for typ, j, d, trig, target, atr in rows:
        cost = (SP[j] if SP[j] > 0 else 0.0225) / atr
        R = simulate(trig, d, atr, H, L, C, j + 1, LIVE) - cost
        hit = None
        if target is not None:
            hit = False
            sl = trig - d * LIVE.sl_atr * atr
            for k in range(j + 1, min(j + 1 + LIVE.max_hold, len(C))):
                if (L[k] <= sl) if d > 0 else (H[k] >= sl):
                    break
                if (H[k] >= target) if d > 0 else (L[k] <= target):
                    hit = True; break
        res[typ].append((R, hit))
    return res


def rep(name, rows):
    if not rows or len(rows) < 8:
        print(f"    {name:<24} n={len(rows):<4} — zu wenige")
        return
    Rs = [r for r, _ in rows]
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    g = sum(x for x in Rs if x > 0); l = -sum(x for x in Rs if x < 0)
    hits = [h for _, h in rows if h is not None]
    zt = f"{100*sum(hits)/len(hits):>3.0f}%" if hits else "  —"
    print(f"    {name:<24} n={n:<4} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} "
          f"PF={(g/l if l > 0 else 9.99):>4.2f} ΣR={s:>+6.0f} Ziel={zt}")


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 120000
    mt5.initialize()
    sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(c)), None)
    bars = None
    for req in (n, 100000, 80000, 60000, 40000):
        bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, req)
        if bars is not None and len(bars) > 3000:
            break
    si = mt5.symbol_info(sym); point = si.point; mt5.shutdown()
    H = [float(b["high"]) for b in bars]; L = [float(b["low"]) for b in bars]
    C = [float(b["close"]) 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("=" * 96)
    print(f"  NEUE HANDBUCH-MUSTER — {sym} M30 ({N} Bars), Echtkosten")
    print(f"  Detektor: core/patterns._detect (der ECHTE) · Exit: core/exit_model.py "
          f"(Trail {LIVE.mult})")
    print("  Urteil: nur wenn ein Typ die KONTROLLE (generischer Swing-Bruch) in BEIDEN")
    print("  Hälften schlägt, trägt das MUSTER etwas bei — sonst ist es Breakout+Trailing.")
    print("=" * 96)

    label = {"triple_top": "Dreifach-Top", "triple_bottom": "Dreifach-Boden",
             "flag": "Flagge", "pennant": "Wimpel",
             "wedge_rise": "Steigender Keil", "wedge_fall": "Fallender Keil"}
    for lbl, a, b in (("H1 (alt)", _PIVOT_K, mid), ("H2 (neu)", mid, N - LIVE.max_hold - 1)):
        print(f"\n{lbl}:")
        res = evaluate(scan(H, L, C, A, SP, a, b), H, L, C, SP)
        ctl = evaluate(control(H, L, C, A, a, b), H, L, C, SP)
        rep("KONTROLLE Swing-Bruch", ctl["control"])
        print("    " + "-" * 74)
        for t in _NEW:
            rep(label[t], res.get(t, []))
        print("    " + "-" * 74)
        for t in sorted(_MEASURED):
            if res.get(t):
                rep(f"(alt) {t}", res[t])


if __name__ == "__main__":
    main()