S/R-Level von M5 auf M30 umgestellt + P(break)-Modell nachtrainiert
User: "DIE S/R Linien liegen oft sehr nahe beieinander" -> Option 2 (beides umstellen, Modell nachtrainieren) gewaehlt. Erst gemessen: backtest_level_tf.py, 80k M5-Bars, 2 Halbjahre, Trades bleiben M5, nur die Level-Quelle variiert. Ergebnis (M30 gewinnt in BEIDEN Haelften): TF AUC(H2) Abstand <1xATR dR bei P<0.55 (H1/H2) M5 0.710 0.61xATR 71% +6698 +7749 (bisher live) M15 0.725 1.16xATR 46% +10138 +12071 M30 0.721 1.95xATR 28% +12424 +13047 <- gewaehlt Die Beschwerde ist damit quantifiziert: 71% der M5-Level lagen naeher als 1xATR. Umsetzung: - wave_rec: _m30_hl gespeichert, neues _pb_levels30 (k=3, letzte 50 M30-Bars = trainingsgleich zum 300-M5-Lookback), im Snapshot. - engine._draw_levels nutzt pb_levels30 (Fallback pb_levels beim Kaltstart); speist Chart, MQL5-CSV, Dashboard UND den P(break)-Auto-Close. - _PB_MU/_PB_SD/_PB_W auf M30 nachtrainiert: out-of-sample validiert (H1->H2), dann auf allen 80k gefittet (34771 Touches, Kalibrierung 23/25, 38/36, 58/59, 86/87). Merkmale bleiben M5. Alte M5-Werte als Kommentar (Rueckweg). - NICHT mitveraendert: Entry-Raum-Gate nutzt weiter M5-pb_levels (entry_room_atr=0.6 ist darauf kalibriert). Verifiziert live: R/S 84.617/83.722 = 2.9xATR Abstand (vorher 0.28xATR), M30-Pivots 4/7 statt M5 27/31, keine Fehler im Log. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
06b37a1474
commit
b77b3d60f5
@@ -0,0 +1,283 @@
|
||||
#!/usr/bin/env python3
|
||||
"""backtest_level_tf.py — Auf WELCHER Timeframe sollen die S/R-Level gebildet werden?
|
||||
|
||||
User-Beobachtung 2026-07-30: „die S/R-Linien liegen oft sehr nahe beieinander".
|
||||
Stimmt — `_draw_levels` nutzt rohe **M5**-Pivots (k=3), und M5 erzeugt viele kleine
|
||||
Zacken. Der User wählte Option 2: Level-TF umstellen UND das P(break)-Modell auf der
|
||||
neuen TF nachtrainieren/prüfen (statt nur die Chart-Anzeige zu ändern).
|
||||
|
||||
⚠ Warum das NICHT nur Kosmetik ist: dieselben Level speisen den **P(break)-Auto-Close**
|
||||
(`engine._sr_close_hint`). Das Modell (`_PB_W`) ist auf M5-Pivots trainiert (AUC 0,71).
|
||||
Mit M15-/M30-Leveln wäre es nicht mehr kalibriert → erst messen, dann umstellen.
|
||||
|
||||
Gemessen wird je Level-TF (M5 / M15 / M30), Trades bleiben M5-basiert:
|
||||
1. **Level-ABSTÄNDE** — wie weit ist das Gegenlevel typisch weg (×ATR)? Beantwortet
|
||||
die Ausgangsbeobachtung mit Zahlen.
|
||||
2. **P(break)-Modell NEU trainiert** (H1) und **out-of-sample geprüft** (H2):
|
||||
AUC + Kalibrierung. Ein Level-Satz taugt nur, wenn Durchbrüche vorhersagbar bleiben.
|
||||
3. **R-Ertrag** der Close-Regel (P<Schwelle → am Level schließen) gegen die
|
||||
Trailing-Baseline, in BEIDEN Hälften, mit Echtkosten.
|
||||
|
||||
Umgestellt wird nur, wenn eine höhere TF in AUC **und** R-Ertrag mindestens
|
||||
gleichwertig ist — sonst bleibt M5 (dann ggf. nur die Chart-Anzeige ändern).
|
||||
|
||||
Aufruf: python backtest_level_tf.py [n_bars]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
import MetaTrader5 as mt5
|
||||
|
||||
from core.analysis import calc_trend_angle
|
||||
from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD,
|
||||
_REVERSAL_STRETCH, _STRETCH_MAX)
|
||||
|
||||
_MAXH = 200; _ATRMIN = 0.12
|
||||
_SL_ATR = 2.0; _TRAILON = 0.3; _MULT = 1.5; _BE = 1.3
|
||||
_LOCK_START = 3.5; _LOCK_SCALE = 0.6; _LOCK_MIN = 1.2; _TP_INIT = 3.5
|
||||
_PIV_K = 3; _LOOKBACK = 300
|
||||
_BRK_W = 12; _BRK_ATR = 0.5 # Bruch-Definition (wie im Live-Modell)
|
||||
_TFS = (("M5", 1), ("M15", 3), ("M30", 6)) # (Label, M5-Bars je HTF-Bar)
|
||||
|
||||
|
||||
def _ema_series(v, p):
|
||||
k = 2.0 / (p + 1); o = []; e = v[0]
|
||||
for i, x in enumerate(v):
|
||||
e = x if i == 0 else x * k + e * (1 - k); o.append(e)
|
||||
return o
|
||||
|
||||
|
||||
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])))
|
||||
out = [None] * len(C); run = 0.0
|
||||
for i in range(1, len(C)):
|
||||
run += t[i]
|
||||
if i > p: run -= t[i - p]
|
||||
out[i] = run / min(i, p)
|
||||
return out
|
||||
|
||||
|
||||
def sim_run(entry, d, atr, H, L, C, j0):
|
||||
"""Live-Exit (Trailing + Phasen-Ratsche) — identisch zu backtest_srclose_prob.py."""
|
||||
sl = entry - d * _SL_ATR * atr; tp = entry + d * _TP_INIT * atr
|
||||
hw = entry; rank = 0
|
||||
end = min(j0 + _MAXH, len(C) - 1)
|
||||
for j in range(j0, end + 1):
|
||||
hi, lo = H[j], L[j]
|
||||
if (lo <= sl) if d > 0 else (hi >= sl): return (sl - entry) * d / atr
|
||||
if (hi >= tp) if d > 0 else (lo <= tp): return (tp - entry) * d / atr
|
||||
hw = max(hw, hi) if d > 0 else min(hw, lo)
|
||||
profit = (hw - entry) * d
|
||||
ph = 0 if profit < _TRAILON * atr else (1 if profit < _LOCK_START * atr else 2)
|
||||
if ph < rank: ph = rank
|
||||
rank = ph
|
||||
if ph == 1:
|
||||
cand = hw - d * _MULT * atr
|
||||
if profit >= _BE * atr: cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||||
sl = max(sl, cand) if d > 0 else min(sl, cand)
|
||||
elif ph == 2:
|
||||
tm = max(_LOCK_MIN, _MULT * _LOCK_SCALE); cand = hw - d * tm * atr
|
||||
cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||||
sl = max(sl, cand) if d > 0 else min(sl, cand)
|
||||
return (C[end] - entry) * d / atr
|
||||
|
||||
|
||||
def pivots_for_tf(H, L, step):
|
||||
"""Pivots auf der aggregierten TF, zurückgegeben als (m5_index_bestätigt, preis).
|
||||
|
||||
LOOK-AHEAD-frei: ein HTF-Pivot bei Index j ist erst bekannt, wenn k Bars DANACH
|
||||
abgeschlossen sind → verfügbar ab M5-Index (j+k+1)·step."""
|
||||
if step == 1:
|
||||
ph = [(i + _PIV_K + 1, H[i]) for i in range(_PIV_K, len(H) - _PIV_K)
|
||||
if H[i] == max(H[i - _PIV_K:i + _PIV_K + 1])]
|
||||
pl = [(i + _PIV_K + 1, L[i]) for i in range(_PIV_K, len(L) - _PIV_K)
|
||||
if L[i] == min(L[i - _PIV_K:i + _PIV_K + 1])]
|
||||
return ph, pl
|
||||
hh, ll = [], []
|
||||
for s in range(0, len(H) - step + 1, step):
|
||||
hh.append(max(H[s:s + step])); ll.append(min(L[s:s + step]))
|
||||
ph, pl = [], []
|
||||
for j in range(_PIV_K, len(hh) - _PIV_K):
|
||||
avail = (j + _PIV_K + 1) * step
|
||||
if hh[j] == max(hh[j - _PIV_K:j + _PIV_K + 1]): ph.append((avail, hh[j]))
|
||||
if ll[j] == min(ll[j - _PIV_K:j + _PIV_K + 1]): pl.append((avail, ll[j]))
|
||||
return ph, pl
|
||||
|
||||
|
||||
def collect(a, b, H, L, C, EF, ES, AT, SP, PH, PL):
|
||||
"""Wellen-Events sammeln; Level aus der übergebenen Pivot-Liste (TF-spezifisch)."""
|
||||
TH = _REVERSAL_STRETCH; out = []
|
||||
ph_i = 0; pl_i = 0
|
||||
ph_live: list[tuple[int, float]] = []; pl_live: list[tuple[int, float]] = []
|
||||
start = max(a, _N_BARS, _LOOKBACK)
|
||||
# Pivots, die vor `start` schon bekannt sind, vorladen
|
||||
while ph_i < len(PH) and PH[ph_i][0] <= start: ph_live.append(PH[ph_i]); ph_i += 1
|
||||
while pl_i < len(PL) and PL[pl_i][0] <= start: pl_live.append(PL[pl_i]); pl_i += 1
|
||||
for i in range(start, min(b, len(C) - _MAXH - 1)):
|
||||
while ph_i < len(PH) and PH[ph_i][0] <= i: ph_live.append(PH[ph_i]); ph_i += 1
|
||||
while pl_i < len(PL) and PL[pl_i][0] <= i: pl_live.append(PL[pl_i]); pl_i += 1
|
||||
atr = AT[i]
|
||||
if not atr or atr <= 0: continue
|
||||
atr = max(atr, _ATRMIN); es = ES[i]; ef = EF[i]
|
||||
stretch = (C[i] - es) / atr
|
||||
ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR); ad = ang - 90.0
|
||||
d = 0
|
||||
if stretch <= -TH and ad >= _ANGLE_DEAD: d = 1
|
||||
elif stretch >= TH and ad <= -_ANGLE_DEAD: d = -1
|
||||
elif abs(stretch) < _STRETCH_MAX: d = 1 if ef > es else -1 if ef < es else 0
|
||||
if not d: continue
|
||||
entry = C[i]; cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||||
R_run = sim_run(entry, d, atr, H, L, C, i + 1) - cost
|
||||
rec = {"R_run": R_run, "touched": False}
|
||||
# nur Pivots aus dem Lookback-Fenster verwenden
|
||||
lo_idx = i - _LOOKBACK
|
||||
if d > 0:
|
||||
cands = [p for (ix, p) in ph_live if ix >= lo_idx and p > entry + 0.3 * atr]
|
||||
level = min(cands) if cands else None
|
||||
else:
|
||||
cands = [p for (ix, p) in pl_live if ix >= lo_idx and p < entry - 0.3 * atr]
|
||||
level = max(cands) if cands else None
|
||||
if level is not None:
|
||||
rec["dist_atr"] = abs(level - entry) / atr
|
||||
jt = None
|
||||
for j in range(i + 1, min(i + _MAXH, len(C) - _BRK_W)):
|
||||
if ((H[j] >= level) if d > 0 else (L[j] <= level)): jt = j; break
|
||||
if ((entry - L[j]) if d > 0 else (H[j] - entry)) >= 2.0 * atr: break
|
||||
if jt is not None:
|
||||
up = level + d * _BRK_ATR * atr; dn = level - d * _BRK_ATR * atr
|
||||
brk = False
|
||||
for j in range(jt, min(jt + _BRK_W, len(H))):
|
||||
if (H[j] >= up) if d > 0 else (L[j] <= up): brk = True; break
|
||||
if (L[j] <= dn) if d > 0 else (H[j] >= dn): brk = False; break
|
||||
rec.update(touched=True, brk=1.0 if brk else 0.0,
|
||||
R_close=(level - entry) * d / atr - cost,
|
||||
feat=[(C[jt] - C[max(0, jt - 6)]) * d / atr,
|
||||
(C[jt] - C[max(0, jt - 3)]) * d / atr,
|
||||
1.0 if (EF[jt] - ES[jt]) * d > 0 else 0.0,
|
||||
abs(level - entry) / atr])
|
||||
out.append(rec)
|
||||
return out
|
||||
|
||||
|
||||
def fit_logreg(X, y, iters=3000, lr=0.3):
|
||||
n, m = X.shape; Xb = np.hstack([np.ones((n, 1)), X]); w = np.zeros(m + 1)
|
||||
for _ in range(iters):
|
||||
p = 1 / (1 + np.exp(-(Xb @ w))); w -= lr * (Xb.T @ (p - y)) / n
|
||||
return w
|
||||
|
||||
|
||||
def auc(y, p):
|
||||
"""ROC-AUC über Rangstatistik (Mann-Whitney)."""
|
||||
y = np.asarray(y); p = np.asarray(p)
|
||||
pos = p[y == 1]; neg = p[y == 0]
|
||||
if len(pos) == 0 or len(neg) == 0: return float("nan")
|
||||
order = np.argsort(np.concatenate([pos, neg]))
|
||||
ranks = np.empty(len(order), float); ranks[order] = np.arange(1, len(order) + 1)
|
||||
return (ranks[:len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))
|
||||
|
||||
|
||||
def main():
|
||||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||||
mt5.initialize(); sym = None
|
||||
for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
||||
if mt5.symbol_info(c): sym = c; break
|
||||
si = mt5.symbol_info(sym); point = si.point
|
||||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
|
||||
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]
|
||||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW); AT = _atr_series(H, L, C)
|
||||
mid = len(C) // 2
|
||||
|
||||
print("=" * 96)
|
||||
print(f" LEVEL-TIMEFRAME-VERGLEICH — {sym} ({len(C)} M5-Bars, 2 Halbjahre)")
|
||||
print(f" Trades bleiben M5; nur die S/R-LEVEL-Quelle variiert. Echtkosten, Live-Exit.")
|
||||
print(f" Aktuell live: M5-Pivots (k={_PIV_K}) — Modell _PB_W darauf trainiert (AUC 0,71)")
|
||||
print("=" * 96)
|
||||
|
||||
summary = {}
|
||||
for lbl, step in _TFS:
|
||||
PH, PL = pivots_for_tf(H, L, step)
|
||||
ev1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, SP, PH, PL)
|
||||
ev2 = collect(mid, len(C), H, L, C, EF, ES, AT, SP, PH, PL)
|
||||
t1 = [e for e in ev1 if e["touched"]]; t2 = [e for e in ev2 if e["touched"]]
|
||||
if len(t1) < 200 or len(t2) < 200:
|
||||
print(f"\n### {lbl}: zu wenige Touches ({len(t1)}/{len(t2)}) — übersprungen")
|
||||
continue
|
||||
X1 = np.array([e["feat"] for e in t1]); y1 = np.array([e["brk"] for e in t1])
|
||||
mu = X1.mean(0); sd = X1.std(0) + 1e-9
|
||||
w = fit_logreg((X1 - mu) / sd, y1)
|
||||
|
||||
def pb(e):
|
||||
x = (np.array(e["feat"]) - mu) / sd
|
||||
return 1 / (1 + np.exp(-(w @ np.concatenate([[1.0], x]))))
|
||||
|
||||
for e in t1 + t2: e["_p"] = pb(e)
|
||||
a2 = auc([e["brk"] for e in t2], [e["_p"] for e in t2])
|
||||
a1 = auc([e["brk"] for e in t1], [e["_p"] for e in t1])
|
||||
d1 = [e["dist_atr"] for e in ev1 if "dist_atr" in e]
|
||||
d2 = [e["dist_atr"] for e in ev2 if "dist_atr" in e]
|
||||
near1 = 100 * sum(1 for x in d1 if x < 1.0) / max(1, len(d1))
|
||||
near2 = 100 * sum(1 for x in d2 if x < 1.0) / max(1, len(d2))
|
||||
print(f"\n### LEVEL-TF {lbl} (Pivots auf {lbl}, Trades M5)")
|
||||
print(f" Level-ABSTAND (×ATR): H1 Median {np.median(d1):.2f} · H2 {np.median(d2):.2f}"
|
||||
f" | <1×ATR entfernt: H1 {near1:.0f}% · H2 {near2:.0f}%")
|
||||
print(f" Touches: H1 {len(t1)} · H2 {len(t2)}"
|
||||
f" | Basisrate P(break): H1 {np.mean(y1):.0%} · H2 {np.mean([e['brk'] for e in t2]):.0%}")
|
||||
print(f" Modell (auf H1 trainiert): AUC in-sample H1 {a1:.3f} · "
|
||||
f"**out-of-sample H2 {a2:.3f}**")
|
||||
print(f" Koeff: mom6={w[1]:+.2f} mom3={w[2]:+.2f} wt={w[3]:+.2f} dist={w[4]:+.2f}")
|
||||
# Kalibrierung out-of-sample (H2)
|
||||
print(f" Kalibrierung H2: ", end="")
|
||||
for lo, hi in ((0.0, 0.3), (0.3, 0.5), (0.5, 0.7), (0.7, 1.01)):
|
||||
g = [e for e in t2 if lo <= e["_p"] < hi]
|
||||
if len(g) >= 20:
|
||||
print(f"[{lo:.1f}-{hi:.1f}) pred {np.mean([e['_p'] for e in g]):.0%}"
|
||||
f"/real {np.mean([e['brk'] for e in g]):.0%} (n={len(g)}) ", end="")
|
||||
print()
|
||||
# R-Ertrag der Close-Regel
|
||||
res = {}
|
||||
for hl, ev, ts in (("H1", ev1, t1), ("H2", ev2, t2)):
|
||||
base = sum(e["R_run"] for e in ev)
|
||||
best = None
|
||||
row = []
|
||||
for X in (0.45, 0.50, 0.55, 0.60, 0.70):
|
||||
tot = (sum((e["R_close"] if e["_p"] < X else e["R_run"]) for e in ts)
|
||||
+ sum(e["R_run"] for e in ev if not e["touched"]))
|
||||
row.append((X, tot - base))
|
||||
if best is None or tot - base > best[1]: best = (X, tot - base)
|
||||
res[hl] = (base, row, best)
|
||||
print(f" R-Ertrag {hl}: Baseline(nur Trailing) ΣR={base:+.0f} → "
|
||||
+ " ".join(f"P<{X:.2f}: {dR:+.0f}" for X, dR in row))
|
||||
summary[lbl] = {"auc_h2": a2, "d_med": (np.median(d1) + np.median(d2)) / 2,
|
||||
"near": (near1 + near2) / 2,
|
||||
"best55": (dict(res["H1"][1])[0.55], dict(res["H2"][1])[0.55]),
|
||||
"touch": (len(t1), len(t2))}
|
||||
|
||||
print("\n" + "=" * 96)
|
||||
print(" URTEIL — umstellen nur, wenn eine höhere TF in AUC UND R-Ertrag mithält:")
|
||||
print(f" {'TF':<6}{'AUC(H2)':>9}{'Level-Abstand':>15}{'<1xATR':>9}"
|
||||
f"{'ΔR bei P<0,55 (H1/H2)':>26}")
|
||||
for lbl in [t[0] for t in _TFS]:
|
||||
if lbl not in summary: continue
|
||||
s = summary[lbl]
|
||||
print(f" {lbl:<6}{s['auc_h2']:>9.3f}{s['d_med']:>13.2f}xATR{s['near']:>8.0f}%"
|
||||
f"{s['best55'][0]:>+13.0f}{s['best55'][1]:>+13.0f}")
|
||||
if "M5" in summary:
|
||||
base = summary["M5"]
|
||||
for lbl in ("M15", "M30"):
|
||||
if lbl not in summary: continue
|
||||
s = summary[lbl]
|
||||
ok_auc = s["auc_h2"] >= base["auc_h2"] - 0.02
|
||||
ok_r = (s["best55"][0] >= base["best55"][0] * 0.9
|
||||
and s["best55"][1] >= base["best55"][1] * 0.9)
|
||||
print(f" → {lbl}: AUC {'ok' if ok_auc else 'SCHLECHTER'} · "
|
||||
f"R-Ertrag {'ok' if ok_r else 'SCHLECHTER'} · "
|
||||
f"{'UMSTELLEN vertretbar' if (ok_auc and ok_r) else 'NICHT umstellen'}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user