backtest_volume_profile.py: Volume Profile / Liquiditaetslinien gemessen -> VERWORFEN
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>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
2996eddd7e
commit
29f53c45c1
@@ -0,0 +1,242 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user