Axel HocksandClaude Opus 5
cddcefa463
Anlass war die Frage nach numpy. Statt zu vermuten wurde profiliert — und das
Ergebnis war ein anderes als erwartet: von 10,07 s Gesamtlaufzeit steckten
8,17 s in der EIGENZEIT der Backtest-Schleife, nur 0,44 s in _build.
Ursache:
[C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K] # je BAR, ganze Liste
Das ist O(Bars x Pivots) — bei 80k Bars hunderte Millionen Vergleiche, und
genau daher kamen die 9-20-Minuten-Laeufe. PH/PL sind sortiert, also genuegen
zwei Binaersuchen fuer dieselbe Menge in O(log n).
Gemessen: backtest_signal_touchfill.py ueber 80k Bars von ~9 min auf 8,9 s
(~60x), bei 8k Bars 10,07 s -> 1,43 s. Das Ergebnis bleibt inhaltlich
identisch (alle sechs Felder negativ); die Handvoll Trades Unterschied ist die
bekannte Live-Bar-Drift, nicht der Umbau.
MIGRATIONS-REGEL ERNST GENOMMEN: ein Vorher/Nachher-Vergleich zweier LAEUFE
taugt hier nicht, weil copy_rates_from_pos am neuesten Bar ankert und
eintreffende Live-Bars das Fenster verschieben. Die Gleichheit wird deshalb
DIREKT auf der Datenstruktur bewiesen — tests/test_pivotfenster.py mit vier
Faellen: 4.000 Bars x 400 Pivots, Randfaelle, leere Liste, beidseitig
inklusive Grenzen.
Angewandt auf 7 Skripte (auto_signal_v3, breakout_gated, dist_gated, htf_vola,
metalabel, sl_width, trail_be) — mechanisch identische Transformation, alle
kompilieren, Stichprobe gegengelaufen (dist_gated 20k in 1,1 s).
Lehre: die naheliegende Antwort (numpy) war die falsche. Ein Profil kostet zwei
Minuten und haette diese Bremse jederzeit gezeigt — sie lag seit Monaten in
jedem gegateten Backtest.
17 Tests gruen, n-Falle-Scan sauber.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
205 lines
8.5 KiB
Python
205 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
||
"""SL-Breite: ATR_M15 (live) gegen ATR_M5 (kanonisch) — 2026-08-04.
|
||
|
||
Anlass: `analyze_manual_close.py` zeigte, dass die 69 per Broker-SL geschlossenen
|
||
Trades mit dem kanonischen Exit **1.229 € weniger** verloren hätten. Der dortige
|
||
Confounder war explizit benannt: `simulate()` setzt den Initial-SL auf
|
||
**2,0×ATR_M5**, live kommt er aus **ATR_M15** (`_calc_sl_tp`, SL_TF fix M15) und
|
||
ist damit oft deutlich weiter — unter Margin-Sizing schlägt das voll durch.
|
||
Diese Messung trennt beides.
|
||
|
||
⚠ `backtest_sl_tf_mismatch.py` hat die M5-Umstellung 2026-07-23 schon geprüft und
|
||
wegen H2 verworfen — aber NUR auf Squeeze-Entries und mit einem eigenen
|
||
Exit-Nachbau. Hier: die **gegatete Live-Signal-Population** und der **kanonische**
|
||
Exit. Damit ist es keine Wiederholung, sondern die Messung unter den Bedingungen,
|
||
unter denen der Bot tatsächlich läuft.
|
||
|
||
Variiert wird AUSSCHLIESSLICH die Breite des Initial-SL; Trailing, Breakeven,
|
||
Time-Stop und TP bleiben identisch — sonst wäre nicht zuzuordnen, was wirkt.
|
||
Alle Ergebnisse in **ATR_M5-Einheiten**, damit die Varianten vergleichbar sind.
|
||
|
||
Aufruf: python backtest_sl_width.py [bars]
|
||
"""
|
||
import dataclasses
|
||
import statistics as stx
|
||
import sys
|
||
from bisect import bisect_left, bisect_right
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
from core.analysis import calc_trend_angle
|
||
from core.exit_model import LIVE, simulate
|
||
from core.wave_rec import (WaveRecommender, _ANGLE_LR, _EMA_FAST, _EMA_SLOW,
|
||
_HTF_DEADBAND, _ema_series)
|
||
|
||
_ATRMIN = 0.06
|
||
_COOL = 3
|
||
_PIVK = 3
|
||
_PIV_LOOK = 300
|
||
|
||
|
||
class _TU:
|
||
def snapshot(self):
|
||
return {}
|
||
|
||
|
||
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])))
|
||
return [(sum(t[max(1, i - p + 1):i + 1]) / max(1, len(t[max(1, i - p + 1):i + 1])))
|
||
if i else None for i in range(len(C))]
|
||
|
||
|
||
def _atr_m15(H, L, C, N):
|
||
"""ATR_M15 je M5-Bar (kausal): aus 3er-Aggregaten, auf M5 zurückgemappt."""
|
||
HH = [max(H[i:i + 3]) for i in range(0, N - 2, 3)]
|
||
LL = [min(L[i:i + 3]) for i in range(0, N - 2, 3)]
|
||
CC = [C[min(i + 2, N - 1)] for i in range(0, N - 2, 3)]
|
||
a15 = _atr_series(HH, LL, CC)
|
||
out = [None] * N
|
||
for j, v in enumerate(a15):
|
||
for k in range(j * 3, min(j * 3 + 3, N)):
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def _htf_sign(C, i, n, atr):
|
||
lo = max(0, i - n * (_EMA_SLOW + 5))
|
||
sub = C[lo:i + 1:n]
|
||
if len(sub) < _EMA_SLOW + 2 or not atr:
|
||
return 0
|
||
d = (_ema_series(sub, _EMA_FAST)[-1] - _ema_series(sub, _EMA_SLOW)[-1]) / atr
|
||
return 0 if abs(d) < _HTF_DEADBAND else (1 if d > 0 else -1)
|
||
|
||
|
||
def st(Rs):
|
||
if not Rs:
|
||
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)
|
||
return dict(n=n, wr=100 * w / n, oR=s / n, pf=(up / dn if dn > 0 else 9.99),
|
||
sum=s, worst=min(Rs))
|
||
|
||
|
||
def main():
|
||
n_req = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||
mt5.initialize()
|
||
sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
|
||
if mt5.symbol_info(c)), None)
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n_req)
|
||
point = mt5.symbol_info(sym).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]
|
||
N = len(C); mid = N // 2
|
||
A = _atr_series(H, L, C)
|
||
A15 = _atr_m15(H, L, C, N)
|
||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
|
||
|
||
verh = [A15[i] / A[i] for i in range(200, N)
|
||
if A[i] and A15[i] and A[i] >= _ATRMIN]
|
||
print(f"ATR_M15 / ATR_M5 — Median {stx.median(verh):.2f} "
|
||
f"10./90. Perzentil {sorted(verh)[len(verh)//10]:.2f} / "
|
||
f"{sorted(verh)[9*len(verh)//10]:.2f}")
|
||
print(f"→ Der Live-SL (2,0×ATR_M15) ist im Median rund "
|
||
f"{2.0*stx.median(verh):.1f}×ATR_M5 breit, der kanonische 2,0.\n")
|
||
|
||
print("Vorberechnung (Winkel, HTF, Pivots) ...", flush=True)
|
||
AD = [90.0] * N
|
||
for i in range(_ANGLE_LR + 2, N):
|
||
AD[i] = calc_trend_angle(C[i - (_ANGLE_LR + 1):i + 1], _ANGLE_LR)
|
||
HTF = [0] * N; H1 = [0] * N
|
||
for i in range(N):
|
||
if A[i]:
|
||
HTF[i] = _htf_sign(C, i, 6, A[i]); H1[i] = _htf_sign(C, i, 12, A[i])
|
||
PH = [j for j in range(_PIVK, N - _PIVK) if H[j] == max(H[j - _PIVK:j + _PIVK + 1])]
|
||
PL = [j for j in range(_PIVK, N - _PIVK) if L[j] == min(L[j - _PIVK:j + _PIVK + 1])]
|
||
|
||
w = WaveRecommender(_TU(), mt5.TIMEFRAME_M5)
|
||
w.set_reversal_enabled(False); w.set_entry_room(0.6)
|
||
w.set_dead_hours([]); w.set_breakout_k(0.3)
|
||
|
||
# Entries EINMAL bestimmen — alle Varianten sehen dieselbe Liste (sonst wäre
|
||
# nicht der SL verglichen, sondern eine andere Trade-Folge).
|
||
print("Entries bestimmen ...", flush=True)
|
||
entries = []
|
||
pend = None
|
||
i = _PIV_LOOK + _EMA_SLOW + 5
|
||
while i < N - LIVE.max_hold - 2:
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN or not A15[i]:
|
||
i += 1; continue
|
||
w._pb_feats = {"atr": max(atr, 0.12)}
|
||
# ⚠ bisect statt linearer Filterung (2026-08-07). Die alte Fassung
|
||
# scannte je Bar die KOMPLETTE Pivot-Liste = O(Bars x Pivots) und war
|
||
# der Engpass des Projekts (80k-Lauf: ~9 min -> 8,9 s). PH/PL sind
|
||
# sortiert, die Binaersuche liefert DIESELBE Menge — bewiesen in
|
||
# tests/test_pivotfenster.py, nicht behauptet.
|
||
_a1 = bisect_left(PH, i - _PIV_LOOK); _b1 = bisect_right(PH, i - _PIVK)
|
||
_a2 = bisect_left(PL, i - _PIV_LOOK); _b2 = bisect_right(PL, i - _PIVK)
|
||
w._pb_levels = {"ph": [C[j] for j in PH[_a1:_b1]],
|
||
"pl": [C[j] for j in PL[_a2:_b2]]}
|
||
rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
|
||
htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
|
||
sig = rec.get("signal")
|
||
if sig in ("LONG", "SHORT"):
|
||
d = 1 if sig == "LONG" else -1
|
||
if pend is None or pend["dir"] != d:
|
||
pend = {"dir": d, "level": C[i] + d * 0.3 * atr,
|
||
"invalid": C[i] - d * 0.3 * atr, "confirmed": False, "bar": i}
|
||
if not pend["confirmed"]:
|
||
broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
|
||
if broke:
|
||
pend["confirmed"] = True
|
||
else:
|
||
if ((C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])) \
|
||
or (i - pend["bar"]) > 12:
|
||
pend = {"dir": d, "level": C[i] + d * 0.3 * atr,
|
||
"invalid": C[i] - d * 0.3 * atr,
|
||
"confirmed": False, "bar": i}
|
||
i += 1; continue
|
||
rec = w._room_gate(rec, C[i])
|
||
if rec.get("signal") in ("LONG", "SHORT"):
|
||
entries.append((i, d, C[i], atr, A15[i]))
|
||
pend = None
|
||
i += LIVE.timestop_bars # fester Vorlauf, für alle Varianten gleich
|
||
continue
|
||
elif sig == "WARTEN":
|
||
pend = None
|
||
i += 1
|
||
print(f"{len(entries)} Entries (gegatet, Reversal aus, k=0,3)\n")
|
||
|
||
varianten = [
|
||
("A SL 2,0×ATR_M5 (kanonisch)", lambda a5, a15: 2.0),
|
||
("B SL 2,0×ATR_M15 (LIVE)", lambda a5, a15: 2.0 * a15 / a5),
|
||
("C SL 1,5×ATR_M5", lambda a5, a15: 1.5),
|
||
("D SL 2,5×ATR_M5", lambda a5, a15: 2.5),
|
||
("E SL 3,0×ATR_M5", lambda a5, a15: 3.0),
|
||
]
|
||
print(f"{'Variante':<32}{'Hälfte':<10}{'n':>6}{'WR':>6}{'ØR':>9}{'PF':>7}"
|
||
f"{'ΣR':>9}{'Worst':>8}")
|
||
print("-" * 88)
|
||
for lbl, f in varianten:
|
||
for hl, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N)):
|
||
Rs = []
|
||
for (i, d, px, a5, a15) in entries:
|
||
if not (lo <= i < hi):
|
||
continue
|
||
p = dataclasses.replace(LIVE, sl_atr=max(0.5, f(a5, a15)))
|
||
r, _ = simulate(px, d, a5, H, L, C, i + 1, p,
|
||
timestop=True, use_tp=True, ret_bar=True)
|
||
Rs.append(r - (SP[i] if SP[i] > 0 else 0.0225) / a5)
|
||
s = st(Rs)
|
||
if s:
|
||
print(f"{lbl:<32}{hl:<10}{s['n']:>6}{s['wr']:>5.0f}%{s['oR']:>+9.3f}"
|
||
f"{s['pf']:>7.2f}{s['sum']:>+9.0f}{s['worst']:>+8.2f}")
|
||
print()
|
||
print(" Regel (vorab): eine engere SL-Basis wird nur empfohlen, wenn sie in BEIDEN")
|
||
print(" Hälften mehr ΣR liefert UND der Worst-Case nicht schlechter wird.")
|
||
print()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|