"""RSI als FILTER auf dem gegateten Wellensignal (2026-08-19) — die neue Achse. User: "neue achse" → "messe". WAS DARAN NEU IST. Alle bisherigen RSI-Laeufe des Projekts massen RSI als EIGENSTAENDIGEN Einstieg (`backtest_rsi_v2.py` heute: 24 Zellen, keine besteht) oder als Konfluenz-Beitrag auf dem UNGEGATETEN `_build`-Signal mit einem alten Exit (`backtest_confluence.py`). Nie gemessen wurde er auf der Population, die live tatsaechlich handelt: den Signalen, die den VOLLEN Gate-Stack passieren (Totband · M30-Filter · Anti-Ueberdehnung · Winkel · min_conf 55 · Breakout-Bestaetigung · Entry-Raum). Das sind nur ~300-500 je Halbjahr statt ~800 - eine andere Grundgesamtheit, und damit eine faire neue Achse. DIE HYPOTHESE STAMMT AUS DEM PROJEKT SELBST und ist kontraintuitiv: dokumentiert ist "RSI *bestaetigend* ist kontraproduktiv, RSI *gegen* die Richtung = doppelter Edge" - der Pullback-Effekt. Gemessen wird deshalb BEIDES, gegen dieselbe Basis. BASIS alle gegateten Signale, kein Filter MIT nur wenn RSI die Richtung BESTAETIGT (LONG: RSI > x) GEGEN nur wenn RSI DAGEGEN steht (LONG: RSI < 100-x) <- Pullback x in 50 / 60 / 70 = Nachbar-Robustheit. Beide Ausfuehrungen (Markt am Bar-Close / ruhende Order am Bestaetigungs-Level mit Fill bei BERUEHRUNG). ⚠ KEIN NACHBAU: der Gate-Stack kommt aus `backtest_entryroom_level.sammle` (dieselbe Pipeline wie `backtest_conf_gated.py`), die RSI-Serie ist gegen die echte `core.analysis.calc_rsi` verifiziert (s. `backtest_rsi_v2.pruefe`). REGEL VORAB FIXIERT: ein Filter traegt nur, wenn die BEHALTENE Gruppe die Basis in BEIDEN Halbjahren im OeR schlaegt, ihr 95-%-KI die Null ausschliesst UND die beiden Nachbar-Schwellen dasselbe Vorzeichen zeigen. Zusaetzlich muss die ENTFERNTE Gruppe schlechter sein als die behaltene - sonst wirft der Filter nicht die schlechten Signale weg, sondern nur irgendwelche. ⚠ GRENZE: ein Filter verkleinert die Stichprobe zwangslaeufig. Ein besseres OeR bei halber Trade-Zahl ist noch kein Ertrag - deshalb steht SR (Summe) daneben. Aufruf: python backtest_rsi_filter.py [bars] """ import sys import MetaTrader5 as mt5 import core.wave_rec as WR from backtest_breakout_gated import _PIV_K, _TU, _atr_series, _htf_sign from backtest_cost_gate import kennzahlen from backtest_entryroom_level import _GATE, _K, handeln, sammle from backtest_rsi_v2 import pruefe, rsi_serie from core.analysis import calc_trend_angle from core.exit_model import LIVE from core.wave_rec import (WaveRecommender, _ANGLE_LR, _EMA_FAST, _EMA_SLOW, _ema_series) SCHWELLEN = [50, 60, 70] def zeig(lbl, k1, k2, b1=None, b2=None): def f(k, b): if not k: return " --" ci = k.get("ci"); m = "" if ci: m = (" [{:+.3f} .. {:+.3f}]".format(ci[0], ci[1]) + ("" if ci[0] * ci[1] > 0 else " ~0")) dlt = "" if b: dlt = " Δ{:+.3f}".format(k["oer"] - b["oer"]) return "n={:>4} OeR={:+.3f} PF={:.2f} SR={:+7.1f}{}{}".format( k["n"], k["oer"], k["pf"], k["sr"], m, dlt) print(" {:<20} H1 {}".format(lbl, f(k1, b1))) print(" {:<20} H2 {}".format("", f(k2, b2))) if not (k1 and k2 and b1 and b2): return False return (k1["oer"] > b1["oer"] and k2["oer"] > b2["oer"] and k1["oer"] > 0 and k2["oer"] > 0 and (k1.get("ci") or (0, 0))[0] > 0 and (k2.get("ci") or (0, 0))[0] > 0) 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) 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) EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW) R = rsi_serie(C) N = len(C); mid = N // 2; ende = N - LIVE.max_hold - 1 fehler = pruefe(C, R) assert fehler < 1e-6, "RSI-Serie weicht von calc_rsi ab (%.2e)" % fehler print("Vorberechnung ...", flush=True) AD = [90.0] * N for i in range(_ANGLE_LR + 2, N): AD[i] = calc_trend_angle(C[i - (_ANGLE_LR + 1):i + 1], _ANGLE_LR) HTF = [0] * N; H1 = [0] * N for i in range(N): if A[i]: HTF[i] = _htf_sign(C, i, 6, A[i]); H1[i] = _htf_sign(C, i, 12, A[i]) PH = [j for j in range(_PIV_K, N - _PIV_K) if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])] PL = [j for j in range(_PIV_K, N - _PIV_K) if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])] w = WaveRecommender(_TU(), mt5.TIMEFRAME_M5) w.set_reversal_enabled(False) w.set_entry_room(_GATE) # ⚠ ohne das ist das Gate ein No-op (Lehre 05.08.) w.set_dead_hours([]) w.set_breakout_k(_K) s1 = sammle(w, H, L, C, A, EF, ES, AD, HTF, H1, PH, PL, 0, mid) s2 = sammle(w, H, L, C, A, EF, ES, AD, HTF, H1, PH, PL, mid, ende) print("=" * 96) print(" RSI als FILTER auf dem GEGATETEN Signal — {} M5 ({} Bars)".format(sym, N)) print(" Gegatete Signale: H1 {} H2 {} (WR/PF/KI je Gruppe)".format(len(s1), len(s2))) print("=" * 96) treffer = [] for modus, titel in (("markt", "A) MARKT am Bar-Close"), ("level", "B) ruhende Order am Level, Fill bei BERUEHRUNG")): print("\n " + titel) print(" " + "-" * 92) def rs(sigs, keep): out = [] for s in sigs: if not keep(s): continue r = handeln(s, H, L, C, SP, ende, modus, 0.0, 12) if r is not None: out.append(r) return out b1 = kennzahlen(rs(s1, lambda s: True)) b2 = kennzahlen(rs(s2, lambda s: True)) zeig("BASIS (kein Filter)", b1, b2) print("") for x in SCHWELLEN: for lbl, keep in ( ("MIT RSI>{}".format(x), lambda s, x=x: (R[s["i"]] > x) if s["d"] > 0 else (R[s["i"]] < 100 - x)), ("GEGEN RSI<{}".format(100 - x), lambda s, x=x: (R[s["i"]] < 100 - x) if s["d"] > 0 else (R[s["i"]] > x)), ): k1 = kennzahlen(rs(s1, keep)); k2 = kennzahlen(rs(s2, keep)) if zeig(lbl, k1, k2, b1, b2): treffer.append((modus, lbl)) print("") print(" " + "=" * 92) print(" Zellen, die die vorab fixierte Regel erfuellen: {}".format( len(treffer) or "KEINE")) for t in treffer: print(" {}".format(t)) print(" ⚠ SR (Summe) mitlesen: ein besseres OeR bei halber Trade-Zahl ist") print(" Handelsvermeidung, kein Ertrag.") print(" " + "=" * 92) if __name__ == "__main__": main()