diff --git a/CLAUDE.md b/CLAUDE.md index 9b80df3..c60a0ca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1371,6 +1371,36 @@ dort bereits nachvalidiert, ØR +0,305.) **Eine** Oberfläche: klein & kaum handelbar; NY Fed: SKS nicht profitabel; out-of-sample schwach) UND der Struktur-Klasse (Muster = Swings+Trendlinien = die gerade verworfene Klasse). **11. verworfener Signal-Eingriff — am Signal ist nichts zu holen.** +- **Konsolidierungs-Setups (False Breakout · Distribution · Accumulation · Retest) + = VERWORFEN (gemessen `backtest_consolidation.py`, 2026-07-30, User-Vorgabe „lerne + für die Gesamtempfehlung die Konsolidierungen", 3 Chart-Beispiele):** Alle auf + DERSELBEN Box wie der validierte Squeeze (10 M5-Bars ≤2,5×ATR), 80k Bars, 2 + Halbjahre, Live-Exit + Echtkosten. + · **A) False Breakout / Liquidity Sweep** (Ausbruch kehrt zurück → Gegenrichtung): + **fällt in JEDER Variante durch** (Rückkehr-Fenster 1/2/3 Bars × Live-/Setup-Exit × + Session) — H1 durchgehend negativ (−0,127…−0,252), H2 leicht positiv = Regime-Kippen. + Der **Setup-eigene Exit** („Stop hinter dem Docht, Ziel = Gegenseite der Range") ist + in ALLEN 6 Kombinationen negativ — dieselbe Beobachtung wie bei den SMC-Sweeps und + den Chartmustern: der Lehrbuch-Exit trägt nicht, der Ertrag kommt (wenn überhaupt) + vom generischen Trailing. ⚠ **Wertvoller Nebenbefund:** **54 % aller Box-Ausbrüche + kehren binnen 3 Bars in die Box zurück** — der Squeeze-Ausbruch scheitert also + mehrheitlich (deckt sich mit seiner 43 %-Trefferquote). Die Gegenrichtung zu handeln + rettet das aber NICHT. + · **B/C) Distribution/Accumulation** (Ausbruchsrichtung aus dem Trend VOR der Box + vorhersagbar?): **die Basisrate widerlegt es direkt** — nur **48 %** der Boxen nach + Aufwärtstrend brechen nach unten, 52 % bei Accumulation (Zufall = 50 %). „Distribution" + sah mit ØR +0,168/+0,270 zunächst tragfähig aus; der **KONTROLLTEST** (derselbe + Vor-Trend, aber Bruch in die GEGENrichtung = Trendfortsetzung) liefert in H2 mit + **+0,364 sogar MEHR** als die Distribution-Hypothese (H1: −0,021). Beide Varianten + sind Box-Ausbrüche mit Trailing; welche führt, wechselt mit dem Regime. **Kein + Richtungs-Edge aus dem Vor-Trend.** + · **Retest-Einstieg** (statt beim Bruch am zurücklaufenden Level einsteigen): + in BEIDEN Hälften schlechter als der direkte Bruch (Distribution −0,071/+0,010, + Accumulation −0,184/−0,131) — der Retest kostet den besseren Einstiegspreis. + **Fazit: nichts davon in die Gesamtempfehlung.** Der Squeeze bleibt die einzige + validierte Box-Verwertung. **17. verworfener Signal-Eingriff.** ⚠ Methodisch wichtig: + Der Kontrolltest („trägt die Hypothese mehr als ein beliebiger Ausbruch?") ist genau + der, der bei `backtest_patterns.py` noch fehlt — hier hat er die Idee gekippt. - **Volume Profile / „Liquiditätslinien" (POC/HVN/LVN, Value Area) = VERWORFEN (gemessen `backtest_volume_profile.py`, 2026-07-30, User-Frage „System zur besseren Vorhersage — Liquiditätsnachfragelinien / wahrsch. Verkaufslinie?"):** Getestet, WEIL diff --git a/backtest_consolidation.py b/backtest_consolidation.py new file mode 100644 index 0000000..0a96608 --- /dev/null +++ b/backtest_consolidation.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""backtest_consolidation.py — KONSOLIDIERUNGS-SETUPS messen (B2, User-Vorgabe +2026-07-30 „lerne für die Gesamtempfehlung die Konsolidierungen"). + +Drei vom User beschriebene Setups, alle auf DERSELBEN Box: + A) **False Breakout / Liquidity Sweep** — Ausbruch kehrt sofort zurück → Einstieg + in die GEGENRICHTUNG (Stops der Ausbruchs-Trader wurden abgeräumt). + B) **Distribution (bearish)** — Range NACH einem Aufwärtstrend: Smart Money + verteilt in die Range hinein, danach Bruch nach UNTEN → Short. + C) **Accumulation (bullish)** — Range NACH einem Abwärtstrend: leises Einsammeln, + danach Bruch nach OBEN → Long. + (+) **Retest-Variante** zu B/C: Einstieg erst, wenn der Kurs zum gebrochenen + Level zurückkommt, statt direkt beim Bruch. + +⚠ Die eigentliche NEUE Frage bei B/C: Ist die **Ausbruchs-RICHTUNG aus dem Trend VOR +der Box vorhersagbar?** Der validierte Squeeze handelt den Ausbruch richtungs-neutral +(er nimmt, was kommt). B/C behaupten das Gegenteil der naiven Trendfolge — Range nach +Aufwärtstrend soll nach UNTEN brechen. Das ist deskriptiv sofort prüfbar und wird hier +zuerst gemessen (Basisraten), bevor überhaupt ein R-Ertrag gerechnet wird. + +Setup laut User: + · Preis bricht scheinbar aus der Konsolidierung aus (über Resistance / unter Support) + · kehrt sofort zurück → Stops der Ausbruchs-Trader sind abgeräumt (Liquidity Sweep) + · **Close zurück IN der Range** = Einstieg in die GEGENRICHTUNG + · WTI-Tipp: gehäuft vor News (EIA/API) und am Session-Open + +⚠ ABGRENZUNG — warum das trotz zweier heute verworfener Sweep-Tests neu ist: + `backtest_liquidity_sweep.py` (2026-07-30) hat den Sweep an **Equal Highs/Lows** + gemessen (zwei ~gleiche Swing-Pivots) → durchgefallen. HIER ist die Zone eine + **echte Konsolidierungs-BOX** (enge Spanne über N Bars, wie beim validierten + Squeeze) — andere Mechanik, eigene Messung. + Der Bezug zum Squeeze ist der eigentliche Reiz: der **Ausbruch** aus der Box ist + validiert (ØR +0,14…+0,23), trifft aber nur ~43 %. Die anderen ~57 % sind genau + diese Fehlausbrüche. Wenn die Gegenrichtung nach dem Rückfall trägt, wäre das die + **komplementäre Hälfte eines bereits belegten Setups** — die stärkste Ausgangslage, + die eine neue Idee in diesem Projekt haben kann. + +Methodik (wie alle Tests hier): M5, sequentielle 1-Positions-Sim, Live-Exit +(SL 2,0×ATR · Trail 1,5 ab 0,3 · BE 1,3 · Time-Stop 120 min), Kosten = Bar-Spread/ATR, +2 Halbjahre. Zusätzlich der Setup-eigene Exit (Stop hinter dem Fake-Break-Docht, +Ziel = Gegenseite der Range) — genau wie im Referenzbild. + +Achsen: Rückkehr-Fenster (1/2/3 Bars) · Ausbruchstiefe · Exit-Variante · Session. +Maßstab: Squeeze ØR +0,14…+0,23 · PF>1 in BEIDEN Hälften. + +Aufruf: python backtest_false_breakout.py [n_bars] +""" +from __future__ import annotations +import sys +from datetime import datetime, timezone + +import MetaTrader5 as mt5 + +# ── Konsolidierungs-Box: dieselbe Definition wie der validierte Squeeze ────── +_BOX_N = 10 # Box = Spanne der letzten N abgeschlossenen M5-Bars +_BOX_MULT = 2.5 # Box ≤ x×ATR = Konsolidierung +_BRK_K = 0.1 # Ausbruch ab k×ATR über/unter die Box-Grenze +_MAX_OVER = 1.0 # mehr als x×ATR jenseits der Box = echter Ausbruch, kein Fake +_COOL = 12 +# Exit (Live-Modell) +_SL_ATR, _TRAIL, _TRAIL_ON, _BE_ON, _TIMESTOP, _MAXH = 2.0, 1.5, 0.3, 1.3, 24, 288 +_ATRMIN = 0.06 +# Sessions in Berlin-Stunden (Broker UTC+3 → Berlin = Broker − 1 h) +_SESS = {"EIA/US-Open (14-17)": (14, 15, 16), + "EU-Open (8-10)": (8, 9, 10), + "beide": (8, 9, 10, 14, 15, 16)} +# Vor-Trend (fuer Distribution/Accumulation): Kursaenderung ueber die N Bars VOR der +# Box, in ATR. Ab _PRE_MIN gilt der Trend als "klar". +_PRE_N = 40 +_PRE_MIN = 1.0 +_RETEST_W = 24 # Fenster fuer den Retest-Einstieg (Bars nach dem Bruch) +_RETEST_TOL = 0.15 # Retest gilt als erreicht, wenn Kurs bis x*ATR ans Level zurueck + + +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 _sim_live(entry, d, atr, H, L, C, j0): + 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_setup(entry, d, atr, stop, target, H, L, C, j0): + """Setup-eigener Exit: Stop hinter dem Fake-Break-Docht, Ziel = Gegenseite + der Range (die Lehrbuch-Variante aus dem Referenzbild).""" + 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) + if hit_stop: px = stop; break # konservativ: Stop vor Ziel + if hit_tgt: px = target; break + return (px - entry) * d / atr + + +def pre_trend(C, i, atr): + """Trend VOR der Box: +1 auf, -1 ab, 0 unklar (Kursaenderung/ATR ueber _PRE_N).""" + a = i - _BOX_N - _PRE_N + if a < 0: return 0 + d = (C[i - _BOX_N] - C[a]) / atr + return 1 if d >= _PRE_MIN else -1 if d <= -_PRE_MIN else 0 + + +def scan_dist_acc(H, L, C, SP, A, lo_i, hi_i, want_pre, retest, stats=None, + flip=False, any_pre=False): + """Distribution (want_pre=+1: Box nach Aufwaertstrend -> Bruch nach UNTEN) bzw. + Accumulation (want_pre=-1: Box nach Abwaertstrend -> Bruch nach OBEN). + `retest`: Einstieg erst beim Ruecklauf ans gebrochene Level. + `stats`: [boxen, brueche_erwartete_richtung, brueche_gegenrichtung].""" + Rs = [] + def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr + i = max(lo_i, _BOX_N + _PRE_N + 20) + 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 + boxHi = max(H[i - _BOX_N:i]); boxLo = min(L[i - _BOX_N:i]) + if (boxHi - boxLo) > _BOX_MULT * atr: i += 1; continue + if not any_pre and pre_trend(C, i, atr) != want_pre: i += 1; continue + up = boxHi + _BRK_K * atr; dn = boxLo - _BRK_K * atr + d = 0; lvl = None + if H[i] >= up: d, lvl = 1, up + elif L[i] <= dn: d, lvl = -1, dn + if d == 0: i += 1; continue + # erwartete Richtung: Distribution (+1 Vor-Trend) -> runter, Accumulation -> hoch + want_dir = (want_pre if flip else -want_pre) + if stats is not None: + stats[0] += 1 + stats[1 if d == want_dir else 2] += 1 + if d != want_dir: i += _COOL; continue + j0 = i + 1; entry = lvl + if retest: + # auf Ruecklauf ans gebrochene Level warten (altes Support = neuer Widerstand) + hit = None + for j in range(i + 1, min(i + _RETEST_W, len(C) - _MAXH - 2)): + back = (L[j] <= lvl + _RETEST_TOL * atr) if d > 0 else (H[j] >= lvl - _RETEST_TOL * atr) + if back: hit = j; break + if hit is None: i += _COOL; continue + j0 = hit + 1; entry = lvl + Rs.append(_sim_live(entry, d, atr, H, L, C, j0) - cost(i, atr)) + i += _COOL + return Rs + + +def _rep(name, Rs): + if not Rs or len(Rs) < 10: + print(f" {name:<34} 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:<34} 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 scan(H, L, C, SP, A, BH, lo_i, hi_i, back_bars, setup_exit, hours=None, + stats=None): + """Ein Durchlauf. `back_bars` = innerhalb wie vieler Bars der Close zurück in + die Box muss. `stats` sammelt (ausbrueche, davon_fake) für die Quote.""" + Rs = [] + def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225) / atr + i = max(lo_i, _BOX_N + 20) + 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 + boxHi = max(H[i - _BOX_N:i]); boxLo = min(L[i - _BOX_N:i]) + if (boxHi - boxLo) > _BOX_MULT * atr: # keine Konsolidierung + i += 1; continue + up = boxHi + _BRK_K * atr; dn = boxLo - _BRK_K * atr + # Ausbruchs-Bar suchen + d0 = 0 + if H[i] >= up: d0 = 1 + elif L[i] <= dn: d0 = -1 + if d0 == 0: i += 1; continue + if stats is not None: stats[0] += 1 + # zu tief jenseits der Box = echter Ausbruch, kein Fake + over = (H[i] - boxHi) if d0 > 0 else (boxLo - L[i]) + if over > _MAX_OVER * atr: i += _COOL; continue + # FAKE-OUT: Close muss innerhalb `back_bars` zurück IN die Box + back_at = None + for j in range(i, min(i + back_bars, len(C) - _MAXH - 2)): + if (C[j] < boxHi) if d0 > 0 else (C[j] > boxLo): + back_at = j; break + if back_at is None: i += _COOL; continue + if stats is not None: stats[1] += 1 + if hours is not None: + hb = (datetime.fromtimestamp(int(BH[back_at]), timezone.utc).hour - 1) % 24 + if hb not in hours: i += _COOL; continue + d = -d0 # GEGENRICHTUNG des Fehlausbruchs + entry = C[back_at] + wick = max(H[i:back_at + 1]) if d0 > 0 else min(L[i:back_at + 1]) + stop = wick + (0.05 * atr if d < 0 else -0.05 * atr) + target = boxLo if d < 0 else boxHi # Gegenseite der Range + R = (_sim_setup(entry, d, atr, stop, target, H, L, C, back_at + 1) + if setup_exit else + _sim_live(entry, d, atr, H, L, C, back_at + 1)) - cost(back_at, atr) + Rs.append(R) + i = back_at + _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] + BH = [int(b["time"]) for b in bars] + A = _atr_series(H, L, C) + N = len(C); mid = N // 2 + + print("=" * 94) + print(f" FALSE BREAKOUT / Liquidity Sweep aus der KONSOLIDIERUNG — {sym} M5 ({N} Bars)") + print(f" Box={_BOX_N} Bars ≤{_BOX_MULT}×ATR · Ausbruch {_BRK_K}×ATR · Fake wenn Close " + f"zurück in die Box") + print(f" Einstieg = GEGENRICHTUNG · Maßstab: Squeeze ØR +0,14…+0,23 · PF>1 beidseitig") + print("=" * 94) + + # Wie oft ist ein Box-Ausbruch überhaupt ein Fehlausbruch? + st = [0, 0] + scan(H, L, C, SP, A, BH, _BOX_N + 20, N, 3, False, None, st) + if st[0]: + print(f"\n Basisrate: {st[1]} von {st[0]} Box-Ausbrüchen kehren binnen 3 Bars " + f"zurück = {100*st[1]/st[0]:.0f} % Fehlausbrüche") + + verdict = {} + for back in (1, 2, 3): + print(f"\n### Rückkehr binnen {back} Bar(s)") + for exl, se in (("Live-Exit", False), ("Setup-Exit (Ziel=Gegenseite)", True)): + print(f" --- {exl} ---") + for hl, lo, hi in (("H1", _BOX_N + 20, mid), ("H2", mid, N)): + Rs = scan(H, L, C, SP, A, BH, lo, hi, back, se) + verdict.setdefault((back, exl), []).append(_rep(f"[{hl}]", Rs)) + + # ── B/C: Distribution & Accumulation ──────────────────────────────────── + print("\n" + "=" * 94) + print("### B/C — DISTRIBUTION & ACCUMULATION: ist die Ausbruchsrichtung aus dem") + print(f" Vor-Trend vorhersagbar? (Vor-Trend = Kursaenderung ueber {_PRE_N} Bars, " + f">= {_PRE_MIN}xATR)") + for lbl, want in (("Distribution (Box nach AUFwaertstrend -> erwartet: Bruch RUNTER)", 1), + ("Accumulation (Box nach ABwaertstrend -> erwartet: Bruch HOCH)", -1)): + st = [0, 0, 0] + scan_dist_acc(H, L, C, SP, A, _BOX_N + _PRE_N + 20, N, want, False, st) + print(f"\n {lbl}") + if st[0]: + print(f" Basisrate: {st[1]} erwartete vs {st[2]} Gegenrichtung " + f"= {100*st[1]/max(1,st[0]):.0f} % treffen die Erwartung " + f"(Zufall waere ~50 %)") + for retest, rl in ((False, "Einstieg beim Bruch"), (True, "Einstieg beim RETEST")): + print(f" --- {rl} ---") + for hl, lo, hi in (("H1", _BOX_N + _PRE_N + 20, mid), ("H2", mid, N)): + Rs = scan_dist_acc(H, L, C, SP, A, lo, hi, want, retest) + key = ("dist" if want == 1 else "acc", rl) + verdict.setdefault(key, []).append(_rep(f"[{hl}]", Rs)) + + print(f"\n### SESSION-Filter (Rückkehr ≤2 Bars, Live-Exit)") + for sess, hours in _SESS.items(): + print(f" --- {sess} ---") + for hl, lo, hi in (("H1", _BOX_N + 20, mid), ("H2", mid, N)): + Rs = scan(H, L, C, SP, A, BH, lo, hi, 2, False, hours) + verdict.setdefault(("sess", sess), []).append(_rep(f"[{hl}]", Rs)) + + print("\n" + "=" * 94) + 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 key, vals in verdict.items(): + if len(vals) != 2 or any(v is None for v in vals): + print(f" {str(key):<44} 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 + if key[0] == "sess": lbl = f"Session {key[1]}" + elif key[0] == "dist": lbl = f"Distribution · {key[1]}" + elif key[0] == "acc": lbl = f"Accumulation · {key[1]}" + else: lbl = f"FalseBreak Rückkehr ≤{key[0]} · {key[1]}" + print(f" {lbl:<44} H1 {vals[0]:+.3f} · H2 {vals[1]:+.3f} {mark}") + print("\n " + ("→ Kandidat: Parameter-Robustheit prüfen, dann als Verdict-Modul " + "(Gewicht analog Squeeze)." + if any_ok else + "→ KEINE Variante trägt beidhälftig im Squeeze-Band → NICHT in die " + "Gesamtempfehlung.")) + + +if __name__ == "__main__": + main()