AH-Oil-Trader/analyze_scalping_costs.py

#!/usr/bin/env python3
"""Traegt ein M1-Scalping-Modul die Kosten? — B3-Check VOR jedem Bau.

WOZU: Die Frage "Scalping auf M1" ist zuerst eine KOSTEN-Frage, nicht eine
Signal-Frage. Der Spread ist bei WTI absolut ~konstant (0,0225 $, gemessen
`analyze_execution.py`) — er skaliert NICHT mit der Zeitebene. Die Bewegung, aus
der ein Trade seinen Ertrag zieht, skaliert sehr wohl. Je feiner die Zeitebene,
desto groesser also der Anteil, den der Spread vom verfuegbaren Weg frisst.

Die Projekt-Huerde (B3): Ø-Edge **netto > ~0,1×ATR** der gehandelten Zeitebene.
Auf M5 liegen die realen Kosten bei Ø 0,265×ATR_M5 — schon dort ist es eng.

Gemessen wird deshalb:
  A) Spread/ATR je Zeitebene (M1 gegen M5/M15) — Median, Ø, je Berliner Stunde
  B) Wieviel Weg gibt eine M1-Bewegung ueberhaupt her? (MFE/MAE in N Bars)
  C) Welche Trefferquote braeuchte ein Scalp, um die Kosten zu bezahlen?

⚠ Das ist KEIN Backtest eines Scalping-Signals — es ist der Test, ob so ein
Signal ueberhaupt existieren KANN. Faellt A/C durch, eruebrigt sich die
Signalsuche (dieselbe Logik wie beim Entry-Raum-Gate: reine Arithmetik,
deshalb regime-unabhaengig robust).

Aufruf:  python analyze_scalping_costs.py [n_bars]
"""
import sys
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import MetaTrader5 as mt5

_BERLIN = ZoneInfo("Europe/Berlin")
_BROKER = timezone(timedelta(hours=3))
_UTC = timezone.utc
_HURDE = 0.10          # B3: Ø-Edge netto muss > 0,1×ATR sein


def _berlin(ts):
    """MT5-Bar-Zeitstempel → echte Berliner Zeit (s. CLAUDE.md, Fix 2026-08-05).

    ⚠ NICHT `fromtimestamp(ts, UTC+3)` — MT5 liefert dort bereits die
    Broker-Wallclock, das ergaebe 3 Stunden Versatz.
    """
    return (datetime.fromtimestamp(int(ts), _UTC)
            .replace(tzinfo=_BROKER).astimezone(_BERLIN))


def _atr(H, L, C, n=14):
    tr = [0.0]
    for i in range(1, len(C)):
        tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    out, s = [0.0] * len(C), 0.0
    for i in range(1, len(C)):
        s += tr[i]
        if i > n:
            s -= tr[i - n]
        out[i] = s / min(i, n)
    return out


def _q(v, p):
    s = sorted(v)
    return s[min(len(s) - 1, int(p * len(s)))] if s else 0.0


