AH-Oil-Trader/backtest_rsi_v2.py

"""RSI ZUM VIERTEN MAL gemessen — diesmal auf der Ausfuehrungs-Achse (2026-08-19)

User: "messe ihn nochmal".

WARUM DAS KEIN BLOSSES WIEDERHOLEN IST. RSI ist im Projekt dreimal
durchgefallen (`backtest_confluence.py` als Konfluenz-Filter,
`backtest_buch_luis.py` als Divergenz, dazu der Pullback-Befund). ALLE drei
stiegen am **Bar-Close zum Marktpreis** ein. Genau diese Annahme hat am
05./06.08. Squeeze und Auto-Signal von negativ auf positiv gedreht - der
Einstiegspreis war dort +0,32 bzw. +0,61 R wert, mehr als alle Signal-Ideen
zusammen. Fuer RSI war diese Achse nie geprueft.

⚠⚠ ABER die Vorhersage ist ungünstig, und sie steht VOR dem Lauf hier: RSI ist
ein REVERSIONS-Einstieg (ueberkauft -> short). Und fuer Reversion ist die
ruhende Order gemessen SCHLECHTER, nicht besser - `backtest_candle_fade2.py`
(05.08.): "ruhende Order hilft bei Momentum-Entries, schadet bei
Reversion-Entries", weil eine Limit-Order am Extrem nur fuellt, wenn sich die
Bewegung weiter ausdehnt - also genau dann, wenn man falsch liegt. Wenn dieser
Lauf dasselbe zeigt, ist das eine Bestaetigung des Mechanismus, keine
Ueberraschung.

GEMESSEN WERDEN BEIDE LESARTEN, weil der Projekt-Befund sagt, dass die
Lehrbuch-Lesart hier invertiert ist ("RSI *gegen* die Richtung = doppelter
Edge"):
    REVERSION   RSI > hi -> SHORT,  RSI < lo -> LONG   (Lehrbuch)
    MOMENTUM    RSI > hi -> LONG,   RSI < lo -> SHORT  (die Umkehrung)
je mit drei Schwellenpaaren (70/30, 75/25, 80/20) = Nachbar-Robustheit, und je
mit zwei Ausfuehrungen (Markt am Bar-Close / ruhende Order am Extrem der
Ausloese-Bar, Fill bei BERUEHRUNG).

⚠ KEIN NACHBAU des RSI: die Serie wird inkrementell gerechnet UND an 200
Stichproben gegen die echte `core.analysis.calc_rsi` verifiziert. Weicht sie ab,
bricht das Skript ab - ein selbst gebauter Indikator waere genau die
Divergenz-Falle, die im Projekt dreimal zugeschlagen hat.

REGEL VORAB FIXIERT: tragfaehig nur, wenn eine Zelle in BEIDEN Halbjahren
OeR > 0 UND PF > 1 liefert, das 95-%-KI die Null ausschliesst UND die beiden
Nachbar-Schwellen dasselbe Vorzeichen haben.

Aufruf:  python backtest_rsi_v2.py [bars]
"""
import random
import sys

import MetaTrader5 as mt5

from backtest_cost_gate import kennzahlen
from core.analysis import calc_rsi
from core.exit_model import LIVE, simulate
from core.squeeze_scan import atr_series

PERIOD = 14
SCHWELLEN = [(70, 30), (75, 25), (80, 20)]
COOLDOWN = 12          # Bars, damit dieselbe Ueberdehnung nicht dutzendfach zaehlt


def rsi_serie(C, period=PERIOD):
    """Wilder-RSI je Bar. ⚠ Gegen `calc_rsi` verifiziert (s. `pruefe`)."""
    n = len(C)
    out = [50.0] * n
    if n < period + 1:
        return out
    g = l = 0.0
    for i in range(1, period + 1):
        d = C[i] - C[i - 1]
        g += max(d, 0.0); l += max(-d, 0.0)
    ag, al = g / period, l / period
    out[period] = 100.0 if al == 0 else 100.0 - 100.0 / (1.0 + ag / al)
    for i in range(period + 1, n):
        d = C[i] - C[i - 1]
        ag = (ag * (period - 1) + max(d, 0.0)) / period
        al = (al * (period - 1) + max(-d, 0.0)) / period
        out[i] = 100.0 if al == 0 else 100.0 - 100.0 / (1.0 + ag / al)
    return out


