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>
151 lines
6.3 KiB
Python
151 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
||
"""Marktstruktur als SIGNAL — Backtest (2026-07-20, User-Idee „Pro-Chart-Setup").
|
||
Testet die handelbaren Varianten der `structure.py`-Anzeige (nutzt DEREN Logik,
|
||
kausal Bar für Bar), damit Signal == Anzeige:
|
||
|
||
A) Kanal-Pullback (der orange Pfeil): im Aufwärts-Regressionskanal Kurs zurück
|
||
ans UNTERE Kanaldrittel (pos ≤ X) → LONG; Abwärtskanal + pos ≥ 1−X → SHORT.
|
||
Frischer Eintritt in die Zone (pos kreuzt X), one-shot.
|
||
A2) wie A, aber ZUSÄTZLICH Swing-Struktur-Filter (Trend = HH/HL bzw. LH/LL).
|
||
B) BOS-Continuation: frischer Break of Structure in Kanalrichtung → Entry.
|
||
|
||
Sequentielle 1-Positions-Sim, Live-Exit (SL 2,0×ATR + Trailing 1,5 + BE 1,3),
|
||
Echtkosten = Bar-Spread/ATR. M30. 2 Halbjahre. Maßstab: Squeeze ØR +0,14…+0,23 &
|
||
PF>1 in BEIDEN Hälften. Verdict-Regel: Einbau nur bei Robustheit über beide Hälften
|
||
UND Parameter.
|
||
"""
|
||
import sys
|
||
import MetaTrader5 as mt5
|
||
from core.structure import _pivots, _classify, _channel, _atr as _atr_win
|
||
|
||
_MAXH = 96; _ATRMIN = 0.06; _COOL = 3; _LOOK = 220; _REG = 60
|
||
|
||
|
||
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])))
|
||
return [(sum(t[max(1, i-p+1):i+1])/max(1, len(t[max(1, i-p+1):i+1]))) if i else None
|
||
for i in range(len(C))]
|
||
|
||
|
||
def st(Rs):
|
||
if not Rs: return None
|
||
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)
|
||
return dict(n=n, wr=100*w/n, oR=s/n, pf=(up/dn if dn > 0 else 9.99), sum=s)
|
||
|
||
|
||
def line(lbl, s):
|
||
if not s: return f" {lbl:<30} —"
|
||
return (f" {lbl:<30} n={s['n']:>4} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
|
||
f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+6.0f}")
|
||
|
||
|
||
def sim(entry, d, atr, H, L, C, j0):
|
||
eff = entry - d*2.0*atr; hw = entry
|
||
end = min(j0+_MAXH, len(C)-1); exit_px = C[end]; exit_j = end
|
||
for j in range(j0, end+1):
|
||
hj, lj = H[j], L[j]
|
||
if (lj <= eff) if d > 0 else (hj >= eff):
|
||
return (eff-entry)*d/atr, j
|
||
hw = max(hw, hj) if d > 0 else min(hw, lj)
|
||
prof = (C[j]-entry)*d
|
||
if prof >= 0.3*atr:
|
||
cand = hw - d*1.5*atr
|
||
if prof >= 1.3*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, exit_j
|
||
|
||
|
||
def _trend_from_swings(C, i):
|
||
"""Swing-Trend (HH/HL→up, LH/LL→down) kausal über die letzten _LOOK Bars."""
|
||
lo = max(0, i-_LOOK)
|
||
piv = _pivots([0]*0 or None, None, 0) if False else None
|
||
return None # (Platzhalter, in run() direkt mit H/L berechnet)
|
||
|
||
|
||
def run(H, L, C, A, SP, lo, hi, rule, X):
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
|
||
Rs = []
|
||
i = max(lo, _LOOK)
|
||
prev_pos = None
|
||
while i < hi:
|
||
atr = A[i]
|
||
if not atr or atr < _ATRMIN:
|
||
i += 1; prev_pos = None; continue
|
||
w0 = max(0, i-_REG+1)
|
||
ch = _channel(C[w0:i+1], atr)
|
||
if not ch:
|
||
i += 1; prev_pos = None; continue
|
||
pos = ch["pos"]; cdir = ch["dir"]
|
||
d = 0
|
||
if rule in ("A", "A2"):
|
||
# frischer Eintritt ins untere (up) bzw. obere (down) Kanaldrittel
|
||
if cdir == "up" and pos <= X and (prev_pos is None or prev_pos > X):
|
||
d = 1
|
||
elif cdir == "down" and pos >= 1-X and (prev_pos is None or prev_pos < 1-X):
|
||
d = -1
|
||
if d != 0 and rule == "A2":
|
||
# zusätzlich Swing-Struktur bestätigen
|
||
pl = _pivots(H[max(0, i-_LOOK):i+1], L[max(0, i-_LOOK):i+1], 3)
|
||
lab = _classify(pl)
|
||
recent = [x["type"] for x in lab[-4:]]
|
||
ups = sum(1 for t in recent if t in ("HH", "HL"))
|
||
dns = sum(1 for t in recent if t in ("LH", "LL"))
|
||
strend = "up" if ups >= 3 and ups > dns else "down" if dns >= 3 and dns > ups else "range"
|
||
if (d > 0 and strend != "up") or (d < 0 and strend != "down"):
|
||
d = 0
|
||
elif rule == "B":
|
||
# frischer BOS in Kanalrichtung
|
||
pl = _pivots(H[max(0, i-_LOOK):i+1], L[max(0, i-_LOOK):i+1], 3)
|
||
lab = _classify(pl)
|
||
from core.structure import _last_bos
|
||
bos = _last_bos(lab, i+1-max(0, i-_LOOK))
|
||
if bos and bos["bars_ago"] <= int(X): # X = max Bars seit BOS
|
||
if bos["dir"] == "up" and cdir != "down": d = 1
|
||
elif bos["dir"] == "down" and cdir != "up": d = -1
|
||
prev_pos = pos
|
||
if d == 0:
|
||
i += 1; continue
|
||
entry = C[i]
|
||
r, xj = sim(entry, d, max(atr, _ATRMIN), H, L, C, i+1)
|
||
Rs.append(r - cost(i, max(atr, _ATRMIN)))
|
||
i = xj + _COOL; prev_pos = None
|
||
return Rs
|
||
|
||
|
||
def main():
|
||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 60000
|
||
mt5.initialize()
|
||
sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(c)), None)
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n)
|
||
point = mt5.symbol_info(sym).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("="*90)
|
||
print(f" Marktstruktur als SIGNAL — {sym} M30 ({N} Bars, seq. Sim, Live-Exit, Echtkosten)")
|
||
print(f" Maßstab: Squeeze ØR +0,14…+0,23 & PF>1 in BEIDEN Hälften. Verdict = beidhälftig robust.")
|
||
print("="*90)
|
||
|
||
tests = [("A Kanal-Pullback X=0.20", "A", 0.20),
|
||
("A Kanal-Pullback X=0.30", "A", 0.30),
|
||
("A2 +Swing-Filter X=0.25", "A2", 0.25),
|
||
("B BOS-Cont. ≤1 Bar", "B", 1),
|
||
("B BOS-Cont. ≤3 Bars", "B", 3)]
|
||
for lbl, rule, X in tests:
|
||
s1 = st(run(H, L, C, A, SP, 0, mid, rule, X))
|
||
s2 = st(run(H, L, C, A, SP, mid, N-_MAXH-1, rule, X))
|
||
print(f"\n {lbl}:")
|
||
print(line("H1 (alt)", s1))
|
||
print(line("H2 (neu)", s2))
|
||
ok = (s1 and s2 and s1['oR'] > 0 and s2['oR'] > 0 and s1['pf'] > 1 and s2['pf'] > 1)
|
||
print(f" → {'ROBUST (beide Hälften positiv)' if ok else 'fällt durch (nicht beidseitig positiv)'}")
|
||
print()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|