"""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"