From fd3c7f57a5402007f2e54b73da7ffebba844264d Mon Sep 17 00:00:00 2001 From: Axel Hocks Date: Fri, 7 Aug 2026 07:21:45 +0200 Subject: [PATCH] Live-Code von Backtest-Skript entkoppelt: core/squeeze_scan.py core/engine.py (squeeze_monitor) und weekly_review.py importierten backtest_breakout_squeeze - die Abhaengigkeitsrichtung war verkehrt: ein Fehler im Backtest konnte den laufenden Bot treffen. Schwerer wog, WIE die Parameter hineinkamen: BQ._N = _SQ_N; BQ._K = _SQ_K Das mutiert Modul-Globale eines Backtests, wirkt prozessweit und damit auf jeden anderen Nutzer desselben Moduls im selben Prozess. Die Werte weichen real ab (Backtest _N=12, live _SQ_N=10). Geloest ueber core/squeeze_scan.py - Parameter stehen in der Signatur. backtest_breakout_squeeze.py leitet nur noch weiter, damit es EINE Implementierung gibt statt zweier (der sim_run/_ema_series-Fehler). Der Exit ist bewusst NICHT mitrepariert: exit_legacy ist eine bitgenaue Kopie inkl. seiner Abweichungen vom kanonischen Exit (Trail 1,5 statt 1,0, kein Lock/Time-Stop/TP, max_hold 288 statt 200). Damit misst der B4-Monitor gegen einen veralteten Exit - separater, offener Befund. Ihn hier still mitzuaendern haette die Vergleichsgrundlage verschoben. Bitgenauigkeit bewiesen (tests/test_squeeze_scan.py): die alte Fassung liegt eingefroren im Test und laeuft gegen die neue ueber 500 synthetische Reihen x 3 Multiplikatoren. Ein Vorher/Nachher-Lauf waere ungueltig gewesen - der Backtest zieht Bars live aus MT5, das Fenster verschiebt sich staendig. Test selbst per Mutationsprobe geprueft: 5 von 6 eingebauten Fehlern schlagen an, der sechste (d>0 -> d>=0) ist beweisbar aequivalent, da d nur +-1 ist. Live-Gegenprobe auf 9000 M5-Bars: n=112, Treffer 44%, OeR +0,178, PF 1,26. Aktivierung beim naechsten Neustart - bei offener Position kein Neustart fuer eine verhaltensneutrale Aenderung. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 62 +++++++++++++ backtest_breakout_squeeze.py | 65 +++++--------- core/engine.py | 15 +++- core/squeeze_scan.py | 126 ++++++++++++++++++++++++++ tests/test_squeeze_scan.py | 170 +++++++++++++++++++++++++++++++++++ weekly_review.py | 11 ++- 6 files changed, 398 insertions(+), 51 deletions(-) create mode 100644 core/squeeze_scan.py create mode 100644 tests/test_squeeze_scan.py diff --git a/CLAUDE.md b/CLAUDE.md index 371ea29..d3710b3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4731,6 +4731,68 @@ Signal-Eingriffe sind gescheitert, und die gemessene Stärke des Users liegt in AUSWAHL, nicht in einer Regel. Ein Modul, das ihm seine eigene Historie zur aktuellen Lage zeigt, spielt auf diese Stärke; eines das für ihn einsteigt, ersetzt sie. +## ⚠⚠ LIVE-CODE HING AN EINEM BACKTEST-SKRIPT — behoben 2026-08-07 + +**Befund:** `core/engine.py` (`squeeze_monitor`) und `weekly_review.py` importierten +**`backtest_breakout_squeeze`**. Die Abhängigkeitsrichtung war damit verkehrt herum: +„Backtests sind nur Dokumentation" stimmte nicht — ein Fehler dort konnte den +laufenden Bot treffen. + +⚠ **Der schwerere Teil war nicht der Import, sondern wie die Parameter hineinkamen:** +```python +BQ._N = _SQ_N; BQ._K = _SQ_K # Modul-Globale eines Backtests mutieren +``` +Das wirkt **prozessweit** und damit auf jeden anderen Nutzer desselben Moduls im +selben Prozess. Dass die Werte real abweichen, ist keine Theorie: der Backtest-Default +ist `_N=12`, live gilt `_SQ_N=**10**`. Wer das Skript nach dem Monitor-Aufruf im selben +Prozess benutzt hätte, hätte still mit Live- statt Backtest-Parametern gerechnet. + +✅ **Gelöst: `core/squeeze_scan.py`** — `atr_series` · `exit_legacy` · `scan`, alle +Parameter in der **Signatur** statt in Modul-Globalen. `engine`/`weekly_review` rufen +`core` auf; `backtest_breakout_squeeze.py` leitet nur noch weiter, damit es **eine** +Implementierung gibt statt zweier, die auseinanderdriften (der Fehler, den das Projekt +bei `sim_run`/`_ema_series` — Kopien in ≥6 Skripten — schon einmal gemacht hat). + +⚠⚠ **DER EXIT WURDE BEWUSST NICHT MITREPARIERT.** `exit_legacy` ist eine bitgenaue +Kopie inklusive seiner Abweichungen vom kanonischen Live-Exit (`exit_model.LIVE`): +| | Squeeze-Backtest / B4-Monitor | LIVE | +|---|---|---| +| Trail | **1,5** | **1,0** (seit 31.07.) | +| Lock-Phase | keine | ja | +| Time-Stop | keiner | 120 min | +| Initial-TP | keiner | 3,5×ATR | +| max_hold | 288 | 200 | +**Das ist ein FÜNFTES Exit-Modell** neben den drei aus dem Stufe-3-Refactor und dem +kanonischen. Der B4-Monitor („Entry-Edge · Mechanik") misst also gegen einen +**veralteten** Exit. Das hier stillschweigend mitzuändern hätte die Vergleichsgrundlage +des Monitors verschoben, ohne dass es jemand merkt — **offener Befund, eigene +Entscheidung.** ⚠ Beim Umstellen ist die Erwartungsspanne (+0,14…+0,23 ØR) mit +neu zu bestimmen; sie stammt aus Messungen mit genau diesem Exit. + +✅ **Bitgenauigkeit BEWIESEN, nicht behauptet** (`tests/test_squeeze_scan.py`, 5 Tests): +die alte Fassung liegt **eingefroren** im Test (wörtlich aus `git show HEAD:…`) und +läuft gegen die neue über **500 synthetische Kursreihen × 3 Multiplikatoren**. +⚠ Ein Vorher/Nachher-Lauf des Backtests wäre **ungültig** gewesen: er zieht seine Bars +live aus MT5, das Fenster verschiebt sich mit jeder Minute — gleiche Zahlen wären +Zufall, verschiedene kein Beleg. Die Reihen wechseln blockweise die Vola, sonst liegt +der ATR immer über oder immer unter `_ATRMIN` und Kompressions-Bedingung, Filter und +Cooldown werden nie durchlaufen. Zusätzlich prüft der Test, dass überhaupt >500 +Ausbrüche entstehen — „beide finden nichts" ist kein Nachweis von Gleichheit. +✅ **Der Test wurde selbst geprüft (Mutationsprobe):** 6 gezielte Fehler in +`squeeze_scan.py` eingebaut — Trail 1,5→1,4 · Breakeven 1,3→1,2 · Cooldown +1 Bar · +Ausbruchs-Level +0,1 % · ATR-Fenster +1 Bar schlagen **alle** an. Der sechste +(`d > 0` → `d >= 0`) nicht — **kein Test-Loch, sondern beweisbar äquivalent**: `d` ist +im ganzen Pfad nur ±1, die beiden Vergleiche unterscheiden sich allein bei `d == 0`. +⚠ Ein grüner Test ohne Mutationsprobe belegt nichts — dieselbe Fehlerklasse +(„Erfolg melden, wo nichts geprüft wurde") ist am 07.08. viermal aufgetreten. + +Live-Gegenprobe auf 9.000 M5-Bars mit den echten Live-Parametern: n=112, Treffer 44 %, +**ØR +0,178, PF 1,26** — im dokumentierten Erwartungsband. +⚠ **Aktivierung steht aus:** zum Zeitpunkt des Umbaus lief eine offene Position +(2,47 Lots); ein Neustart hätte ihr 10–20 s Trailing/Time-Stop/Notfall-Stop genommen +für eine Änderung, die per Test **nichts** am Verhalten ändert. Greift beim nächsten +regulären Neustart. + ## Backtest-Audit 2026-08-04 — rechnen die Skripte noch korrekt? **Anlass:** User-Frage nach dem `angle=`-Fund in `backtest_auto_signal.py`. Prüfung diff --git a/backtest_breakout_squeeze.py b/backtest_breakout_squeeze.py index 2117001..10c4251 100644 --- a/backtest_breakout_squeeze.py +++ b/backtest_breakout_squeeze.py @@ -19,30 +19,27 @@ _COOL = 12 # Cooldown nach einem Trade (Bars) gegen Überlappung _K = 0.1 # Ausbruch k×ATR über/unter die Box-Grenze +# ⚠⚠ DIE MECHANIK LIEGT SEIT 2026-08-07 IN `core/squeeze_scan.py`. +# Grund: `core/engine.py` und `weekly_review.py` haben sie von HIER importiert — +# LIVE-Code, der an einem Backtest-Skript haengt, und die Live-Parameter wurden +# ueber MUTIERTE Modul-Globale (`BQ._N = ...`) eingeschleust. Beides ist behoben. +# Hier bleiben nur duenne Weiterleitungen, damit es EINE Implementierung gibt und +# nicht zwei, die auseinanderdriften — der Fehler, den dieses Projekt bei +# `sim_run`/`_ema_series` (Kopien in >=6 Skripten) schon einmal gemacht hat. +# ⚠ Die Zahlen dieses Backtests bleiben unveraendert: bitgenau geprueft in +# `tests/test_squeeze_scan.py` (500 synthetische Kursreihen). +from core.squeeze_scan import atr_series as _core_atr +from core.squeeze_scan import exit_legacy as _core_exit +from core.squeeze_scan import scan as _core_scan + + def _atr_series(H, L, C, p=14): - t = [0.0] - for i in range(1, len(C)): - t.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 = t[max(1, i-p+1):i+1] - out.append(sum(seg)/len(seg)) - return out + return _core_atr(H, L, C, p) def sim(entry, d, atr, H, L, C, j0, sl_atr=2.0, trail=1.5, trail_on=0.3, be_on=1.3): - eff = entry - d*sl_atr*atr; hw = entry - end = min(j0+_MAXH, len(C)-1); exit_px = C[end] - for j in range(j0, end+1): - hi, lo = H[j], L[j] - if (lo <= eff) if d > 0 else (hi >= eff): exit_px = eff; break - hw = max(hw, hi) if d > 0 else min(hw, lo) - prof = (C[j]-entry)*d - if prof >= trail_on*atr: - 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) - return (exit_px-entry)*d/atr + return _core_exit(entry, d, atr, H, L, C, j0, sl_atr=sl_atr, trail=trail, + trail_on=trail_on, be_on=be_on, max_hold=_MAXH) def rep(name, Rs): @@ -55,29 +52,11 @@ def rep(name, Rs): def scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult): - """Ein Durchlauf: sammelt R (netto) je Breakout. Rs_any = ALLE Box-Ausbrüche, - Rs_sq = nur die mit Squeeze (Box<=squeeze_mult×ATR). i springt nach jedem - Breakout um _COOL vor (Überlappungsschutz) — für beide Listen identisch.""" - def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr - Rs_sq = []; Rs_any = [] - i = max(lo_i, _N+15) - while i < min(hi_i, len(C)-_MAXH-1): - atr = A[i] - if not atr or atr < _ATRMIN: i += 1; continue - boxHi = max(H[i-_N:i]); boxLo = min(L[i-_N:i]); box = boxHi-boxLo - squeezed = box <= squeeze_mult*atr - hit = None - for j in range(i, min(i+_W, len(C)-_MAXH-1)): - up = boxHi + _K*atr; dn = boxLo - _K*atr - if H[j] >= up: hit = (j, 1, up); break - if L[j] <= dn: hit = (j, -1, dn); break - if hit is None: i += 1; continue - j, d, lvl = hit - R = sim(lvl, d, atr, H, L, C, j+1) - cost(j, atr) - Rs_any.append(R) - if squeezed: Rs_sq.append(R) - i = j + _COOL - return Rs_sq, Rs_any + """Weiterleitung auf `core.squeeze_scan.scan` mit den Modul-Globalen dieses + Skripts als Parameter (`main()` variiert `_N`/`_K` beim Sweep).""" + return _core_scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult, + box_n=_N, k=_K, atr_min=_ATRMIN, fenster=_W, + cooldown=_COOL, max_hold=_MAXH) def main(): diff --git a/core/engine.py b/core/engine.py index 7ccdcf5..6dce6c7 100644 --- a/core/engine.py +++ b/core/engine.py @@ -3524,12 +3524,19 @@ class TradingEngine: try: m5 = self._fetch_m5_monitor() if m5 and len(m5["C"]) >= 300: - import backtest_breakout_squeeze as BQ + # ⚠ Frueher: `import backtest_breakout_squeeze as BQ` PLUS + # `BQ._N = ...; BQ._K = ...`. Damit haing LIVE-Code an einem + # Backtest-Skript, und die Live-Parameter wurden ueber MUTIERTE + # MODUL-GLOBALE eingeschleust (prozessweit wirksam). Jetzt liegt + # die Mechanik in `core/` und die Parameter stehen in der + # Signatur. Verhalten unveraendert (bitgenau geprueft, + # `tests/test_squeeze_scan.py`). + from core.squeeze_scan import atr_series, scan from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K - BQ._N = _SQ_N; BQ._K = _SQ_K H, L, C, SP = m5["H"], m5["L"], m5["C"], m5["SP"] - A = BQ._atr_series(H, L, C) - sq, _ = BQ.scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT) + A = atr_series(H, L, C) + sq, _ = scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT, + box_n=_SQ_N, k=_SQ_K) n = len(sq) if n: w = sum(1 for x in sq if x > 0); s = sum(sq) diff --git a/core/squeeze_scan.py b/core/squeeze_scan.py new file mode 100644 index 0000000..eda9503 --- /dev/null +++ b/core/squeeze_scan.py @@ -0,0 +1,126 @@ +"""Box-Ausbruchs-Scan — die geteilte Mechanik hinter dem Squeeze-Setup. + +⚠⚠ WARUM ES DIESES MODUL GIBT (2026-08-07). `core/engine.py` und +`weekly_review.py` importierten bis dahin **`backtest_breakout_squeeze`**, also +LIVE-Code, der von einem Backtest-Skript abhing. Zwei Folgen: + 1. Die Abhaengigkeitsrichtung war verkehrt. „Backtests sind nur Dokumentation" + stimmte nicht — ein Fehler dort konnte den laufenden Bot treffen. + 2. Schlimmer: die Aufrufer haben **Modul-Globale des Backtests mutiert** + (`BQ._N = ...; BQ._K = ...`), um Live-Parameter einzuschleusen. Das wirkt + prozessweit und damit auf jeden anderen Nutzer desselben Moduls im selben + Prozess. +Hier stehen die Parameter deshalb in der Signatur, nicht in Modul-Globalen. + +⚠⚠ DER EXIT IST EIN PARAMETER, UND DAS IST ABSICHT. `exit_legacy` unten ist eine +BITGENAUE Kopie des Exits aus `backtest_breakout_squeeze.sim` — inklusive seiner +Abweichungen vom kanonischen Live-Exit (`core/exit_model.LIVE`): + Trail 1,5 statt 1,0 · KEINE Lock-Phase · KEIN Time-Stop · KEIN TP · + max_hold 288 statt 200 +Damit bleiben die dokumentierten Zahlen des Backtests UND das Verhalten des +B4-Monitors zunaechst unveraendert — der Umbau ist reine Struktur. +⚠ Dass der B4-Monitor damit gegen einen VERALTETEN Exit misst, ist ein separater, +offener Befund (s. CLAUDE.md). Ihn hier stillschweigend mitzuaendern haette die +Vergleichsgrundlage des Monitors verschoben, ohne dass es jemand merkt. +""" +from __future__ import annotations + +from typing import Callable + +# Vorgaben = der bisherige Stand aus `backtest_breakout_squeeze.py`, damit +# Aufrufer ohne explizite Angabe exakt dasselbe bekommen wie vorher. +BOX_N_STD = 12 # Box-Laenge in Bars +FENSTER_STD = 24 # Fenster nach der Box, in dem der Ausbruch fallen muss +COOLDOWN_STD = 12 # Vorsprung nach einem Trade (Ueberlappungsschutz) +K_STD = 0.1 # Ausbruch k×ATR ueber/unter die Box-Grenze +ATRMIN_STD = 0.12 +MAXH_STD = 288 # Haltedauer-Obergrenze des Legacy-Exits + + +def atr_series(H: list, L: list, C: list, p: int = 14) -> list: + """Gleitender ATR — bitgenau wie `backtest_breakout_squeeze._atr_series`.""" + 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 exit_legacy(entry, d, atr, H, L, C, j0, *, sl_atr=2.0, trail=1.5, + trail_on=0.3, be_on=1.3, max_hold=MAXH_STD) -> float: + """⚠ BITGENAUE Kopie von `backtest_breakout_squeeze.sim` — bewusst inklusive + ihrer Abweichungen vom kanonischen Exit (s. Modul-Docstring). Nicht fuer + neuen Code verwenden; dort gehoert `core.exit_model.simulate` hin.""" + eff = entry - d * sl_atr * atr + hw = entry + end = min(j0 + max_hold, len(C) - 1) + exit_px = C[end] + for j in range(j0, end + 1): + hi, lo = H[j], L[j] + if (lo <= eff) if d > 0 else (hi >= eff): + exit_px = eff + break + hw = max(hw, hi) if d > 0 else min(hw, lo) + prof = (C[j] - entry) * d + if prof >= trail_on * atr: + 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) + return (exit_px - entry) * d / atr + + +def scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult, *, + box_n: int = BOX_N_STD, k: float = K_STD, atr_min: float = ATRMIN_STD, + fenster: int = FENSTER_STD, cooldown: int = COOLDOWN_STD, + max_hold: int = MAXH_STD, + exit_fn: Callable | None = None) -> tuple[list, list]: + """Ein Durchlauf: sammelt R (netto) je Ausbruch. + + Rueckgabe `(Rs_sq, Rs_any)` — `Rs_any` = ALLE Box-Ausbrueche (die Kontrolle), + `Rs_sq` = nur die aus einer KOMPRESSION (Box ≤ `squeeze_mult`×ATR). Der + Vergleich der beiden ist es, der den Squeeze ueberhaupt validiert hat. + `i` springt nach jedem Ausbruch um `cooldown` vor — fuer beide Listen gleich. + + ⚠ `point` wird nicht mehr benutzt (SP kommt bereits in Preis), bleibt aber in + der Signatur, damit bestehende Aufrufe unveraendert funktionieren. + """ + ex = exit_fn or (lambda e, d, a, h, l, c, j: exit_legacy( + e, d, a, h, l, c, j, max_hold=max_hold)) + + def kosten(i, atr): + return (SP[i] if SP[i] > 0 else 0.0225) / atr + + Rs_sq: list = [] + Rs_any: list = [] + i = max(lo_i, box_n + 15) + while i < min(hi_i, len(C) - max_hold - 1): + atr = A[i] + if not atr or atr < atr_min: + i += 1 + continue + box_hi = max(H[i - box_n:i]) + box_lo = min(L[i - box_n:i]) + komprimiert = (box_hi - box_lo) <= squeeze_mult * atr + treffer = None + for j in range(i, min(i + fenster, len(C) - max_hold - 1)): + oben = box_hi + k * atr + unten = box_lo - k * atr + if H[j] >= oben: + treffer = (j, 1, oben) + break + if L[j] <= unten: + treffer = (j, -1, unten) + break + if treffer is None: + i += 1 + continue + j, d, lvl = treffer + R = ex(lvl, d, atr, H, L, C, j + 1) - kosten(j, atr) + Rs_any.append(R) + if komprimiert: + Rs_sq.append(R) + i = j + cooldown + return Rs_sq, Rs_any diff --git a/tests/test_squeeze_scan.py b/tests/test_squeeze_scan.py new file mode 100644 index 0000000..20c0650 --- /dev/null +++ b/tests/test_squeeze_scan.py @@ -0,0 +1,170 @@ +"""Beweist, dass `core/squeeze_scan.py` BITGENAU dasselbe rechnet wie der Code, +der bis 2026-08-07 in `backtest_breakout_squeeze.py` stand. + +⚠⚠ WARUM DIESER TEST SO GEBAUT IST. Der uebliche Weg — den Backtest einmal vor +und einmal nach dem Umbau laufen lassen und die Zahlen vergleichen — ist hier +UNGUELTIG: er zieht seine Bars live aus MT5, und dieses Fenster verschiebt sich +mit jeder Minute. Gleiche Zahlen waeren Zufall, verschiedene kein Beleg. +Deshalb liegt die ALTE Fassung unten als eingefrorene Referenz (woertlich aus +`git show HEAD:backtest_breakout_squeeze.py`, Stand vor dem Umbau) und wird auf +SYNTHETISCHEN, deterministischen Kursreihen gegen die neue gehalten. + +Das ist die Migrations-Regel aus `core/exit_model.py`, angewandt: ein Backtest +darf nur dann auf geteilten Code umgestellt werden, wenn er danach DIESELBEN +Zahlen liefert — sonst ist es keine Refaktorierung, sondern eine unbemerkte +Aenderung der Messgrundlage. +""" +import random + +from core.squeeze_scan import atr_series, exit_legacy, scan + +# ─────────────────────────────────────────────────────────────────────────── +# REFERENZ — eingefroren, NICHT anfassen. Aenderungen gehoeren nach core/. +# ─────────────────────────────────────────────────────────────────────────── +_MAXH = 288 +_ATRMIN = 0.12 +_N = 12 +_W = 24 +_COOL = 12 +_K = 0.1 + + +def _ref_atr(H, L, C, p=14): + t = [0.0] + for i in range(1, len(C)): + t.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 = t[max(1, i-p+1):i+1] + out.append(sum(seg)/len(seg)) + return out + + +def _ref_sim(entry, d, atr, H, L, C, j0, sl_atr=2.0, trail=1.5, trail_on=0.3, + be_on=1.3): + eff = entry - d*sl_atr*atr; hw = entry + end = min(j0+_MAXH, len(C)-1); exit_px = C[end] + for j in range(j0, end+1): + hi, lo = H[j], L[j] + if (lo <= eff) if d > 0 else (hi >= eff): exit_px = eff; break + hw = max(hw, hi) if d > 0 else min(hw, lo) + prof = (C[j]-entry)*d + if prof >= trail_on*atr: + 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) + return (exit_px-entry)*d/atr + + +def _ref_scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult): + def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr + Rs_sq = []; Rs_any = [] + i = max(lo_i, _N+15) + while i < min(hi_i, len(C)-_MAXH-1): + atr = A[i] + if not atr or atr < _ATRMIN: i += 1; continue + boxHi = max(H[i-_N:i]); boxLo = min(L[i-_N:i]); box = boxHi-boxLo + squeezed = box <= squeeze_mult*atr + hit = None + for j in range(i, min(i+_W, len(C)-_MAXH-1)): + up = boxHi + _K*atr; dn = boxLo - _K*atr + if H[j] >= up: hit = (j, 1, up); break + if L[j] <= dn: hit = (j, -1, dn); break + if hit is None: i += 1; continue + j, d, lvl = hit + R = _ref_sim(lvl, d, atr, H, L, C, j+1) - cost(j, atr) + Rs_any.append(R) + if squeezed: Rs_sq.append(R) + i = j + _COOL + return Rs_sq, Rs_any + + +# ─────────────────────────────────────────────────────────────────────────── +def _reihe(seed, n=900): + """Zufaellige, aber deterministische OHLC-Reihe in WTI-Groessenordnung. + + ⚠ Die Vola wechselt bewusst blockweise — sonst liegt der ATR entweder immer + ueber oder immer unter `_ATRMIN` (0,12) und der Filter, die Kompressions- + Bedingung und der Cooldown werden nie durchlaufen. Ein Test, der nur den + ruhigen Fall trifft, prueft die Verzweigungen nicht, die hier zaehlen. + """ + rnd = random.Random(seed) + px = 75.0 + H = []; L = []; C = []; SP = [] + for i in range(n): + vola = (0.05, 0.18, 0.45)[(i // 120) % 3] + px += rnd.gauss(0, vola) + spanne = abs(rnd.gauss(0, vola)) + 0.01 + hi = px + spanne * rnd.random() + lo = px - spanne * rnd.random() + H.append(hi); L.append(lo); C.append(px) + SP.append(0.0 if rnd.random() < 0.1 else rnd.uniform(0.015, 0.05)) + return H, L, C, SP + + +def test_atr_bitgenau(): + for seed in range(50): + H, L, C, _ = _reihe(seed, 400) + assert atr_series(H, L, C) == _ref_atr(H, L, C), f"ATR weicht ab (seed {seed})" + + +def test_exit_bitgenau(): + """Beide Richtungen, damit die d>0/d<0-Zweige des Trailings mitgeprueft + werden — der Exit ist in Richtungen gespiegelt und dort schleicht sich ein + Vorzeichenfehler am leichtesten ein.""" + faelle = 0 + for seed in range(60): + H, L, C, _ = _reihe(seed, 400) + A = atr_series(H, L, C) + for j0 in (60, 140, 250): + atr = A[j0] + if not atr: + continue + for d in (1, -1): + a = exit_legacy(C[j0], d, atr, H, L, C, j0 + 1) + b = _ref_sim(C[j0], d, atr, H, L, C, j0 + 1) + assert a == b, f"Exit weicht ab (seed {seed}, j0 {j0}, d {d})" + faelle += 1 + assert faelle > 300, f"zu wenige Faelle geprueft: {faelle}" + + +def test_scan_bitgenau(): + """Der eigentliche Beleg: der komplette Scan ueber 500 Kursreihen.""" + trades = 0 + for seed in range(500): + H, L, C, SP = _reihe(seed) + A = atr_series(H, L, C) + assert A == _ref_atr(H, L, C) + for mult in (2.0, 2.5, 3.5): + neu = scan(H, L, C, A, SP, 1.0, 0, len(C), mult) + alt = _ref_scan(H, L, C, A, SP, 1.0, 0, len(C), mult) + assert neu == alt, f"Scan weicht ab (seed {seed}, mult {mult})" + trades += len(alt[1]) + # ⚠ Ohne diese Zeile koennte der Test gruen sein, weil BEIDE Fassungen + # nichts finden — „gleich leer" ist kein Nachweis von Gleichheit. + assert trades > 500, f"zu wenige Ausbrueche erzeugt: {trades}" + + +def test_parameter_wirken(): + """Die Parameter muessen ueber die Signatur ankommen — genau das war vorher + ueber mutierte Modul-Globale geloest (`BQ._N = ...`). Ein stiller Default + waere ein Live-Verhalten, das sich nicht mehr steuern laesst.""" + H, L, C, SP = _reihe(7) + A = atr_series(H, L, C) + std = scan(H, L, C, A, SP, 1.0, 0, len(C), 2.5) + lang = scan(H, L, C, A, SP, 1.0, 0, len(C), 2.5, box_n=40) + weit = scan(H, L, C, A, SP, 1.0, 0, len(C), 2.5, k=1.5) + assert std != lang, "box_n bleibt wirkungslos" + assert std != weit, "k bleibt wirkungslos" + + +def test_live_aufrufer_liefert_ergebnisse(): + """Der Live-Aufruf aus `engine.squeeze_monitor` / `weekly_review`, exakt so + wie er dort steht — mit den ECHTEN Live-Parametern aus `wave_rec`.""" + from core.wave_rec import _SQ_K, _SQ_MULT, _SQ_N + H, L, C, SP = _reihe(11, 1500) + A = atr_series(H, L, C) + sq, alle = scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT, + box_n=_SQ_N, k=_SQ_K) + assert len(alle) > 0, "keine Ausbrueche — Live-Pfad liefert nichts" + assert len(sq) <= len(alle), "Squeeze-Teilmenge groesser als die Gesamtmenge" diff --git a/weekly_review.py b/weekly_review.py index dabbae9..8df3e11 100644 --- a/weekly_review.py +++ b/weekly_review.py @@ -43,12 +43,15 @@ def _squeeze_section(m5) -> list: if not m5 or len(m5.get("C") or []) < 300: return ["", "SQUEEZE-BREAKOUT (B4): keine M5-Daten — übersprungen."] try: - import backtest_breakout_squeeze as BQ + # ⚠ s. Kommentar in `engine.squeeze_monitor`: die Mechanik liegt jetzt in + # `core/squeeze_scan.py`, die Live-Parameter stehen in der Signatur statt + # als mutierte Modul-Globale eines Backtests. + from core.squeeze_scan import atr_series, scan from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K - BQ._N = _SQ_N; BQ._K = _SQ_K # live-Params in den Scanner H, L, C, SP = m5["H"], m5["L"], m5["C"], m5["SP"] - A = BQ._atr_series(H, L, C) - sq, _ = BQ.scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT) # SP schon in Preis + A = atr_series(H, L, C) + sq, _ = scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT, + box_n=_SQ_N, k=_SQ_K) # SP schon in Preis out = ["", "-" * 92, f"SQUEEZE-BREAKOUT (B4-Monitor · ~30 Tage M5 neu gerechnet · live-Params " f"Box={_SQ_N}/k={_SQ_K}/≤{_SQ_MULT}×ATR):"]