Nicht ausgeschlossen — aber in JEDER Variante schlechter je Trade als die M5-Kontrolle auf demselben Fenster (beste M1-Variante +0,474/+0,387 gegen +0,605/+0,491). Alle Nachbar-Parameter halten, Einstiegs-Slippage bis 0,20xATR haelt, der skalierte ATR-Floor erzeugt den Befund nicht (Kontrolle ohne Floor bestaetigt), Broker blockt nicht (trade_stops_level = 0). Entscheidend ist die EXIT-Slippage, und sie trifft M1 doppelt so hart, weil sie absolut ist: M1 kippt bei ~0,04 $, M5 erst bei ~0,09 $ (Faktor 2,2 = das ATR-Verhaeltnis). Gemessener Median 0,017 $ -> beide halten; gemessener Ø 0,234 $ -> beide fallen. Die entscheidende Zahl — Slippage bedingt auf Haltedauern <30 min — ist ungemessen. Urteil: nicht umstellen. M1s Vorteil kommt allein aus der 4,3-fachen Frequenz, und genau die multipliziert die ungemessene Groesse. Zwei Methodik-Lehren dokumentiert: der Haertetest lief zuerst auf der schwaechsten statt der besten Variante (Auswahl nach Reihenfolge statt nach Ergebnis), und die Regel "nicht schlechter als die Kontrolle" war wieder unterspezifiziert (ØR und ΣR zeigen gegenlaeufig) — deshalb gibt zeig() jetzt beide Spalten aus. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
294 lines
13 KiB
Python
294 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Traegt der Squeeze auch auf M1? — die offene „Variante B", jetzt gemessen.
|
||
|
||
ANLASS (2026-08-06, User): „Wie sinnvoll waere ein Scalping-Modul auf 1M-Basis?"
|
||
Der B3-Vorcheck (`analyze_scalping_costs.py`) hat ein eigenstaendiges M1-Signal
|
||
verworfen (Spread/ATR_M1 Median 0,319 → noetige Trefferquote 65,9 % bei 1:1).
|
||
Offen blieb die ENGERE Variante: **dieselbe validierte Squeeze-Regel, nur auf M1** —
|
||
kein neues Signal, sondern eine bekannte Regel auf feinerer Ebene.
|
||
|
||
⚠⚠ WAS DIESER TEST LEISTEN KANN — UND WAS NICHT.
|
||
Die M1-Historie des Brokers ist hart bei **80.000 Bars = 80 Tage** gedeckelt.
|
||
Zwei Haelften à 40 Tage liegen damit im **selben Regime**. Das reicht fuer ein
|
||
AUSSCHLUSS-Urteil (faellt es durch, ist es erledigt), aber **NICHT fuer eine
|
||
Bestaetigung** — genau daran sind ORB und die „profitable" Stunde 13 gescheitert.
|
||
Deshalb laeuft die **M5-Kontrolle auf DEMSELBEN Fenster** mit: sie sagt, ob M1 in
|
||
DIESEM Regime schlechter ist als die live laufende Ebene. Dieser Vergleich ist
|
||
design-intern und damit belastbarer als die Absolutwerte.
|
||
|
||
⚠ ATR-FLOOR: `kandidaten()` in `backtest_cost_gate.py` verwirft Bars mit
|
||
ATR < 0,06 (Live-Floor aus `core/trailing.py`). Auf M1 liegt die MITTLERE ATR bei
|
||
0,087 — derselbe Floor wuerde dort einen grossen Teil der ruhigen Phasen wegwerfen
|
||
und damit **den Befund erzeugen** (exakt die Fehlerklasse aus `analyze_tageszeit.py`
|
||
und dem ORB-Artefakt). Der Floor wird deshalb **massstabsgerecht skaliert** und
|
||
zusaetzlich ein Lauf OHNE Floor als Kontrolle gefahren.
|
||
|
||
VORAB FIXIERTE ENTSCHEIDUNGSREGEL (vor dem ersten Lauf festgelegt):
|
||
M1 wird nur dann als tragfaehig gewertet, wenn ALLE vier Punkte zutreffen —
|
||
(1) ØR > 0 UND PF > 1 in BEIDEN Haelften
|
||
(2) nicht schlechter als die M5-Kontrolle auf demselben Fenster
|
||
(3) die Nachbar-Parameter (Box 10/15/20, k 0,1/0,2) kippen das Vorzeichen nicht
|
||
(4) haelt dem Slippage-Haertetest bis 0,20×ATR stand
|
||
Sonst: verworfen.
|
||
|
||
Einstieg = **ruhende Stop-Order am Ausbruchs-Level** (so laeuft es seit 05.08. live),
|
||
Exit = kanonisch (`core/exit_model.LIVE`), Kosten = echter Bar-Spread/ATR,
|
||
sequentiell mit EINEM Positions-Slot.
|
||
|
||
Aufruf: python backtest_squeeze_m1.py
|
||
"""
|
||
import bisect
|
||
import sys
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
import core.wave_rec as wr
|
||
from backtest_cost_gate import _atr_series, _berlin, kennzahlen
|
||
from core.exit_model import LIVE, simulate
|
||
from core.wave_rec import WaveRecommender
|
||
|
||
_M1_BARS = 80000 # Broker-Deckel, mehr liefert MT5 nicht
|
||
_STUNDEN_OK = set(range(10, 18)) # 10–17 Uhr: dort kostet M1 wie M5 (0,21–0,27)
|
||
|
||
|
||
def hole(sym, tf, n):
|
||
b = mt5.copy_rates_from_pos(sym, tf, 0, n)
|
||
if b is None or len(b) < 500:
|
||
return None
|
||
point = mt5.symbol_info(sym).point
|
||
return ([float(x["high"]) for x in b], [float(x["low"]) for x in b],
|
||
[float(x["close"]) for x in b], [int(x["time"]) for x in b],
|
||
[float(x["spread"]) * point for x in b])
|
||
|
||
|
||
def kandidaten_tf(H, L, C, T, SP, A, floor, box=None, k=None):
|
||
"""Wie `backtest_cost_gate.kandidaten`, aber mit EXPLIZITEM ATR-Floor und
|
||
optional abweichender Box/k — sonst waere der Nachbar-Test nicht moeglich.
|
||
|
||
⚠ Bewusst eine eigene Kopie statt einer Aenderung an `backtest_cost_gate`:
|
||
das wuerde dessen dokumentierte Zahlen still verschieben (Migrations-Regel).
|
||
"""
|
||
alt_n, alt_m = wr._SQ_N, wr._SQ_K
|
||
if box is not None:
|
||
wr._SQ_N = box
|
||
if k is not None:
|
||
wr._SQ_K = k
|
||
out, last = [], None
|
||
try:
|
||
for i in range(wr._SQ_N + 2, len(C) - 2):
|
||
atr = A[i]
|
||
if not atr or atr < floor:
|
||
continue
|
||
w0 = i - wr._SQ_N - 2
|
||
st = WaveRecommender._squeeze_one(H[w0:i + 1], L[w0:i + 1],
|
||
C[w0:i + 1], atr)
|
||
if st.get("state") != "active":
|
||
last = None
|
||
continue
|
||
sig = (st["dir"], st["level"])
|
||
if sig == last:
|
||
continue
|
||
last = sig
|
||
out.append({"i": i, "d": 1 if st["dir"] == "LONG" else -1,
|
||
"lvl": float(st["level"]), "atr": atr,
|
||
"kosten": (SP[i] if SP[i] > 0 else 0.0225) / atr,
|
||
"stunde": _berlin(T[i]).hour})
|
||
finally:
|
||
wr._SQ_N, wr._SQ_K = alt_n, alt_m
|
||
return out
|
||
|
||
|
||
def lauf(K, H, L, C, lo, hi, slip=0.0, stunden=None, atr_exit=None,
|
||
exit_slip_usd=0.0):
|
||
"""Sequentiell, EIN Slot. Einstieg = Level + `slip`×ATR (Stop-Order).
|
||
|
||
`atr_exit` (optional) = abweichende Risiko-Einheit fuer den Exit, z. B. ATR_M5
|
||
bei einem M1-Einstieg („schneller Einstieg, normaler Stop"). Kosten werden
|
||
dann auf DIESELBE Einheit bezogen, damit R konsistent bleibt.
|
||
|
||
⚠⚠ `exit_slip_usd` in DOLLAR, nicht in ATR — und das ist der Kern der Frage.
|
||
Slippage ist eine ABSOLUTE Groesse (ein Gap fragt nicht, auf welcher Zeitebene
|
||
man handelt). Sie frisst deshalb von einem kleinen M1-Stop einen viel groesseren
|
||
Anteil als von einem M5-Stop: gemessen (`analyze_execution.py`) Median 0,017 $,
|
||
Ø 0,234 $ — bei ATR_M1 0,079 sind das 0,21 bzw. 2,95 R. Angesetzt auf JEDEN
|
||
Exit, weil alle Ausstiegsarten (SL, Trail, Time-Stop, TP) zum Markt fuellen.
|
||
"""
|
||
rs, frei_ab = [], lo
|
||
for kd in K:
|
||
i = kd["i"]
|
||
if not (lo <= i < hi) or i < frei_ab:
|
||
continue
|
||
if stunden is not None and kd["stunde"] not in stunden:
|
||
continue
|
||
d = kd["d"]
|
||
a_ein = kd["atr"] # Einstiegs-/Level-Massstab
|
||
a_ex = (atr_exit[i] if atr_exit else a_ein)
|
||
if not a_ex or a_ex <= 0:
|
||
continue
|
||
entry = kd["lvl"] + d * slip * a_ein
|
||
r, xb = simulate(entry, d, a_ex, H, L, C, i + 1, LIVE,
|
||
timestop=True, use_tp=True, ret_bar=True)
|
||
kosten = kd["kosten"] * a_ein / a_ex # Spread in Einheiten von a_ex
|
||
rs.append(r - kosten - exit_slip_usd / a_ex)
|
||
frei_ab = xb + 1
|
||
return rs
|
||
|
||
|
||
def zeig(lbl, r1, r2, warn=""):
|
||
"""Gibt ØR **und** ΣR aus — beides, absichtlich.
|
||
|
||
⚠ Am 05.08. ist genau hier eine Entscheidungsregel unbrauchbar geworden
|
||
(„ΣR UND ØR ≥ Referenz"): laufen die Varianten unterschiedlich viele Trades,
|
||
ziehen ØR und ΣR mechanisch gegeneinander. Wer nur eine Spalte zeigt, waehlt
|
||
die Antwort aus. Deshalb stehen hier beide, und die Auswahl passiert nicht
|
||
stillschweigend.
|
||
"""
|
||
k1, k2 = kennzahlen(r1), kennzahlen(r2)
|
||
if not k1 or not k2:
|
||
print(f" {lbl:<40} — zu wenige Trades")
|
||
return None
|
||
ok = k1["oer"] > 0 and k2["oer"] > 0 and k1["pf"] > 1 and k2["pf"] > 1
|
||
print(f" {lbl:<40} H1 ØR {k1['oer']:+.3f} PF {k1['pf']:.2f} ΣR {k1['sr']:+7.0f} "
|
||
f"n{k1['n']:<5} | H2 ØR {k2['oer']:+.3f} PF {k2['pf']:.2f} "
|
||
f"ΣR {k2['sr']:+7.0f} n{k2['n']:<5} {'✅' if ok else '❌'}{warn}")
|
||
return (min(k1["oer"], k2["oer"]) if ok else None)
|
||
|
||
|
||
def main():
|
||
if not mt5.initialize():
|
||
print("MT5 nicht erreichbar"); return 1
|
||
sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
|
||
if mt5.symbol_info(c)), None)
|
||
m1 = hole(sym, mt5.TIMEFRAME_M1, _M1_BARS)
|
||
if not m1:
|
||
mt5.shutdown(); print("keine M1-Daten"); return 1
|
||
H1_, L1_, C1_, T1_, SP1 = m1
|
||
A1 = _atr_series(H1_, L1_, C1_)
|
||
|
||
# M5 fuer (a) die Kontrolle und (b) die „M5-Stop"-Variante — gleiches Fenster
|
||
n_m5 = int(len(C1_) / 5) + 500
|
||
m5 = hole(sym, mt5.TIMEFRAME_M5, n_m5)
|
||
H5, L5, C5, T5, SP5 = m5
|
||
A5 = _atr_series(H5, L5, C5)
|
||
mt5.shutdown()
|
||
|
||
# M1-Index → ATR_M5 (fuer die Hybrid-Variante)
|
||
a5_bei_m1 = []
|
||
for t in T1_:
|
||
j = bisect.bisect_right(T5, t) - 1
|
||
a5_bei_m1.append(A5[j] if 0 <= j < len(A5) else 0.0)
|
||
|
||
med = lambda v: sorted(v)[len(v) // 2]
|
||
mA1 = med([a for a in A1[20:] if a > 0])
|
||
mA5 = med([a for a in A5[20:] if a > 0])
|
||
floor_m1 = round(0.06 * mA1 / mA5, 4)
|
||
|
||
print("=" * 94)
|
||
print(f" SQUEEZE AUF M1 — die offene Variante B, jetzt gemessen ({sym})")
|
||
print("=" * 94)
|
||
print(f" M1: {len(C1_)} Bars {_berlin(T1_[0]):%d.%m.%Y} – {_berlin(T1_[-1]):%d.%m.%Y}"
|
||
f" Median ATR {mA1:.4f}")
|
||
print(f" M5: {len(C5)} Bars {_berlin(T5[0]):%d.%m.%Y} – {_berlin(T5[-1]):%d.%m.%Y}"
|
||
f" Median ATR {mA5:.4f}")
|
||
print(f" ATR-Floor M1 massstabsgerecht skaliert: {floor_m1} (M5 live: 0.06)")
|
||
print()
|
||
print(" VORAB FIXIERT: tragfaehig nur bei ØR>0 UND PF>1 in BEIDEN Haelften,")
|
||
print(" nicht schlechter als die M5-Kontrolle, Nachbarn stabil, Slippage bis 0,20.")
|
||
print(" ⚠ 80 Tage = EIN Regime → das kann AUSSCHLIESSEN, nicht bestaetigen.")
|
||
|
||
lo1, hi1 = 0, len(C1_) // 2
|
||
lo2, hi2 = len(C1_) // 2, len(C1_)
|
||
lo5, hi5 = 0, len(C5) // 2
|
||
lo6, hi6 = len(C5) // 2, len(C5)
|
||
|
||
K1 = kandidaten_tf(H1_, L1_, C1_, T1_, SP1, A1, floor_m1)
|
||
K1_nofloor = kandidaten_tf(H1_, L1_, C1_, T1_, SP1, A1, 0.0)
|
||
K5 = kandidaten_tf(H5, L5, C5, T5, SP5, A5, 0.06)
|
||
print()
|
||
print(f" Ausbrueche gefunden: M1 {len(K1)} (ohne Floor {len(K1_nofloor)}) · "
|
||
f"M5-Kontrolle {len(K5)}")
|
||
|
||
print()
|
||
print("=" * 94)
|
||
print(" KONTROLLE — dieselbe Regel auf M5, DASSELBE Fenster")
|
||
print("=" * 94)
|
||
ref = zeig("M5 (live laufende Ebene)",
|
||
lauf(K5, H5, L5, C5, lo5, hi5),
|
||
lauf(K5, H5, L5, C5, lo6, hi6))
|
||
|
||
print()
|
||
print("=" * 94)
|
||
print(" M1-VARIANTEN (Einstieg als ruhende Stop-Order am Level)")
|
||
print("=" * 94)
|
||
erg = {}
|
||
erg["A"] = zeig("A) Stop M1-ATR, alle Stunden",
|
||
lauf(K1, H1_, L1_, C1_, lo1, hi1),
|
||
lauf(K1, H1_, L1_, C1_, lo2, hi2))
|
||
erg["B"] = zeig("B) Stop M5-ATR (Hybrid), alle Stunden",
|
||
lauf(K1, H1_, L1_, C1_, lo1, hi1, atr_exit=a5_bei_m1),
|
||
lauf(K1, H1_, L1_, C1_, lo2, hi2, atr_exit=a5_bei_m1))
|
||
erg["C"] = zeig("C) A + nur 10–17 Uhr",
|
||
lauf(K1, H1_, L1_, C1_, lo1, hi1, stunden=_STUNDEN_OK),
|
||
lauf(K1, H1_, L1_, C1_, lo2, hi2, stunden=_STUNDEN_OK))
|
||
erg["D"] = zeig("D) B + nur 10–17 Uhr",
|
||
lauf(K1, H1_, L1_, C1_, lo1, hi1, stunden=_STUNDEN_OK,
|
||
atr_exit=a5_bei_m1),
|
||
lauf(K1, H1_, L1_, C1_, lo2, hi2, stunden=_STUNDEN_OK,
|
||
atr_exit=a5_bei_m1))
|
||
erg["A0"] = zeig("A ohne ATR-Floor (Selektions-Kontrolle)",
|
||
lauf(K1_nofloor, H1_, L1_, C1_, lo1, hi1),
|
||
lauf(K1_nofloor, H1_, L1_, C1_, lo2, hi2))
|
||
|
||
# ⚠ Die beste Variante wird nach dem ERGEBNIS gewaehlt (schlechtere der beiden
|
||
# Haelften), nicht nach Reihenfolge — der erste Entwurf nahm stur D und hat
|
||
# damit ausgerechnet die SCHWAECHSTE M1-Variante haertegetestet.
|
||
beste = max((v for v in erg.items() if v[1] is not None),
|
||
key=lambda v: v[1], default=("A", 0))[0]
|
||
ax = a5_bei_m1 if beste in ("B", "D") else None
|
||
st = _STUNDEN_OK if beste in ("C", "D") else None
|
||
Kb = K1_nofloor if beste == "A0" else K1
|
||
print()
|
||
print("=" * 94)
|
||
print(f" SLIPPAGE-HAERTETEST auf der BESTEN M1-Variante ({beste})")
|
||
print("=" * 94)
|
||
print(" (1) EINSTIEG — die Stop-Order loest aus, fuellt aber zum Markt:")
|
||
for s in (0.0, 0.05, 0.10, 0.20):
|
||
zeig(f" {beste} · Einstiegs-Slippage +{s:.2f}×ATR",
|
||
lauf(Kb, H1_, L1_, C1_, lo1, hi1, slip=s, stunden=st, atr_exit=ax),
|
||
lauf(Kb, H1_, L1_, C1_, lo2, hi2, slip=s, stunden=st, atr_exit=ax))
|
||
print()
|
||
print(" (2) AUSSTIEG — in DOLLAR, weil Slippage nicht mit der Zeitebene skaliert.")
|
||
print(" Gemessen (analyze_execution.py): Median 0,017 $ · Ø 0,234 $ · Max 4,192 $.")
|
||
print(" Dieselben Betraege auch auf die M5-Kontrolle, damit der Vergleich haelt:")
|
||
for s_usd in (0.0, 0.017, 0.05, 0.10):
|
||
zeig(f" M1 {beste} · Exit-Slippage {s_usd:.3f} $",
|
||
lauf(Kb, H1_, L1_, C1_, lo1, hi1, stunden=st, atr_exit=ax,
|
||
exit_slip_usd=s_usd),
|
||
lauf(Kb, H1_, L1_, C1_, lo2, hi2, stunden=st, atr_exit=ax,
|
||
exit_slip_usd=s_usd))
|
||
for s_usd in (0.017, 0.05, 0.10):
|
||
zeig(f" M5-Kontrolle · Exit-Slippage {s_usd:.3f} $",
|
||
lauf(K5, H5, L5, C5, lo5, hi5, exit_slip_usd=s_usd),
|
||
lauf(K5, H5, L5, C5, lo6, hi6, exit_slip_usd=s_usd))
|
||
|
||
print()
|
||
print("=" * 94)
|
||
print(" NACHBAR-PARAMETER (Box / k) — kippt das Vorzeichen?")
|
||
print("=" * 94)
|
||
for box, kk in ((10, 0.1), (15, 0.1), (20, 0.1), (10, 0.2), (15, 0.2)):
|
||
Kx = kandidaten_tf(H1_, L1_, C1_, T1_, SP1, A1, floor_m1, box=box, k=kk)
|
||
zeig(f"Box {box} · k {kk} (n={len(Kx)})",
|
||
lauf(Kx, H1_, L1_, C1_, lo1, hi1, stunden=st, atr_exit=ax),
|
||
lauf(Kx, H1_, L1_, C1_, lo2, hi2, stunden=st, atr_exit=ax))
|
||
|
||
print()
|
||
print("=" * 94)
|
||
print(" Erinnerung: ✅ heisst NICHT bestaetigt — beide Haelften liegen im")
|
||
print(" selben Regime. Ein ❌ ist dagegen belastbar (Ausschluss).")
|
||
print("=" * 94)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|