Axel HocksandClaude Opus 4.8
75d28827e8
Headless FastAPI-Backend (server.py + core/engine.py) mit Mobile-PWA (web/), Strategie-/Backtest-Suite und Doku. Secrets, DB, Logs und Laufzeit-State sind via .gitignore ausgeschlossen; Config-Vorlage: oil_widget_config.ini.example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
135 lines
6.6 KiB
Python
135 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
||
"""S/R-Wellen-Analyse (Track B):
|
||
TEIL 1 Containment: Welcher %-Anteil der Wellen (EMA-Richtungsmoves) bleibt
|
||
INNERHALB der S/R-Linien — d. h. erreicht das gegenüberliegende Level
|
||
und PRALLT AB, statt durchzubrechen? (Basisrate P(break) am Touch.)
|
||
TEIL 2 P(Durchbruch)-Modell: logistische Regression auf beobachtbaren Merkmalen
|
||
am Touch (Anlauf-Momentum 6/3 Bars, EMA-mit-Trend, Level-Distanz), auf H1
|
||
trainiert und auf H2 KALIBRIERT geprüft (predicted vs. echte Bruchrate je
|
||
Bin) — ein P ist nur brauchbar, wenn es out-of-sample kalibriert ist.
|
||
Break = nach Touch +0,5×ATR JENSEITS des Levels binnen 12 Bars, bevor 0,5×ATR zurück.
|
||
"""
|
||
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; _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 collect(a,b,H,L,C,EF,ES,AT):
|
||
"""Sammelt Wellen-Touches → Liste (break?, feat-vector, reached?)."""
|
||
TH=_REVERSAL_STRETCH; out=[]; n_wave=0; n_reach=0
|
||
for i in range(max(a,_N_BARS,_LOOKBACK), min(b,len(C)-_MAXH-1)):
|
||
atr=AT[i]
|
||
if not atr or atr<=0: continue
|
||
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
|
||
d=0
|
||
if stretch<=-TH and ad>=_ANGLE_DEAD: d=1
|
||
elif stretch>=TH and ad<=-_ANGLE_DEAD: d=-1
|
||
elif abs(stretch)<_STRETCH_MAX: d=1 if ef>es else -1 if ef<es else 0
|
||
if not d: continue
|
||
n_wave+=1
|
||
entry=C[i]
|
||
# nächstes gegenüberliegendes Pivot-Level
|
||
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])
|
||
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
|
||
if level is None: continue
|
||
# Touch suchen (bis MAXH), solange die Welle "lebt" (kein 2×ATR-Gegenlauf)
|
||
jt=None
|
||
for j in range(i+1, min(i+_MAXH, len(C)-_BRK_W)):
|
||
if ((H[j]>=level) if d>0 else (L[j]<=level)): jt=j; break
|
||
if ((entry-L[j]) if d>0 else (H[j]-entry))*1 >= 2.0*atr: break # Welle tot
|
||
if jt is None: continue
|
||
n_reach+=1
|
||
up=level+d*_BRK_ATR*atr; dn=level-d*_BRK_ATR*atr; brk=False
|
||
for j in range(jt, min(jt+_BRK_W, len(H))):
|
||
if (H[j]>=up) if d>0 else (L[j]<=up): brk=True; break
|
||
if (L[j]<=dn) if d>0 else (H[j]>=dn): brk=False; break
|
||
mom6=(C[jt]-C[max(0,jt-6)])*d/atr
|
||
mom3=(C[jt]-C[max(0,jt-3)])*d/atr
|
||
wt=1.0 if (EF[jt]-ES[jt])*d>0 else 0.0
|
||
distlvl=abs(level-entry)/atr
|
||
out.append((1.0 if brk else 0.0, [mom6,mom3,wt,distlvl]))
|
||
return out, n_wave, n_reach
|
||
|
||
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 predict(w,X):
|
||
return 1/(1+np.exp(-(np.hstack([np.ones((X.shape[0],1)),X])@w)))
|
||
|
||
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
|
||
ev1,w1,r1=collect(_N_BARS,mid,H,L,C,EF,ES,AT)
|
||
ev2,w2,r2=collect(mid,len(C),H,L,C,EF,ES,AT)
|
||
print("="*90)
|
||
print(f" S/R-WELLEN — {sym} M5 ({len(C)} Bars) Level=Pivot(k{_PIV_K}), Break=+{_BRK_ATR}×ATR in {_BRK_W} Bars")
|
||
print("="*90)
|
||
print("\nTEIL 1 — Containment (bleibt die Welle innerhalb der S/R-Linien?):")
|
||
for lbl,ev,nw,nr in (("H1 (alt)",ev1,w1,r1),("H2 (neu)",ev2,w2,r2)):
|
||
if not ev: continue
|
||
pbreak=100*sum(e[0] for e in ev)/len(ev)
|
||
print(f" {lbl}: {nw} Wellen · {100*nr/max(1,nw):.0f}% erreichen das Level · "
|
||
f"davon {pbreak:.0f}% DURCHBRUCH → {100-pbreak:.0f}% bleiben drin (Abpraller)")
|
||
|
||
print("\nTEIL 2 — P(Durchbruch)-Modell (Merkmale: mom6, mom3, mit-Trend, Level-Dist):")
|
||
X1=np.array([e[1] for e in ev1]); y1=np.array([e[0] for e in ev1])
|
||
X2=np.array([e[1] for e in ev2]); y2=np.array([e[0] for e in ev2])
|
||
mu=X1.mean(0); sd=X1.std(0)+1e-9
|
||
w=fit_logreg((X1-mu)/sd,y1)
|
||
p2=predict(w,(X2-mu)/sd)
|
||
# AUC (schnell, via Rangvergleich)
|
||
order=np.argsort(p2); ranks=np.empty_like(order,dtype=float); ranks[order]=np.arange(1,len(p2)+1)
|
||
npos=y2.sum(); nneg=len(y2)-npos
|
||
auc=(ranks[y2==1].sum()-npos*(npos+1)/2)/(npos*nneg) if npos and nneg else float('nan')
|
||
print(f" Trainiert auf H1, geprüft auf H2 · AUC={auc:.3f} (0,5=Zufall; >0,6=brauchbar trennscharf)")
|
||
print(f" Koeffizienten (standardisiert): mom6={w[1]:+.2f} mom3={w[2]:+.2f} "
|
||
f"mitTrend={w[3]:+.2f} dist={w[4]:+.2f} · Basis(intercept)={w[0]:+.2f}")
|
||
print("\n Kalibrierung auf H2 (vorhergesagtes P vs. echte Bruchrate):")
|
||
print(f" {'P-Bin':<12}{'n':>7}{'Ø-Vorhersage':>14}{'echt-Bruch':>12}")
|
||
edges=[0,0.25,0.30,0.35,0.40,0.45,0.55,1.01]
|
||
for k in range(len(edges)-1):
|
||
m=(p2>=edges[k])&(p2<edges[k+1])
|
||
if m.sum()<30: continue
|
||
print(f" {edges[k]:.2f}–{edges[k+1]:.2f} {int(m.sum()):>7}{p2[m].mean():>13.0%}{y2[m].mean():>12.0%}")
|
||
# Nutz-Regel: Trenn-Schwelle
|
||
for thr in (0.40,0.45,0.50):
|
||
run=p2>=thr
|
||
print(f"\n Regel „laufen lassen wenn P(break)≥{thr:.0%}\": {100*run.mean():.0f}% der Touches laufen, "
|
||
f"davon {100*y2[run].mean():.0f}% brechen wirklich durch · "
|
||
f"Rest schließen: {100*y2[~run].mean():.0f}% brechen (Fehl-Close)")
|
||
|
||
if __name__=="__main__":
|
||
main()
|