Axel HocksandClaude Opus 5
3314b6e9b1
User-Frage "sollten wir den S/R close auch deaktivieren?" -> nein, die
beiden sind gegensaetzlich gemessen. Aber sie konkurrieren seit dem Einbau
um dieselben Trades, und die S/R-Freigabe lief OHNE Flip-Close.
backtest_exit_combo.py (NEU, 80k M5, 2 Halbjahre, Echtkosten, Exit-Kern aus
core/exit_model.py, 5 Varianten auf IDENTISCHEN Entries):
nur S/R-Close H1 +59 · H2 -14 (535/284 S/R-Closes)
nur Flip-Close H1 -1 · H2 -43
BEIDE H1 +56 · H2 -50 (507/252)
BEIDE, Flip nachrangig H1 +62 · H2 -49 (516/260)
KOLLISION BELEGT: der Flip nahm dem S/R-Close 28 bzw. 32 Trades weg
(5-11 %) - genau die, fuer die dieser validiert ist. Mit Nachrang ist es in
BEIDEN Haelften besser als ohne.
Umsetzung: liegt ein S/R-ZIEL <=0,6xATR entfernt (dieselbe Schwelle wie der
Live-Hinweis), haelt der Flip sich zurueck.
[trading] auto_flip_close_subordinate (Default true).
Mit 4 Szenarien getestet: in Reichweite, ausser Reichweite, kein Level,
Nachrang aus.
EHRLICH DAZU: auch mit Nachrang kostet der Flip in H2 rund 35 R gegenueber
"nur S/R-Close" (-49 vs -14). Der Nachrang begrenzt den Schaden, er dreht
ihn nicht um.
METHODIK-FEHLER BEIM BAU, KORRIGIERT: der erste Lauf liess die sequentielle
Sim nach dem Exit bei xb+1 weiterlaufen - dadurch hatte JEDE Variante eine
ANDERE Trade-Folge und die Zahlen waren nicht vergleichbar. Jetzt feste,
geteilte Entry-Liste wie in backtest_pbreak_rvalue.py. (Die zunaechst
gemeldeten Deltas -195/-144 stammten aus diesem konfundierten Lauf.)
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
178 lines
7.6 KiB
Python
178 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
||
"""S/R-Close und Flip-Close im DIREKTEN Wettbewerb (User-Frage 2026-07-31).
|
||
|
||
WARUM: Seit dem Einbau des Auto-Flip-Close konkurrieren zwei Exits um dieselben
|
||
Trades — **wer zuerst feuert, gewinnt.** Beide sind einzeln gemessen, aber gegen
|
||
UNTERSCHIEDLICHE Baselines:
|
||
S/R-Close (neues Modell, Schwelle 0,35): H1 +357 · H2 +1418 gegen reines Trailing
|
||
Flip-Close (0,5×ATR): H1 −92 · H2 −160 gegen reines Trailing
|
||
Die S/R-Messung lief OHNE Flip-Close. Die Baseline, gegen die der validierte Exit
|
||
freigegeben wurde, existiert seit dem Einbau nicht mehr.
|
||
|
||
BEFÜRCHTUNG: Ein Trade läuft ins Ziel-Level, wäre dort (P(break) niedrig) gut
|
||
geschlossen worden — aber die nachlaufende EMA dreht 0,5×ATR vorher und der
|
||
Flip-Close nimmt ihn früher und schlechter raus.
|
||
|
||
GEMESSEN WERDEN 5 VARIANTEN auf IDENTISCHEN Entries (sequentiell, 1 Position,
|
||
Echtkosten, Exit-Kern aus `core/exit_model.py`):
|
||
1 nur Trailing (Baseline)
|
||
2 nur S/R-Close (der validierte Stand vor heute)
|
||
3 nur Flip-Close (das neue Feature allein)
|
||
4 BEIDE (was seit heute live ist)
|
||
5 BEIDE, Flip NACHRANGIG (Flip nur, wenn KEIN S/R-Ziel in Reichweite)
|
||
Variante 5 ist der Vorschlag (b): den validierten Exit vor dem unvalidierten schützen.
|
||
|
||
Alles über den `stop_when`-Hook des gemeinsamen Exit-Kerns — die Phasen-Mechanik
|
||
wird NICHT kopiert (diese Kopiererei hat im Projekt vier divergierende Exit-Modelle
|
||
erzeugt).
|
||
"""
|
||
import sys
|
||
from bisect import bisect_left
|
||
|
||
import MetaTrader5 as mt5
|
||
|
||
from core.analysis import calc_trend_angle
|
||
from core.exit_model import LIVE, simulate
|
||
from core.engine import _p_break
|
||
from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD,
|
||
_REVERSAL_STRETCH, _STRETCH_MAX)
|
||
from backtest_pbreak_retrain import (_ema_series, _atr_series, build_levels,
|
||
_cluster, pick_level, _ZONE, _M30)
|
||
|
||
_ATRMIN = 0.12
|
||
_PBRK = 0.35 # live `sr_close_pbreak`
|
||
_FLIP_MIN_R = 0.5 # live `auto_flip_close_min_r`
|
||
_NEAR_ATR = 0.6 # „S/R-Ziel in Reichweite" (wie der Live-Hinweis)
|
||
|
||
|
||
def signal_series(H, L, C, EF, ES, AT):
|
||
TH = _REVERSAL_STRETCH
|
||
sig = [0] * len(C)
|
||
for i in range(_N_BARS, len(C)):
|
||
atr = AT[i]
|
||
if not atr or atr <= 0:
|
||
continue
|
||
atr = max(atr, _ATRMIN); es = ES[i]; ef = EF[i]
|
||
stretch = (C[i] - es) / atr
|
||
ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR); ad = ang - 90.0
|
||
if stretch <= -TH and ad >= _ANGLE_DEAD: sig[i] = 1
|
||
elif stretch >= TH and ad <= -_ANGLE_DEAD: sig[i] = -1
|
||
elif abs(stretch) < _STRETCH_MAX: sig[i] = 1 if ef > es else -1 if ef < es else 0
|
||
return sig
|
||
|
||
|
||
def entries(sig, AT, lo, hi, cool=12):
|
||
"""Einstiegs-Liste EINMAL berechnen — alle Varianten teilen sie sich.
|
||
|
||
⚠ WICHTIG (Korrektur 2026-07-31): ein sequentieller Lauf, der nach dem Exit bei
|
||
`xb+1` weitermacht, erzeugt für JEDE Variante eine ANDERE Trade-Folge — sobald
|
||
eine früher aussteigt, verschiebt sich alles Nachfolgende. Die Varianten wären
|
||
dann nicht vergleichbar (Konfundierung). Deshalb feste Entries mit Cooldown,
|
||
wie in `backtest_pbreak_rvalue.py`.
|
||
"""
|
||
out = []
|
||
i = lo
|
||
while i < hi:
|
||
if sig[i] != 0:
|
||
out.append((i, sig[i], max(AT[i] or 0, _ATRMIN)))
|
||
i += cool
|
||
else:
|
||
i += 1
|
||
return out
|
||
|
||
|
||
def run(ent, sig, H, L, C, AT, EF, ES, SP, LV, sr=False, flip=False,
|
||
flip_subordinate=False, stats=None):
|
||
"""`stats` sammelt, WER geschlossen hat (für die Kollisions-Analyse)."""
|
||
Rs = []
|
||
for (i, d, atr) in ent:
|
||
entry = C[i]
|
||
cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||
state = {"lvl": None, "who": None}
|
||
|
||
def hook(j, prof_atr, _d=d, _e=entry, _st=state):
|
||
aj = max(AT[j] or 0, _ATRMIN)
|
||
lv = LV[min(j // _M30, len(LV) - 1)]
|
||
level = pick_level(_cluster(lv, aj), C[j], _d, aj, _st["lvl"])
|
||
_st["lvl"] = level
|
||
near = level is not None and abs(C[j] - level) <= _NEAR_ATR * aj
|
||
# ── S/R-Close: AM Level (±0,15×ATR) und P(break) unter der Schwelle
|
||
if sr and level is not None and abs(C[j] - level) <= _ZONE * aj:
|
||
m6 = (C[j] - C[max(0, j - 6)]) * _d / aj
|
||
m3 = (C[j] - C[max(0, j - 3)]) * _d / aj
|
||
wt = 1.0 if (EF[j] - ES[j]) * _d > 0 else 0.0
|
||
dist = abs(level - _e) / aj
|
||
if _p_break(m6, m3, wt, dist) < _PBRK and prof_atr > 0:
|
||
_st["who"] = "sr"; return True
|
||
# ── Flip-Close: Signal dreht, mind. min_r im Plus
|
||
if flip and sig[j] == -_d and prof_atr > _FLIP_MIN_R:
|
||
if flip_subordinate and near:
|
||
return False # S/R-Ziel in Reichweite → Vorrang für den S/R
|
||
_st["who"] = "flip"; return True
|
||
return False
|
||
|
||
R = simulate(entry, d, atr, H, L, C, i + 1, LIVE,
|
||
stop_when=(hook if (sr or flip) else None))
|
||
Rs.append(R - cost)
|
||
if stats is not None:
|
||
stats[state["who"] or "exit"] = stats.get(state["who"] or "exit", 0) + 1
|
||
return Rs
|
||
|
||
|
||
def st(Rs):
|
||
if not Rs:
|
||
return " —"
|
||
n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
|
||
g = sum(x for x in Rs if x > 0); l = -sum(x for x in Rs if x < 0)
|
||
return (f"n={n:>4} WR={100*w/n:>3.0f}% ØR={s/n:+.3f} PF={(g/l if l > 0 else 9.99):>4.2f} "
|
||
f"ΣR={s:>+6.0f}")
|
||
|
||
|
||
def main():
|
||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||
mt5.initialize(); sym = None
|
||
for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
||
if mt5.symbol_info(c):
|
||
sym = c; break
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
|
||
si = mt5.symbol_info(sym); point = si.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]
|
||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW); AT = _atr_series(H, L, C)
|
||
LV = build_levels(H, L)
|
||
sig = signal_series(H, L, C, EF, ES, AT)
|
||
mid = len(C) // 2
|
||
halves = [("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C) - LIVE.max_hold - 1)]
|
||
|
||
print("=" * 98)
|
||
print(f" S/R-CLOSE vs FLIP-CLOSE im Wettbewerb — {sym} M5 ({len(C)} Bars), Echtkosten")
|
||
print(f" Exit-Kern core/exit_model.py · Trail {LIVE.mult} · P(break)<{_PBRK} · "
|
||
f"Flip ab {_FLIP_MIN_R}×ATR")
|
||
print("=" * 98)
|
||
ENT = {lbl: entries(sig, AT, a, b) for lbl, a, b in halves}
|
||
for lbl, _, _ in halves:
|
||
print(f" {lbl}: {len(ENT[lbl])} Entries (fuer ALLE Varianten identisch)")
|
||
|
||
variants = [("1 nur Trailing (Baseline)", dict()),
|
||
("2 nur S/R-Close", dict(sr=True)),
|
||
("3 nur Flip-Close", dict(flip=True)),
|
||
("4 BEIDE (seit heute live)", dict(sr=True, flip=True)),
|
||
("5 BEIDE, Flip NACHRANGIG", dict(sr=True, flip=True,
|
||
flip_subordinate=True))]
|
||
base = {}
|
||
for name, kw in variants:
|
||
print(f"\n{name}")
|
||
for lbl, a, b in halves:
|
||
stats = {}
|
||
Rs = run(ENT[lbl], sig, H, L, C, AT, EF, ES, SP, LV, stats=stats, **kw)
|
||
s = sum(Rs)
|
||
if not kw:
|
||
base[lbl] = s
|
||
who = " · ".join(f"{k}={v}" for k, v in sorted(stats.items()))
|
||
print(f" {lbl} {st(Rs)} Δ Basis {s - base.get(lbl, 0):+.0f} [{who}]")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|