AH-Oil-Trader/backtest_pbreak_calibration.py

#!/usr/bin/env python3
"""Kalibrierungs-Test des P(break)-Modells auf der GEGEN-/STOP-Seite (User-Frage:
Kurs naehert sich einem Level GEGEN Trade/Trend — ist die P(break)-Zahl dort
belastbar?). Vorgehen wie backtest_srclose_prob.py:

  - Modell auf H1 ZIEL-Seiten-Touches trainieren (identisch zum Live-Modell).
  - Auswertung auf H1 UND H2, getrennt nach:
      * Quelle: ZIEL-Seite (mit-Trade, wie trainiert) vs STOP-Seite (gegen-Trade)
      * wt: mit-Trend (1) vs gegen-Trend (0)
  - Metriken je Gruppe: Kalibrierung (Ø-P vs echte Break-Rate je P-Bin) + AUC + n.
Break/Bounce-Definition IDENTISCH zum Training: _BRK_ATR=0.5, _BRK_W=12 (=60min).
"""
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
_PIV_K=3; _LOOKBACK=300; _BRK_W=12; _BRK_ATR=0.5

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])))
    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 _signal_dir(i,C,EF,ES,AT):
    atr=AT[i]
    if not atr or atr<=0: return 0,None
    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
    if stretch<=-_REVERSAL_STRETCH and ad>=_ANGLE_DEAD: return 1,atr
    if stretch>=_REVERSAL_STRETCH and ad<=-_ANGLE_DEAD: return -1,atr
    if abs(stretch)<_STRETCH_MAX: return (1 if ef>es else -1 if ef<es else 0),atr
    return 0,atr

def _pivots(i,H,L):
    phis,plos=[],[]
    for j in range(i-_LOOKBACK+_PIV_K, i-_PIV_K):
        if H[j]==max(H[j-_PIV_K:j+_PIV_K+1]): phis.append(H[j])
        if L[j]==min(L[j-_PIV_K:j+_PIV_K+1]): plos.append(L[j])
    return phis,plos

def _touch_and_break(i,entry,dd,atr,level,H,L,C,EF,ES):
    """Sucht ab i+1 den ersten Touch des Levels in Richtung dd (dd=Break-Richtung),
    dann Break/Bounce per _BRK_ATR/_BRK_W. → dict(feat, brk) oder None."""
    jt=None
    for j in range(i+1, min(i+_MAXH, len(C)-_BRK_W)):
        # Touch = Kurs erreicht das Level in Break-Richtung dd
        if (H[j]>=level) if dd>0 else (L[j]<=level): jt=j; break
        # Abbruch, wenn Kurs vorher 2*atr in Gegenrichtung (weg vom Level) laeuft
        if ((C[j]-entry)*(-dd)) >= 2.0*atr: break
    if jt is None: return None
    up=level+dd*_BRK_ATR*atr; dn=level-dd*_BRK_ATR*atr; brk=None
    for j in range(jt, min(jt+_BRK_W, len(H))):
        if (H[j]>=up) if dd>0 else (L[j]<=up): brk=1.0; break
        if (L[j]<=dn) if dd>0 else (H[j]>=dn): brk=0.0; break
    if brk is None: brk=0.0   # Timeout = Bounce (wie im Training)
    feat=[(C[jt]-C[max(0,jt-6)])*dd/atr,(C[jt]-C[max(0,jt-3)])*dd/atr,
          1.0 if (EF[jt]-ES[jt])*dd>0 else 0.0, abs(level-entry)/atr]
    return {"feat":feat,"brk":brk,"wt":feat[2]}

