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>
150 lines
7.5 KiB
Python
150 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
||
"""R-Ertrag der P(break)-Schwellen-Regel messen (Track B, Echtkosten, 2 Hälften):
|
||
Regel: erreicht die Welle das gegenüberliegende Level, berechne P(Durchbruch) mit
|
||
dem kalibrierten Logit-Modell (auf H1 trainiert). Ist P < Schwelle X → am Level
|
||
CLOSEN; sonst laufen lassen (Live-Trailing). Baseline = immer laufen lassen.
|
||
Findet das R-Optimum und prüft die User-Schwelle 60 % (= closen wenn P<0,60).
|
||
"""
|
||
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; _TRAILON=0.3; _MULT=1.5; _BE=1.3
|
||
_LOCK_START=3.5; _LOCK_SCALE=0.6; _LOCK_MIN=1.2; _TP_INIT=3.5
|
||
_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 sim_run(entry,d,atr,H,L,C,j0):
|
||
sl=entry-d*_SL_ATR*atr; tp=entry+d*_TP_INIT*atr; hw=entry; rank=0
|
||
end=min(j0+_MAXH,len(C)-1)
|
||
for j in range(j0,end+1):
|
||
hi,lo=H[j],L[j]
|
||
if (lo<=sl) if d>0 else (hi>=sl): return (sl-entry)*d/atr
|
||
if (hi>=tp) if d>0 else (lo<=tp): return (tp-entry)*d/atr
|
||
hw=max(hw,hi) if d>0 else min(hw,lo)
|
||
profit=(hw-entry)*d
|
||
ph=0 if profit<_TRAILON*atr else (1 if profit<_LOCK_START*atr else 2)
|
||
if ph<rank: ph=rank
|
||
rank=ph
|
||
if ph==1:
|
||
cand=hw-d*_MULT*atr
|
||
if profit>=_BE*atr: cand=max(cand,entry) if d>0 else min(cand,entry)
|
||
sl=max(sl,cand) if d>0 else min(sl,cand)
|
||
elif ph==2:
|
||
tm=max(_LOCK_MIN,_MULT*_LOCK_SCALE); cand=hw-d*tm*atr
|
||
cand=max(cand,entry) if d>0 else min(cand,entry)
|
||
sl=max(sl,cand) if d>0 else min(sl,cand)
|
||
return (C[end]-entry)*d/atr
|
||
|
||
def collect(a,b,H,L,C,EF,ES,AT,SP,point):
|
||
"""→ Liste dicts: R_run (baseline), touched, break, feat, R_close (bei Touch)."""
|
||
TH=_REVERSAL_STRETCH; out=[]
|
||
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
|
||
entry=C[i]; cost=(SP[i] if SP[i]>0 else 0.0225)/atr
|
||
R_run=sim_run(entry,d,atr,H,L,C,i+1)-cost
|
||
rec={"R_run":R_run,"touched":False}
|
||
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 not None:
|
||
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)) >= 2.0*atr: break
|
||
if jt is not None:
|
||
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
|
||
rec.update(touched=True, brk=1.0 if brk else 0.0,
|
||
R_close=(level-entry)*d/atr-cost,
|
||
feat=[(C[jt]-C[max(0,jt-6)])*d/atr,(C[jt]-C[max(0,jt-3)])*d/atr,
|
||
1.0 if (EF[jt]-ES[jt])*d>0 else 0.0, abs(level-entry)/atr])
|
||
out.append(rec)
|
||
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 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
|
||
si=mt5.symbol_info(sym); point=si.point
|
||
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]
|
||
SP=[float(b["spread"])*point 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=collect(_N_BARS,mid,H,L,C,EF,ES,AT,SP,point)
|
||
ev2=collect(mid,len(C),H,L,C,EF,ES,AT,SP,point)
|
||
# Modell auf H1-Touches trainieren
|
||
t1=[e for e in ev1 if e["touched"]]
|
||
X1=np.array([e["feat"] for e in t1]); y1=np.array([e["brk"] for e in t1])
|
||
mu=X1.mean(0); sd=X1.std(0)+1e-9
|
||
w=fit_logreg((X1-mu)/sd,y1)
|
||
def pbreak(e):
|
||
x=(np.array(e["feat"])-mu)/sd
|
||
return 1/(1+np.exp(-(w@np.concatenate([[1.0],x]))))
|
||
print("="*94)
|
||
print(f" P(break)-Close-Regel — R-Ertrag (Echtkosten, Trailing-Baseline) — {sym} M5")
|
||
print(f" Modell auf H1 trainiert · Koeff mom6={w[1]:+.2f} mom3={w[2]:+.2f} wt={w[3]:+.2f} dist={w[4]:+.2f}")
|
||
print("="*94)
|
||
for lbl,ev in (("H1 (alt)",ev1),("H2 (neu)",ev2)):
|
||
base=sum(e["R_run"] for e in ev)
|
||
tset=[e for e in ev if e["touched"]]
|
||
for e in tset: e["_p"]=pbreak(e)
|
||
print(f"\n{lbl}: {len(ev)} Wellen · {len(tset)} Level-Touches · Baseline (immer laufen) ΣR={base:+.0f}")
|
||
print(f" {'Schwelle X':<12}{'%laufen':>9}{'ΣR Regel':>10}{'Δ vs Baseline':>15}")
|
||
for X in (0.30,0.35,0.40,0.45,0.50,0.55,0.60,0.70):
|
||
run=sum(1 for e in tset if e["_p"]>=X)
|
||
total=sum((e["R_close"] if e["_p"]<X else e["R_run"]) for e in tset) \
|
||
+ sum(e["R_run"] for e in ev if not e["touched"])
|
||
mark=" ← deine 60%" if abs(X-0.60)<1e-9 else ""
|
||
print(f" P≥{X:.2f} {100*run/max(1,len(tset)):>7.0f}%{total:>10.0f}{total-base:>+15.0f}{mark}")
|
||
# Mindestgewinn-Variante (User-Frage): am Level nur closen, wenn der Close
|
||
# mindestens minR (netto, ×ATR) bringt — sonst weiterlaufen lassen.
|
||
print(f" {'Mindestgewinn (P<0.60)':<24}{'%geclosed':>10}{'ΣR Regel':>10}{'Δ vs minR=0':>13}")
|
||
base_rule=sum((e["R_close"] if e["_p"]<0.60 else e["R_run"]) for e in tset) \
|
||
+ sum(e["R_run"] for e in ev if not e["touched"])
|
||
for minR in (0.0,0.1,0.2,0.3,0.5,0.8):
|
||
closed=sum(1 for e in tset if e["_p"]<0.60 and e["R_close"]>=minR)
|
||
total=sum((e["R_close"] if (e["_p"]<0.60 and e["R_close"]>=minR) else e["R_run"])
|
||
for e in tset) + sum(e["R_run"] for e in ev if not e["touched"])
|
||
print(f" minR={minR:.1f}×ATR {100*closed/max(1,len(tset)):>8.0f}%{total:>10.0f}{total-base_rule:>+13.0f}")
|
||
|
||
if __name__=="__main__":
|
||
main()
|