AH-Oil-Trader/backtest_structure_tf.py

#!/usr/bin/env python3
"""Marktstruktur-Kanal als SIGNAL — über MEHRERE Zeitebenen (2026-08-03).

Anlass: User-Beobachtung im M5-Chart „der Kurs folgt dem Marktstrukturkanal und
bricht die untere Linie nicht — können wir das nutzen?". Die vorhandene Messung
(`backtest_structure.py`, 2026-07-20) lief ausschliesslich auf **M30** und fiel
in beiden Hälften durch. M5 ist damit streng genommen eine UNGETESTETE Variante.

Unterschiede zu `backtest_structure.py` — bewusst, und beide relevant:

 1. **Kanonischer Exit** (`core/exit_model.simulate`, LIVE) statt der dortigen
    Exit-KOPIE. Jene nutzt Trail **1,5** und kennt weder Time-Stop noch TP; live
    läuft seit 2026-07-31 Trail **1,0** MIT Time-Stop. Für eine NEUE Messung ist
    der kanonische Kern richtig (die alte Datei bleibt unverändert — sie umzu-
    stellen wäre laut Migrations-Regel nur erlaubt, wenn die Zahlen gleich blieben).
 2. **Zeitebene als Parameter**, plus zwei Lesarten des „M5-Kanals":
      · `bars`  — gleiche BAR-Anzahl wie auf M30 (=schnellerer, kürzerer Kanal;
                  das, was man im M5-Chart tatsächlich sieht)
      · `time`  — gleiche ZEIT-Spanne (Fenster ×6 auf M5) = derselbe Kanal, nur
                  feiner aufgelöst
    Beide zu testen ist Pflicht: kippt das Vorzeichen zwischen ihnen, ist es
    Rauschen (die ORB-/SMC-Lehre — Nachbar-Parameter müssen mithalten).

Maßstab wie immer: beidhälftig positiv UND parameter-robust. Vergleich:
Squeeze ØR +0,14…+0,23 bei PF>1.

Aufruf:  python backtest_structure_tf.py [bars]
"""
import sys

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.structure import _pivots, _classify, _channel

_ATRMIN = 0.06
_COOL = 3


def _atr_series(H, L, C, p=14):
    t = [0.0]
    for i in range(1, len(C)):
        t.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    return [(sum(t[max(1, i - p + 1):i + 1]) / max(1, len(t[max(1, i - p + 1):i + 1])))
            if i else None for i in range(len(C))]


def st(Rs):
    if not Rs:
        return None
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
    return dict(n=n, wr=100 * w / n, oR=s / n, pf=(up / dn if dn > 0 else 9.99), sum=s)


def line(lbl, s):
    if not s:
        return f"    {lbl:<26} — (keine Trades)"
    return (f"    {lbl:<26} n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
            f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}")


def run(H, L, C, A, SP, lo, hi, rule, X, look, reg):
    """Sequentielle 1-Positions-Sim mit dem KANONISCHEN Live-Exit."""
    def cost(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    Rs = []
    i = max(lo, look)
    prev_pos = None
    while i < hi:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; prev_pos = None; continue
        ch = _channel(C[max(0, i - reg + 1):i + 1], atr)
        if not ch:
            i += 1; prev_pos = None; continue
        pos, cdir = ch["pos"], ch["dir"]
        d = 0
        if rule in ("A", "A2"):
            # frischer Eintritt ins untere (Aufwärtskanal) bzw. obere (Abwärts) Drittel
            if cdir == "up" and pos <= X and (prev_pos is None or prev_pos > X):
                d = 1
            elif cdir == "down" and pos >= 1 - X and (prev_pos is None or prev_pos < 1 - X):
                d = -1
            if d != 0 and rule == "A2":
                pl = _pivots(H[max(0, i - look):i + 1], L[max(0, i - look):i + 1], 3)
                lab = _classify(pl)
                recent = [x["type"] for x in lab[-4:]]
                ups = sum(1 for t in recent if t in ("HH", "HL"))
                dns = sum(1 for t in recent if t in ("LH", "LL"))
                strend = ("up" if ups >= 3 and ups > dns else
                          "down" if dns >= 3 and dns > ups else "range")
                if (d > 0 and strend != "up") or (d < 0 and strend != "down"):
                    d = 0
        prev_pos = pos
        if d == 0:
            i += 1; continue
        entry = C[i]
        a = max(atr, _ATRMIN)
        r, xj = simulate(entry, d, a, H, L, C, i + 1, LIVE,
                         timestop=True, use_tp=True, ret_bar=True)
        Rs.append(r - cost(i, a))
        i = xj + _COOL
        prev_pos = None
    return Rs


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)
    point = mt5.symbol_info(sym).point
    TFS = [("M5", mt5.TIMEFRAME_M5), ("M15", mt5.TIMEFRAME_M15),
           ("M30", mt5.TIMEFRAME_M30)]
    data = {}
    for lbl, tf in TFS:
        b = mt5.copy_rates_from_pos(sym, tf, 0, n)
        if b is not None and len(b) > 5000:
            data[lbl] = ([float(x["high"]) for x in b], [float(x["low"]) for x in b],
                         [float(x["close"]) for x in b],
                         [float(x["spread"]) * point for x in b])
    mt5.shutdown()

    print("=" * 92)
    print(f"  Kanal-Pullback über Zeitebenen — {sym}, kanonischer Live-Exit "
          f"(SL {LIVE.sl_atr}× · Trail {LIVE.mult}× · BE {LIVE.be}× · Time-Stop), Echtkosten")
    print("  Regel: im Aufwärtskanal am UNTEREN Drittel long / im Abwärtskanal am OBEREN short")
    print("  Maßstab: beidhälftig ØR>0 UND PF>1 UND Nachbar-Varianten halten mit")
    print("=" * 92)

    # (reg, look) je Lesart. M30-Referenz = 60/220 wie in backtest_structure.py.
    plans = [
        ("M5  Kanal 60 Bars  (=5 h,  'was man im M5-Chart sieht')", "M5", 60, 220),
        ("M5  Kanal 360 Bars (=30 h, zeitgleich zum M30-Kanal)",    "M5", 360, 1320),
        ("M15 Kanal 120 Bars (=30 h, Nachbar-Zeitebene)",           "M15", 120, 440),
        ("M30 Kanal 60 Bars  (=30 h, die ALTE Messung)",            "M30", 60, 220),
    ]
    for title, tfl, reg, look in plans:
        if tfl not in data:
            print(f"\n  {title}: keine Daten"); continue
        H, L, C, SP = data[tfl]
        A = _atr_series(H, L, C)
        N = len(C); mid = N // 2
        print(f"\n  {title}   [{N} Bars]")
        for rl, X, sub in (("A", 0.20, "Kanal-Pullback X=0.20"),
                           ("A", 0.30, "Kanal-Pullback X=0.30"),
                           ("A2", 0.25, "+Swing-Filter X=0.25")):
            s1 = st(run(H, L, C, A, SP, 0, mid, rl, X, look, reg))
            s2 = st(run(H, L, C, A, SP, mid, N - LIVE.max_hold - 1, rl, X, look, reg))
            print(f"    · {sub}")
            print(line("H1 (alt)", s1))
            print(line("H2 (neu)", s2))
            ok = (s1 and s2 and s1["oR"] > 0 and s2["oR"] > 0
                  and s1["pf"] > 1 and s2["pf"] > 1)
            print(f"      → {'ROBUST' if ok else 'fällt durch'}")
    print()


if __name__ == "__main__":
    main()