def collect(a,b,H,L,C,EF,ES,AT,side):
    """side='target' → mit-Trade Ziel-Level (wie Training); 'stop' → Gegen-Level."""
    out=[]
    for i in range(max(a,_N_BARS,_LOOKBACK), min(b,len(C)-_MAXH-1)):
        d,atr=_signal_dir(i,C,EF,ES,AT)
        if not d: continue
        entry=C[i]; phis,plos=_pivots(i,H,L)
        if side=="target":
            if d>0:
                cands=[p for p in phis if p>entry+0.3*atr]; level=min(cands) if cands else None
            else:
                cands=[p for p in plos if p<entry-0.3*atr]; level=max(cands) if cands else None
            dd=d
        else:  # stop-Seite: Gegen-Level, Break-Richtung = -d
            if d>0:
                cands=[p for p in plos if p<entry-0.3*atr]; level=max(cands) if cands else None
            else:
                cands=[p for p in phis if p>entry+0.3*atr]; level=min(cands) if cands else None
            dd=-d
        if level is None: continue
        r=_touch_and_break(i,entry,dd,atr,level,H,L,C,EF,ES)
        if r: out.append(r)
    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(p,y):
    p=np.asarray(p); y=np.asarray(y)
    pos=p[y==1]; neg=p[y==0]
    if len(pos)==0 or len(neg)==0: return float('nan')
    # Mann-Whitney via Rang
    order=np.argsort(np.concatenate([pos,neg]))
    ranks=np.empty_like(order,dtype=float); ranks[order]=np.arange(1,len(order)+1)
    r_pos=ranks[:len(pos)].sum()
    return (r_pos-len(pos)*(len(pos)+1)/2)/(len(pos)*len(neg))

def calib_table(name,ev,pf):
    if not ev:
        print(f"  {name:<28} (keine Events)"); return
    P=np.array([pf(e) for e in ev]); Y=np.array([e["brk"] for e in ev])
    a=auc(P,Y); base=Y.mean()
    print(f"  {name:<28} n={len(ev):>5}  Break-Basisrate={base*100:>4.0f}%  AUC={a:.2f}")
    bins=[(0,0.3),(0.3,0.5),(0.5,0.7),(0.7,1.01)]
    for lo,hi in bins:
        m=(P>=lo)&(P<hi)
        if m.sum()==0: continue
        print(f"      P {lo:.1f}-{hi:.1f}:  vorhergesagt Ø{P[m].mean()*100:>4.0f}%  →  echt {Y[m].mean()*100:>4.0f}%  (n{m.sum()})")

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
    bars=None
    for req in (n,80000,60000,40000):
        bars=mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_M5,0,req)
        if bars is not None and len(bars)>2000: break
    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]
    EF=_ema_series(C,_EMA_FAST); ES=_ema_series(C,_EMA_SLOW); AT=_atr_series(H,L,C)
    mid=len(C)//2
    print("="*94)
    print(f"  P(break)-Kalibrierung Ziel- vs Stop-Seite — {sym} M5, {len(bars)} Bars")
    print("="*94)
    tgt1=collect(_N_BARS,mid,H,L,C,EF,ES,AT,"target")
    tgt2=collect(mid,len(C),H,L,C,EF,ES,AT,"target")
    stp1=collect(_N_BARS,mid,H,L,C,EF,ES,AT,"stop")
    stp2=collect(mid,len(C),H,L,C,EF,ES,AT,"stop")
    # Modell auf H1-ZIEL-Touches trainieren (identisch zum Live-Modell)
    X1=np.array([e["feat"] for e in tgt1]); y1=np.array([e["brk"] for e in tgt1])
    mu=X1.mean(0); sd=X1.std(0)+1e-9
    w=fit_logreg((X1-mu)/sd,y1)
    def pf(e):
        x=(np.array(e["feat"])-mu)/sd
        return 1/(1+np.exp(-(w@np.concatenate([[1.0],x]))))
    print(f"  Modell (auf H1-Ziel trainiert): mom6={w[1]:+.2f} mom3={w[2]:+.2f} wt={w[3]:+.2f} dist={w[4]:+.2f}\n")
    for lbl,tgt,stp in (("H1 (Trainings-Halfte)",tgt1,stp1),("H2 (OUT-OF-SAMPLE)",tgt2,stp2)):
        print(f"── {lbl} ──")
        calib_table("ZIEL-Seite (mit-Trade)",tgt,pf)
        calib_table("STOP-Seite (gegen-Trade)",stp,pf)
        # STOP-Seite zusaetzlich nach wt gesplittet (der User-Fall ist wt=0)
        calib_table("  davon gegen-Trend (wt=0)",[e for e in stp if e["wt"]==0.0],pf)
        calib_table("  davon mit-Trend  (wt=1)",[e for e in stp if e["wt"]==1.0],pf)
        print()
    print("  Kalibriert = 'vorhergesagt Ø' ≈ 'echt' in jedem Bin, UND das in H2 (out-of-sample).")

if __name__=="__main__":
    main()