#!/usr/bin/env python3 """Entry-Raum-Gate (Lösung gegen Klein-Close-Trades, 2026-07-16): Kein Entry, wenn das nächste GEGENLEVEL (Pivot in Trade-Richtung) < X×ATR entfernt ist — der Ertrag ist dort durch den S/R-Auto-Close gedeckelt (63 % Containment), die Kosten (real ~0,265×ATR) fressen den Rest → strukturell negativer Erwartungswert. Sim = Live-Politik: sequentiell (1 Position), Exit via SL 2,0×ATR + Trailing 1,5 + BE 1,3 UND S/R-Auto-Close (am Gegenlevel, wenn P(break)<0,6 — kalibriertes Modell aus engine._p_break, Features trainingsgleich M5). Gruppierung der Trades nach Entry-Distanz zum Gegenlevel. 2 Halbjahre, Kosten = Bar-Spread/ATR. Gate-Schwelle X nur setzen, wo die Gruppe in BEIDEN Hälften negativ ist. """ import sys, bisect import MetaTrader5 as mt5 from core.analysis import calc_trend_angle from core.engine import _p_break from core.wave_rec import (WaveRecommender, _atr, _ema_last, _ema_series, _EMA_FAST, _EMA_SLOW, _N_BARS, _HTF_DEADBAND, _ANGLE_LR) _MAXH = 288; _ATRMIN = 0.12; _PIV_K = 3; _LOOKBACK = 300; _PTHR = 0.60 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 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 line(lbl, s): if not s: return f" {lbl:<26} —" return (f" {lbl:<26} n={s['n']:>4} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} " f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+6.0f}") 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+_N_BARS+_MAXH+5) m30 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n//6+500) h1 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_H1, 0, n//12+500) si = mt5.symbol_info(sym); point = si.point mt5.shutdown() T = [int(b["time"]) for b in bars]; 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) def series(rr): t = [int(b["time"]) for b in rr]; c = [float(b["close"]) for b in rr] hh = [float(b["high"]) for b in rr]; ll = [float(b["low"]) for b in rr] return t, _ema_series(c, _EMA_FAST), _ema_series(c, _EMA_SLOW), _atr_series(hh, ll, c) mT, mEf, mEs, mA = series(m30) hT, hEf, hEs, hA = series(h1) def tf_sign(tt, ef, es, aa, ts): k = bisect.bisect_right(tt, ts)-1 if k < _EMA_SLOW or aa[k] is None or aa[k] <= 0: return 0 dd = ef[k]-es[k] return 0 if abs(dd) < _HTF_DEADBAND*aa[k] else (1 if dd > 0 else -1) # Signale je Bar (echte _build-Logik) + volle EMA-Serien für P(break)-Features w = WaveRecommender(type("T", (), {"snapshot": lambda s: {"intervals": {}}})(), mt5.TIMEFRAME_M5) EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW) sig = [0]*len(C) for i in range(_N_BARS, len(C)-1): wc = C[i-_N_BARS:i]; wh = H[i-_N_BARS:i]; wl = L[i-_N_BARS:i] atr = _atr(wh, wl, wc) if not atr or atr <= 0: continue ef = _ema_last(wc, _EMA_FAST); es = _ema_last(wc, _EMA_SLOW) a5 = calc_trend_angle(C[i-_ANGLE_LR-2:i], _ANGLE_LR) rec, _ = w._build(ef, es, C[i-1], atr, "M5", 0, htf_trend=tf_sign(mT, mEf, mEs, mA, T[i]), h1_trend=tf_sign(hT, hEf, hEs, hA, T[i]), angle=a5) s_ = rec["signal"] sig[i] = 1 if s_ == "LONG" else -1 if s_ == "SHORT" else 0 # Pivot-Events: (bestätigt_ab_bar, preis) — Pivot an Bar p ist ab p+_PIV_K bekannt phE = [(p+_PIV_K, H[p]) for p in range(_PIV_K, len(C)-_PIV_K) if H[p] == max(H[p-_PIV_K:p+_PIV_K+1])] plE = [(p+_PIV_K, L[p]) for p in range(_PIV_K, len(C)-_PIV_K) if L[p] == min(L[p-_PIV_K:p+_PIV_K+1])] phT = [e[0] for e in phE]; plT = [e[0] for e in plE] def next_level(i, px, d): """Nächstes Gegenlevel (Pivot in Trade-Richtung) aus Bars [i-LOOKBACK, i].""" if d > 0: k = bisect.bisect_right(phT, i) cands = [pr for (t_, pr) in phE[max(0, k-80):k] if t_ >= i-_LOOKBACK and pr > px] return min(cands) if cands else None k = bisect.bisect_right(plT, i) cands = [pr for (t_, pr) in plE[max(0, k-80):k] if t_ >= i-_LOOKBACK and pr < px] return max(cands) if cands else None def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr def run(lo, hi_): """Sequentielle Sim mit Live-Exit inkl. S/R-Auto-Close. Liefert Liste (dist_entry_atr|None, R).""" out = [] i = lo while i < hi_: d = sig[i] if d == 0: i += 1; continue atr = max(A[i] or 0, _ATRMIN) entry = C[i] lvl = next_level(i, entry, d) dist0 = (lvl-entry)*d/atr if lvl else None eff = entry - d*2.0*atr; hw = entry end = min(i+_MAXH, len(C)-1); exit_px = C[end]; exit_j = end for j in range(i+1, end+1): hj, lj = H[j], L[j] if (lj <= eff) if d > 0 else (hj >= eff): exit_px = eff; exit_j = j; break hw = max(hw, hj) if d > 0 else min(hw, lj) prof = (C[j]-entry)*d # S/R-Auto-Close: Touch des Gegenlevels + P(break)<0,6 + im Plus if lvl is not None and ((hj >= lvl) if d > 0 else (lj <= lvl)): a_j = max(A[j] or atr, _ATRMIN) if len(C) > 7 and j >= 7: mom6 = (C[j]-C[j-6])/a_j*d; mom3 = (C[j]-C[j-3])/a_j*d wt = 1.0 if (EF[j]-ES[j])*d > 0 else 0.0 p = _p_break(mom6, mom3, wt, abs(lvl-entry)/a_j) if p < _PTHR and (lvl-entry)*d > 0: exit_px = lvl; exit_j = j; break lvl = next_level(j, C[j], d) # Durchbruch → nächstes Level if prof >= 0.3*atr: cand = hw - d*1.5*atr if prof >= 1.3*atr: cand = max(cand, entry) if d > 0 else min(cand, entry) eff = max(eff, cand) if d > 0 else min(eff, cand) out.append((dist0, (exit_px-entry)*d/atr - cost(i, atr))) i = exit_j + 1 return out mid = len(C)//2 print("="*92) print(f" Entry-Raum-Gate — {sym} M5 (seq. Sim, Exit=SL/Trail/BE + S/R-Close P<0,6 · Echtkosten)") print(f" Gruppen nach Entry-Distanz zum GEGENLEVEL (×ATR). Gate dort, wo BEIDE Hälften rot.") print("="*92) B = [(0.0, 0.3), (0.3, 0.6), (0.6, 1.0), (1.0, 2.0), (2.0, 99.0)] for lbl, lo, hi_ in (("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C)-_MAXH-1)): res = run(lo, hi_) print(f"\n{lbl}: ({len(res)} Trades)") print(line("GESAMT", st([r for _, r in res]))) for a, b in B: grp = [r for dd, r in res if dd is not None and a <= dd < b] print(line(f"Raum {a:.1f}-{b:.1f}xATR", st(grp))) print(line("kein Gegenlevel (frei)", st([r for dd, r in res if dd is None]))) if __name__ == "__main__": main()