#!/usr/bin/env python3 """Event-Blackout-Messung (Track B): Sind Signale rund um die planbaren Öl-Events schlechter? EIA-Lagerbestände = Mittwoch 16:30 Berlin (10:30 ET, Offset ganzjährig 6 h), API = Dienstag 22:30 Berlin. Misst Signale (Trend+Reversal, Live-Exit-Sim) je Fenster: eia_pre Mi 15:30–16:30 · eia_post Mi 16:30–18:00 api_pre Di 21:30–22:30 · api_post Di 22:30–23:30 gegen 'rest', über 2 History-Hälften. Blackout nur bauen, wenn ein Fenster in BEIDEN Hälften klar negativ ist (netto, Kosten 0,1×ATR). """ import sys, datetime as dt from zoneinfo import ZoneInfo 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; _TRAILON=0.3; _TPTRAIL=0.5; _ATRMIN=0.12; _SL_ATR=2.0; _COST=0.10 _BROKER_OFF=3*3600; _BERLIN=ZoneInfo("Europe/Berlin") 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 simulate(entry,d,atr,sl,H,L,C,j0): eff=sl; hw=entry; trail=False end=min(j0+_MAXH,len(C)-1); exit_px=C[end] for j in range(j0,end+1): hi,lo=H[j],L[j] if (lo<=eff) if d>0 else (hi>=eff): return (eff-entry)*d/atr hw=max(hw,hi) if d>0 else min(hw,lo) if (C[j]-entry)*d>=_TRAILON*atr: trail=True if trail: lock=hw-d*_TPTRAIL*atr eff=max(eff,lock) if d>0 else min(eff,lock) return (exit_px-entry)*d/atr def st(v): if not v: return "n=0" n=len(v); w=sum(1 for x in v if x>0) return (f"n={n:>5} WR={100*w/n:>3.0f}% Ø-R={sum(v)/n:+.3f} " f"netto={sum(v)/n-_COST:+.3f} ΣR={sum(v):+.0f}") def bdt(raw): return dt.datetime.fromtimestamp(int(raw)-_BROKER_OFF, tz=dt.timezone.utc).astimezone(_BERLIN) def wclass(t): wd=t.weekday(); hm=t.hour*60+t.minute if wd==2: # Mittwoch if 15*60+30<=hm<16*60+30: return "eia_pre" if 16*60+30<=hm<18*60: return "eia_post" if wd==1: # Dienstag if 21*60+30<=hm<22*60+30: return "api_pre" if 22*60+30<=hm<23*60+30: return "api_post" return "rest" 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,100000,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]; T=[int(b["time"]) 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; TH=_REVERSAL_STRETCH print("="*92) print(f" Event-Fenster (EIA Mi 16:30 · API Di 22:30 Berlin) — {sym} M5 ({len(C)} Bars)") print("="*92) for lbl,a,b in (("H1 (alt)",_N_BARS,mid),("H2 (neu)",mid,len(C))): B={k:[] for k in ("eia_pre","eia_post","api_pre","api_post","rest")} for i in range(max(a,_N_BARS), 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