Axel HocksandClaude Opus 4.8
29f53c45c1
User-Frage "System zur besseren Vorhersage - Liquiditaetsnachfragelinien?".
Getestet, weil es die einzige Klasse mit ANDERER Datenquelle war (Volumen = wo
wurde gehandelt, statt Pivots = wo war ein Docht). SMC/Order Blocks selbst sind
strukturell Pivot-Zonen = die schon 5x verworfene Klasse.
Aufbau: feste Preis-Bins 0.05$, rollierendes 24h-Fenster, tick_volume ueber
[low,high] verteilt, Value Area 70%; 80k M5-Bars, 2 Halbjahre, Live-Exit+Echtkosten.
Ergebnis - nichts traegt, die Kern-Idee am eindeutigsten:
A) Reversion am POC ("Kurs prallt an der Liquiditaetslinie ab")
H1 -0.178 / H2 -0.121 (PF 0.79/0.84) -> in BEIDEN Haelften negativ
A) Reversion am HVN H1 -0.067 / H2 +0.044 -> kippt
B) Momentum durch LVN H1 -0.031 / H2 +0.003 -> Breakeven-Rauschen
Informations-Test (Forward-Return 2h, relativ zur mid-Baseline): kein Bucket
weicht konsistent ab. Tiefere Einsicht: die Volume-Profile-Position ist ein
TREND-PROXY ("ausserhalb Value Area" = Ausbruch laeuft = folgt dem Regime),
keine unabhaengige Information -> erbt die Regime-Anfaelligkeit von Momentum.
15. verworfener Signal-Eingriff. CLAUDE.md aktualisiert.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
243 lines
11 KiB
Python
243 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""backtest_volume_profile.py — Volume Profile / echte Liquiditätszonen (B2,
|
||
User-Frage 2026-07-30 „System zur besseren Vorhersage — Liquiditätslinien?").
|
||
|
||
WARUM diese Klasse überhaupt getestet wird (die anderen sind durch): Order Blocks /
|
||
SMC-„Liquiditätslinien" sind strukturell Pivot-/Swing-Zonen — dieselbe Klasse, die
|
||
hier schon 5× durchfiel (`backtest_structure.py`, `backtest_doubletop.py`,
|
||
`backtest_patterns.py`, P(break)-Entry, `backtest_gaps.py`). Das Volume Profile ist
|
||
eine ANDERE Datenquelle: nicht „wo war ein Docht", sondern **wo wurde tatsächlich
|
||
gehandelt** (Akzeptanz). Damit ist es legitim testbar statt Runde 6.
|
||
|
||
Begriffe:
|
||
· POC (Point of Control) = Preis-Bin mit dem MEISTEN Volumen im Fenster
|
||
· HVN (High Volume Node) = viel Volumen → „Akzeptanz", Kurs verweilt/prallt ab
|
||
· LVN (Low Volume Node) = wenig Volumen → „Ablehnung", Kurs läuft schnell durch
|
||
· VA (Value Area) = engster Preisbereich, der ~70 % des Volumens enthält
|
||
|
||
Aufbau: feste globale Preis-Bins (`_BIN`) → rollierendes Fenster (`_WIN` Bars),
|
||
Bar-Volumen gleichmäßig über [low, high] verteilt (Standard-Approximation).
|
||
Volumen = `tick_volume` (CFD liefert kein real_volume — bei MT5-CFDs Standard-Proxy,
|
||
korreliert gut mit echtem Volumen).
|
||
|
||
TEIL 1 — INFORMATION (hat die Volumen-Position überhaupt Vorhersagewert?):
|
||
Forward-Return (2 h, in ATR) gebucketet nach Kursposition im Profil. Kein Exit,
|
||
keine Kosten → isoliert die reine Information. Trägt nur, wenn ein Bucket in
|
||
BEIDEN Hälften konsistent abweicht.
|
||
|
||
TEIL 2 — HANDELBAR (sequenzielle 1-Positions-Sim, Live-Exit, Echtkosten):
|
||
(A) Mean-Reversion am POC/HVN: Kurs erreicht die Zone → Gegenrichtung
|
||
(= die „Liquiditätslinie hält"-Idee)
|
||
(B) Momentum durch LVN: Kurs betritt eine Leerzone → in Bewegungsrichtung
|
||
(= „Liquidity Void, läuft schnell durch")
|
||
Maßstab: der validierte Squeeze (ØR +0,14…+0,23, PF>1). Beide Hälften positiv,
|
||
sonst raus.
|
||
|
||
Aufruf: python backtest_volume_profile.py [n_bars]
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
from collections import defaultdict
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
_BIN = 0.05 # Preis-Bin-Größe ($) — bei WTI ~85 = 0,06 %, ~0,25×ATR
|
||
_WIN = 288 # Profil-Fenster (M5-Bars) = 24 h
|
||
_FWD = 24 # Forward-Return-Horizont (Bars) = 2 h
|
||
_MAXH = 288 # Sim-Horizont
|
||
_ATRMIN = 0.06
|
||
_SL_ATR, _TRAIL, _TRAIL_ON, _BE_ON, _TIMESTOP = 2.0, 1.5, 0.3, 1.3, 24
|
||
_COOL = 12 # Cooldown zwischen Trades (Bars)
|
||
_VA_FRAC = 0.70 # Value Area = 70 % des Volumens
|
||
|
||
|
||
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] * len(C); run = 0.0
|
||
for i in range(1, len(C)):
|
||
run += tr[i]
|
||
if i > p: run -= tr[i - p]
|
||
out[i] = run / min(i, p)
|
||
return out
|
||
|
||
|
||
def _bins_of(lo, hi):
|
||
"""Bin-Indizes, die eine Bar-Range überdeckt (mind. 1)."""
|
||
a, b = int(lo / _BIN), int(hi / _BIN)
|
||
return range(a, b + 1)
|
||
|
||
|
||
class Profile:
|
||
"""Rollierendes Volume Profile über feste Preis-Bins (effizient: add/remove)."""
|
||
def __init__(self):
|
||
self.vol = defaultdict(float)
|
||
|
||
def add(self, lo, hi, v):
|
||
bs = _bins_of(lo, hi); share = v / max(1, len(bs))
|
||
for b in bs: self.vol[b] += share
|
||
|
||
def remove(self, lo, hi, v):
|
||
bs = _bins_of(lo, hi); share = v / max(1, len(bs))
|
||
for b in bs:
|
||
self.vol[b] -= share
|
||
if self.vol[b] <= 1e-9: del self.vol[b]
|
||
|
||
def stats(self):
|
||
"""(poc_bin, total, sorted_bins_desc) — None wenn leer."""
|
||
if not self.vol: return None
|
||
items = sorted(self.vol.items(), key=lambda kv: -kv[1])
|
||
return items[0][0], sum(self.vol.values()), items
|
||
|
||
def classify(self, price):
|
||
"""Wo liegt `price` im Profil? → (label, rel_vol, in_va)
|
||
rel_vol = Volumen des Preis-Bins / POC-Volumen (1,0 = am POC)."""
|
||
s = self.stats()
|
||
if not s: return None, 0.0, False
|
||
poc, total, items = s
|
||
pv = items[0][1]
|
||
b = int(price / _BIN)
|
||
v = self.vol.get(b, 0.0)
|
||
rel = v / pv if pv > 0 else 0.0
|
||
# Value Area: Bins absteigend nach Volumen aufsummieren bis _VA_FRAC
|
||
acc = 0.0; va = set()
|
||
for bb, vv in items:
|
||
va.add(bb); acc += vv
|
||
if acc >= _VA_FRAC * total: break
|
||
in_va = b in va
|
||
if b == poc: lab = "POC"
|
||
elif rel >= 0.60: lab = "HVN"
|
||
elif rel <= 0.15: lab = "LVN"
|
||
else: lab = "mid"
|
||
return lab, rel, in_va
|
||
|
||
|
||
def _sim(entry, d, atr, H, L, C, j0):
|
||
eff = entry - d * _SL_ATR * atr; hw = entry; started = False
|
||
end = min(j0 + _MAXH, len(C) - 1); exit_px = C[end]
|
||
for j in range(j0, end + 1):
|
||
if (L[j] <= eff) if d > 0 else (H[j] >= eff): exit_px = eff; break
|
||
hw = max(hw, H[j]) if d > 0 else min(hw, L[j])
|
||
prof = (C[j] - entry) * d
|
||
if prof >= _TRAIL_ON * atr:
|
||
started = True
|
||
cand = hw - d * _TRAIL * atr
|
||
if prof >= _BE_ON * atr:
|
||
cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||
eff = max(eff, cand) if d > 0 else min(eff, cand)
|
||
if not started and (j - j0) >= _TIMESTOP: exit_px = C[j]; break
|
||
return (exit_px - entry) * d / atr
|
||
|
||
|
||
def _rep(name, Rs, kind="R"):
|
||
if not Rs or len(Rs) < 5:
|
||
print(f" {name:<30} n={len(Rs):>4} (zu wenige)"); return None
|
||
n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
|
||
up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
|
||
pf = up / dn if dn > 0 else 9.99
|
||
print(f" {name:<30} n={n:>5} pos={100*w/n:>3.0f}% Ø{kind}={s/n:+.3f} "
|
||
f"PF={pf:.2f} Σ={s:+.0f}")
|
||
return s / n
|
||
|
||
|
||
def main():
|
||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||
mt5.initialize()
|
||
sym = None
|
||
for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
||
if mt5.symbol_info(c): sym = c; break
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
|
||
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]; V = [float(b["tick_volume"]) 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("=" * 92)
|
||
print(f" VOLUME PROFILE / Liquiditätszonen — {sym} M5 ({N} Bars)")
|
||
print(f" Bin={_BIN}$ · Fenster={_WIN} Bars (24h) · Volumen=tick_volume · VA={_VA_FRAC:.0%}")
|
||
print("=" * 92)
|
||
|
||
# ── Profil einmal durchlaufen und je Bar die Klassifikation cachen ──────────
|
||
prof = Profile()
|
||
lab_at = [None] * N; rel_at = [0.0] * N; poc_at = [None] * N; inva_at = [False] * N
|
||
for i in range(N):
|
||
prof.add(L[i], H[i], V[i])
|
||
if i >= _WIN: prof.remove(L[i - _WIN], H[i - _WIN], V[i - _WIN])
|
||
if i >= _WIN:
|
||
lab, rel, in_va = prof.classify(C[i])
|
||
lab_at[i] = lab; rel_at[i] = rel; inva_at[i] = in_va
|
||
s = prof.stats()
|
||
poc_at[i] = (s[0] + 0.5) * _BIN if s else None
|
||
|
||
# ── TEIL 1: INFORMATION (Forward-Return je Bucket, ohne Exit/Kosten) ────────
|
||
print("\nTEIL 1 — INFORMATION: Forward-Return 2 h (in ATR) je Kursposition im Profil")
|
||
print(" (positiv/negativ = Richtungs-Drift; 'pos%' = Anteil Aufwärts)")
|
||
for label, lo, hi in (("H1 (alt)", _WIN, mid), ("H2 (neu)", mid, N - _FWD - 1)):
|
||
print(f"\n{label}:")
|
||
buckets = defaultdict(list)
|
||
for i in range(lo, hi):
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN or lab_at[i] is None: continue
|
||
fwd = (C[i + _FWD] - C[i]) / atr
|
||
buckets[lab_at[i]].append(fwd)
|
||
buckets["in VA" if inva_at[i] else "außerhalb VA"].append(fwd)
|
||
for k in ("POC", "HVN", "mid", "LVN", "in VA", "außerhalb VA"):
|
||
if k in buckets: _rep(k, buckets[k], kind="fwd")
|
||
|
||
# ── TEIL 2: HANDELBAR ──────────────────────────────────────────────────────
|
||
print("\n" + "=" * 92)
|
||
print("TEIL 2 — HANDELBAR (seq. Sim · Live-Exit · Echtkosten) · Maßstab Squeeze +0,14…+0,23")
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||
|
||
verdict = defaultdict(list)
|
||
for label, lo, hi in (("H1 (alt)", _WIN, mid), ("H2 (neu)", mid, N)):
|
||
print(f"\n{label}:")
|
||
# (A) Mean-Reversion am POC/HVN — Kurs erreicht die Zone von außen
|
||
for zone in ("POC", "HVN"):
|
||
Rs = []; i = max(lo, _WIN + 2)
|
||
while i < min(hi, N - _MAXH - 2):
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN or lab_at[i] is None: i += 1; continue
|
||
# frischer Eintritt in die Zone (vorher NICHT drin)
|
||
if lab_at[i] == zone and lab_at[i - 1] != zone:
|
||
# Gegenrichtung zur Anlaufbewegung (prallt ab)
|
||
d = -1 if C[i] > C[i - 3] else 1
|
||
Rs.append(_sim(C[i], d, atr, H, L, C, i + 1) - cost(i, atr))
|
||
i += _COOL; continue
|
||
i += 1
|
||
verdict[f"A {zone}-Reversion"].append(_rep(f"A) Reversion am {zone}", Rs))
|
||
# (B) Momentum durch LVN — Kurs betritt Leerzone, läuft weiter
|
||
Rs = []; i = max(lo, _WIN + 2)
|
||
while i < min(hi, N - _MAXH - 2):
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN or lab_at[i] is None: i += 1; continue
|
||
if lab_at[i] == "LVN" and lab_at[i - 1] != "LVN":
|
||
d = 1 if C[i] > C[i - 3] else -1 # in Bewegungsrichtung
|
||
Rs.append(_sim(C[i], d, atr, H, L, C, i + 1) - cost(i, atr))
|
||
i += _COOL; continue
|
||
i += 1
|
||
verdict["B LVN-Momentum"].append(_rep("B) Momentum durch LVN", Rs))
|
||
|
||
print("\n" + "=" * 92)
|
||
print(" URTEIL (trägt nur, wenn ØR in BEIDEN Hälften > 0 UND >= +0,14 = Squeeze-Band):")
|
||
any_ok = False
|
||
for k, vals in verdict.items():
|
||
if len(vals) != 2 or any(v is None for v in vals):
|
||
print(f" {k:<28} unvollständig"); continue
|
||
ok = all(v > 0 for v in vals); band = all(v >= 0.14 for v in vals)
|
||
mark = "✅ TRÄGT" if (ok and band) else ("⚠ positiv, unter Band" if ok else "❌ fällt durch")
|
||
if ok and band: any_ok = True
|
||
print(f" {k:<28} H1 {vals[0]:+.3f} · H2 {vals[1]:+.3f} {mark}")
|
||
print("\n " + ("→ Kandidat gefunden: Parameter-Robustheit prüfen, dann klein live (B4)."
|
||
if any_ok else
|
||
"→ KEINE Variante trägt beidhälftig im Squeeze-Band → Volume Profile "
|
||
"bleibt Kontext, kein Signal."))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|