AH-Oil-Trader/backtest_rsi_filter.py

"""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()