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>
99 lines
4.4 KiB
Python
99 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
"""Misst einen Höher-TF-WINKEL-Filter: Signal verwerfen (→WARTEN), wenn der
|
||
M30- UND/ODER H1-Regressionswinkel klar GEGEN die Signalrichtung steht (führender
|
||
Winkel statt nachlaufender EMA). Genau der Fall, der die −37/−10-Shorts erzeugte:
|
||
Welle short, aber M30/H1-Winkel stark aufwärts.
|
||
|
||
Frage: Haben die so entfernten Signale negativen Edge (gut weg) und trägt der
|
||
Gesamtertrag der behaltenen Signale? (Pauschal-Filter kann auch Gewinner killen.)
|
||
"""
|
||
import sys, bisect
|
||
import MetaTrader5 as mt5
|
||
from core.analysis import calc_trend_angle
|
||
from core.wave_rec import (WaveRecommender, _atr, _ema_last, _EMA_FAST, _EMA_SLOW,
|
||
_N_BARS, _HTF_DEADBAND, _ANGLE_LR)
|
||
|
||
|
||
class _TU:
|
||
def snapshot(self): return {"intervals": {}}
|
||
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 _rep(name, r):
|
||
if not r: print(f" {name:<30} -"); return
|
||
n=len(r); w=sum(1 for x in r if x>0)
|
||
print(f" {name:<30} n={n:>4} Treffer={100*w/n:>3.0f}% Ø-Edge={sum(r)/n:+.4f} Summe={sum(r):+.1f}")
|
||
|
||
def main():
|
||
n=int(sys.argv[1]) if len(sys.argv)>1 else 12000
|
||
K=10
|
||
mt5.initialize()
|
||
sym=None
|
||
for c in ("SpotCrude","USOIL","WTI","XTIUSD"):
|
||
if mt5.symbol_info(c): sym=c; break
|
||
bars=mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_M5,0,n+_N_BARS+K+5)
|
||
m30=mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_M30,0,n//6+400)
|
||
h1 =mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_H1,0,n//12+400)
|
||
mt5.shutdown()
|
||
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 m30]; mc=[float(b["close"]) for b in m30]
|
||
mh=[float(b["high"]) for b in m30]; ml=[float(b["low"]) for b in m30]
|
||
mEf=_ema_series(mc,_EMA_FAST); mEs=_ema_series(mc,_EMA_SLOW); mA=_atr_series(mh,ml,mc)
|
||
hT=[int(b["time"]) for b in h1]; hc=[float(b["close"]) for b in h1]
|
||
def m30sign(ts):
|
||
idx=bisect.bisect_right(mT,ts)-1
|
||
if idx<_EMA_SLOW or mA[idx] is None or mA[idx]<=0: return 0
|
||
d=mEf[idx]-mEs[idx]
|
||
return 0 if abs(d)<_HTF_DEADBAND*mA[idx] else (1 if d>0 else -1)
|
||
def ang_at(ts, TT, CC):
|
||
idx=bisect.bisect_right(TT,ts)-1
|
||
if idx<_ANGLE_LR+2: return 90.0
|
||
return calc_trend_angle(CC[idx-_ANGLE_LR-1:idx+1], _ANGLE_LR)
|
||
|
||
w=WaveRecommender(_TU(), mt5.TIMEFRAME_M5)
|
||
sigs=[] # (d, fwd, m30dev, h1dev) dev = Winkel-90 (>0 auf, <0 ab)
|
||
for i in range(_N_BARS, len(C)-K):
|
||
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)
|
||
a5=calc_trend_angle(C[i-_ANGLE_LR-2:i],_ANGLE_LR)
|
||
rec,_=w._build(ef,es,C[i-1],atr,"M5",0,htf_trend=m30sign(T[i]),angle=a5)
|
||
if rec["signal"]=="WARTEN": continue
|
||
d=1 if rec["signal"]=="LONG" else -1
|
||
fwd=(C[i+K]-C[i])*d
|
||
sigs.append((d, fwd, ang_at(T[i],mT,mc)-90.0, ang_at(T[i],hT,hc)-90.0))
|
||
|
||
base=[f for (_,f,_,_) in sigs]
|
||
print("="*78)
|
||
print(f" Höher-TF-Winkel-Filter — {sym} M5 Signale={len(sigs)} Vorlauf={K}")
|
||
print("="*78)
|
||
_rep("BASELINE (alle Signale)", base)
|
||
print(" dev = Regressionswinkel − 90 (>0 aufwärts, <0 abwärts)\n")
|
||
|
||
for T_ in (2, 10, 20, 40):
|
||
# "gegen die Richtung": Short & Winkel auf (dev>+T) bzw. Long & Winkel ab (dev<−T)
|
||
def against(d, dev): return (d<0 and dev> T_) or (d>0 and dev< -T_)
|
||
for mode in ("BEIDE", "EINER"):
|
||
removed=[]; kept=[]
|
||
for (d,f,m,h) in sigs:
|
||
am=against(d,m); ah=against(d,h)
|
||
cut = (am and ah) if mode=="BEIDE" else (am or ah)
|
||
(removed if cut else kept).append(f)
|
||
print(f"Schwelle T={T_}° · '{mode} TF gegen Signal':")
|
||
_rep(" entfernt (würde →WARTEN)", removed)
|
||
_rep(" behalten (gehandelt)", kept)
|
||
db=(sum(kept)-sum(base))
|
||
print(f" Δ Gesamtertrag vs Baseline: {db:+.1f} "
|
||
f"({'BESSER' if db>0 else 'schlechter'}) · behalten {len(kept)}/{len(sigs)}")
|
||
print()
|
||
|
||
if __name__=="__main__":
|
||
main()
|