AH-Oil-Trader/backtest_trailmult.py

#!/usr/bin/env python3
"""Trailing-Multiplikator SAUBER gemessen — Initial-SL fest bei 2,0×ATR wie live.

⚠ WARUM DIESES SKRIPT EXISTIERT (2026-07-31, Fund beim Stufe-3-Refactor):
`backtest_trailing.py` setzt den Initial-SL auf **`mult`×ATR** statt auf die feste
Live-Größe 2,0×ATR:

    sl = entry - d*mult*atr        # <-- Initial-SL = mult

Der dortige Vergleich „mult 1,5 vs 2,5" misst also in Wahrheit
*(SL 1,5 · Trail 1,5)* gegen *(SL 2,5 · Trail 2,5)* — zwei Änderungen auf einmal.
Deshalb war der Worst-Case dort immer exakt gleich dem Multiplikator (−1,50 / −2,50):
das war der **SL**, nicht der Trail. Live ist der SL dagegen fest (Band 1,8–2,2 →
2,0, `backtest_exit.py`), nur der Trail hing an der Zeitebene.

Die auf dieser Grundlage getroffene Stufe-1-Entscheidung (`_MULT_BY_TF` überall auf
1,5) muss deshalb neu belegt werden — HIER, mit fixem SL und variablem Trail, also
genau der Änderung, um die es live geht.

Aufbau ansonsten identisch zu `backtest_trailing.py` (gleiches Signal-Set
Trend+Reversal, 2 Halbjahre), Exit aus `core/exit_model.py` — der gemeinsamen Quelle,
damit dieser Fehler nicht noch einmal passieren kann.
"""
import sys

import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD,
                           _REVERSAL_STRETCH, _STRETCH_MAX)

_ATRMIN = 0.12


def _ema_series(v, p):
    k = 2.0 / (p + 1); o = []; e = v[0]
    for i, x in enumerate(v):
        e = x if i == 0 else x * k + e * (1 - k)
        o.append(e)
    return o


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])))
    out = []
    for i in range(len(C)):
        if not i:
            out.append(None); continue
        w = t[max(1, i - p + 1):i + 1]
        out.append(sum(w) / max(1, len(w)))
    return out


def st(Rs):
    if not Rs:
        return "  -"
    n = len(Rs); w = sum(1 for r in Rs if r > 0)
    g = sum(r for r in Rs if r > 0); ls = -sum(r for r in Rs if r < 0)
    return (f"WR={100*w/n:>3.0f}% Ø-R={sum(Rs)/n:+.3f} PF={(g/ls if ls > 0 else 99):>4.2f} "
            f"Worst={min(Rs):+.2f} ΣR={sum(Rs):+.0f}")


def signals(H, L, C, ES, EF, AT, lo, hi, step):
    TH = _REVERSAL_STRETCH; out = []
    for i in range(max(lo, _N_BARS), min(hi, len(C) - LIVE.max_hold - 1), step):
        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
        d = 0
        if stretch <= -TH and ad >= _ANGLE_DEAD: d = 1
        elif stretch >= TH and ad <= -_ANGLE_DEAD: d = -1
        elif abs(stretch) < _STRETCH_MAX: d = 1 if ef > es else -1 if ef < es else 0
        if d:
            out.append((i, d, atr))
    return out


def squeeze_signals(H, L, C, AT, lo, hi):
    """Nur SQUEEZE-Ausbrüche — der EINZIGE Setup mit belegtem Edge, und er lebt von
    LÄUFERN (43 % Treffer, Ertrag im Tail). Gegentest zum Wave-Signal: ein enger
    Trail, der dort gewinnt, muss hier verlieren, wenn er nur „früher aussteigen"
    belohnt. Regel exakt wie live (`wave._squeeze_one`)."""
    from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K
    _W = 24; _COOL = 12
    out = []
    i = max(lo, _N_BARS, _SQ_N + 15)
    while i < min(hi, len(C) - LIVE.max_hold - 1):
        atr = AT[i]
        if not atr or atr < 0.06:
            i += 1; continue
        boxHi = max(H[i - _SQ_N:i]); boxLo = min(L[i - _SQ_N:i])
        if (boxHi - boxLo) > _SQ_MULT * atr:
            i += 1; continue
        hit = None
        for j in range(i, min(i + _W, len(C) - LIVE.max_hold - 1)):
            up = boxHi + _SQ_K * atr; dn = boxLo - _SQ_K * atr
            if H[j] >= up: hit = (j, 1, up); break
            if L[j] <= dn: hit = (j, -1, dn); break
        if hit is None:
            i += 1; continue
        j, d, lvl = hit
        out.append((j, d, atr, lvl))
        i = j + _COOL
    return out


def main():
    squeeze = "--squeeze" in sys.argv
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    n = int(args[0]) if args else 80000
    step = int(args[1]) if len(args) > 1 else 2
    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)
    mid = len(C) // 2
    halves = [("H1 (alt)", _N_BARS, mid), ("H2 (neu)", mid, len(C))]
    print("=" * 92)
    print(f"  TRAIL-MULTIPLIKATOR — {sym} M5 ({len(C)} Bars)")
    print(f"  Initial-SL FEST {LIVE.sl_atr}×ATR (wie live) · nur der Trail variiert")
    print(f"  Exit aus core/exit_model.py · Time-Stop {LIVE.timestop_bars} Bars · BE {LIVE.be}")
    if squeeze:
        print("  Signal-Set: NUR SQUEEZE-Ausbrueche (laeuferabhaengig) · ECHTKOSTEN")
    print("=" * 92)
    if squeeze:
        SIG = {lbl: [(i, d, atr) for (i, d, atr, _lv)
                     in squeeze_signals(H, L, C, AT, a, b)] for lbl, a, b in halves}
    else:
        SIG = {lbl: signals(H, L, C, ES, EF, AT, a, b, step) for lbl, a, b in halves}
    for lbl, _, _ in halves:
        print(f"  {lbl}: {len(SIG[lbl])} Signale")
    # ⚠ Auch die kleinen Werte prüfen: läuft es nach unten monoton weiter besser,
    # liegt das „Optimum" am Rand des getesteten Bereichs = Warnsignal (dann misst
    # man womöglich nur „macht weniger" statt „macht es besser").
    for mult in (0.5, 0.75, 1.0, 1.25, 1.5, 2.0, 2.5, 3.0):
        tag = "  ← live (nach Stufe 1)" if abs(mult - 1.5) < 1e-9 else ""
        print(f"\nTrail {mult:.1f}×ATR{tag}")
        for lbl, _, _ in halves:
            p = LIVE.with_(mult=mult)
            Rs = [simulate(C[i], d, atr, H, L, C, i + 1, p)
                  - ((SP[i] if SP[i] > 0 else 0.0225) / atr if squeeze else 0.0)
                  for (i, d, atr) in SIG[lbl]]
            print(f"   {lbl}  {st(Rs)}")


if __name__ == "__main__":
    main()