AH-Oil-Trader/backtest_vwap_m15.py

#!/usr/bin/env python3
"""VWAP-Bänder auf M15 — taugen sie als LEVEL für ein Daytrading-Modul?

ANLASS (2026-08-08, User-Wunsch nach einem M15-Modul auf Profi-Methodik). Vom
eingereichten Werkzeugkasten ist fast alles in diesem Projekt schon gemessen und
durchgefallen — Volume Profile (POC-Reversion beidhälftig negativ), Liquidity
Sweeps (alle 9 Kombis), Market Structure (24/24 Zellen), ORB (alle 4 Varianten
mit realer Fill-Annahme), RSI/MACD, Inter-Market. Footprint/CVD/DOM gibt es hier
gar nicht (Market-Maker-Broker ohne Orderbuch).
**VWAP ist die einzige Zutat aus der Liste, die nie gemessen wurde** —
`core/analysis/indicators.calc_vwap` existiert, taucht in CLAUDE.md aber nirgends
als Signal auf.

⚠⚠ DIE VORFRAGE ENTSCHEIDET ALLES — und sie steht deshalb ganz vorne:
**STEHT das Band still?** Der einzige im Projekt belegte Hebel ist die
Ausführung: eine ruhende Order an einem STEHENDEN Level (Squeeze +0,32 R
gegenüber Market; SIG +0,35/+0,32 am Level gegen −0,09/−0,11 am Markt). Und der
Gegenbeleg ist genauso klar: das Momentum-Level WANDERT jede Bar, deshalb fällt
es durch (+0,45 R Aufschlag war reines Look-ahead). Ein VWAP-Band bewegt sich
per Konstruktion mit jedem Tick. Wandert es so stark wie das Momentum-Level, ist
das Modul erledigt, bevor irgendein Ertrag gerechnet wird.

GEMESSEN WIRD, in dieser Reihenfolge:
  0) BEWEGLICHKEIT  — wie weit rückt das Band je Bar (in ×ATR)? Gegen zwei
     Referenzen: die Squeeze-Box (steht, trägt) und `C[i−N]` (wandert, trägt nicht).
  1) HALTEQUOTE     — prallt der Kurs am Band ab? ⚠ MIT KONTROLLGRUPPE: eine
     Zufallslinie gleicher Distanz. Die Fibonacci-Messung hat gezeigt, dass ein
     BELIEBIGES Level hier zu ~55 % hält — ohne Kontrolle misst man die Basisrate.
  2) ERTRAG         — Reversion (Limit am Band) und Ausbruch (Stop hinter dem
     Band), beide mit FILL BEI BERÜHRUNG (die ehrliche Annahme, s. 06.08.),
     kanonischer Exit, Echtkosten, 2 Halbjahre.

⚠⚠ ENTSCHEIDUNGSREGEL, VORAB FESTGELEGT:
  (a) Beweglichkeit ≤ Squeeze-Box-Niveau, sonst Abbruch — ein wanderndes Level
      ist als ruhende Order nicht handelbar.
  (b) Haltequote muss die Zufallskontrolle in BEIDEN Hälften schlagen.
  (c) ØR > 0 und PF > 1 in BEIDEN Hälften, und die Nachbar-k dürfen nicht kippen.
Sonst: kein Einbau. Bei 23 gescheiterten Signal-Eingriffen gilt hohe Beweislast.

Aufruf:  python backtest_vwap_m15.py [n_M15_bars]      (Default 26000 ≈ 9 Monate)
"""
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

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))     # Brokerzeit = UTC+3
_ATRMIN = 0.12
_BAENDER = (1.0, 1.5, 2.0, 2.5, 3.0)
_ZONE = 0.15        # „am Band" in ×ATR (wie `at_level` live)


