#!/usr/bin/env python3 """Reversal-Konfidenz-Boni (offener Befund Review 2026-07-15): Im Reversal-Zweig vergeben „starker Trend" (+10/+5, |sep|-abhängig) und „⭐ tiefer Pullback" (+15, bei REV per Definition IMMER) Konfidenz-Boni, obwohl beide beim Reversal GEGEN das Signal stehen. Effekt: Reversals passieren das _MIN_CONF=55-Gate leichter. Messung: Signale aus der echten `_build`-Logik (M30-Filter + H1-Konfluenz + Winkel, hour=None = gate-frei wie alle Backtests). Für REV-Signale wird eine korrigierte Konfidenz conf_fix = conf − (15 + Trendstärke-Bonus) gerechnet. Vergleich Gate LIVE (conf≥55) vs. Gate FIX (conf_fix≥55) — entscheidend ist der Edge der Trades, die durch den Fix NEU RAUSFALLEN: nur wenn die in BEIDEN Hälften negativ sind, ist der Fix berechtigt. Exit = live (SL2,0/Trail1,5/BE1,3), Kosten = Bar-Spread/ATR. R = Profit/ATR. """ import sys, bisect import MetaTrader5 as mt5 from core.analysis import calc_trend_angle 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; _GATE = 55 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 sim(entry, d, atr, H, L, C, j0, sl_atr=2.0, trail=1.5, trail_on=0.3, be_on=1.3): eff = entry - d*sl_atr*atr; hw = entry end = min(j0+_MAXH, len(C)-1); exit_px = C[end] for j in range(j0, end+1): hi, lo = H[j], L[j] if (lo <= eff) if d > 0 else (hi >= eff): exit_px = eff; break hw = max(hw, hi) if d > 0 else min(hw, lo) prof = (C[j]-entry)*d if prof >= trail_on*atr: cand = hw - d*trail*atr if prof >= be_on*atr: cand = max(cand, entry) if d > 0 else min(cand, entry) eff = max(eff, cand) if d > 0 else min(eff, cand) return (exit_px-entry)*d/atr 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 99.9), sum=s) def line(lbl, s): if not s: return f" {lbl:<30} —" return (f" {lbl:<30} 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] 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): i = bisect.bisect_right(tt, ts)-1 if i < _EMA_SLOW or aa[i] is None or aa[i] <= 0: return 0 dd = ef[i]-es[i] return 0 if abs(dd) < _HTF_DEADBAND*aa[i] else (1 if dd > 0 else -1) w = WaveRecommender(type("T", (), {"snapshot": lambda s: {"intervals": {}}})(), mt5.TIMEFRAME_M5) def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr sigs = [] # (i, d, atr, conf, conf_fix, is_rev) for i in range(_N_BARS, len(C)-_MAXH-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) if rec["signal"] == "WARTEN": continue d = 1 if rec["signal"] == "LONG" else -1 conf = int(rec.get("conf_pct") or 0) is_rev = "REV" in (rec.get("setup") or "") conf_fix = conf if is_rev: sep = (ef - es) / atr bonus = 15 + (10 if abs(sep) >= 0.5 else 5 if abs(sep) >= 0.25 else 0) conf_fix = conf - bonus sigs.append((i, d, max(atr, _ATRMIN), conf, conf_fix, is_rev)) mid = len(C)//2 print("="*86) print(f" Reversal-Konfidenz-Boni — {sym} M5 ({len(sigs)} Signale · Gate {_GATE} · Echtkosten)") print(f" FIX = beim Reversal ohne '+15 Pullback' und '+10/5 Trendstärke'") print("="*86) for lbl, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, len(C))): seg = [x for x in sigs if lo <= x[0] < hi] def R(x): return sim(C[x[0]], x[1], x[2], H, L, C, x[0]+1) - cost(x[0], x[2]) rev_all = [x for x in seg if x[5]] rev_live = [x for x in rev_all if x[3] >= _GATE] rev_fix = [x for x in rev_all if x[4] >= _GATE] dropped = [x for x in rev_all if x[3] >= _GATE and x[4] < _GATE] print(f"\n{lbl}: ({len(seg)} Signale, davon {len(rev_all)} Reversal)") print(line("REV alle (ohne Gate)", st([R(x) for x in rev_all]))) print(line("REV durch Gate LIVE (jetzt)", st([R(x) for x in rev_live]))) print(line("REV durch Gate FIX", st([R(x) for x in rev_fix]))) sd = st([R(x) for x in dropped]) print(line("→ FÄLLT NEU RAUS (Fix-Opfer)", sd) + (" <- Edge<0 = Fix berechtigt" if sd and sd["oR"] < 0 else "")) print(f"\n Fix NUR umsetzen, wenn die 'fällt neu raus'-Gruppe in BEIDEN Hälften ØR<0 hat") print(f" (sonst würden profitable Reversals unters Gate gedrückt — 7× gelernte Lektion).") if __name__ == "__main__": main()