AH-Oil-Trader/backtest_squeeze_m15.py

#!/usr/bin/env python3
"""Trägt der Squeeze-Breakout auf M15? — Voraussetzung für das M15-Modul.

ANLASS (2026-08-08, User: „baue mir das M15-Empfehlungsmodul"). Der Squeeze ist
der EINZIGE beidhälftig validierte Einstieg des Projekts — gemessen auf **M5**
(ØR +0,14…+0,23) und auf **M1** (nicht ausgeschlossen, aber je Trade schlechter).
**M15 wurde nie gemessen.** Die Live-Parameter (Box 10, k 0,1, ≤2,5×ATR) sind auf
M5 kalibriert; sie ungeprüft auf eine andere Zeitebene zu übernehmen wäre genau
der Deployment-Drift, der hier achtmal dokumentiert ist.

⚠ WARUM M15 AUSSICHTSREICH IST — Kosten-Arithmetik, kein Gefühl: der Spread ist
bei WTI absolut ~konstant, die Bewegung skaliert mit der Zeitebene. Gemessen
(`analyze_scalping_costs.py`): Median Spread/ATR **M1 0,319 · M5 0,193 ·
M15 0,133**. M15 ist also die GÜNSTIGSTE der drei — dieselbe Achse, an der M1
gescheitert ist, nur in die andere Richtung.

AUFBAU:
  · Mechanik aus `core/squeeze_scan.py` (geteilter Code, kein Nachbau)
  · Exit = **kanonisch** (`core/exit_model.LIVE`, Trail 1,0 + Time-Stop + TP) —
    NICHT der `exit_legacy` des alten Squeeze-Backtests (Trail 1,5, kein
    Time-Stop, kein TP). Für ein NEUES Modul gilt der Live-Exit.
  · Kosten = echter Bar-Spread/ATR
  · ATR-Floor **relativ** (0,542 × Median-ATR, wie `backtest_squeeze_multi.py`) —
    ein absoluter M5-Floor wäre auf M15 ein Selektions-Artefakt (Lehre aus
    `analyze_tageszeit.py`)
  · 2 Halbjahre · Nachbar-Parameter (Box × k) · **M5-Kontrolle auf DEMSELBEN
    Fenster** — nur der interne Vergleich ist belastbar, nicht die Absolutwerte

⚠⚠ ENTSCHEIDUNGSREGEL, VORAB FESTGELEGT:
  (1) ØR > 0 UND PF > 1 in BEIDEN Hälften
  (2) alle Nachbar-Kombinationen (Box 10/15/20 × k 0,1/0,2) ebenfalls positiv —
      kippt ein Nachbar, ist es Rauschen (die Regel, an der ORB starb)
  (3) hält Einstiegs-Slippage bis +0,20×ATR aus (der live gemessene Bereich)
Sonst: NICHT bauen.

Aufruf:  python backtest_squeeze_m15.py [n_M15_bars]     (Default 26000)
"""
from __future__ import annotations

import statistics as st
import sys

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.squeeze_scan import atr_series, scan

_FLOOR_REL = 0.542      # wie backtest_squeeze_multi.py
_BOXEN = (10, 15, 20)
_KS = (0.1, 0.2)


def kanonischer_exit(entry, d, atr, H, L, C, j0):
    """Der LIVE-Exit — Trail 1,0, Lock-Phase, Time-Stop, Initial-TP.
    ⚠ Bewusst NICHT `squeeze_scan.exit_legacy`: das ist die eingefrorene alte
    Fassung (Trail 1,5, kein Time-Stop/TP), gegen die der B4-Monitor misst."""
    return simulate(entry, d, atr, H, L, C, j0, LIVE)


def hole(tf, n):
    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, tf, 0, n)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    if bars is None:
        return None, sym
    return ({"T": [int(b["time"]) 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],
             "SP": [float(b["spread"]) * point for b in bars]}, sym)


def kz(Rs):
    if not Rs:
        return "n=0"
    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); v = -sum(x for x in Rs if x < 0)
    return (f"n={n:>4} Treffer={100*w/n:>3.0f}% ØR={s/n:+.3f} "
            f"PF={(g/v if v > 0 else 9.99):>4.2f} ΣR={s:>+6.0f}")