def _berlin(ts: int) -> datetime:
    """⚠ MT5 liefert die BROKER-Wallclock, keinen echten UTC-Epoch. Genau hier
    ist `analyze_tageszeit.py` am 05.08. um drei Stunden verrutscht."""
    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 vwap_baender(T, H, L, C, V):
    """Session-verankerter VWAP + kumulierte Standardabweichung je M15-Bar.

    Anker = Berliner Kalendertag (Reset 00:00). ⚠ KAUSAL: an Bar i fliessen nur
    Bars bis EINSCHLIESSLICH i ein — das Band an i ist also aus i bekannt, und
    eine Order kann fruehestens ab i+1 liegen (so wird unten auch gehandelt).
    """
    vw = [None] * len(C)
    sd = [None] * len(C)
    cum_pv = cum_v = cum_p2v = 0.0
    tag = None
    for i in range(len(C)):
        d = _berlin(T[i]).date()
        if d != tag:
            tag = d
            cum_pv = cum_v = cum_p2v = 0.0
        tp = (H[i] + L[i] + C[i]) / 3.0
        v = float(V[i] or 1)
        cum_pv += tp * v
        cum_v += v
        cum_p2v += tp * tp * v
        if cum_v <= 0:
            continue
        m = cum_pv / cum_v
        var = max(0.0, cum_p2v / cum_v - m * m)
        vw[i] = m
        sd[i] = var ** 0.5
    return vw, sd