def pruefe(C, R, proben=200):
    """Die Serie MUSS der echten Funktion entsprechen — sonst Abbruch."""
    rnd = random.Random(42)
    schlimmster = 0.0
    for _ in range(proben):
        i = rnd.randrange(PERIOD + 5, len(C))
        schlimmster = max(schlimmster, abs(calc_rsi(C[:i + 1], PERIOD) - R[i]))
    return schlimmster


def handeln(i, d, H, L, C, SP, A, ende, modus):
    """modus 'markt' = Fill am Close von i · 'level' = ruhende Order am Extrem."""
    atr = A[i]
    if not atr or atr <= 0 or i + 1 >= ende:
        return None
    if modus == "markt":
        f, preis = i + 1, C[i]
    else:
        # Ruhende Order am Extrem der Ausloese-Bar, Fill bei BERUEHRUNG.
        preis = H[i] if d > 0 else L[i]
        f = None
        for x in range(i + 1, min(i + 13, ende)):
            if (H[x] >= preis) if d > 0 else (L[x] <= preis):
                f = x
                break
        if f is None or f + 1 >= ende:
            return None
        f += 1
    r = simulate(preis, d, atr, H, L, C, f, LIVE, timestop=True, use_tp=True)
    return r - ((SP[i] if SP[i] > 0 else 0.0225) / atr)


def lauf(H, L, C, SP, A, R, lo_i, hi_i, ende, hi, lo, lesart, modus):
    rs, letzte = [], -10**9
    for i in range(max(lo_i, PERIOD + 1), hi_i):
        if i - letzte < COOLDOWN:
            continue
        if R[i] >= hi:
            d = -1 if lesart == "reversion" else 1
        elif R[i] <= lo:
            d = 1 if lesart == "reversion" else -1
        else:
            continue
        letzte = i
        r = handeln(i, d, H, L, C, SP, A, ende, modus)
        if r is not None:
            rs.append(r)
    return rs


def zeig(lbl, k1, k2):
    def f(k):
        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"))
        return "n={:>4} WR={:>3.0f}% OeR={:+.3f} PF={:.2f}{}".format(
            k["n"], k["wr"], k["oer"], k["pf"], m)
    ok = (k1 and k2 and k1["oer"] > 0 and k2["oer"] > 0 and k1["pf"] > 1
          and k2["pf"] > 1 and (k1.get("ci") or (0, 0))[0] > 0
          and (k2.get("ci") or (0, 0))[0] > 0)
    print("    {:<26} H1 {}".format(lbl, f(k1)))
    print("    {:<26} H2 {}   {}".format("", f(k2), "✅" if ok else "❌"))
    return ok


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)
    R = rsi_serie(C)
    N = len(C); mid = N // 2; ende = N - LIVE.max_hold - 1

    fehler = pruefe(C, R)
    print("=" * 96)
    print("  RSI({}) auf {} M5 ({} Bars) · kanonischer Exit · Echtkosten · 95-%-KI"
          .format(PERIOD, sym, N))
    print("  Serie gegen die echte `calc_rsi` geprueft: groesste Abweichung {:.2e}"
          .format(fehler))
    assert fehler < 1e-6, "RSI-Serie weicht von calc_rsi ab — Abbruch"
    print("=" * 96)

    treffer = []
    for modus, titel in (("markt", "A) MARKT am Bar-Close (= die drei alten Laeufe)"),
                         ("level", "B) ruhende Order am Extrem, Fill bei BERUEHRUNG")):
        print("\n  " + titel)
        print("  " + "-" * 92)
        for lesart in ("reversion", "momentum"):
            for hi, lo in SCHWELLEN:
                k1 = kennzahlen(lauf(H, L, C, SP, A, R, 0, mid, ende, hi, lo, lesart, modus))
                k2 = kennzahlen(lauf(H, L, C, SP, A, R, mid, ende, ende, hi, lo, lesart, modus))
                if zeig("{} {}/{}".format(lesart.upper()[:3], hi, lo), k1, k2):
                    treffer.append((modus, lesart, hi, lo))
            print("")

    print("  " + "=" * 92)
    print("    Zellen, die die vorab fixierte Regel erfuellen: {}".format(
        len(treffer) or "KEINE"))
    for t in treffer:
        print("      {}".format(t))
    print("    ⚠ Nachbar-Robustheit zusaetzlich pruefen: eine einzelne Zelle unter")
    print("      12 getesteten ist bei 5 % Irrtum die ERWARTETE Zahl an Fehlalarmen.")
    print("  " + "=" * 92)


if __name__ == "__main__":
    main()