breakout_k mit vollem Live-Gate-Stack gemessen
Nachgeholt, was in der Vollpruefung offen blieb: die ungegatete Messung zeigte k=0,5/0,8/1,0 besser als den Live-Wert 0,3 - aber ohne min_conf, HTF-Filter und Entry-Raum. backtest_breakout_gated.py schliesst die Luecke mit den ECHTEN Methoden (_build, _confirm_breakout-Mechanik, _room_gate mit kausalen M5-Pivots). k=0,0 H1 -0,173/-193 H2 +0,003/ +7 k=0,2 H1 -0,185/-149 H2 -0,012/ -16 k=0,3 H1 -0,160/-122 H2 -0,036/ -47 (LIVE) k=0,5 H1 -0,127/ -85 H2 -0,041/ -47 k=0,8 H1 -0,101/ -54 H2 -0,012/ -11 k=1,0 H1 -0,096/ -45 H2 -0,016/ -12 Bestaetigt: k=0,5/0,8/1,0 schlagen 0,3 in BEIDEN Haelften, die Nachbarn halten mit, und die OR verbessert sich in H1 monoton - also nicht nur Handelsvermeidung. Die vorab fixierte Regel ist erfuellt. TROTZDEM keine Umstellung empfohlen, und der Grund ist kein Messfehler sondern eine Zweckfrage: alle Werte bleiben negativ (PF 0,70-1,01), das Gate begrenzt Schaden statt Edge zu erzeugen; der Auto-Signal-Pfad ist ohnehin aus, das Signal ist ein Vorschlag fuer den Menschen, und dessen gemessener Vorteil entstand auf der k=0,3-Population; k=0,8 kuerzt die Empfehlungen um ~39 % und treibt den WARTEN-Anteil weiter hoch, der mit 93 % ohnehin im Fokus steht. Zwei benannte Abweichungen der Simulation: - _confirm_breakout misst seinen Timeout mit time.time() gegen 3600 s. Im Bar-Loop vergeht keine Wall-Clock-Zeit, der Timeout wuerde nie feuern. 3600 s sind auf M5 exakt 12 Bars, so wird gezaehlt. Merke: jede time.time()-Messung im Live-Code ist in einer Bar-Simulation stumm. - hour=None, weil die EIA-Pruefung in _build ueber datetime.now(_BERLIN) laeuft - mit Bar-Stunde wuerde ein Lauf am Mittwoch 15:30-16:30 alle Bars blocken. 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
80a5be7fe5
commit
3dc56e8571
@@ -0,0 +1,210 @@
|
||||
#!/usr/bin/env python3
|
||||
"""`breakout_k` mit dem VOLLEN Live-Gate-Stack (2026-08-04).
|
||||
|
||||
Offener Punkt aus der Vollprüfung: `backtest_breakout_canon.py` zeigte, dass
|
||||
k=0,5/0,8/1,0 den Live-Wert 0,3 in beiden Hälften schlagen — aber dort lief das
|
||||
**ungegatete** EMA-Signal. Live stehen mehrere Gates davor, und die verschieben
|
||||
die Population erheblich. Diese Messung schließt die Lücke.
|
||||
|
||||
Modelliert (alles über die ECHTEN Methoden, kein Nachbau):
|
||||
· `_build` → Totband · HTF-Gegen-Trend (M30) · Anti-Überdehnung · Winkel-
|
||||
Konfidenz · Mindest-Konfidenz 55 · Reversal (per Schalter)
|
||||
· Breakout-Bestätigung → Mechanik 1:1 aus `_confirm_breakout` übernommen,
|
||||
aber **bar-basiert** statt Wall-Clock (s. u.)
|
||||
· `_room_gate` → mit echten M5-Pivots (`_pb_levels`) und `_pb_feats`
|
||||
|
||||
⚠ ZWEI bewusste Abweichungen, beide unvermeidbar und hier benannt:
|
||||
1. **Timeout bar-basiert.** `_confirm_breakout` misst den Timeout mit
|
||||
`time.time()` gegen `_breakout_timeout_s`=3600 s. In einem Bar-Loop vergeht
|
||||
keine Wall-Clock-Zeit — der Timeout würde NIE feuern. 3600 s sind auf M5
|
||||
exakt **12 Bars**; genau so wird hier gezählt. (Das ist zugleich ein Hinweis
|
||||
für künftige Backtests: jede Zeitmessung über `time.time()` im Live-Code ist
|
||||
in einer Bar-Simulation stumm.)
|
||||
2. **`hour=None`.** Würde man die Bar-Stunde übergeben, liefe die EIA-Prüfung in
|
||||
`_build` über `datetime.now(_BERLIN)` — also die ECHTE Uhrzeit des Laufs, für
|
||||
JEDEN Bar gleich. Ein Lauf am Mittwoch 15:30–16:30 würde damit alle Bars
|
||||
blocken. `dead_hours` ist live ohnehin leer; es fehlt also nur das
|
||||
EIA-Fenster (ein seltenes Ereignis-Gate).
|
||||
|
||||
Exit: `core/exit_model.LIVE` · Echtkosten (Bar-Spread/ATR) · 2 Halbjahre.
|
||||
⚠ Regel VORAB: ein anderes k wird nur empfohlen, wenn es in BEIDEN Hälften mehr
|
||||
ΣR liefert UND die Nachbarwerte mithalten.
|
||||
|
||||
Aufruf: python backtest_breakout_gated.py [bars]
|
||||
"""
|
||||
import sys
|
||||
|
||||
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
|
||||
_TIMEOUT_BARS = 12 # = _breakout_timeout_s 3600 s auf M5
|
||||
_PIV_K = 3 # Pivot-Fenster wie in `refresh_market`
|
||||
_PIV_LOOK = 300 # Pivot-Rückblick wie live
|
||||
_ATR_FLOOR_PB = 0.12 # `_pb_feats`-Floor wie live
|
||||
|
||||
|
||||
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 _htf_sign(C, i, n, atr):
|
||||
"""M30-/H1-Trend aus M5 aggregiert: EMA12/50 auf jedem n-ten Close."""
|
||||
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
|
||||
ef = _ema_series(sub, _EMA_FAST)[-1]; es = _ema_series(sub, _EMA_SLOW)[-1]
|
||||
d = (ef - es) / 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)
|
||||
|
||||
|
||||
def run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL, lo, hi, k):
|
||||
"""Ein Durchlauf mit dem vollen Gate-Stack für ein bestimmtes k."""
|
||||
def cost(i, atr):
|
||||
return (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||||
|
||||
Rs = []
|
||||
pend = None # spiegelt `_pend` aus `_confirm_breakout`
|
||||
i = max(lo, _PIV_LOOK + _EMA_SLOW + 5)
|
||||
while i < hi:
|
||||
atr = A[i]
|
||||
if not atr or atr < _ATRMIN:
|
||||
i += 1; continue
|
||||
# Raum-Gate mit echten, KAUSALEN M5-Pivots füttern (ein Pivot bei j ist
|
||||
# erst ab j+_PIV_K bekannt — sonst schaut das Gate in die Zukunft).
|
||||
w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
|
||||
w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
|
||||
"pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
|
||||
snap = {}
|
||||
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")
|
||||
|
||||
# ── Breakout-Bestätigung: Mechanik 1:1 aus `_confirm_breakout`,
|
||||
# Timeout bar-basiert statt Wall-Clock (s. Modulkopf).
|
||||
if k > 0 and 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 * k * atr,
|
||||
"invalid": C[i] - d * k * atr, "confirmed": False, "bar": i}
|
||||
if not pend["confirmed"]:
|
||||
broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
|
||||
against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
|
||||
if broke:
|
||||
pend["confirmed"] = True
|
||||
else:
|
||||
if against or (i - pend["bar"]) > _TIMEOUT_BARS:
|
||||
pend = {"dir": d, "level": C[i] + d * k * atr,
|
||||
"invalid": C[i] - d * k * atr,
|
||||
"confirmed": False, "bar": i}
|
||||
i += 1; continue # noch nicht bestätigt → WARTEN
|
||||
elif sig == "WARTEN":
|
||||
pend = None
|
||||
|
||||
if sig not in ("LONG", "SHORT"):
|
||||
i += 1; continue
|
||||
rec = w._room_gate(rec, C[i]) # Entry-Raum-Gate (echte Methode)
|
||||
if rec.get("signal") not in ("LONG", "SHORT"):
|
||||
i += 1; continue
|
||||
|
||||
d = 1 if rec["signal"] == "LONG" else -1
|
||||
a = max(atr, _ATRMIN)
|
||||
r, xj = simulate(C[i], d, a, H, L, C, i + 1, LIVE,
|
||||
timestop=True, use_tp=True, ret_bar=True)
|
||||
Rs.append(r - cost(i, a))
|
||||
pend = None
|
||||
i = xj + _COOL
|
||||
return Rs
|
||||
|
||||
|
||||
def main():
|
||||
n = 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)
|
||||
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]
|
||||
A = _atr_series(H, L, C)
|
||||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)
|
||||
N = len(C); mid = N // 2
|
||||
|
||||
print("Vorberechnung (Winkel, HTF-Trends, 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]) # M30 = 6 × M5
|
||||
H1[i] = _htf_sign(C, i, 12, A[i]) # H1 = 12 × M5
|
||||
PH = [j for j in range(_PIV_K, N - _PIV_K)
|
||||
if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
|
||||
PL = [j for j in range(_PIV_K, N - _PIV_K)
|
||||
if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]
|
||||
|
||||
w = WaveRecommender(_TU(), mt5.TIMEFRAME_M5)
|
||||
w.set_reversal_enabled(False) # Live-Stand seit 04.08.
|
||||
w.set_entry_room(0.6) # Live-Wert
|
||||
w.set_dead_hours([]) # live leer
|
||||
|
||||
print("=" * 94)
|
||||
print(f" breakout_k mit VOLLEM Gate-Stack — {sym} M5 ({N} Bars, 2 Halbjahre)")
|
||||
print(f" Gates: Totband · M30-Filter · Anti-Überdehnung · Winkel · min_conf 55 ·")
|
||||
print(f" Entry-Raum 0,6×ATR (echte M5-Pivots) · Reversal AUS")
|
||||
print(f" Exit: SL {LIVE.sl_atr}× · Trail {LIVE.mult}× · BE {LIVE.be}× · Time-Stop · TP")
|
||||
print("=" * 94)
|
||||
erg = {}
|
||||
for k in (0.0, 0.2, 0.3, 0.5, 0.8, 1.0):
|
||||
w.set_breakout_k(k)
|
||||
s1 = st(run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL, 0, mid, k))
|
||||
s2 = st(run(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL,
|
||||
mid, N - LIVE.max_hold - 1, k))
|
||||
erg[k] = (s1, s2)
|
||||
tag = " ← LIVE" if abs(k - 0.3) < 1e-9 else ""
|
||||
print(f"\n k = {k:.1f}{tag}")
|
||||
for lbl, s in (("H1 (alt)", s1), ("H2 (neu)", s2)):
|
||||
print(f" {lbl} " + (f"n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
|
||||
f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}"
|
||||
if s else "— keine Trades"))
|
||||
live = erg.get(0.3)
|
||||
if live and live[0] and live[1]:
|
||||
print("\n" + "=" * 94)
|
||||
print(" ΣR-Differenz gegen den Live-Wert k=0,3:")
|
||||
for k, (s1, s2) in sorted(erg.items()):
|
||||
if abs(k - 0.3) < 1e-9 or not (s1 and s2):
|
||||
continue
|
||||
beide = s1["sum"] > live[0]["sum"] and s2["sum"] > live[1]["sum"]
|
||||
print(f" k={k:.1f}: H1 {s1['sum'] - live[0]['sum']:+7.0f} "
|
||||
f"H2 {s2['sum'] - live[1]['sum']:+7.0f}"
|
||||
+ (" << in BEIDEN besser" if beide else ""))
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user