AH-Oil-Trader/backtest_trail_start.py

#!/usr/bin/env python3
"""Wann soll das Trailing STARTEN? (`trail_start`, live 0,3×ATR)

ANLASS (2026-08-14). Der User fragte, warum seine Trades automatisch geschlossen
wurden. Befund aus dem Log: **9 von 11 `sl`-Closes seit dem 13.08. waren der
NACHGEZOGENE Stop**, nicht der Initial-SL — zusammen −75,90 EUR.

⚠⚠ DER MECHANISMUS IST ARITHMETIK, KEIN DEFEKT:
  Der Trailing-Stop sitzt `mult`×ATR (=1,0) hinter dem Bestkurs. Er erreicht den
  EINSTIEG damit erst bei **1,0×ATR Gewinn**. Aktiviert wird er aber schon bei
  **0,3×ATR**. Im Band dazwischen liegt der Stop zwangslaeufig auf der
  VERLUSTSEITE — ein normaler Ruecklauf schliesst dann mit kleinem Minus.
  Real (Ticket 50023965): Einstieg 82,283 · Bestkurs 82,141 (0,47×ATR im Plus)
  · Stop 82,141+0,305 = 82,456 · das sind 0,173 UEBER dem Einstieg.

GEMESSEN WIRD: was passiert, wenn das Trailing SPAETER startet? Bis dahin haelt
der Initial-SL (2,0×ATR) — die Trades bekommen mehr Luft, riskieren aber mehr.

VORAB FIXIERTE REGEL: eine Aenderung nur, wenn ein Wert die Basis (0,3) in
BEIDEN Haelften in ØR schlaegt UND die Nachbarn nicht kippen.

⚠⚠ RICHTUNG DER VERZERRUNG — hier ausnahmsweise GUENSTIG fuer die Aussage:
  `analyze_trail_resolution.py` (05.08.) hat gemessen, dass eine Bar-Simulation
  **enge Stops ueberschaetzt**. Ein spaeter Trail-Start bedeutet frueh einen
  WEITEREN Stop — die Verzerrung begueenstigt also die BASIS (0,3). Gewinnt ein
  spaeterer Start trotzdem, ist das belastbar.

⚠ POPULATION: Squeeze-Entries (der autonome Pfad). Die Trades des Users sind
  manuell (`setup=WAVE`) und als Einstiegs-AUSWAHL nicht backtestbar. Der Exit
  weiss zwar nicht, wie eingestiegen wurde — aber welche Phasen ein Trade
  ueberhaupt erreicht, haengt von der Einstiegsqualitaet ab. Die Zahlen gelten
  also streng nur fuer diese Population.

Aufruf:  python backtest_trail_start.py [bars]
"""
from __future__ import annotations

import random
import sys
from dataclasses import replace

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.squeeze_scan import atr_series, scan

_SQ_N, _SQ_K, _SQ_MULT = 10, 0.1, 2.5
_ATRMIN = 0.06        # ⚠ LIVE-Wert (Modul-Default 0,12 waere eine andere Population)
_WARTE = 8


def entries(H, L, C, A, SP, point, lo, hi):
    got = []
    scan(H, L, C, A, SP, point, lo, hi, _SQ_MULT, box_n=_SQ_N, k=_SQ_K,
         atr_min=_ATRMIN, exit_fn=lambda e, d, a, h, l, c, j: 0.0,
         on_entry=lambda j, d, lvl, a, komp: (
             got.append({"j": j, "d": d, "lvl": lvl, "atr": a}) if komp else None))
    return got