def lauf(d, lo, hi, box, k, floor, slip=0.0):
    """Ein Durchlauf. `slip` = Einstiegs-Slippage in ×ATR (Härtetest)."""
    A = atr_series(d["H"], d["L"], d["C"])
    ex = kanonischer_exit
    if slip:
        def ex(entry, dd, atr, H, L, C, j0, _s=slip):
            return kanonischer_exit(entry + dd * _s * atr, dd, atr, H, L, C, j0)
    sq, alle = scan(d["H"], d["L"], d["C"], A, d["SP"], 1.0, lo, hi, 2.5,
                    box_n=box, k=k, atr_min=floor, max_hold=LIVE.max_hold,
                    exit_fn=ex)
    return sq, alle


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 26000
    m15, sym = hole(mt5.TIMEFRAME_M15, n)
    if not m15 or len(m15["C"]) < 5000:
        print("zu wenige M15-Bars"); return
    A15 = atr_series(m15["H"], m15["L"], m15["C"])
    med15 = st.median([a for a in A15 if a])
    floor15 = _FLOOR_REL * med15
    mid = len(m15["C"]) // 2

    print("=" * 96)
    print(f"  SQUEEZE auf M15 — {sym}, {len(m15['C'])} Bars")
    print(f"  Median-ATR {med15:.4f} $ → relativer Floor {floor15:.4f} "
          f"(M5-Referenz: 0,12 bei Median ~0,168)")
    print("  Exit = KANONISCH (Trail 1,0 + Time-Stop + TP), Kosten = echter Bar-Spread")
    print("=" * 96)

    sp_rel = st.median([s / max(a, floor15) for s, a in zip(m15["SP"], A15) if a and s > 0])
    print(f"\n  Kosten-Kontrolle: Median Spread/ATR auf M15 = {sp_rel:.3f}")
    print("    (erwartet ~0,133 laut analyze_scalping_costs.py; M5 0,193 · M1 0,319)")

    # ── Hauptlauf mit den Live-Parametern ────────────────────────────────
    print(f"\n  ── LIVE-PARAMETER (Box 10, k 0,1, ≤2,5×ATR) ──")
    for nm, lo, hi in (("H1", 40, mid), ("H2", mid, len(m15["C"]))):
        sq, alle = lauf(m15, lo, hi, 10, 0.1, floor15)
        print(f"    {nm}  Squeeze   {kz(sq)}")
        print(f"    {'':<4}Kontrolle {kz(alle)}   (beliebige Box — trägt die Kompression?)")

    # ── Nachbar-Robustheit ───────────────────────────────────────────────
    print(f"\n  ── NACHBAR-ROBUSTHEIT (kippt eine Kombi, ist es Rauschen) ──")
    print(f"    {'Box':>5}{'k':>6}{'H1 ØR':>10}{'H1 PF':>8}{'H2 ØR':>10}{'H2 PF':>8}")
    for box in _BOXEN:
        for k in _KS:
            z = []
            for lo, hi in ((40, mid), (mid, len(m15["C"]))):
                sq, _ = lauf(m15, lo, hi, box, k, floor15)
                if not sq:
                    z += ["—", "—"]; continue
                s = sum(sq) / len(sq)
                g = sum(x for x in sq if x > 0); v = -sum(x for x in sq if x < 0)
                z += [f"{s:+.3f}", f"{(g/v if v > 0 else 9.99):.2f}"]
            print(f"    {box:>5}{k:>6.1f}{z[0]:>10}{z[1]:>8}{z[2]:>10}{z[3]:>8}")

    # ── Härtetest: Einstiegs-Slippage ────────────────────────────────────
    print(f"\n  ── HÄRTETEST Einstiegs-Slippage (ruhende Stop-Order füllt zum Markt) ──")
    print(f"    {'Slippage':>10}{'H1 ØR':>10}{'H2 ØR':>10}")
    for slip in (0.0, 0.05, 0.10, 0.20, 0.30):
        z = []
        for lo, hi in ((40, mid), (mid, len(m15["C"]))):
            sq, _ = lauf(m15, lo, hi, 10, 0.1, floor15, slip=slip)
            z.append(f"{sum(sq)/len(sq):+.3f}" if sq else "—")
        print(f"    {slip:>9.2f}×{z[0]:>10}{z[1]:>10}")

    # ── M5-Kontrolle auf DEMSELBEN Fenster ───────────────────────────────
    m5, _ = hole(mt5.TIMEFRAME_M5, n * 3)
    if m5:
        von, bis = m15["T"][0], m15["T"][-1]
        idx = [i for i, t in enumerate(m5["T"]) if von <= t <= bis]
        if len(idx) > 5000:
            a5, b5 = idx[0], idx[-1]
            A5 = atr_series(m5["H"], m5["L"], m5["C"])
            f5 = _FLOOR_REL * st.median([x for x in A5 if x])
            m5m = (a5 + b5) // 2
            print(f"\n  ── M5-KONTROLLE, gleiches Fenster ({b5-a5} Bars) ──")
            for nm, lo, hi in (("H1", a5 + 40, m5m), ("H2", m5m, b5)):
                sq, _ = lauf(m5, lo, hi, 10, 0.1, f5)
                print(f"    {nm}  {kz(sq)}")


if __name__ == "__main__":
    main()