#!/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()