#!/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()