AH-Oil-Trader/backtest_squeeze_woche.py

#!/usr/bin/env python3
"""Haette der Auto-Squeeze in der letzten Woche Gewinn gemacht?

ANLASS (2026-08-11, User: „analysiere den Breakout-Auto-Trader, ob er zuverlaessig
funktioniert; simuliere ueber die letzte Woche, ob er Gewinn erzielt haette").

⚠⚠ WAS DIESE SIMULATION IST UND WAS NICHT. **Eine Woche ist KEIN Edge-Nachweis.**
Der Squeeze feuert 2–3×/Woche; das sind Stichproben von n≈2–6, und im Projekt sind
Einzelzeitraeume schon dreimal faelschlich zum Beleg erhoben worden (15-Minuten-
Regel 31.07., Nacht-Guard 05.08., das Veto im Wochentest 08.08.). Gemessen wird
hier die **MECHANIK**: feuert die Regel ueberhaupt, zu welchen Preisen, und was
haette der kanonische Exit daraus gemacht.

MODELLIERT WIRD DER LIVE-PFAD (Stand seit 05.08.):
  · echte Live-Funktion `WaveRecommender._squeeze_one` — kein Nachbau
  · **ruhende Stop-Order** an der Box-Grenze, solange `armed`; Fuellung bei
    BERUEHRUNG (nicht bei Bar-Schluss). Das ist der teurere und richtige Fall:
    (a) „Close bricht Level" ist Look-ahead und rund +0,32 R zu optimistisch
    (`backtest_squeeze_touchfill.py`).
  · Nacht-Sperre 3–6 Uhr Berlin (`auto_squeeze_night_hours`)
  · ATR-Floor, EIN Positions-Slot (sequentiell), Echtkosten aus dem Bar-Spread
  · **kanonischer Exit** `core.exit_model.LIVE` — ausdruecklich NICHT
    `squeeze_scan.exit_legacy` (Trail 1,5 · kein Lock · kein Time-Stop · kein TP),
    gegen den der B4-Monitor bis heute misst.

Aufruf:  python backtest_squeeze_woche.py [tage]        (Default 7)
"""
from __future__ import annotations

import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.wave_rec import WaveRecommender

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_NACHT = {3, 4, 5, 6}
_ATR_MIN = 0.06
_SLIPPAGE = (0.0, 0.05, 0.10, 0.20)     # ×ATR Aufschlag auf den Fill


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


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 lauf(T, H, L, C, SP, A, lo_i, slip=0.0, nacht=True):
    """Sequentiell EIN Slot, ruhende Stop-Order, Fuellung bei Beruehrung."""
    trades = []
    i = lo_i
    n = len(C)
    while i < n - LIVE.max_hold - 2:
        atr = A[i]
        if not atr or atr < _ATR_MIN:
            i += 1; continue
        if nacht and _berlin(T[i]).hour in _NACHT:
            i += 1; continue
        sq = WaveRecommender._squeeze_one(H[:i + 1], L[:i + 1], C[:i + 1], atr)
        if not sq or sq.get("box_hi") is None or sq.get("state") is None:
            i += 1; continue
        # ⚠ Die Box stammt aus den N bereits GESCHLOSSENEN Bars vor i — eine
        #   ruhende Order darf waehrend Bar i also legitim gefuellt werden.
        up = sq["box_hi"] + 0.1 * atr
        dn = sq["box_lo"] - 0.1 * atr
        traf_up = H[i] >= up
        traf_dn = L[i] <= dn
        if not (traf_up or traf_dn):
            i += 1; continue
        if traf_up and traf_dn:
            # Beide Seiten in EINEM Bar — aus OHLC nicht aufloesbar. Die naeher
            # am Open liegende Seite wird zuerst beruehrt worden sein.
            traf_up = abs(up - C[i - 1]) <= abs(C[i - 1] - dn)
            traf_dn = not traf_up
        d = 1 if traf_up else -1
        fill = (up + slip * atr) if traf_up else (dn - slip * atr)
        kost = (SP[i] if SP[i] > 0 else 0.0225) / atr
        R, xb = simulate(fill, d, atr, H, L, C, i + 1, LIVE, ret_bar=True)
        trades.append({"i": i, "t": T[i], "d": d, "fill": fill, "atr": atr,
                       "R": R - kost, "bars": xb - i,
                       "box_atr": sq.get("box_atr")})
        i = xb + 1                      # Slot war belegt
    return trades