def hole(sym, tf, n):
    b = mt5.copy_rates_from_pos(sym, tf, 0, n)
    if b is None or len(b) < 200:
        return None
    point = mt5.symbol_info(sym).point
    return {
        "H": [float(x["high"]) for x in b],
        "L": [float(x["low"]) for x in b],
        "C": [float(x["close"]) for x in b],
        "SP": [float(x["spread"]) * point for x in b],
        "T": [int(x["time"]) for x in b],
    }


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    if not mt5.initialize():
        print("MT5 nicht erreichbar"); return 1
    sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
                if mt5.symbol_info(c)), None)
    if not sym:
        mt5.shutdown(); print("Symbol nicht gefunden"); return 1

    tfs = [("M1", mt5.TIMEFRAME_M1), ("M5", mt5.TIMEFRAME_M5),
           ("M15", mt5.TIMEFRAME_M15)]
    daten = {}
    for lbl, tf in tfs:
        d = hole(sym, tf, n)
        if d:
            d["ATR"] = _atr(d["H"], d["L"], d["C"])
            daten[lbl] = d

    print("=" * 86)
    print(f"  SCALPING auf M1 — traegt es die Kosten?   Symbol {sym}")
    print("=" * 86)
    for lbl in ("M1", "M5", "M15"):
        if lbl in daten:
            d = daten[lbl]
            print(f"  {lbl}: {len(d['C'])} Bars  "
                  f"{_berlin(d['T'][0]):%d.%m.%Y} – {_berlin(d['T'][-1]):%d.%m.%Y}")

    # ── A) Spread/ATR je Zeitebene ───────────────────────────────────────
    print()
    print("=" * 86)
    print("  A) KOSTEN — Spread als Anteil der ATR der jeweiligen Zeitebene")
    print("=" * 86)
    print(f"  {'TF':<5} {'Ø ATR ($)':>10} {'Ø Spread ($)':>13} "
          f"{'Median Sp/ATR':>14} {'Ø Sp/ATR':>10} {'75 %':>8}")
    kosten = {}
    for lbl in ("M1", "M5", "M15"):
        if lbl not in daten:
            continue
        d = daten[lbl]
        r = [d["SP"][i] / d["ATR"][i] for i in range(20, len(d["C"]))
             if d["ATR"][i] > 0 and d["SP"][i] > 0]
        if not r:
            continue
        kosten[lbl] = (_q(r, 0.5), sum(r) / len(r))
        ga = [a for a in d["ATR"][20:] if a > 0]
        gs = [s for s in d["SP"][20:] if s > 0]
        print(f"  {lbl:<5} {sum(ga)/len(ga):>10.4f} {sum(gs)/len(gs):>13.4f} "
              f"{_q(r, 0.5):>14.3f} {sum(r)/len(r):>10.3f} {_q(r, 0.75):>8.3f}")
    print()
    print(f"  Referenz: B3-Huerde = Ø-Edge netto > {_HURDE:.2f}×ATR ·")
    print("  gemessen auf M5 (backtest_realcosts.py): Ø 0,265×ATR = schon dort eng.")

    # ── A2) je Berliner Stunde (M1) ──────────────────────────────────────
    if "M1" in daten:
        d = daten["M1"]
        prostd = {}
        for i in range(20, len(d["C"])):
            if d["ATR"][i] <= 0 or d["SP"][i] <= 0:
                continue
            prostd.setdefault(_berlin(d["T"][i]).hour, []).append(
                d["SP"][i] / d["ATR"][i])
        print()
        print("  Spread/ATR_M1 je Berliner Stunde (die guenstigsten zuerst):")
        rang = sorted(((sum(v) / len(v), h, len(v)) for h, v in prostd.items()
                       if len(v) > 50))
        for m, h, k in rang[:5]:
            print(f"    {h:02d}:00   {m:.3f}×ATR_M1   (n={k})")
        print("    ...")
        for m, h, k in rang[-3:]:
            print(f"    {h:02d}:00   {m:.3f}×ATR_M1   (n={k})")

    # ── B) Wieviel Weg gibt M1 her? ──────────────────────────────────────
    print()
    print("=" * 86)
    print("  B) VERFUEGBARER WEG — beste Auslenkung (MFE) in N M1-Bars, ×ATR_M1")
    print("=" * 86)
    if "M1" in daten:
        d = daten["M1"]
        H, L, C, A = d["H"], d["L"], d["C"], d["ATR"]
        print(f"  {'N Bars':<8} {'Median MFE':>12} {'Ø MFE':>9} {'Ø MAE':>9} "
              f"{'MFE > Kosten?':>16}")
        k_med = kosten.get("M1", (0, 0))[0]
        for N in (3, 5, 10, 15, 30):
            mfe, mae = [], []
            # Richtungsfrei: die BESTE der beiden Seiten (Obergrenze — ein
            # echtes Signal muesste die Seite vorher treffen)
            for i in range(20, len(C) - N, 7):
                if A[i] <= 0:
                    continue
                hi = max(H[i + 1:i + N + 1]); lo = min(L[i + 1:i + N + 1])
                mfe.append(max(hi - C[i], C[i] - lo) / A[i])
                mae.append(min(hi - C[i], C[i] - lo) / A[i])
            if not mfe:
                continue
            anteil = 100.0 * sum(1 for x in mfe if x > k_med) / len(mfe)
            print(f"  {N:<8} {_q(mfe, 0.5):>12.2f} {sum(mfe)/len(mfe):>9.2f} "
                  f"{sum(mae)/len(mae):>9.2f} {anteil:>15.0f} %")
        print()
        print("  ⚠ Das ist eine OBERGRENZE: gemessen wird die beste der beiden")
        print("    Seiten im Nachhinein. Ein echtes Signal muss die Richtung")
        print("    VORHER treffen — und das ist im Projekt 23× gescheitert.")

    # ── C) noetige Trefferquote ──────────────────────────────────────────
    print()
    print("=" * 86)
    print("  C) WELCHE TREFFERQUOTE braeuchte ein M1-Scalp?")
    print("=" * 86)
    if "M1" in kosten:
        k = kosten["M1"][0]
        print(f"  Kosten je Trade (Median) = {k:.3f}×ATR_M1")
        print(f"  {'Ziel':<8} {'Stop':<8} {'CRV':>6} {'noetige WR':>12} "
              f"{'noetige WR OHNE Kosten':>24}")
        for ziel, stop in ((0.5, 0.5), (1.0, 1.0), (1.0, 0.5),
                           (1.5, 1.5), (2.0, 2.0)):
            g, v = ziel - k, stop + k
            wr = 100.0 * v / (g + v) if g > 0 else float("nan")
            wr0 = 100.0 * stop / (ziel + stop)
            gg = f"{wr:.1f} %" if g > 0 else "unmoeglich"
            print(f"  {ziel:<8.1f} {stop:<8.1f} {ziel/stop:>6.2f} "
                  f"{gg:>12} {wr0:>23.1f} %")
        print()
        print("  Die letzte Spalte ist der KOSTENFREIE Fall. Der Abstand")
        print("  dazwischen ist der Preis, den das Signal zusaetzlich verdienen muss.")

    mt5.shutdown()
    return 0


if __name__ == "__main__":
    sys.exit(main())