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>
132 lines
6.0 KiB
Python
132 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
||
"""Auto-Flip-Close (User-Taktik-Kern, 2026-07-16): Position im PLUS automatisch
|
||
schließen, sobald die Empfehlung auf die GEGENRICHTUNG dreht (heute nur Alarm).
|
||
Sequentielle Trade-Sim (EINE Position wie live): Entry bei Signal, Exit via
|
||
SL 2,0×ATR + Trailing 1,5 + BE 1,3 (Basis) — Varianten zusätzlich mit Flip-Close
|
||
(nur im Plus; optional Mindestgewinn in ×ATR ~ der '3%-Margin'-Idee; Referenz:
|
||
Flip auch im Minus). Signale aus der echten `_build`-Logik (M30-Filter +
|
||
H1-Konfluenz + Winkel, hour=None = gate-frei). Kosten = Bar-Spread/ATR.
|
||
2 Halbjahre. R = Profit/ATR.
|
||
|
||
Entscheidung: Auto-Flip-Close nur bauen, wenn ØR & ΣR in BEIDEN Hälften ≥ Basis.
|
||
"""
|
||
import sys, bisect
|
||
import MetaTrader5 as mt5
|
||
from core.analysis import calc_trend_angle
|
||
from core.wave_rec import (WaveRecommender, _atr, _ema_last, _ema_series,
|
||
_EMA_FAST, _EMA_SLOW, _N_BARS, _HTF_DEADBAND, _ANGLE_LR)
|
||
_MAXH = 288; _ATRMIN = 0.12
|
||
|
||
|
||
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 99.9),
|
||
sum=s, worst=min(Rs))
|
||
|
||
|
||
def line(lbl, s):
|
||
if not s: return f" {lbl:<34} —"
|
||
return (f" {lbl:<34} n={s['n']:>4} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
|
||
f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+6.0f} Worst={s['worst']:+.1f}")
|
||
|
||
|
||
def run_seq(sig, H, L, C, A, SP, lo, hi, flip=False, min_r=0.0, flip_neg=False):
|
||
"""Sequentielle Sim: eine Position, Entry bei Signal, Exit SL/Trail/(Flip)."""
|
||
def cost(i, atr): return (SP[i] if SP[i] > 0 else 0.0225)/atr
|
||
Rs = []
|
||
i = lo
|
||
while i < hi:
|
||
d = sig[i]
|
||
if d == 0:
|
||
i += 1; continue
|
||
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):
|
||
hi_, lo_ = H[j], L[j]
|
||
if (lo_ <= eff) if d > 0 else (hi_ >= eff):
|
||
exit_px = eff; exit_j = j; break
|
||
hw = max(hw, hi_) if d > 0 else min(hw, lo_)
|
||
prof = (C[j]-entry)*d
|
||
# Flip-Close: Signal dreht auf Gegenrichtung → Close zum Bar-Close
|
||
if flip and sig[j] == -d and (flip_neg or prof > min_r*atr):
|
||
exit_px = C[j]; exit_j = j; break
|
||
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)
|
||
Rs.append((exit_px-entry)*d/atr - cost(i, atr))
|
||
i = exit_j + 1
|
||
return Rs
|
||
|
||
|
||
def main():
|
||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||
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_M5, 0, n+_N_BARS+_MAXH+5)
|
||
m30 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M30, 0, n//6+500)
|
||
h1 = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_H1, 0, n//12+500)
|
||
si = mt5.symbol_info(sym); point = si.point
|
||
mt5.shutdown()
|
||
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]
|
||
A = _atr_series(H, L, C)
|
||
|
||
def series(rr):
|
||
t = [int(b["time"]) for b in rr]; c = [float(b["close"]) for b in rr]
|
||
hh = [float(b["high"]) for b in rr]; ll = [float(b["low"]) for b in rr]
|
||
return t, _ema_series(c, _EMA_FAST), _ema_series(c, _EMA_SLOW), _atr_series(hh, ll, c)
|
||
mT, mEf, mEs, mA = series(m30)
|
||
hT, hEf, hEs, hA = series(h1)
|
||
|
||
def tf_sign(tt, ef, es, aa, ts):
|
||
k = bisect.bisect_right(tt, ts)-1
|
||
if k < _EMA_SLOW or aa[k] is None or aa[k] <= 0: return 0
|
||
dd = ef[k]-es[k]
|
||
return 0 if abs(dd) < _HTF_DEADBAND*aa[k] else (1 if dd > 0 else -1)
|
||
|
||
w = WaveRecommender(type("T", (), {"snapshot": lambda s: {"intervals": {}}})(), mt5.TIMEFRAME_M5)
|
||
sig = [0]*len(C)
|
||
for i in range(_N_BARS, len(C)-1):
|
||
wc = C[i-_N_BARS:i]; wh = H[i-_N_BARS:i]; wl = L[i-_N_BARS:i]
|
||
atr = _atr(wh, wl, wc)
|
||
if not atr or atr <= 0: continue
|
||
ef = _ema_last(wc, _EMA_FAST); es = _ema_last(wc, _EMA_SLOW)
|
||
a5 = calc_trend_angle(C[i-_ANGLE_LR-2:i], _ANGLE_LR)
|
||
rec, _ = w._build(ef, es, C[i-1], atr, "M5", 0,
|
||
htf_trend=tf_sign(mT, mEf, mEs, mA, T[i]),
|
||
h1_trend=tf_sign(hT, hEf, hEs, hA, T[i]), angle=a5)
|
||
s_ = rec["signal"]
|
||
sig[i] = 1 if s_ == "LONG" else -1 if s_ == "SHORT" else 0
|
||
mid = len(C)//2
|
||
|
||
print("="*92)
|
||
print(f" Auto-Flip-Close — {sym} M5 (sequentielle 1-Positions-Sim · Exit live · Echtkosten)")
|
||
print(f" Flip = Signal dreht auf Gegenrichtung → Close zum Bar-Close (heute nur Alarm)")
|
||
print("="*92)
|
||
for lbl, lo, hi_ in (("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C)-_MAXH-1)):
|
||
print(f"\n{lbl}:")
|
||
print(line("BASIS (nur SL/Trailing)", st(run_seq(sig, H, L, C, A, SP, lo, hi_))))
|
||
print(line("+ Flip-Close im Plus (minR=0)", st(run_seq(sig, H, L, C, A, SP, lo, hi_, flip=True))))
|
||
print(line("+ Flip-Close ab +0.3xATR", st(run_seq(sig, H, L, C, A, SP, lo, hi_, flip=True, min_r=0.3))))
|
||
print(line("+ Flip-Close ab +0.5xATR", st(run_seq(sig, H, L, C, A, SP, lo, hi_, flip=True, min_r=0.5))))
|
||
print(line("+ Flip-Close IMMER (auch Minus)", st(run_seq(sig, H, L, C, A, SP, lo, hi_, flip=True, flip_neg=True))))
|
||
print(f"\n Bauen nur, wenn eine Flip-Variante in BEIDEN Hälften ØR & ΣR ≥ BASIS hält.")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|