AH-Oil-Trader/backtest_trail_be.py

#!/usr/bin/env python3
"""Trailing × Breakeven — sind die beiden noch GEMEINSAM kalibriert? (2026-08-05)

ANLASS: Frage des Users nach der Trailing-Kalibrierung, ausgelöst vom Trade
T=49408993 (05.08.): er lief +0,50 $ ins Plus (0,85×ATR), wurde vom Trail
ausgestoppt und endete bei −15,33 €.

⚠ ZUR PRÄMISSE „mein Einsatz ist jetzt 3 Lot": **die Positionsgröße ändert das
Trailing-Optimum NICHT.** R ist ATR-normiert; Lots skalieren nur das EUR-Ergebnis,
nicht den optimalen Multiplikator. Gemessen wird deshalb etwas anderes — und
Wichtigeres:

**TRAIL UND BREAKEVEN WURDEN NIE GEMEINSAM OPTIMIERT.**
  * `be = 1.3` stammt aus `backtest_exit.py` — gemessen, als der Trail noch **1,5** war.
  * `mult` wurde am 31.07. auf **1,0** gesenkt (`backtest_trailmult.py`), **ohne**
    das Breakeven neu zu prüfen.
Die beiden hängen aber zusammen: solange der Gewinn unter `be` liegt, steht der
Stop noch auf der Verlustseite; ist der Trail eng, wird dieses Band schneller
durchlaufen. Genau in diesem Band ist der Trade von heute gestorben — Gewinn
0,85×ATR, Breakeven-Schwelle 1,3×ATR, also **0,45×ATR zu wenig**, und der Trail
(1,0×ATR) lag zwangsläufig noch über dem Einstieg.

MESSUNG: 2D-Sweep mult × be, kanonischer Exit (`exit_model.simulate`), Echtkosten,
80k M5, 2 Halbjahre, auf ZWEI Populationen (Squeeze-Ausbrüche und gegatete
Wellensignale) über eine JE POPULATION FESTE Entry-Liste — nur die Exit-Parameter
variieren, sonst wäre nicht vergleichbar, was verglichen wird.

VORAB FIXIERTE ENTSCHEIDUNGSREGEL:
  Umstellen nur, wenn ein Paar (mult, be) den Ist-Stand (1,0 / 1,3)
    (a) in ΣR **und** ØR in BEIDEN Halbjahren schlägt,
    (b) auf BEIDEN Populationen,
    (c) und die direkten Nachbarwerte mithalten (kein Parameter-Zufall),
    (d) ohne den Worst-Case zu verschlechtern.
  Sonst bleibt es, wie es ist.

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

import MetaTrader5 as mt5

from backtest_cost_gate import _atr_series, kandidaten
from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _EMA_FAST, _EMA_SLOW, _N_BARS,
                           _ANGLE_LR, _ANGLE_DEAD, _HTF_DEADBAND)

_ATRMIN = 0.06
_TIMEOUT_BARS = 12
_PIV_K, _PIV_LOOK = 3, 300
_ATR_FLOOR_PB = 0.12
_K = 0.3
# ⚠ Bereich bewusst NACH UNTEN erweitert: der Vorlauf zeigte das Optimum bei 0,6,
# also am RAND — dann misst man moeglicherweise nur die Grenze des Suchraums.
# Genau diese Warnung steht seit 31.07. in CLAUDE.md (backtest_trailmult.py).
_MULT = (0.3, 0.4, 0.5, 0.6, 0.8, 1.0, 1.5)
# Bei engem Trail ist das Breakeven wirkungslos (der Trail erreicht den Einstand
# ohnehin frueher) — deshalb ein grobes Raster, das reicht fuer die Interaktion.
_BE = (0.6, 1.0, 1.3, 2.0)


def _ema_series(v, p):
    k = 2 / (p + 1); out = [v[0]]
    for x in v[1:]:
        out.append(out[-1] + k * (x - out[-1]))
    return out


def _htf_sign(C, i, n, atr):
    if i < n * _EMA_SLOW:
        return 0
    seg = C[i - n * _EMA_SLOW:i + 1:n]
    if len(seg) < _EMA_SLOW + 2:
        return 0
    ef = _ema_series(seg, _EMA_FAST)[-1]; es = _ema_series(seg, _EMA_SLOW)[-1]
    dead = _HTF_DEADBAND * atr * (n ** 0.5)
    return 1 if ef - es > dead else (-1 if es - ef > dead else 0)


def wellen_entries(w, H, L, C, A, SP, EF, ES, AD, HTF, H1, PH, PL):
    """Gegatete Wellensignale — volle Live-Kette, EINE feste Liste."""
    out = []
    pend = None
    for i in range(_PIV_LOOK + _EMA_SLOW + 5, len(C) - 2):
        atr = A[i]
        if not atr or atr < _ATRMIN:
            continue
        w._pb_feats = {"atr": max(atr, _ATR_FLOOR_PB)}
        w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIV_K],
                        "pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIV_K]}
        rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
                          htf_trend=HTF[i], h1_trend=H1[i], angle=AD[i])
        sig = rec.get("signal")
        frisch = False
        if sig in ("LONG", "SHORT"):
            d = 1 if sig == "LONG" else -1
            if pend is None or pend["dir"] != d:
                pend = {"dir": d, "level": C[i] + d * _K * atr,
                        "invalid": C[i] - d * _K * atr, "confirmed": False, "bar": i}
            if not pend["confirmed"]:
                broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
                against = (C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])
                if broke:
                    pend["confirmed"] = True; frisch = True
                else:
                    if against or (i - pend["bar"]) > _TIMEOUT_BARS:
                        pend = {"dir": d, "level": C[i] + d * _K * atr,
                                "invalid": C[i] - d * _K * atr,
                                "confirmed": False, "bar": i}
                    continue
        else:
            pend = None
            continue
        if not frisch:
            continue
        rec = w._room_gate(rec, C[i])
        if rec.get("signal") not in ("LONG", "SHORT"):
            continue
        out.append({"i": i, "d": 1 if rec["signal"] == "LONG" else -1,
                    "lvl": pend["level"], "atr": max(atr, _ATRMIN),
                    "kosten": (SP[i] if SP[i] > 0 else 0.0225) / max(atr, _ATRMIN)})
        pend = None
    return out


def sweep(K, H, L, C, lo, hi, mult, be):
    """Sequentiell, EIN Slot — nur die Exit-Parameter variieren."""
    p = LIVE.with_(mult=mult, be=be)
    rs = []
    frei = lo
    for k in K:
        if not (lo <= k["i"] < hi) or k["i"] < frei:
            continue
        r, xb = simulate(k["lvl"], k["d"], k["atr"], H, L, C, k["i"] + 1, p,
                         timestop=True, use_tp=True, ret_bar=True)
        rs.append(r - k["kosten"])
        frei = xb + 1
    return rs


def kenn(rs):
    if not rs:
        return None
    n = len(rs); s = sum(rs)
    g = sum(x for x in rs if x > 0); v = -sum(x for x in rs if x < 0)
    return {"n": n, "oer": s / n, "sr": s,
            "pf": (g / v) if v > 0 else float("inf"), "worst": min(rs)}


def main():
    n_bars = 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_bars)
    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]; T = [int(b["time"]) 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)
    AD = [0.0] * len(C)
    for i in range(_N_BARS, len(C)):
        AD[i] = calc_trend_angle(C[i - _N_BARS:i + 1], _ANGLE_LR) - 90.0
        if abs(AD[i]) < _ANGLE_DEAD:
            AD[i] = 0.0
    HTF = [_htf_sign(C, i, 6, A[i] or 0.1) for i in range(len(C))]
    H1S = [_htf_sign(C, i, 12, A[i] or 0.1) for i in range(len(C))]
    PH = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if H[j] == max(H[j - _PIV_K:j + _PIV_K + 1])]
    PL = [j for j in range(_PIV_K, len(C) - _PIV_K)
          if L[j] == min(L[j - _PIV_K:j + _PIV_K + 1])]

    w = WaveRecommender(lambda: None)
    w.set_reversal_enabled(False)
    w.set_entry_room(0.6)          # ⚠ ohne diese Zeile ist das Gate ein No-op
    w.set_dead_hours("")
    mid = len(C) // 2

    print("=" * 96)
    print(f"  TRAILING x BREAKEVEN — gemeinsam kalibriert? · {sym} M5, {len(C)} Bars")
    print(f"  Ist-Stand: mult {LIVE.mult} · be {LIVE.be}   ·   2 Halbjahre · Echtkosten")
    print("  Hinweis: die Positionsgroesse aendert das Optimum NICHT (R ist ATR-normiert).")
    print("=" * 96)

    pops = [("SQUEEZE", kandidaten(H, L, C, T, SP, A)),
            ("WELLE (gegated)",
             wellen_entries(w, H, L, C, A, SP, EF, ES, AD, HTF, H1S, PH, PL))]

    ist = {}
    for name, K in pops:
        print(f"\n  {name} — {len(K)} Entries")
        print(f"    {'be \\ mult':<11}" + "".join(f"{m:>15.1f}" for m in _MULT))
        for be in _BE:
            zeile = f"    {be:<11.1f}"
            for m in _MULT:
                k1 = kenn(sweep(K, H, L, C, 0, mid, m, be))
                k2 = kenn(sweep(K, H, L, C, mid, len(C), m, be))
                if m == LIVE.mult and be == LIVE.be:
                    ist[name] = (k1, k2)
                zeile += f"{(k1['oer'] if k1 else 0):>+7.3f}/{(k2['oer'] if k2 else 0):>+7.3f}"
            print(zeile)
        print(f"    (Zellen = OeR H1 / OeR H2; Ist-Stand {LIVE.mult}/{LIVE.be} ist die Referenz)")

    # ── Urteil nach der VORAB fixierten Regel ─────────────────────────────
    print("\n" + "=" * 96)
    print("  URTEIL (Regel VOR dem Lauf fixiert: schlaegt Ist-Stand in OeR UND SigmaR,")
    print("          in BEIDEN Haelften, auf BEIDEN Populationen, Nachbarn halten mit,")
    print("          Worst-Case nicht schlechter)")
    print("=" * 96)
    treffer = []
    for m in _MULT:
        for be in _BE:
            if m == LIVE.mult and be == LIVE.be:
                continue
            ok = True
            for name, K in pops:
                i1, i2 = ist[name]
                k1 = kenn(sweep(K, H, L, C, 0, mid, m, be))
                k2 = kenn(sweep(K, H, L, C, mid, len(C), m, be))
                if not (k1 and k2 and i1 and i2):
                    ok = False; break
                if not (k1["oer"] > i1["oer"] and k2["oer"] > i2["oer"]
                        and k1["sr"] > i1["sr"] and k2["sr"] > i2["sr"]
                        and k1["worst"] >= i1["worst"] - 1e-9
                        and k2["worst"] >= i2["worst"] - 1e-9):
                    ok = False; break
            if ok:
                treffer.append((m, be))
    if treffer:
        print(f"    Kandidaten (vor Nachbar-Pruefung): {treffer}")
        print("    ⚠ Nachbarn von Hand gegenpruefen — eine einzelne bessere Zelle")
        print("      zwischen schlechteren ist Rauschen.")
    else:
        print("    => KEIN Paar schlaegt den Ist-Stand auf beiden Populationen.")
        print("       Trailing 1,0 / Breakeven 1,3 bleibt.")
    print()


if __name__ == "__main__":
    main()