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>
205 lines
7.9 KiB
Python
205 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Inter-Market-Kontext als Verdict-Kandidat (2026-07-19, User-Frage „weitere
|
||
Indikatoren?"): die EINZIGE noch ungetestete Indikator-Klasse — Information von
|
||
AUSSERHALB des WTI-Charts. Getestet werden zwei Hypothesen:
|
||
|
||
(A) Brent-Bestätigung: Brent-Trend (EMA12/50 M5) bestätigt das WTI-Signal →
|
||
besserer Edge; Brent DAGEGEN → schlechter. (Brent-WTI laufen ~parallel,
|
||
Divergenz = möglicher Fehlausbruch.)
|
||
(B) DXY-Gegenwind: USDX-Trend gegen die Trade-Richtung (Öl in USD:
|
||
Dollar rauf = Öl-Gegenwind für LONG, Dollar runter = Gegenwind für SHORT).
|
||
|
||
Methodik wie die 8 verworfenen Signal-Filter: ereignisbasierte EMA-Signale auf
|
||
WTI M5 (frisches Kreuzen, Totband 0,15×ATR), sequentielle 1-Positions-Sim mit
|
||
Live-Exit (SL 2,0×ATR + Trailing 1,5 + BE 1,3), Echtkosten = Bar-Spread/ATR.
|
||
Jeder Trade wird nach dem Inter-Market-Zustand AM ENTRY gebucketed. 2 Halbjahre.
|
||
|
||
Verdict-Regel (vorab festgelegt, gegen Parameter-Fishing): ein Gate/Konfidenz-
|
||
Einbau kommt NUR in Frage, wenn der „dagegen"-Bucket in BEIDEN Hälften klar
|
||
schlechter ist als „dafür" UND das über beide Feature-Varianten (EMA-Stand und
|
||
Momentum N=12/36) robust ist. Maßstab der 6×-Lektion: Signal-Filter tragen kaum.
|
||
"""
|
||
import sys
|
||
import MetaTrader5 as mt5
|
||
|
||
_MAXH = 288; _ATRMIN = 0.12; _COOL = 6; _DEAD = 0.15
|
||
|
||
|
||
def _ema_series(C, p):
|
||
k = 2.0 / (p + 1); e = C[0]; out = [e]
|
||
for x in C[1:]:
|
||
e = e + k * (x - e); out.append(e)
|
||
return out
|
||
|
||
|
||
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 _load(sym, n):
|
||
mt5.symbol_select(sym, True)
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
|
||
if bars is None or len(bars) < 1000:
|
||
print(f"FEHLER: keine Daten für {sym}"); sys.exit(1)
|
||
point = mt5.symbol_info(sym).point
|
||
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]
|
||
SP = [float(b["spread"])*point for b in bars]
|
||
return T, H, L, C, SP
|
||
|
||
|
||
def _dir_series(H, L, C):
|
||
"""EMA12/50-Richtung mit Totband: +1/-1/0 je Bar."""
|
||
e12 = _ema_series(C, 12); e50 = _ema_series(C, 50); A = _atr_series(H, L, C)
|
||
out = []
|
||
for i in range(len(C)):
|
||
a = A[i] or 0
|
||
if a <= 0: out.append(0); continue
|
||
d = e12[i] - e50[i]
|
||
out.append(1 if d > _DEAD*a else (-1 if d < -_DEAD*a else 0))
|
||
return out, A
|
||
|
||
|
||
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:<26} —"
|
||
return (f" {lbl:<26} 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_trades(T, H, L, C, A, SP):
|
||
"""Sequentielle Sim: frisches EMA-Signal → Trade mit Live-Exit. Liefert
|
||
Liste (entry_index, dir, netto_R)."""
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
|
||
sig, _ = _dir_series(H, L, C)
|
||
out = []; i = 60
|
||
while i < len(C) - 2:
|
||
if sig[i] == 0 or sig[i] == sig[i-1]:
|
||
i += 1; continue
|
||
d = sig[i]
|
||
atr = max(A[i] or 0, _ATRMIN)
|
||
entry = C[i]; eff = entry - d*2.0*atr; hw = entry
|
||
end = min(i+_MAXH, len(C)-1); exit_px = C[end]; exit_j = end
|
||
for j in range(i+1, end+1):
|
||
hj, lj = H[j], L[j]
|
||
if (lj <= eff) if d > 0 else (hj >= eff):
|
||
exit_px = eff; exit_j = j; break
|
||
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)
|
||
out.append((i, d, (exit_px-entry)*d/atr - cost(i, atr)))
|
||
i = exit_j + _COOL
|
||
return out
|
||
|
||
|
||
def main():
|
||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||
mt5.initialize()
|
||
Tw, Hw, Lw, Cw, SPw = _load("SpotCrude", n+_MAXH+60)
|
||
Tb, Hb, Lb, Cb, _ = _load("SpotBrent", n+_MAXH+60)
|
||
Tx, Hx, Lx, Cx, _ = _load("USDX", n+_MAXH+60)
|
||
mt5.shutdown()
|
||
|
||
dir_b, Ab = _dir_series(Hb, Lb, Cb)
|
||
dir_x, Ax = _dir_series(Hx, Lx, Cx)
|
||
idx_b = {t: i for i, t in enumerate(Tb)}
|
||
idx_x = {t: i for i, t in enumerate(Tx)}
|
||
Aw = _atr_series(Hw, Lw, Cw)
|
||
|
||
def look(idx, t):
|
||
"""Index zum WTI-Zeitstempel (bis 2 Gitterschritte zurück — Feed-Lücken)."""
|
||
for dt_ in (0, 300, 600):
|
||
j = idx.get(t - dt_)
|
||
if j is not None: return j
|
||
return None
|
||
|
||
def mom_dir(C_, A_, j, N, thr=0.3, floor=None):
|
||
if j is None or j < N: return None
|
||
a = A_[j] or 0
|
||
a = max(a, floor) if floor else (a if a > 0 else None)
|
||
if not a: return None
|
||
m = (C_[j] - C_[j-N]) / a
|
||
return 1 if m > thr else (-1 if m < -thr else 0)
|
||
|
||
trades = sim_trades(Tw, Hw, Lw, Cw, Aw, SPw)
|
||
print("="*88)
|
||
print(f" Inter-Market-Kontext — WTI M5 ({len(trades)} Trades, seq. Sim, Live-Exit, Echtkosten)")
|
||
print(f" Buckets nach Zustand AM ENTRY. Verdict nur bei Robustheit in BEIDEN Hälften.")
|
||
print("="*88)
|
||
|
||
mid_t = Tw[len(Tw)//2]
|
||
feats = [] # (half, dir, R, brent_ema, dxy_ema, brent_m12, brent_m36, dxy_m12, dxy_m36)
|
||
miss_b = miss_x = 0
|
||
for i, d, r in trades:
|
||
t = Tw[i]
|
||
jb = look(idx_b, t); jx = look(idx_x, t)
|
||
if jb is None: miss_b += 1
|
||
if jx is None: miss_x += 1
|
||
h = 1 if t < mid_t else 2
|
||
feats.append((h, d, r,
|
||
dir_b[jb] if jb is not None else None,
|
||
dir_x[jx] if jx is not None else None,
|
||
mom_dir(Cb, Ab, jb, 12, floor=0.12),
|
||
mom_dir(Cb, Ab, jb, 36, floor=0.12),
|
||
mom_dir(Cx, Ax, jx, 12),
|
||
mom_dir(Cx, Ax, jx, 36)))
|
||
if miss_b or miss_x:
|
||
print(f" (ohne Inter-Market-Match: Brent {miss_b} · USDX {miss_x} — übersprungen)")
|
||
|
||
def bucket3(val, d):
|
||
"""dafür / dagegen / neutral relativ zur Trade-Richtung."""
|
||
if val is None: return None
|
||
if val == 0: return "neutral"
|
||
return "dafür" if val == d else "dagegen"
|
||
|
||
def bucket_dxy(val, d):
|
||
"""DXY: Dollar MIT der Trade-Richtung = Gegenwind (Öl invers zum Dollar)."""
|
||
if val is None: return None
|
||
if val == 0: return "neutral"
|
||
return "Gegenwind" if val == d else "Rückenwind"
|
||
|
||
sections = [
|
||
("(A) Brent-Trend (EMA12/50)", 3, bucket3),
|
||
("(A2) Brent-Momentum 1h (N=12)", 5, bucket3),
|
||
("(A3) Brent-Momentum 3h (N=36)", 6, bucket3),
|
||
("(B) DXY-Trend (EMA12/50)", 4, bucket_dxy),
|
||
("(B2) DXY-Momentum 1h (N=12)", 7, bucket_dxy),
|
||
("(B3) DXY-Momentum 3h (N=36)", 8, bucket_dxy),
|
||
]
|
||
order = ("dafür", "dagegen", "neutral", "Rückenwind", "Gegenwind")
|
||
for title, col, bfn in sections:
|
||
print(f"\n {title}:")
|
||
for h in (1, 2):
|
||
print(f" H{h}:")
|
||
groups = {}
|
||
for row in feats:
|
||
if row[0] != h: continue
|
||
b = bfn(row[col], row[1])
|
||
if b is None: continue
|
||
groups.setdefault(b, []).append(row[2])
|
||
for name in order:
|
||
if name in groups:
|
||
print(line(name, st(groups[name])))
|
||
|
||
print("\n Verdict-Regel: Einbau NUR wenn 'dagegen'/'Gegenwind' in BEIDEN Hälften")
|
||
print(" klar schlechter als 'dafür'/'Rückenwind' UND über die Varianten robust.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|