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>
206 lines
8.2 KiB
Python
206 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Exit-Simulation: bringt ein WEITERER Initial-SL netto mehr? (Befund B)
|
||
|
||
Replays die echten Signale (M5 + M30-Filter) und simuliert den tatsaechlichen
|
||
Exit-Ablauf bar-fuer-bar, originalgetreu zu core/trailing.py:
|
||
- Initial-SL = X×ATR (die getestete Variable)
|
||
- Teil-Exit 50 % bei +1,5×ATR (einmalig)
|
||
- Phasen: Init (<0,3×ATR halte Initial-SL) · Trail (SL = HW∓mult×ATR,
|
||
Breakeven-Floor ab +0,6×ATR, mult=1,5 fuer M5) · Lock (>=3,5×ATR enger)
|
||
- Phasen-Ratsche (nie zurueck)
|
||
Pessimistische Intrabar-Annahme: Gegenlauf VOR Mitlauf (zaehlt SL zuerst) —
|
||
ueberschaetzt den Nutzen eines weiten SL also NICHT.
|
||
|
||
PnL in R (= ATR-Vielfache, vergleichbar ueber Trades). Der weite Init-SL wirkt
|
||
nur in der Init-Phase: sobald Trailing greift, kappt HW∓1,5×ATR ihn ohnehin.
|
||
Hinweis: der feste Init-TP (+3,5×ATR) wird weggelassen — der Runner-Exit laeuft
|
||
praktisch ueber den Trailing-SL; das ist die konservative, dominante Mechanik.
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
import MetaTrader5 as mt5
|
||
from core.wave_rec import (WaveRecommender, _atr, _ema_last,
|
||
_EMA_FAST, _EMA_SLOW, _N_BARS, _HTF_DEADBAND)
|
||
|
||
_MULT = 1.5 # _MULT_BY_TF[M5]
|
||
_BE_ATR = 0.6 # _BREAKEVEN_ATR
|
||
_TRAIL_ON = 0.3 # _TRAIL_START_ATR
|
||
_LOCK_ATR = 3.5 # _PHASE4_ATR
|
||
_LOCK_MULT = max(1.2, _MULT * 0.6)
|
||
_PART_ATR = 1.5 # _PARTIAL_TP_ATR
|
||
_PART_FRAC = 0.0 # Teil-Exit AUS (entspricht Live: _PARTIAL_TP_FRAC=0)
|
||
_ATR_MIN = 0.12
|
||
_MAXH = 240 # max. Haltedauer in M5-Bars (~20 h)
|
||
|
||
|
||
class _NeutralTU:
|
||
def snapshot(self): return {"intervals": {}}
|
||
|
||
|
||
def _ema_series(vals, period):
|
||
k = 2.0/(period+1); out=[]; e=vals[0]
|
||
for i,v in enumerate(vals):
|
||
e = v if i==0 else v*k + e*(1-k); out.append(e)
|
||
return out
|
||
|
||
|
||
def _atr_series(H,L,C,period=14):
|
||
trs=[0.0]
|
||
for i in range(1,len(C)):
|
||
trs.append(max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1])))
|
||
out=[]
|
||
for i in range(len(C)):
|
||
w=trs[max(1,i-period+1):i+1]; out.append(sum(w)/len(w) if w else None)
|
||
return out
|
||
|
||
|
||
def _htf_sign_at(ts, T, Ef, Es, ATR):
|
||
lo,hi,idx=0,len(T)-1,-1
|
||
while lo<=hi:
|
||
m=(lo+hi)//2
|
||
if T[m]<=ts: idx=m; lo=m+1
|
||
else: hi=m-1
|
||
if idx<_EMA_SLOW or ATR[idx] is None or ATR[idx]<=0: return 0
|
||
d=Ef[idx]-Es[idx]
|
||
return 0 if abs(d)<_HTF_DEADBAND*ATR[idx] else (1 if d>0 else -1)
|
||
|
||
|
||
def simulate(entry, d, atr, X, H, L, C, j0, be=_BE_ATR):
|
||
"""Ein Trade. Gibt (R_total, stopped_in_init) zurueck. d=+1 long/-1 short.
|
||
be = Breakeven-Schwelle in xATR (ab welchem Profit der SL auf Entry rueckt)."""
|
||
mult = _MULT
|
||
sl = entry - d * X * atr
|
||
hw = entry
|
||
size = 1.0
|
||
realized = 0.0 # in Preis-Einheiten
|
||
partial = False
|
||
phase_rank = 0 # 0 Init, 1 Trail, 2 Lock
|
||
init_stop = 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]
|
||
# 1) Gegenlauf zuerst → SL-Treffer?
|
||
hit = (lo <= sl) if d > 0 else (hi >= sl)
|
||
if hit:
|
||
exit_px = sl
|
||
if phase_rank == 0:
|
||
init_stop = True
|
||
break
|
||
# 2) HW mit Mitlauf
|
||
hw = max(hw, hi) if d > 0 else min(hw, lo)
|
||
# 3) Teil-Exit 50 % bei +1,5×ATR (Mitlauf-Extrem)
|
||
fav = ((hi if d > 0 else lo) - entry) * d
|
||
if not partial and fav >= _PART_ATR * atr:
|
||
lvl = entry + d * _PART_ATR * atr
|
||
realized += _PART_FRAC * (lvl - entry) * d
|
||
size -= _PART_FRAC
|
||
partial = True
|
||
# 4) Phase aus Close-Profit + Ratsche
|
||
prof = (C[j] - entry) * d
|
||
rank = 0 if prof < _TRAIL_ON * atr else (1 if prof < _LOCK_ATR * atr else 2)
|
||
phase_rank = max(phase_rank, rank)
|
||
# 5) Trailing-SL nachziehen
|
||
if phase_rank == 1:
|
||
cand = hw - d * mult * atr
|
||
cand = (max(cand, entry - mult * atr) if d > 0
|
||
else min(cand, entry + mult * atr))
|
||
if prof >= 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 phase_rank == 2:
|
||
cand = hw - d * _LOCK_MULT * atr
|
||
cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||
sl = max(sl, cand) if d > 0 else min(sl, cand)
|
||
R = (realized + size * (exit_px - entry) * d) / atr
|
||
return R, init_stop
|
||
|
||
|
||
def main():
|
||
args = sys.argv[1:]
|
||
mode = "be" if (args and args[0] == "be") else "width"
|
||
if mode == "be": args = args[1:]
|
||
n_bars = int(args[0]) if args else 8000
|
||
widths = [1.2, 1.5, 1.8, 2.0, 2.5, 3.0, 4.0]
|
||
if not mt5.initialize(): print("init", mt5.last_error()); sys.exit(1)
|
||
sym=None
|
||
for c in ("SpotCrude","USOIL","WTI","XTIUSD"):
|
||
if mt5.symbol_info(c): sym=c; break
|
||
sym=sym or "SpotCrude"
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n_bars+_N_BARS+_MAXH+5)
|
||
m30b = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n_bars//6+400)
|
||
mt5.shutdown()
|
||
if bars is None or m30b is None: print("Bars fehlen"); sys.exit(1)
|
||
|
||
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]
|
||
mT=[int(b["time"]) for b in m30b]; mc=[float(b["close"]) for b in m30b]
|
||
mh=[float(b["high"]) for b in m30b]; ml=[float(b["low"]) for b in m30b]
|
||
mEf=_ema_series(mc,_EMA_FAST); mEs=_ema_series(mc,_EMA_SLOW); mATR=_atr_series(mh,ml,mc)
|
||
|
||
w=WaveRecommender(_NeutralTU(), mt5.TIMEFRAME_M5)
|
||
# Signale einmal sammeln
|
||
sigs=[]
|
||
for i in range(_N_BARS, len(C)-_MAXH-1):
|
||
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)
|
||
m=_htf_sign_at(T[i], mT,mEf,mEs,mATR)
|
||
rec,_=w._build(ef,es,C[i-1],atr,"M5",5,htf_trend=m)
|
||
s=rec["signal"]
|
||
if s=="WARTEN": continue
|
||
sigs.append((i, 1 if s=="LONG" else -1, max(atr,_ATR_MIN)))
|
||
|
||
def metrics(Rs):
|
||
n=len(Rs); win=sum(1 for r in Rs if r>0)
|
||
gains=sum(r for r in Rs if r>0); losses=-sum(r for r in Rs if r<0)
|
||
losers=[r for r in Rs if r<0]
|
||
pf=gains/losses if losses>0 else float('inf')
|
||
avg_loss=sum(losers)/len(losers) if losers else 0.0
|
||
return n,win,pf,avg_loss
|
||
|
||
if mode == "be":
|
||
X = 2.0
|
||
bes = [0.6, 0.8, 1.0, 1.3, 1.5, 99.0]
|
||
print("="*66)
|
||
print(f" Breakeven-Test — {sym} M5+M30 SL={X}xATR Signale={len(sigs)}")
|
||
print("="*66)
|
||
print(f" {'Breakeven':<11}{'Treffer':>8}{'Oe-R':>8}{'Summe-R':>9}"
|
||
f"{'PF':>6}{'Oe-Verl.':>9}{'Scratch%':>9}")
|
||
for be in bes:
|
||
Rs=[]; scratch=0
|
||
for (i,d,atr) in sigs:
|
||
R,_=simulate(C[i], d, atr, X, H, L, C, i+1, be=be)
|
||
Rs.append(R)
|
||
if -0.15 < R < 0.05: scratch+=1 # ~Breakeven gescratcht
|
||
n,win,pf,avg_loss=metrics(Rs)
|
||
lbl = "aus (nie)" if be>10 else f"{be:.1f}"
|
||
print(f" {lbl:<11}{100*win/n:>7.0f}%{sum(Rs)/n:>8.3f}{sum(Rs):>9.1f}"
|
||
f"{pf:>6.2f}{avg_loss:>9.2f}{100*scratch/n:>8.0f}%")
|
||
print("\n Breakeven = ab wieviel xATR Profit der SL auf Entry rückt (aus=nie)")
|
||
print(" Scratch% = Anteil ~Breakeven-Ausgänge (R zw. −0,15 und +0,05)")
|
||
return
|
||
|
||
print("="*70)
|
||
print(f" Exit-Simulation — {sym} M5+M30 Signale={len(sigs)} Halt<= {_MAXH} Bars")
|
||
print(" (Teil-Exit AUS · Breakeven@0.6 · Trail HW∓1.5ATR · pessimistisch)")
|
||
print("="*70)
|
||
print(f" {'Init-SL':<9}{'Treffer':>8}{'Oe-R':>8}{'Summe-R':>9}{'PF':>6}"
|
||
f"{'Oe-Verl.':>9}{'Worst-R':>9}{'Init-Stop':>10}")
|
||
for X in widths:
|
||
Rs=[]; init_stops=0
|
||
for (i,d,atr) in sigs:
|
||
R, istop = simulate(C[i], d, atr, X, H, L, C, i+1)
|
||
Rs.append(R)
|
||
if istop: init_stops+=1
|
||
n,win,pf,avg_loss=metrics(Rs)
|
||
print(f" {X:<9.1f}{100*win/n:>7.0f}%{sum(Rs)/n:>8.3f}{sum(Rs):>9.1f}"
|
||
f"{pf:>6.2f}{avg_loss:>9.2f}{min(Rs):>9.2f}{100*init_stops/n:>9.0f}%")
|
||
print("\n Oe-R = Ø/Trade in ATR-Vielfachen · PF = Profit-Faktor")
|
||
print(" Oe-Verl. = Ø verlierender Trade · Worst-R = größter Einzelverlust (Tail)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|