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>
111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
"""Volatilitäts-Squeeze-Breakout (B2, User-Idee 'Breakouterkennung'):
|
||
Struktureller Ausbruch aus einer KOMPRESSION — NICHT aus dem EMA-Trend.
|
||
Setup: die Spanne der letzten N Bars (Box) ist eng relativ zum ATR (Squeeze);
|
||
danach bricht der Kurs k×ATR über/unter die Box → Einstieg in Ausbruchsrichtung.
|
||
Exit = live-Modell (SL 2,0×ATR + Trailing 1,5, Breakeven 1,3), pessimistisch.
|
||
Kosten = echter Bar-Spread/ATR (Ø ~0,265). 2 Halbjahre. R = Profit/ATR.
|
||
|
||
Kernfrage (B3): Trägt der SQUEEZE-Filter netto in BEIDEN Hälften — schlägt er den
|
||
Ausbruch aus einer BELIEBIGEN Box (ohne Kompression)? Sonst raus.
|
||
"""
|
||
import sys
|
||
import MetaTrader5 as mt5
|
||
|
||
_MAXH = 288; _ATRMIN = 0.12
|
||
_N = 12 # Box-Länge (Bars) = 1 h auf M5
|
||
_W = 24 # Fenster nach der Box, in dem der Breakout erfolgen muss (2 h)
|
||
_COOL = 12 # Cooldown nach einem Trade (Bars) gegen Überlappung
|
||
_K = 0.1 # Ausbruch k×ATR über/unter die Box-Grenze
|
||
|
||
|
||
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])))
|
||
out = [None]
|
||
for i in range(1, len(C)):
|
||
seg = t[max(1, i-p+1):i+1]
|
||
out.append(sum(seg)/len(seg))
|
||
return out
|
||
|
||
|
||
def sim(entry, d, atr, H, L, C, j0, sl_atr=2.0, trail=1.5, trail_on=0.3, be_on=1.3):
|
||
eff = entry - d*sl_atr*atr; hw = entry
|
||
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): exit_px = eff; break
|
||
hw = max(hw, hi) if d > 0 else min(hw, lo)
|
||
prof = (C[j]-entry)*d
|
||
if prof >= trail_on*atr:
|
||
cand = hw - d*trail*atr
|
||
if prof >= be_on*atr: cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||
eff = max(eff, cand) if d > 0 else min(eff, cand)
|
||
return (exit_px-entry)*d/atr
|
||
|
||
|
||
def rep(name, Rs):
|
||
if not Rs: print(f" {name:<32} -"); return
|
||
n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
|
||
up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
|
||
pf = up/dn if dn > 0 else 9.99
|
||
print(f" {name:<32} Trades={n:>4} Treffer={100*w/n:>3.0f}% "
|
||
f"ØR={s/n:+.3f} PF={pf:.2f} ΣR={s:+.0f}")
|
||
|
||
|
||
def scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult):
|
||
"""Ein Durchlauf: sammelt R (netto) je Breakout. Rs_any = ALLE Box-Ausbrüche,
|
||
Rs_sq = nur die mit Squeeze (Box<=squeeze_mult×ATR). i springt nach jedem
|
||
Breakout um _COOL vor (Überlappungsschutz) — für beide Listen identisch."""
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
|
||
Rs_sq = []; Rs_any = []
|
||
i = max(lo_i, _N+15)
|
||
while i < min(hi_i, len(C)-_MAXH-1):
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN: i += 1; continue
|
||
boxHi = max(H[i-_N:i]); boxLo = min(L[i-_N:i]); box = boxHi-boxLo
|
||
squeezed = box <= squeeze_mult*atr
|
||
hit = None
|
||
for j in range(i, min(i+_W, len(C)-_MAXH-1)):
|
||
up = boxHi + _K*atr; dn = boxLo - _K*atr
|
||
if H[j] >= up: hit = (j, 1, up); break
|
||
if L[j] <= dn: hit = (j, -1, dn); break
|
||
if hit is None: i += 1; continue
|
||
j, d, lvl = hit
|
||
R = sim(lvl, d, atr, H, L, C, j+1) - cost(j, atr)
|
||
Rs_any.append(R)
|
||
if squeezed: Rs_sq.append(R)
|
||
i = j + _COOL
|
||
return Rs_sq, Rs_any
|
||
|
||
|
||
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 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n+_MAXH+30)
|
||
si = mt5.symbol_info(sym); point = si.point
|
||
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]
|
||
A = _atr_series(H, L, C)
|
||
N = len(C); mid = N//2
|
||
print("="*88)
|
||
print(f" Volatilitäts-Squeeze-Breakout — {sym} M5 ({N} Bars · Box={_N} · Ausbruch={_K}×ATR)")
|
||
print(f" Exit=live(SL2,0/Trail1,5/BE1,3) · Kosten=Bar-Spread/ATR · 2 Halbjahre")
|
||
print("="*88)
|
||
for label, lo, hi in (("H1 (alt)", 0, mid), ("H2 (neu)", mid, N)):
|
||
print(f"\n{label}:")
|
||
_, any_R = scan(H, L, C, A, SP, point, lo, hi, 99.0)
|
||
rep("Ausbruch OHNE Squeeze-Filter", any_R)
|
||
for sm in (2.5, 2.0, 1.5):
|
||
sq_R, _ = scan(H, L, C, A, SP, point, lo, hi, sm)
|
||
rep(f"Squeeze Box<={sm}×ATR", sq_R)
|
||
print(f"\n Squeeze trägt nur, wenn ØR & PF in BEIDEN Hälften > 'ohne Filter' UND ØR netto > 0.")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|