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>
108 lines
4.9 KiB
Python
108 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Flatter-Analyse der Durchbruchs-Empfehlung (letzte 2 Tage, M1-Auflösung):
|
||
Rekonstruiert P(break) minütlich WIE DIE ENGINE (M5-Serie inkl. LAUFENDER Kerze,
|
||
deren Close = aktueller Kurs — genau das lässt mom3/mom6 im Bar springen) und misst
|
||
in Level-Nähe (≤0,6×ATR):
|
||
- P-Spanne innerhalb einzelner M5-Kerzen (wie stark springt P intra-Bar)
|
||
- Schwellen-Kreuzungen (close↔laufen bei 60 %) je Annäherungs-Episode
|
||
- Zerlegung: Sprünge durch mom3 (laufende Kerze) vs. Level-Wechsel
|
||
"""
|
||
import sys, math
|
||
import MetaTrader5 as mt5
|
||
from core.engine import _p_break
|
||
from core.wave_rec import _EMA_FAST, _EMA_SLOW
|
||
_PIV_K=3; _LOOKBACK=300; _THR=0.60
|
||
|
||
def ema_last(vals,p):
|
||
k=2.0/(p+1); e=vals[0]
|
||
for x in vals[1:]: e=x*k+e*(1-k)
|
||
return e
|
||
def atr14(H,L,C):
|
||
tr=[max(H[i]-L[i],abs(H[i]-C[i-1]),abs(L[i]-C[i-1])) for i in range(1,len(C))]
|
||
return sum(tr[-14:])/14
|
||
|
||
def main():
|
||
days=int(sys.argv[1]) if len(sys.argv)>1 else 2
|
||
mt5.initialize(); sym='SpotCrude'
|
||
m1=mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_M1,0,days*1440+50)
|
||
m5=mt5.copy_rates_from_pos(sym,mt5.TIMEFRAME_M5,0,days*288+_LOOKBACK+60)
|
||
mt5.shutdown()
|
||
T5=[int(b["time"]) for b in m5]
|
||
O5=[float(b["open"]) for b in m5]; H5=[float(b["high"]) for b in m5]
|
||
L5=[float(b["low"]) for b in m5]; C5=[float(b["close"]) for b in m5]
|
||
T1=[int(b["time"]) for b in m1]; C1=[float(b["close"]) for b in m1]
|
||
|
||
import bisect
|
||
ep=[] # aktive Annäherungs-Episode: Liste von (P, level, mom3)
|
||
episodes=[] # abgeschlossene Episoden
|
||
per_bar={} # (m5-index) -> [P...] für Intra-Bar-Spanne
|
||
n_eval=0
|
||
for i1 in range(30, len(m1)):
|
||
t=T1[i1]; px=C1[i1]
|
||
j=bisect.bisect_right(T5,t)-1 # laufende/letzte M5-Kerze
|
||
if j < _LOOKBACK+10: continue
|
||
# M5-Serie wie die Engine sie sieht: abgeschlossene Bars + laufende Kerze
|
||
# mit Close = aktuellem M1-Kurs
|
||
closes=C5[j-_LOOKBACK:j]+[px]
|
||
highs =H5[j-_LOOKBACK:j]+[max(px,O5[j])]
|
||
lows =L5[j-_LOOKBACK:j]+[min(px,O5[j])]
|
||
atr=max(atr14(highs,lows,closes),0.12)
|
||
mom6=(closes[-1]-closes[-7])/atr
|
||
mom3=(closes[-1]-closes[-4])/atr
|
||
ediff=(ema_last(closes,_EMA_FAST)-ema_last(closes,_EMA_SLOW))/atr
|
||
# Pivot-Level aus abgeschlossenen Bars
|
||
ph=[H5[k] for k in range(j-_LOOKBACK+_PIV_K, j-_PIV_K)
|
||
if H5[k]==max(H5[k-_PIV_K:k+_PIV_K+1])]
|
||
# LONG-Sicht: nächster Widerstand über Kurs (analog gilt Short symmetrisch)
|
||
cands=[p for p in ph if p>px]
|
||
level=min(cands) if cands else None
|
||
if level is None or (level-px) > 0.6*atr:
|
||
if ep: episodes.append(ep); ep=[]
|
||
continue
|
||
wt=1.0 if ediff>0 else 0.0
|
||
dist=abs(level-px)/atr
|
||
P=_p_break(mom6,mom3,wt,dist)
|
||
ep.append((P,level,mom3)); n_eval+=1
|
||
per_bar.setdefault(j,[]).append(P)
|
||
if ep: episodes.append(ep)
|
||
|
||
print("="*84)
|
||
print(f" P(break)-Flattern — letzte {days} Tage, {n_eval} Minuten in Level-Nähe (≤0,6×ATR)")
|
||
print("="*84)
|
||
# Intra-Bar-Spanne
|
||
spans=[max(v)-min(v) for v in per_bar.values() if len(v)>=3]
|
||
if spans:
|
||
spans.sort()
|
||
print(f"\nP-Spanne INNERHALB einer M5-Kerze (n={len(spans)} Kerzen):")
|
||
print(f" Median {100*spans[len(spans)//2]:.0f} Pp · P90 {100*spans[int(len(spans)*0.9)]:.0f} Pp · Max {100*max(spans):.0f} Pp")
|
||
# Episoden: Schwellen-Flips
|
||
flips=[]; levsw=[]
|
||
for e in episodes:
|
||
if len(e)<3: continue
|
||
f=sum(1 for k in range(1,len(e)) if (e[k][0]<_THR)!=(e[k-1][0]<_THR))
|
||
sw=sum(1 for k in range(1,len(e)) if abs(e[k][1]-e[k-1][1])>1e-9)
|
||
flips.append(f); levsw.append(sw)
|
||
if flips:
|
||
flips.sort()
|
||
print(f"\nAnnäherungs-Episoden: {len(flips)} · Ø-Dauer "
|
||
f"{sum(len(e) for e in episodes if len(e)>=3)/len(flips):.0f} min")
|
||
print(f" close↔laufen-Flips (60%-Schwelle) je Episode: "
|
||
f"Median {flips[len(flips)//2]} · P90 {flips[int(len(flips)*0.9)]} · Max {max(flips)}")
|
||
print(f" Episoden mit ≥2 Flips: {100*sum(1 for f in flips if f>=2)/len(flips):.0f}% "
|
||
f"· Level-Wechsel je Episode Ø {sum(levsw)/len(levsw):.1f}")
|
||
# mom3-Beitrag: P-Änderung je Minute vs mom3-Änderung
|
||
dP=[]; dM=[]
|
||
for e in episodes:
|
||
for k in range(1,len(e)):
|
||
if abs(e[k][1]-e[k-1][1])<1e-9: # gleiches Level → reiner Feature-Effekt
|
||
dP.append(abs(e[k][0]-e[k-1][0])); dM.append(abs(e[k][2]-e[k-1][2]))
|
||
if dP:
|
||
print(f"\nMinuten-Sprünge bei GLEICHEM Level (n={len(dP)}):")
|
||
print(f" Ø |ΔP|/min = {100*sum(dP)/len(dP):.1f} Pp · Ø |Δmom3|/min = {sum(dM)/len(dM):.2f}×ATR")
|
||
big=[p for p,m in zip(dP,dM) if m>0.3]
|
||
print(f" bei |Δmom3|>0,3: Ø |ΔP| = {100*sum(big)/max(1,len(big)):.1f} Pp "
|
||
f"({100*len(big)/len(dP):.0f}% der Minuten) → mom3 (laufende Kerze) ist der Treiber")
|
||
|
||
if __name__=="__main__":
|
||
main()
|