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>
109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
backtest_signal.py — Edge-Selbsttest der Empfehlung
|
||
====================================================
|
||
Rechnet die ECHTE Empfehlungslogik (core/wave_rec.WaveRecommender._build)
|
||
über die letzten N historischen Bars nach und misst, ob der Kurs DANACH in
|
||
Signalrichtung läuft. So siehst du jederzeit, ob die Empfehlung auf dem
|
||
aktuellen Markt einen Vorhersagewert (Edge) hat.
|
||
|
||
Ø-Edge > 0 → Empfehlung trägt (Kurs folgt dem Signal)
|
||
Ø-Edge ≈ 0 → kein Edge (Zufall)
|
||
Ø-Edge < 0 → Signal läuft verkehrt
|
||
|
||
Aufruf (läuft parallel zum Server, nur Lese-Zugriff auf die MT5-History):
|
||
python backtest_signal.py # M15, 3000 Bars, Vorlauf 4/6/10
|
||
python backtest_signal.py M5 2000 # andere TF / Bar-Anzahl
|
||
python backtest_signal.py M15 3000 8 # fester Vorlauf 8 Bars
|
||
|
||
TU-Bestätigung wird im Test neutral gesetzt (historisch nicht verfügbar) —
|
||
gemessen wird also der reine Richtungs-Edge der Logik; live kommt die
|
||
TU-Bestätigung noch obendrauf.
|
||
"""
|
||
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)
|
||
|
||
_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 _report(name: str, rets: list[float]) -> None:
|
||
if not rets:
|
||
print(f" {name:<8} keine Signale"); return
|
||
n = len(rets)
|
||
win = sum(1 for x in rets if x > 0)
|
||
print(f" {name:<8} n={n:>4} Treffer={100*win/n:>3.0f}% "
|
||
f"Ø-Edge={sum(rets)/n:+.4f} Summe={sum(rets):+.2f}")
|
||
|
||
|
||
def main():
|
||
tf_lbl = (sys.argv[1].upper() if len(sys.argv) > 1 else "M15")
|
||
n_bars = int(sys.argv[2]) if len(sys.argv) > 2 else 3000
|
||
ks = [int(sys.argv[3])] if len(sys.argv) > 3 else [4, 6, 10]
|
||
if tf_lbl not in _TF:
|
||
print(f"Unbekannte TF '{tf_lbl}'. Erlaubt: {', '.join(_TF)}"); sys.exit(1)
|
||
tf = _TF[tf_lbl]
|
||
|
||
if not mt5.initialize():
|
||
print(f"MT5-Init fehlgeschlagen: {mt5.last_error()}"); sys.exit(1)
|
||
sym = None
|
||
for cand in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
||
if mt5.symbol_info(cand):
|
||
sym = cand; break
|
||
if sym is None:
|
||
from core.config import load_config
|
||
sym = load_config()["trading"].get("last_symbol", "").strip() or "SpotCrude"
|
||
bars = mt5.copy_rates_from_pos(sym, tf, 0, n_bars + _N_BARS + max(ks) + 5)
|
||
mt5.shutdown()
|
||
if bars is None or len(bars) < _N_BARS + 50:
|
||
print("Zu wenige Bars von MT5 erhalten."); 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]
|
||
w = WaveRecommender(_NeutralTU(), tf)
|
||
|
||
print("=" * 60)
|
||
print(f" Empfehlungs-Edge — {sym} {tf_lbl}, {len(C)} Bars")
|
||
print("=" * 60)
|
||
|
||
for K in ks:
|
||
res = {"LONG": [], "SHORT": [], "WARTEN": 0}
|
||
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]
|
||
if len(win_c) < _EMA_SLOW + 5:
|
||
continue
|
||
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)
|
||
rec, _ = w._build(ef, es, C[i - 1], atr, tf_lbl, 5)
|
||
sig = rec["signal"]
|
||
if sig == "WARTEN":
|
||
res["WARTEN"] += 1; continue
|
||
fwd = C[i + K] - C[i]
|
||
res[sig].append(fwd if sig == "LONG" else -fwd)
|
||
|
||
print(f"\nVorlauf {K} Bars (~{K} × {tf_lbl}):")
|
||
_report("LONG", res["LONG"])
|
||
_report("SHORT", res["SHORT"])
|
||
_report("ALLE", res["LONG"] + res["SHORT"])
|
||
print(f" WARTEN (kein Trade): {res['WARTEN']}")
|
||
|
||
print("\nØ-Edge > 0 = Empfehlung hat Vorhersagewert | ≈0 = Zufall | <0 = verkehrt")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|