Axel HocksandClaude Opus 4.8
919fe61dc3
1) Entry-Checkliste (User: "blende einen Hinweis ein wenn die Checkliste gegen
meine Eroeffnung ist"): engine._entry_checklist prueft 7 Punkte mit Live-Daten
-- Signal-Deckung, HTF-Trend M30+H1, Entry-Raum, News-Konflikt, Kosten,
Verlust-Serie heute (Mental-Check), Nacht-Stunde. Snapshot checklist{long,short},
Anzeige im Bestaetigungs-Dialog, blockt nicht. Kommt jetzt auch bei vorhandenem
Signal (alter Dialog nur bei WARTEN/Gegen-Signal).
Real beim Einbau: beide Richtungen "stop" (11 Verluste heute = 80% des Kontos).
2) MQL5 v1.14 (User: "blende die liquiditaetslinien aus, nur die Pfeile sollen
rechtsbuendig angezeigt werden"): LQ-Linien + Text-Labels raus, nur noch der
Pfeil auf dem Wand-Preis, rechtsbuendig am aktuellen Rand (in RepositionLabels
nachgefuehrt, bleibt beim Scrollen rechts). Details im Tooltip. Pinke
Liquiditaets-Trendlinie unveraendert. Kompiliert.
3) backtest_liquidity_sweep.py (SMC Equal Highs/Lows + Swing Liquidity, Multi-TF-
und Session-Achse): faellt durch -- Details folgen in der Doku.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
317 lines
16 KiB
Python
317 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""backtest_liquidity_sweep.py — Liquiditäts-Sweeps (SMC/ICT) messen (B2).
|
||
|
||
User-Vorgabe 2026-07-30 (mit Chart-Beispielen): die drei „wichtigsten Liquiditäts-
|
||
muster" für WTI in die Gesamtempfehlung aufnehmen —
|
||
A) **Equal Highs → Sell-Side-Sweep**: zwei ~gleiche Hochs bilden eine Stop-Zone;
|
||
der Kurs schießt KURZ darüber (Sweep), schließt zurück darunter → SHORT.
|
||
B) **Equal Lows → Buy-Side-Sweep**: Spiegelbild → LONG.
|
||
C) **Swing Liquidity**: markanter Swing-Punkt wird gesweept/retestet, dann Trade
|
||
in Richtung des höheren Timeframe-Trends (hier M30-EMA12/50 als HTF-Proxy).
|
||
|
||
⚠ WARUM das trotz 15 verworfener Signal-Ideen einen eigenen Test verdient: Der
|
||
**Sweep ist ein FEHLAUSBRUCH** (Level kurz durchstochen, Schluss zurück im Bereich) —
|
||
eine andere Mechanik als die bereits verworfenen Muster (`backtest_doubletop.py` misst
|
||
den Nackenlinien-BRUCH, nicht den Fehlausbruch) und als die Struktur-/Volume-Profile-
|
||
Klasse. Verwandt ist er mit dem validierten P(break)-Modell: ein Sweep ist genau der
|
||
Fall „Level wird berührt/überschritten, hält aber" — dort ist die Vorhersage bereits
|
||
belegt (AUC 0,72), allerdings nur als EXIT, nicht als Entry.
|
||
|
||
Methodik (wie alle Tests hier): Signal-TF M5, sequentielle 1-Positions-Sim, Live-Exit
|
||
(SL 2,0×ATR · Trail 1,5 ab 0,3 · Breakeven 1,3 · Time-Stop 120 min), Kosten = echter
|
||
Bar-Spread/ATR, 2 Halbjahre. Zusätzlich wird der SMC-eigene Exit geprüft
|
||
(Stop über/unter dem Sweep-Docht, Ziel = gegenüberliegende Liquidität), weil das
|
||
Setup laut Lehre damit steht und fällt.
|
||
|
||
Maßstab: der einzige validierte Auto-Entry (Squeeze) liefert ØR +0,14…+0,23 · PF>1
|
||
in BEIDEN Hälften. Alles darunter ist kein Kandidat für die Gesamtempfehlung.
|
||
|
||
Die vom User nachgereichten WTI-Praxisregeln sind als ACHSEN eingebaut, statt sie
|
||
zu glauben:
|
||
· **Multi-TF** („Liquidität auf 4H/1H identifizieren, Entry auf 15M/5M") → die
|
||
Level-TF ist variabel (M5 / M30 / H1). Passt zum eigenen Befund von 2026-07-30
|
||
(`backtest_level_tf.py`): M30-Level schlagen M5-Level deutlich.
|
||
· **Session-Timing** („Sweeps um 8:30–10:30 NY = 14:30–16:30 Berlin, oder 2–4 Uhr
|
||
asiatische Liquiditätsjagd") → Session-Filter als eigene Variante.
|
||
· **„Nicht jeder Sweep dreht — warte auf den Close zurück"** → ist die harte
|
||
Bedingung dieses Tests (Close muss im selben Bar zurück in der Zone sein).
|
||
· **DXY-Korrelation** → NICHT eingebaut: die Inter-Market-Klasse (Brent/DXY) wurde
|
||
2026-07-19 gemessen und verworfen (`backtest_intermarket.py`, kippt zwischen den
|
||
Hälften). Ohne neuen Beleg keine zweite Runde.
|
||
· **„Sweep-Wick 0,50–1,00 $, kleiner sizen"** → im Test implizit über den
|
||
SMC-Exit (Stop am Docht) abgebildet; Sizing ist hier neutral (R-Rechnung).
|
||
|
||
Aufruf: python backtest_liquidity_sweep.py [n_bars]
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
# ── Muster-Parameter (bewusst wenige, sonst Overfitting) ─────────────────────
|
||
_PIV_K = 3 # Swing-Pivot-Fenster
|
||
_EQ_TOL = 0.15 # „gleich" = beide Hochs/Tiefs innerhalb x×ATR
|
||
_EQ_MINGAP = 3 # min. Bars zwischen den beiden Pivots (sonst dieselbe Zacke)
|
||
_EQ_MAXGAP = 60 # max. Bars — danach ist die Zone kalt
|
||
_SWEEP_MIN = 0.05 # Sweep muss das Level um ≥ x×ATR überschießen (echter Stop-Grab)
|
||
_SWEEP_MAX = 1.00 # aber nicht mehr als x×ATR (sonst echter Ausbruch, kein Sweep)
|
||
_BACK_BARS = 3 # innerhalb so vieler Bars muss der Close zurück in die Zone
|
||
_LOOKBACK = 300 # Pivot-Suchfenster
|
||
_COOL = 12 # Cooldown zwischen Trades (Bars)
|
||
# Session-Fenster in BERLIN-Stunden (Broker ist UTC+3 → Berlin = Broker − 1 h).
|
||
# NY 8:30–10:30 ET ≈ 14:30–16:30 Berlin (EIA/Market-Open), Asia-Jagd 2–4 Uhr.
|
||
_SESS = {"NY": (14, 15, 16), "Asia": (2, 3, 4), "NY+Asia": (2, 3, 4, 14, 15, 16)}
|
||
_LEVEL_TFS = (("M5", 1), ("M30", 6), ("H1", 12)) # Level-Quelle (Multi-TF-Regel)
|
||
|
||
# ── Exit (Live-Modell, identisch zu den anderen Backtests) ───────────────────
|
||
_SL_ATR, _TRAIL, _TRAIL_ON, _BE_ON, _TIMESTOP, _MAXH = 2.0, 1.5, 0.3, 1.3, 24, 288
|
||
_ATRMIN = 0.06
|
||
|
||
|
||
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 _ema_series(v, p):
|
||
k = 2.0 / (p + 1); o = [v[0]]
|
||
for x in v[1:]:
|
||
o.append(x * k + o[-1] * (1 - k))
|
||
return o
|
||
|
||
|
||
def _sim_live(entry, d, atr, H, L, C, j0):
|
||
"""Live-Exit: SL 2×ATR, Trailing 1,5 ab 0,3, Breakeven 1,3, Time-Stop."""
|
||
eff = entry - d * _SL_ATR * atr; hw = entry; started = False
|
||
end = min(j0 + _MAXH, len(C) - 1); px = C[end]
|
||
for j in range(j0, end + 1):
|
||
if (L[j] <= eff) if d > 0 else (H[j] >= eff): 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: px = C[j]; break
|
||
return (px - entry) * d / atr
|
||
|
||
|
||
def _sim_smc(entry, d, atr, stop, target, H, L, C, j0):
|
||
"""SMC-Exit laut Lehre: Stop über/unter dem Sweep-Docht, Ziel = gegenüber-
|
||
liegende Liquidität. Ohne Ziel → Live-Exit-Fallback."""
|
||
if target is None:
|
||
return _sim_live(entry, d, atr, H, L, C, j0)
|
||
end = min(j0 + _MAXH, len(C) - 1); px = C[end]
|
||
for j in range(j0, end + 1):
|
||
hit_stop = (L[j] <= stop) if d > 0 else (H[j] >= stop)
|
||
hit_tgt = (H[j] >= target) if d > 0 else (L[j] <= target)
|
||
# konservativ: bei beidem im selben Bar zählt der Stop
|
||
if hit_stop: px = stop; break
|
||
if hit_tgt: px = target; break
|
||
return (px - entry) * d / atr
|
||
|
||
|
||
def _rep(name, Rs):
|
||
if not Rs or len(Rs) < 10:
|
||
print(f" {name:<40} 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:<40} n={n:>4} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} "
|
||
f"PF={pf:.2f} ΣR={s:+.0f} worst={min(Rs):+.1f}")
|
||
return s / n
|
||
|
||
|
||
def find_pivots(H, L, step=1):
|
||
"""Pivots auf der aggregierten Level-TF; Index = M5-Index, ab dem der Pivot
|
||
BEKANNT ist (Bestätigung braucht k Bars danach) → kein Look-ahead."""
|
||
if step == 1:
|
||
ph = [(i, H[i]) for i in range(_PIV_K, len(H) - _PIV_K)
|
||
if H[i] == max(H[i - _PIV_K:i + _PIV_K + 1])]
|
||
pl = [(i, L[i]) for i in range(_PIV_K, len(L) - _PIV_K)
|
||
if L[i] == min(L[i - _PIV_K:i + _PIV_K + 1])]
|
||
return ph, pl
|
||
hh, ll = [], []
|
||
for s0 in range(0, len(H) - step + 1, step):
|
||
hh.append(max(H[s0:s0 + step])); ll.append(min(L[s0:s0 + step]))
|
||
ph, pl = [], []
|
||
for j in range(_PIV_K, len(hh) - _PIV_K):
|
||
avail = (j + _PIV_K + 1) * step
|
||
if hh[j] == max(hh[j - _PIV_K:j + _PIV_K + 1]): ph.append((avail, hh[j]))
|
||
if ll[j] == min(ll[j - _PIV_K:j + _PIV_K + 1]): pl.append((avail, ll[j]))
|
||
return ph, pl
|
||
|
||
|
||
def scan(H, L, C, SP, A, EF30, ES30, lo_i, hi_i, mode, smc_exit,
|
||
step=1, hours=None, BH=None):
|
||
"""mode: 'eqh' (Equal Highs→Short), 'eql' (Equal Lows→Long), 'swing' (HTF-Trend).
|
||
`step` = Level-TF (1=M5, 6=M30, 12=H1) · `hours` = erlaubte Berlin-Stunden
|
||
(None = alle). Ein Pivot ist erst ab (idx+k+1) bekannt → kein Look-ahead."""
|
||
ph, pl = find_pivots(H, L, step)
|
||
Rs = []
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||
|
||
i = max(lo_i, _LOOKBACK)
|
||
hi_i = min(hi_i, len(C) - _MAXH - 2)
|
||
while i < hi_i:
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN: i += 1; continue
|
||
if hours is not None and BH is not None and BH[i] not in hours:
|
||
i += 1; continue
|
||
tol = _EQ_TOL * atr
|
||
# ── Liquiditätszone bestimmen ────────────────────────────────────────
|
||
level = None; d = 0; zone_lo = None
|
||
if mode in ("eqh", "swing"):
|
||
# bekannte Hoch-Pivots im Fenster (bestätigt vor i)
|
||
hs = [(ix, p) for ix, p in ph if i - _LOOKBACK <= ix <= i - _PIV_K - 1]
|
||
if mode == "eqh" and len(hs) >= 2:
|
||
# zwei ~gleiche Hochs mit passendem Abstand
|
||
for a in range(len(hs) - 1, 0, -1):
|
||
for b in range(a - 1, -1, -1):
|
||
gap = hs[a][0] - hs[b][0]
|
||
if gap < _EQ_MINGAP: continue
|
||
if gap > _EQ_MAXGAP: break
|
||
if abs(hs[a][1] - hs[b][1]) <= tol:
|
||
level = max(hs[a][1], hs[b][1]); d = -1; break
|
||
if level is not None: break
|
||
elif mode == "swing" and hs:
|
||
level = hs[-1][1]; d = -1
|
||
if mode in ("eql", "swing") and level is None:
|
||
ls = [(ix, p) for ix, p in pl if i - _LOOKBACK <= ix <= i - _PIV_K - 1]
|
||
if mode == "eql" and len(ls) >= 2:
|
||
for a in range(len(ls) - 1, 0, -1):
|
||
for b in range(a - 1, -1, -1):
|
||
gap = ls[a][0] - ls[b][0]
|
||
if gap < _EQ_MINGAP: continue
|
||
if gap > _EQ_MAXGAP: break
|
||
if abs(ls[a][1] - ls[b][1]) <= tol:
|
||
level = min(ls[a][1], ls[b][1]); d = 1; break
|
||
if level is not None: break
|
||
elif mode == "swing" and ls:
|
||
level = ls[-1][1]; d = 1
|
||
if level is None or d == 0: i += 1; continue
|
||
|
||
# ── Sweep prüfen: Bar i durchsticht das Level, Close kommt zurück ────
|
||
if d < 0: # Equal Highs: High über Level, Close zurück darunter
|
||
over = H[i] - level
|
||
swept = (_SWEEP_MIN * atr <= over <= _SWEEP_MAX * atr) and C[i] < level
|
||
wick = H[i]
|
||
else: # Equal Lows: Low unter Level, Close zurück darüber
|
||
over = level - L[i]
|
||
swept = (_SWEEP_MIN * atr <= over <= _SWEEP_MAX * atr) and C[i] > level
|
||
wick = L[i]
|
||
if not swept:
|
||
i += 1; continue
|
||
# „Close zurück in die Zone" darf auch 1–_BACK_BARS Bars dauern; hier ist er
|
||
# bereits im Sweep-Bar erfüllt (strengste, sauberste Variante).
|
||
|
||
# Swing-Modus: nur MIT dem HTF-Trend (M30-EMA12/50) handeln
|
||
if mode == "swing":
|
||
htf = 1 if EF30[i] > ES30[i] else -1 if EF30[i] < ES30[i] else 0
|
||
if htf != d: i += 1; continue
|
||
|
||
entry = C[i]
|
||
stop = wick + d * -1 * (0.05 * atr) # knapp jenseits des Sweep-Dochts
|
||
stop = (wick - 0.05 * atr) if d > 0 else (wick + 0.05 * atr)
|
||
# Ziel = gegenüberliegende Liquidität (nächster Pivot in Trade-Richtung)
|
||
if d < 0:
|
||
cands = [p for ix, p in pl if i - _LOOKBACK <= ix <= i - _PIV_K - 1
|
||
and p < entry - 0.3 * atr]
|
||
target = max(cands) if cands else None
|
||
else:
|
||
cands = [p for ix, p in ph if i - _LOOKBACK <= ix <= i - _PIV_K - 1
|
||
and p > entry + 0.3 * atr]
|
||
target = min(cands) if cands else None
|
||
|
||
R = (_sim_smc(entry, d, atr, stop, target, H, L, C, i + 1) if smc_exit
|
||
else _sim_live(entry, d, atr, H, L, C, i + 1)) - cost(i, atr)
|
||
Rs.append(R)
|
||
i += _COOL
|
||
return Rs
|
||
|
||
|
||
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]; SP = [float(b["spread"]) * point for b in bars]
|
||
A = _atr_series(H, L, C)
|
||
# HTF-Proxy für den Swing-Modus: M30 aus M5 aggregiert (6 Bars), auf M5 gemappt
|
||
c30 = [C[s + 5] for s in range(0, len(C) - 5, 6)]
|
||
e12 = _ema_series(c30, 12); e50 = _ema_series(c30, 50)
|
||
EF30 = [0.0] * len(C); ES30 = [0.0] * len(C)
|
||
for i in range(len(C)):
|
||
j = i // 6 - 1
|
||
if 0 <= j < len(e12): EF30[i] = e12[j]; ES30[i] = e50[j]
|
||
# Berlin-Stunde je Bar (Broker UTC+3 → Berlin = Broker − 1 h, Sommerzeit)
|
||
from datetime import datetime, timezone
|
||
BH = [(datetime.fromtimestamp(int(b["time"]), timezone.utc).hour - 1) % 24
|
||
for b in bars]
|
||
N = len(C); mid = N // 2
|
||
|
||
print("=" * 96)
|
||
print(f" LIQUIDITÄTS-SWEEPS (SMC) — {sym} M5 ({N} Bars, 2 Halbjahre)")
|
||
print(f" Sweep = Level um {_SWEEP_MIN}–{_SWEEP_MAX}×ATR durchstochen, Close zurück in der Zone")
|
||
print(f" Maßstab: validierter Squeeze ØR +0,14…+0,23 · PF>1 in BEIDEN Hälften")
|
||
print("=" * 96)
|
||
|
||
verdict = {}
|
||
MODES = (("eqh", "A) Equal Highs -> SHORT"),
|
||
("eql", "B) Equal Lows -> LONG"),
|
||
("swing", "C) Swing-Liquidity (HTF)"))
|
||
# ── Achse 1: LEVEL-TF (Multi-TF-Regel „Liquidität auf höherem TF") × Exit ──
|
||
for tf_lbl, step in _LEVEL_TFS:
|
||
print("\n" + "=" * 96)
|
||
print(f"### LEVEL-TF {tf_lbl} — Liquidität auf {tf_lbl} identifiziert, Entry auf M5")
|
||
for exit_lbl, smc in (("Live-Exit", False), ("SMC-Exit", True)):
|
||
print(f"\n --- EXIT: {exit_lbl} ---")
|
||
for mode, mlabel in MODES:
|
||
for hl, lo, hi in (("H1", _LOOKBACK, mid), ("H2", mid, N)):
|
||
Rs = scan(H, L, C, SP, A, EF30, ES30, lo, hi, mode, smc, step)
|
||
oR = _rep(f"{mlabel} [{hl}]", Rs)
|
||
verdict.setdefault((f"{tf_lbl}/{exit_lbl}", mlabel), []).append(oR)
|
||
# ── Achse 2: SESSION-Filter (Level-TF M30, Live-Exit) ────────────────────
|
||
print("\n" + "=" * 96)
|
||
print("### SESSION-FILTER (Level-TF M30, Live-Exit) — „Sweeps passieren um 14–16 / 2–4 Uhr\"")
|
||
for sess, hours in _SESS.items():
|
||
print(f"\n --- Session: {sess} (Berlin-Stunden {hours}) ---")
|
||
for mode, mlabel in MODES[:2]:
|
||
for hl, lo, hi in (("H1", _LOOKBACK, mid), ("H2", mid, N)):
|
||
Rs = scan(H, L, C, SP, A, EF30, ES30, lo, hi, mode, False, 6, hours, BH)
|
||
oR = _rep(f"{mlabel} [{hl}]", Rs)
|
||
verdict.setdefault((f"M30/Sess-{sess}", mlabel), []).append(oR)
|
||
|
||
print("\n" + "=" * 96)
|
||
print(" URTEIL (Kandidat für die Gesamtempfehlung nur, wenn ØR in BEIDEN Hälften")
|
||
print(" > 0 UND ≥ +0,14 = Squeeze-Band):")
|
||
any_ok = False
|
||
for (ex, m), vals in verdict.items():
|
||
if len(vals) != 2 or any(v is None for v in vals):
|
||
print(f" {m:<38} {ex[:22]:<24} 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" {m:<38} {ex[:22]:<24} H1 {vals[0]:+.3f} · H2 {vals[1]:+.3f} {mark}")
|
||
print("\n " + ("→ Kandidat gefunden: Parameter-Robustheit prüfen, dann als Verdict-Modul."
|
||
if any_ok else
|
||
"→ KEINE Variante trägt beidhälftig im Squeeze-Band → nicht in die "
|
||
"Gesamtempfehlung (höchstens als Anzeige)."))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|