#!/usr/bin/env python3 """Misst, ob eigene Indikatoren (MACD / ADX / RSI) als Konfluenz den Edge der Live-Empfehlung (M5 + M30-Filter) heben — VOR jeder Integration ("erst messen"). Idee: keinen externen Anbieter, sondern lokal aus den MT5-Bars rechnen (Trend=EMA/ADX, Momentum=MACD/RSI). Pro EMA-Signal wird der Vorlauf-Edge je Indikator-Übereinstimmung gebucketet + ein kombiniertes Gate getestet. """ from __future__ import annotations import sys import MetaTrader5 as mt5 from core.wave_rec import (WaveRecommender, _atr, _ema_last, _EMA_FAST, _EMA_SLOW, _N_BARS, _HTF_DEADBAND) _TF = {"M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15, "M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1} class _NeutralTU: def snapshot(self): return {"intervals": {}} def _ema_series(vals, period): k = 2.0/(period+1); out=[]; e=vals[0] for i,v in enumerate(vals): e = v if i==0 else v*k + e*(1-k); out.append(e) return out def _atr_series(H,L,C,period=14): trs=[0.0] for i in range(1,len(C)): trs.append(max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1]))) out=[] for i in range(len(C)): w=trs[max(1,i-period+1):i+1]; out.append(sum(w)/len(w) if w else None) return out def _rsi_series(C, period=14): n=len(C); out=[None]*n if n < period+1: return out ag=al=0.0 for i in range(1, period+1): ch=C[i]-C[i-1]; ag+=max(ch,0); al+=max(-ch,0) ag/=period; al/=period out[period]=100 - 100/(1+(ag/al if al>0 else 1e9)) for i in range(period+1, n): ch=C[i]-C[i-1] ag=(ag*(period-1)+max(ch,0))/period al=(al*(period-1)+max(-ch,0))/period out[i]=100 - 100/(1+(ag/al if al>0 else 1e9)) return out def _macd_hist(C, fast=12, slow=26, sig=9): ef=_ema_series(C,fast); es=_ema_series(C,slow) line=[ef[i]-es[i] for i in range(len(C))] signal=_ema_series(line,sig) return [line[i]-signal[i] for i in range(len(C))] def _adx(H,L,C,period=14): """Wilder ADX. Gibt (adx[], plus_di[], minus_di[]) (None vor Warmup).""" n=len(C); pdm=[0.0]*n; mdm=[0.0]*n; tr=[0.0]*n for i in range(1,n): up=H[i]-H[i-1]; dn=L[i-1]-L[i] pdm[i]=up if (up>dn and up>0) else 0.0 mdm[i]=dn if (dn>up and dn>0) else 0.0 tr[i]=max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1])) adx=[None]*n; pdi=[None]*n; mdi=[None]*n if n < 2*period+1: return adx,pdi,mdi s_tr=sum(tr[1:period+1]); s_p=sum(pdm[1:period+1]); s_m=sum(mdm[1:period+1]) dxs=[] for i in range(period, n): if i>period: s_tr=s_tr - s_tr/period + tr[i] s_p =s_p - s_p/period + pdm[i] s_m =s_m - s_m/period + mdm[i] p=100*s_p/s_tr if s_tr>0 else 0.0 m=100*s_m/s_tr if s_tr>0 else 0.0 pdi[i]=p; mdi[i]=m dx=100*abs(p-m)/(p+m) if (p+m)>0 else 0.0 dxs.append((i,dx)) # ADX = Wilder-Glättung der DX if len(dxs) >= period: first_bar, _=dxs[period-1] a=sum(d for _,d in dxs[:period])/period adx[first_bar]=a for k in range(period, len(dxs)): bar,d=dxs[k] a=(a*(period-1)+d)/period adx[bar]=a return adx,pdi,mdi def _rep(name, rets): if not rets: print(f" {name:<26} -"); return n=len(rets); win=sum(1 for x in rets if x>0) print(f" {name:<26} n={n:>4} Treffer={100*win/n:>3.0f}% Oe-Edge={sum(rets)/n:+.4f}") def main(): n_bars=int(sys.argv[1]) if len(sys.argv)>1 else 8000 K=10 if not mt5.initialize(): print("init",mt5.last_error()); sys.exit(1) sym=None for c in ("SpotCrude","USOIL","WTI","XTIUSD"): if mt5.symbol_info(c): sym=c; break sym=sym or "SpotCrude" bars=mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n_bars+_N_BARS+K+5) m30b=mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n_bars//6+400) mt5.shutdown() if bars is None or m30b is None: print("Bars fehlen"); sys.exit(1) 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] mT=[int(b["time"]) for b in m30b]; mc=[float(b["close"]) for b in m30b] mh=[float(b["high"]) for b in m30b]; ml=[float(b["low"]) for b in m30b] mEf=_ema_series(mc,_EMA_FAST); mEs=_ema_series(mc,_EMA_SLOW); mATR=_atr_series(mh,ml,mc) rsi=_rsi_series(C); hist=_macd_hist(C); adx,pdi,mdi=_adx(H,L,C) def m30_sign(ts): lo,hi,idx=0,len(mT)-1,-1 while lo<=hi: mid=(lo+hi)//2 if mT[mid]<=ts: idx=mid; lo=mid+1 else: hi=mid-1 if idx<_EMA_SLOW or mATR[idx] is None or mATR[idx]<=0: return 0 d=mEf[idx]-mEs[idx] return 0 if abs(d)<_HTF_DEADBAND*mATR[idx] else (1 if d>0 else -1) w=WaveRecommender(_NeutralTU(), mt5.TIMEFRAME_M5) base=[]; macd_ok=[]; macd_no=[]; di_ok=[]; di_no=[] adx_strong=[]; adx_weak=[]; rsi_ok=[]; rsi_no=[]; gate=[] for i in range(_N_BARS, len(C)-K): j=i-1 if adx[j] is None or rsi[j] is None: continue 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) rec,_=w._build(ef,es,C[j],atr,"M5",5,htf_trend=m30_sign(T[i])) if rec["signal"]=="WARTEN": continue d=1 if rec["signal"]=="LONG" else -1 r=(C[i+K]-C[i])*d base.append(r) (macd_ok if (hist[j]>0)==(d>0) else macd_no).append(r) (di_ok if (pdi[j]-mdi[j]>0)==(d>0) else di_no).append(r) (adx_strong if adx[j]>=25 else adx_weak).append(r) rsi_aligned = (rsi[j]>50) if d>0 else (rsi[j]<50) (rsi_ok if rsi_aligned else rsi_no).append(r) if (hist[j]>0)==(d>0) and adx[j]>=20: # kombiniertes Gate gate.append(r) print("="*64) print(f" Konfluenz-Test — {sym} M5+M30 Vorlauf={K}") print("="*64) _rep("BASIS (alle Signale)", base) print("-- MACD-Histogramm --"); _rep("MACD dafür", macd_ok); _rep("MACD dagegen", macd_no) print("-- ADX-Richtung (DI) --"); _rep("DI dafür", di_ok); _rep("DI dagegen", di_no) print("-- ADX-Stärke --"); _rep("ADX>=25 (Trend stark)", adx_strong); _rep("ADX<25 (schwach)", adx_weak) print("-- RSI 50-Linie --"); _rep("RSI dafür", rsi_ok); _rep("RSI dagegen", rsi_no) print("-- Gate: MACD dafür & ADX>=20 --"); _rep("Gate", gate) if __name__ == "__main__": main()