def lauf(ein, H, L, C, SP, start):
    """Sequentiell EIN Slot. Variiert wird NUR `trail_start`."""
    p = replace(LIVE, trail_start=start)
    Rs, war_plus_schloss_minus, mfes = [], 0, []
    frei_ab, grenze = -1, len(C) - LIVE.max_hold - 2
    for e in ein:
        i, d, lvl, a = e["j"], e["d"], e["lvl"], e["atr"]
        if i < frei_ab or i >= grenze or not a:
            continue
        fill = None
        for k in range(i + 1, min(i + 1 + _WARTE, grenze)):
            if L[k] <= lvl <= H[k]:
                fill = k; break
        if fill is None:
            continue
        R, xb = simulate(lvl, d, a, H, L, C, fill + 1, p, ret_bar=True)
        netto = R - SP[i] / a
        Rs.append(netto)
        # ⚠ Die Kennzahl, um die es dem User geht: war der Trade im Plus und
        #   schloss trotzdem im Minus? MFE = beste Auslenkung bis zum Exit.
        mfe = 0.0
        for j in range(fill + 1, xb + 1):
            g = ((H[j] - lvl) if d > 0 else (lvl - L[j])) / a
            mfe = max(mfe, g)
        mfes.append(mfe)
        if mfe >= start and netto < 0:
            war_plus_schloss_minus += 1
        frei_ab = xb + 1
    return Rs, war_plus_schloss_minus, mfes


def ci(Rs, n=2500, seed=42):
    if len(Rs) < 20:
        return None, None
    r = random.Random(seed); m = len(Rs)
    b = sorted(sum(Rs[r.randrange(m)] for _ in range(m)) / m for _ in range(n))
    return b[int(.025 * n)], b[int(.975 * n)]


def kenn(Rs, wpm, mfes):
    if not Rs:
        return None
    g = sum(x for x in Rs if x > 0); v = -sum(x for x in Rs if x < 0)
    u, o = ci(Rs)
    return {"n": len(Rs), "oer": sum(Rs) / len(Rs), "sr": sum(Rs),
            "wr": 100 * sum(1 for x in Rs if x > 0) / len(Rs),
            "pf": (g / v if v else float("inf")), "worst": min(Rs),
            "u": u, "o": o, "wpm": wpm,
            "wpmq": 100 * wpm / len(Rs)}


