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>
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Misst, ob ein Pullback-Einstieg (Kurs nahe der EMA = 'Wellenspitze' der
|
|
Gegenbewegung) den Edge gegenueber dem aktuellen Dauer-Trendsignal hebt.
|
|
|
|
Live-Konfig: Basis M5 + M30-Gegen-Trend-Filter (wie eingebaut). Fuer jedes
|
|
Signal wird der Abstand zur EMA50 in ATR ('near') berechnet und der Edge je
|
|
Naehe-Bucket gemessen. near klein = frisch am EMA (Pullback beendet),
|
|
near gross = weit weg gechased.
|
|
"""
|
|
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)
|
|
|
|
_TF = {"M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15,
|
|
"M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1}
|
|
|
|
|
|
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.0 - k)
|
|
out.append(e)
|
|
return out
|
|
|
|
|
|
def _rep(name, rets):
|
|
if not rets:
|
|
print(f" {name:<22} keine Signale"); return
|
|
n = len(rets); win = sum(1 for x in rets if x > 0)
|
|
print(f" {name:<22} n={n:>4} Treffer={100*win/n:>3.0f}% "
|
|
f"Oe-Edge={sum(rets)/n:+.4f} Summe={sum(rets):+.2f}")
|
|
|
|
|
|
def main():
|
|
base_lbl = (sys.argv[1].upper() if len(sys.argv) > 1 else "M5")
|
|
n_bars = int(sys.argv[2]) if len(sys.argv) > 2 else 4000
|
|
K = int(sys.argv[3]) if len(sys.argv) > 3 else 10
|
|
tf, htf = _TF[base_lbl], mt5.TIMEFRAME_M30
|
|
|
|
if not mt5.initialize():
|
|
print("MT5-Init:", mt5.last_error()); sys.exit(1)
|
|
sym = None
|
|
for cand in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
|
if mt5.symbol_info(cand):
|
|
sym = cand; break
|
|
sym = sym or "SpotCrude"
|
|
bars = mt5.copy_rates_from_pos(sym, tf, 0, n_bars + _N_BARS + K + 5)
|
|
hbars = mt5.copy_rates_from_pos(sym, htf, 0, n_bars // 2 + 300)
|
|
mt5.shutdown()
|
|
if bars is None or hbars is None:
|
|
print("Zu wenige Bars."); sys.exit(1)
|
|
|
|
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]
|
|
hT = [int(b["time"]) for b in hbars]
|
|
hC = [float(b["close"]) for b in hbars]
|
|
hH = [float(b["high"]) for b in hbars]; hL = [float(b["low"]) for b in hbars]
|
|
hEf = _ema_series(hC, _EMA_FAST); hEs = _ema_series(hC, _EMA_SLOW)
|
|
|
|
def htf_sign(ts):
|
|
lo, hi, idx = 0, len(hT) - 1, -1
|
|
while lo <= hi:
|
|
mid = (lo + hi) // 2
|
|
if hT[mid] <= ts: idx = mid; lo = mid + 1
|
|
else: hi = mid - 1
|
|
if idx < _EMA_SLOW: return 0
|
|
atr = _atr(hH[:idx+1], hL[:idx+1], hC[:idx+1])
|
|
if not atr: return 0
|
|
d = hEf[idx] - hEs[idx]
|
|
return 0 if abs(d) < _HTF_DEADBAND * atr else (1 if d > 0 else -1)
|
|
|
|
w = WaveRecommender(_NeutralTU(), tf)
|
|
buckets = {"(-inf,0)": [], "[0,0.3)": [], "[0.3,1)": [], "[1,2)": [], "[2,inf)": []}
|
|
gate03, nogate = [], []
|
|
for i in range(_N_BARS, len(C) - K):
|
|
win_c = C[i-_N_BARS:i]; win_h = H[i-_N_BARS:i]; win_l = L[i-_N_BARS:i]
|
|
atr = _atr(win_h, win_l, win_c)
|
|
if not atr or atr <= 0: continue
|
|
ef = _ema_last(win_c, _EMA_FAST); es = _ema_last(win_c, _EMA_SLOW)
|
|
ht = htf_sign(T[i])
|
|
rec, _ = w._build(ef, es, C[i-1], atr, base_lbl, 5, htf_trend=ht)
|
|
sig = rec["signal"]
|
|
if sig == "WARTEN": continue
|
|
fwd = C[i+K] - C[i]
|
|
r = fwd if sig == "LONG" else -fwd
|
|
stretch = (C[i-1] - es) / atr
|
|
near = stretch if sig == "LONG" else -stretch
|
|
nogate.append(r)
|
|
if near < 0: buckets["(-inf,0)"].append(r)
|
|
elif near < 0.3: buckets["[0,0.3)"].append(r)
|
|
elif near < 1.0: buckets["[0.3,1)"].append(r)
|
|
elif near < 2.0: buckets["[1,2)"].append(r)
|
|
else: buckets["[2,inf)"].append(r)
|
|
if near <= 0.3: gate03.append(r)
|
|
|
|
print("=" * 70)
|
|
print(f" Pullback-Test — {sym} Basis={base_lbl}+M30-Filter Vorlauf={K}")
|
|
print("=" * 70)
|
|
print("Edge je Abstand zur EMA (near = wie nah am EMA in ATR; klein = frisch):")
|
|
for k in ["(-inf,0)", "[0,0.3)", "[0.3,1)", "[1,2)", "[2,inf)"]:
|
|
_rep("near "+k, buckets[k])
|
|
print("\nVergleich Gate:")
|
|
_rep("OHNE Gate (alle)", nogate)
|
|
_rep("MIT Gate near<=0.3", gate03)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|