def kz(tr, lbl):
    if not tr:
        return f"  {lbl:<34} keine Ausbrueche"
    n = len(tr); Rs = [x["R"] for x in tr]
    w = sum(1 for x in Rs if x > 0)
    g = sum(x for x in Rs if x > 0); v = -sum(x for x in Rs if x < 0)
    ow = g / w if w else 0.0
    ol = v / (n - w) if n - w else 0.0
    return (f"  {lbl:<34} n={n:>3}  Treffer {100*w/n:>3.0f} %  OeR {sum(Rs)/n:>+6.3f}  "
            f"PF {(g/v if v else 9.99):>4.2f}  Verh {(ow/ol if ol else 9.99):>4.2f}  "
            f"SumR {sum(Rs):>+6.2f}")


def main():
    tage = int(sys.argv[1]) if len(sys.argv) > 1 else 7
    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, 20000)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    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]
    A = atr_series(H, L, C)

    grenze = T[-1] - tage * 86400
    lo_i = next(i for i, t in enumerate(T) if t >= grenze)
    lo_i = max(lo_i, 60)

    print("=" * 100)
    print(f"  AUTO-SQUEEZE — Simulation der letzten {tage} Tage ({sym})")
    print(f"  {_berlin(T[lo_i]):%d.%m %H:%M} bis {_berlin(T[-1]):%d.%m %H:%M}   "
          f"{len(C)-lo_i} M5-Bars")
    print("  Live-Pfad: ruhende Stop-Order an der Box, Fuellung bei BERUEHRUNG,")
    print("  Nacht 3-6 gesperrt, EIN Slot, kanonischer Exit, Echtkosten.")
    print("=" * 100)

    basis = lauf(T, H, L, C, SP, A, lo_i)
    print("\n  ── Fill-Slippage (eine Stop-Order fuellt zum Markt, nicht zum Ideal) ──")
    for s in _SLIPPAGE:
        print(kz(lauf(T, H, L, C, SP, A, lo_i, slip=s), f"Slippage +{s:.2f}xATR"))

    print("\n  ── Gegenprobe: ohne Nacht-Sperre ──")
    print(kz(lauf(T, H, L, C, SP, A, lo_i, nacht=False), "alle Stunden"))

    print(f"\n  ── die einzelnen Ausbrueche (Slippage 0) ──")
    if not basis:
        print("    KEINER — die Regel hat in diesem Fenster nicht gefeuert.")
    for x in basis:
        print(f'    {_berlin(x["t"]):%d.%m %H:%M}  {"LONG " if x["d"]>0 else "SHORT"} '
              f'@ {x["fill"]:.3f}  Box {x["box_atr"]:.2f}xATR  '
              f'{x["bars"]*5:>4} min  R {x["R"]:>+6.2f}')

    if basis:
        # EUR-Naeherung: R x ATR x Lots x ~87,6 EUR/$/Lot (robust geschaetzt, s. CLAUDE.md)
        lots = 2.5
        eur = sum(x["R"] * x["atr"] for x in basis) * lots * 87.6
        print(f"\n  grob in EUR bei {lots} Lots: {eur:>+8.0f} EUR")
        print("  ⚠ Naeherung — die echte Lot-Groesse schwankt mit dem Konto.")

    print("\n  " + "-" * 96)
    print("  ⚠⚠ EINE WOCHE IST KEIN EDGE-NACHWEIS. Der Squeeze feuert 2-3x/Woche;")
    print("     n dieser Groesse ist Rauschen. Die 2-Stichproben-Regel gilt weiter.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())