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 <noreply@anthropic.com>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 5
Claude Opus 5
parent
a35a0e7006
commit
fd3c7f57a5
@@ -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"
|
||||
Reference in New Issue
Block a user