def zeig(name, k, basis=None):
    if not k:
        print(f"  {name:<26} —"); return
    kib = "      —      " if k["u"] is None else f"[{k['u']:+.3f} … {k['o']:+.3f}]"
    dl = "" if basis is None else f" Δ{k['oer']-basis['oer']:+.3f}"
    print(f"  {name:<26} n={k['n']:>4} WR {k['wr']:>4.1f}% ØR {k['oer']:>+.3f} "
          f"PF {k['pf']:>5.2f} worst {k['worst']:>+6.2f} "
          f"„war im Plus, Minus\" {k['wpm']:>4} ({k['wpmq']:>4.1f}%) {kib}{dl}")


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()
    if bars is None:
        print("  keine Bars"); return 1
    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)
    mid = len(C) // 2
    e1 = entries(H, L, C, A, SP, point, 60, mid)
    e2 = entries(H, L, C, A, SP, point, mid, len(C))
    print(f"\n  {sym} · {len(C)} M5-Bars · Squeeze-Entries H1 {len(e1)} · H2 {len(e2)}")
    print(f"  Variiert wird NUR `trail_start`. Trail-Abstand bleibt 1,0×ATR, "
          f"Initial-SL 2,0×ATR, Breakeven 1,3.")
    print(f"  ⚠ Der Trailing-Stop erreicht den Einstieg erst bei 1,0×ATR Gewinn — "
          f"darunter liegt er im Minus.\n")

    b1 = b2 = None
    ergebnisse = []
    for st in (0.3, 0.5, 0.8, 1.0, 1.3, 1.5, 2.0):
        k1 = kenn(*lauf(e1, H, L, C, SP, st))
        k2 = kenn(*lauf(e2, H, L, C, SP, st))
        if st == 0.3:
            b1, b2 = k1, k2
        mark = "  ← LIVE" if st == 0.3 else ""
        zeig(f"trail_start {st:.1f}  H1{mark}", k1, None if st == 0.3 else b1)
        zeig(f"trail_start {st:.1f}  H2{mark}", k2, None if st == 0.3 else b2)
        ergebnisse.append((st, k1, k2))
        print()

    # ── GEPAARTER Vergleich ──────────────────────────────────────────────
    # ⚠⚠ Die Einzel-Konfidenzintervalle oben taugen NICHT, um zwei Varianten zu
    #    trennen: sie messen die Streuung der Trades, nicht die des UNTERSCHIEDS.
    #    Da alle Varianten auf DENSELBEN Einstiegen laufen, ist der Unterschied
    #    je Trade paarweise bestimmbar — und dessen Intervall ist um ein
    #    Vielfaches enger. Dafuer muss die Slot-Sperre aus (sonst nehmen die
    #    Varianten verschiedene Teilmengen und die Paarung bricht).
    print("  ── GEPAART (alle Entries, ohne Slot-Sperre) ───────────────────")
    print("     Der statistisch richtige Test: Differenz JE TRADE gegen 0,3.\n")

    def alle(ein, start):
        p = replace(LIVE, trail_start=start)
        out = {}
        grenze = len(C) - LIVE.max_hold - 2
        for e in ein:
            i, d, lvl, a = e["j"], e["d"], e["lvl"], e["atr"]
            if i >= grenze or not a:
                continue
            fill = None
            for k in range(i + 1, min(i + 1 + _WARTE, grenze)):
                if L[k] <= lvl <= H[k]:
                    fill = k; break
            if fill is None:
                continue
            R = simulate(lvl, d, a, H, L, C, fill + 1, p)
            out[i] = R - SP[i] / a
        return out

    for lbl, ein in (("H1", e1), ("H2", e2)):
        basis = alle(ein, 0.3)
        print(f"    {lbl} (n={len(basis)} gepaarte Trades)")
        for st in (0.5, 0.8, 1.0, 1.3, 1.5, 2.0):
            v = alle(ein, st)
            dif = [v[k] - basis[k] for k in basis if k in v]
            if len(dif) < 20:
                continue
            m = sum(dif) / len(dif)
            r = random.Random(7); nn = len(dif)
            bs = sorted(sum(dif[r.randrange(nn)] for _ in range(nn)) / nn
                        for _ in range(3000))
            lo, hi = bs[75], bs[2925]
            sig = "✅ ohne 0" if (lo > 0 or hi < 0) else "⚠ enthaelt 0"
            print(f"      {st:.1f} gegen 0,3:  Δ je Trade {m:>+.4f}  "
                  f"KI [{lo:+.4f} … {hi:+.4f}]  {sig}")
        print()

    print("  ── URTEIL ─────────────────────────────────────────────────────")
    besser = [(st, k1, k2) for st, k1, k2 in ergebnisse
              if st != 0.3 and k1 and k2 and k1["oer"] > b1["oer"] and k2["oer"] > b2["oer"]]
    if besser:
        werte = ", ".join(f"{st:.1f}" for st, _, _ in besser)
        print(f"    SEQUENTIELL sehen {len(besser)} Werte besser aus ({werte}) —")
        print("    ⚠⚠ ABER DAS IST EIN SELEKTIONSEFFEKT, KEIN EXIT-EFFEKT.")
        print("       Ein spaeterer Trail-Start haelt Trades laenger, damit passen")
        print("       ANDERE Trades in den einen Slot. Der gepaarte Test oben —")
        print("       gleiche Entries, nur der Exit variiert — dreht in H2 sogar das")
        print("       Vorzeichen. Genau die Falle aus `backtest_exit_combo.py`")
        print("       (31.07.): verschiedene Trade-Folgen sind nicht vergleichbar.")
    print("    ➜ MASSGEBLICH ist der GEPAARTE Vergleich. Enthaelt dort JEDES")
    print("      Intervall die Null, ist kein Wert von der Basis 0,3 zu trennen")
    print("      → keine Aenderung.")
    print("    ⚠ Davon UNBERUEHRT ist die Spalte „war im Plus, Minus\": das ist eine")
    print("      Auszaehlung, keine Ertragsschaetzung — sie faellt monoton von")
    print("      ~48 % (0,3) auf 0 % (ab 1,3) und erklaert das beobachtete Verhalten.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())