def beweglichkeit(vw, sd, AT, k):
    """Wie weit rueckt das obere Band je Bar, in ×ATR? Median + 90-Perzentil."""
    d = []
    for i in range(1, len(vw)):
        if not (vw[i] and vw[i - 1] and sd[i] is not None and sd[i - 1] is not None):
            continue
        a = AT[i]
        if not a or a <= 0:
            continue
        o1 = vw[i - 1] + k * sd[i - 1]
        o2 = vw[i] + k * sd[i]
        d.append(abs(o2 - o1) / max(a, _ATRMIN))
    if not d:
        return None
    d.sort()
    return d[len(d) // 2], d[int(0.9 * len(d))]


def halte_quote(H, L, C, AT, vw, sd, k, lo, hi, fenster=8, zufall=False):
    """Prallt der Kurs am Band ab? Beruehrung = Bar-Spanne trifft die Zone.
    Ausgang: erreicht er zuerst Band∓0,5×ATR (haelt) oder Band±0,5×ATR (bricht)?
    `zufall=True` = Kontrolllinie in derselben Distanz, aber ohne VWAP-Bezug."""
    import random
    rnd = random.Random(7)
    haelt = bricht = 0
    war = False
    for i in range(max(lo, 1), min(hi, len(C) - fenster - 1)):
        # ⚠⚠ BAND AUS i-1, NICHT AUS i. Der VWAP an i enthaelt Bar i selbst —
        # gegen die Spanne DERSELBEN Bar zu testen ist Look-ahead: ein Ausschlag
        # zieht das Band zu sich hin und erzeugt so seine eigene „Beruehrung".
        # Live liegt die Order zwangslaeufig auf dem Stand des VORBARS.
        if not vw[i - 1] or sd[i - 1] is None:
            war = False
            continue
        a = AT[i]
        if not a or a <= 0:
            war = False
            continue
        a = max(a, _ATRMIN)
        lvl = vw[i - 1] + k * sd[i - 1]
        if zufall:
            # gleiche Distanz zum Kurs, aber willkuerliche Lage
            dist = abs(lvl - C[i])
            lvl = C[i] + dist * rnd.choice((1, -1))
        oben = lvl > C[i]
        beruehrt = (L[i] <= lvl + _ZONE * a) and (H[i] >= lvl - _ZONE * a)
        if beruehrt and not war:
            durch = lvl + (0.5 * a if oben else -0.5 * a)
            zurueck = lvl - (0.5 * a if oben else -0.5 * a)
            aus = None
            for j in range(i + 1, i + 1 + fenster):
                if (H[j] >= durch) if oben else (L[j] <= durch):
                    aus = "break"; break
                if (L[j] <= zurueck) if oben else (H[j] >= zurueck):
                    aus = "halt"; break
            if aus == "break":
                bricht += 1
            else:
                haelt += 1          # Timeout zaehlt als Halt (wie im P(break)-Training)
        war = beruehrt
    n = haelt + bricht
    return (n, haelt / n if n else None)


def ertrag(H, L, C, AT, SP, vw, sd, k, lo, hi, art):
    """`art`: 'reversion' = Limit AM Band, Richtung zurueck zum VWAP ·
    'ausbruch' = Stop HINTER dem Band, Richtung weiter.
    ⚠ FILL BEI BERUEHRUNG — die einzige live erreichbare Annahme (s. 06.08.:
    der Close-Bruch-Fill ist Look-ahead und schenkt 0,3–0,85 R)."""
    Rs = []
    i = max(lo, 1)
    war = False
    while i < min(hi, len(C) - LIVE.max_hold - 2):
        # ⚠⚠ Band aus i-1 (s. `halte_quote`): das an i berechnete Band enthaelt
        # Bar i selbst und waere Look-ahead.
        if not vw[i - 1] or sd[i - 1] is None:
            war = False; i += 1; continue
        a = AT[i]
        if not a or a <= 0:
            war = False; i += 1; continue
        a = max(a, _ATRMIN)
        for seite in (1, -1):                    # oberes und unteres Band
            lvl = vw[i - 1] + seite * k * sd[i - 1]
            beruehrt = (L[i] <= lvl + _ZONE * a) and (H[i] >= lvl - _ZONE * a)
            if not beruehrt or war:
                continue
            d = -seite if art == "reversion" else seite
            kosten = (SP[i] if SP[i] > 0 else 0.0225) / a
            R, xb = simulate(lvl, d, a, H, L, C, i + 1, LIVE, ret_bar=True)
            Rs.append(R - kosten)
            i = xb
            break
        if vw[i - 1] and sd[i - 1] is not None:
            _o = vw[i - 1] + k * sd[i - 1]
            war = (L[i] <= _o + _ZONE * a) and (H[i] >= _o - _ZONE * a)
        else:
            war = False
        i += 1
    return Rs


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} WR={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 main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 26000
    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_M15, 0, n)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    if bars is None or len(bars) < 5000:
        print(f"zu wenige M15-Bars ({0 if bars is None else len(bars)})"); 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]
    V = [float(b["tick_volume"]) for b in bars]
    SP = [float(b["spread"]) * point for b in bars]
    AT = atr_series(H, L, C)
    vw, sd = vwap_baender(T, H, L, C, V)
    mid = len(C) // 2

    print("=" * 92)
    print(f"  VWAP-BÄNDER auf M15 — {sym}, {len(C)} Bars "
          f"({_berlin(T[0]):%d.%m.%Y} bis {_berlin(T[-1]):%d.%m.%Y})")
    print("=" * 92)

    # ── 0) DIE VORFRAGE ───────────────────────────────────────────────────
    print("\n  0) BEWEGLICHKEIT — steht das Band still? (Verschiebung je Bar in ×ATR)")
    print(f"     {'k':>5}{'Median':>10}{'90-Perzentil':>15}")
    for k in _BAENDER:
        b = beweglichkeit(vw, sd, AT, k)
        if b:
            print(f"     {k:>5.1f}{b[0]:>10.3f}{b[1]:>15.3f}")
    print("     Referenz: Squeeze-Box steht (0,000 zwischen den Ausbrüchen) und TRÄGT;")
    print("     das Momentum-Level `C[i−N]` wandert und TRÄGT NICHT.")

    # ── 1) HALTEQUOTE mit Kontrolle ───────────────────────────────────────
    print("\n  1) HALTEQUOTE am oberen Band — gegen eine Zufallslinie gleicher Distanz")
    print(f"     {'k':>5}{'H1 VWAP':>12}{'H1 Zufall':>12}{'H2 VWAP':>12}{'H2 Zufall':>12}")
    for k in _BAENDER:
        z = []
        for lo, hi in ((0, mid), (mid, len(C))):
            for zuf in (False, True):
                nn, q = halte_quote(H, L, C, AT, vw, sd, k, lo, hi, zufall=zuf)
                z.append(f"{100*q:.1f}% ({nn})" if q is not None else "—")
        print(f"     {k:>5.1f}" + "".join(f"{x:>12}" for x in z))

    # ── 2) ERTRAG ─────────────────────────────────────────────────────────
    for art in ("reversion", "ausbruch"):
        print(f"\n  2) ERTRAG — {art.upper()} (Fill bei Berührung, kanonischer Exit, Echtkosten)")
        for k in _BAENDER:
            r1 = ertrag(H, L, C, AT, SP, vw, sd, k, 0, mid, art)
            r2 = ertrag(H, L, C, AT, SP, vw, sd, k, mid, len(C), art)
            print(f"     k={k:<4} H1 {kz(r1)}")
            print(f"     {'':<7}H2 {kz(r2)}")


if __name__ == "__main__":
    main()