AH-Oil-Trader/backtest_confluence.py

#!/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()