AH-Oil-Trader/backtest_m15_auto.py

#!/usr/bin/env python3
"""Kann der Autotrader die M15-KARTE automatisch handeln? (User-Auftrag 2026-08-13)

FRAGE: Der Bot soll eroeffnen, sobald die Karte GRUEN steht und eine gerichtete
Lesart zeigt. Das ist eine **Track-B-Frage** (Signal/Edge) — hohe Beweislast.

⚠⚠ WAS HIER NEU IST. Die Karte wurde ausdruecklich als BEDINGUNGS-Anzeige gebaut,
nicht als Richtungsquelle. Ihre Richtung entsteht erst aus der `lesart`, und die
ist bisher NIE als Handelsregel gemessen worden:
  · `backtest_m15_gates.py` (08.08.) handelt die vom HTF ERLAUBTE Richtung —
    nicht die Lesart. Ergebnis dort: H1 ueber ALLE sechs Abstands-Buckets negativ.
  · `backtest_squeeze_m15.py` / `backtest_vwap_m15.py` messen andere Signale.
Neu gemessen wird also genau der Zusatz: **traegt die P(break)-Lesart als
Richtungsgeber?**

VORAB FIXIERTE ENTSCHEIDUNGSREGEL (damit die Latte hinterher nicht wandert):
  (1) OeR > 0 UND PF > 1 in BEIDEN Haelften
  (2) 95-%-Bootstrap-KI schliesst in BEIDEN Haelften die Null aus
  (3) n >= 100 je Haelfte
  (4) schlaegt die Kontrolle C (HTF-Richtung) in BEIDEN Haelften
  (5) Nachbar-Schwellen (gap 0,4/0,5/0,6) kippen das Vorzeichen nicht
Verfehlt eine Bedingung → NICHT bauen.

AUFBAU — live-treu, mit den GETEILTEN Kernen statt Nachbauten:
  · Bars M5 (die Ebene, auf der P(break) trainiert ist — auf M15 waere das
    Modell falsch skaliert, Fix 2026-07-13).
  · ATR_M15 / M30-Level / HTF-Trend durch Aggregation aus M5 (3 / 6 / 12).
  · Level ueber `core.sr_levels` (linien/naechste/halten) = DERSELBE Code wie
    `_draw_levels` — der Nachbau wich am 07.08. nachweislich ab.
  · P(break) ueber `core.engine._p_break` mit den echten Live-Gewichten.
  · Status/Lesart inkl. der **Hysterese** vom 12.08. (sonst misst man das
    Flattern statt der Regel).
  · Exit `core.exit_model.LIVE`, Echtkosten aus dem Bar-Spread.
  · Sequentiell EIN Slot, 1x je Episode (wie der Live-Dedup).
  · Fill bei BERUEHRUNG — nicht zum Level-Preis, egal wo der Bar schliesst.
    Der Look-ahead-Aufschlag lag im Projekt bei 0,32-0,85 R.

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

import random
import sys
from bisect import bisect_left, bisect_right

import MetaTrader5 as mt5

from core.engine import _p_break
from core.exit_model import LIVE, simulate
from core.sr_levels import halten, linien, naechste
from test_m15_karte_tag import atr_series, ema

_PIV_K = 3
_LOOK = 50            # M30-Pivot-Fenster wie live
_HTF_DEAD = 0.15
_ATR5_FLOOR = 0.12    # wie im P(break)-Training
_WARTE = 8            # Bars, die die ruhende Order am Level liegt


def agg(H, L, C, step):
    """Hoehere TF aus M5 aggregieren (nur abgeschlossene Bars)."""
    hh, hl, hc = [], [], []
    for s in range(0, len(C) - step + 1, step):
        hh.append(max(H[s:s + step])); hl.append(min(L[s:s + step]))
        hc.append(C[s + step - 1])
    return hh, hl, hc


def htf_sign(H, L, C, step):
    """Trend-Vorzeichen der hoeheren TF je M5-Index — LOOK-AHEAD-frei."""
    hh, hl, hc = agg(H, L, C, step)
    out = [0] * len(C)
    if len(hc) < 55:
        return out
    ef, es, ha = ema(hc, 12), ema(hc, 50), atr_series(hh, hl, hc)
    sg = []
    for j in range(len(hc)):
        a = ha[j]
        if j < 52 or not a:
            sg.append(0); continue
        d = ef[j] - es[j]
        sg.append(0 if abs(d) < _HTF_DEAD * a else (1 if d > 0 else -1))
    for i in range(len(C)):
        j = i // step - 1
        out[i] = sg[j] if 0 <= j < len(sg) else 0
    return out


def atr_hoeher(H, L, C, step):
    """ATR der hoeheren TF je M5-Index (kausal: letzter ABGESCHLOSSENER Bar)."""
    hh, hl, hc = agg(H, L, C, step)
    ha = atr_series(hh, hl, hc)
    out = [None] * len(C)
    for i in range(len(C)):
        j = i // step - 1
        out[i] = ha[j] if 0 <= j < len(ha) else None
    return out


def pivots(H, L, C, step):
    """ALLE M30-Pivots EINMAL berechnen (Index + Preis), statt je Bar neu.

    ⚠ Das ist die dokumentierte O(log n)-Form: die quadratische Fassung war der
    Engpass des Projekts (9 min → 8,9 s). Das Pivot-Kriterium ist LOKAL, eine
    globale Berechnung liefert daher dieselbe Menge wie das rollende Fenster.
    Kausal: ein Pivot bei x ist erst bei x+K bekannt.
    """
    hh, hl, _ = agg(H, L, C, step)
    ph_i, ph_v, pl_i, pl_v = [], [], [], []
    for x in range(_PIV_K, len(hh) - _PIV_K):
        if hh[x] == max(hh[x - _PIV_K:x + _PIV_K + 1]):
            ph_i.append(x); ph_v.append(round(hh[x], 3))
        if hl[x] == min(hl[x - _PIV_K:x + _PIV_K + 1]):
            pl_i.append(x); pl_v.append(round(hl[x], 3))
    return (ph_i, ph_v), (pl_i, pl_v)


def fenster(idx, val, lo, hi):
    """Pivots mit lo <= index <= hi — zwei Binaersuchen statt Listenlauf."""
    a, b = bisect_left(idx, lo), bisect_right(idx, hi)
    return val[a:b]


def lauf(H, L, C, SP, lo, hi, gap_gruen, modus):
    """Sequentielle Sim EINES Slots ueber [lo, hi).

    modus: 'karte'   = Richtung aus der Lesart (das, was gefragt ist)
           'veto'    = Lesart, aber nur wenn sie der HTF-Erlaubnis entspricht
           'htf'     = KONTROLLE: HTF-erlaubte Richtung (= backtest_m15_gates)
           'gegen'   = KONTROLLE: Gegenrichtung der Lesart
    """
    a5 = atr_series(H, L, C)
    a15 = atr_hoeher(H, L, C, 3)
    s30 = htf_sign(H, L, C, 6)
    (ph_i, ph_v), (pl_i, pl_v) = pivots(H, L, C, 6)
    ef, es = ema(C, 12), ema(C, 50)

    Rs, n_ep = [], 0
    prev_r = prev_s = None
    hy_status, hy_lesart = None, None
    arm = None                    # letzte gehandelte Episode (dir, level)
    frei_ab = lo                  # Slot belegt bis ...
    grenze = min(hi, len(C) - LIVE.max_hold - 2)

    for i in range(lo, grenze):
        a = a5[i]; a15i = a15[i]
        if not a or not a15i:
            continue
        a5f = max(a, _ATR5_FLOOR)
        j30 = i // 6 - 1
        if j30 < _PIV_K + 5:
            continue
        ph = fenster(ph_i, ph_v, j30 - _LOOK, j30 - _PIV_K)
        pl = fenster(pl_i, pl_v, j30 - _LOOK, j30 - _PIV_K)
        if not ph and not pl:
            continue
        # --- Level: DERSELBE Code wie _draw_levels (Clustern mit ATR_M5) ---
        lv = linien(ph, pl, a5f)
        r, s = naechste(lv, C[i])
        r = halten(r, prev_r, C[i], a5f, True)
        s = halten(s, prev_s, C[i], a5f, False)
        prev_r, prev_s = r, s

        # --- Zonen + P(break), Merkmale trainingsgleich aus M5 ---
        mom6 = (C[i] - C[i - 6]) / a5f
        mom3 = (C[i] - C[i - 3]) / a5f
        ediff = (ef[i] - es[i]) / a5f
        zonen = []
        for lvl, seite in ((r, "res"), (s, "sup")):
            if lvl is None:
                continue
            d = 1 if seite == "res" else -1
            g = abs(lvl - C[i])
            wt = 1.0 if ediff * d > 0 else 0.0
            p = _p_break(mom6 * d, mom3 * d, wt, g / a5f)
            zonen.append({"seite": seite, "level": lvl,
                          "gap15": g / a15i, "pb": round(100 * p)})
        if not zonen:
            continue
        nah = min(zonen, key=lambda x: x["gap15"])

        # --- Status + Lesart MIT der Live-Hysterese (12.08.) ---
        kost = SP[i] / a15i
        if kost >= 0.32:
            status = "rot"
        else:
            schw = (gap_gruen + 0.10) if hy_status == "gruen" else gap_gruen
            status = "gruen" if nah["gap15"] <= schw else "gelb"
        lesart = "unklar"
        pb = nah["pb"]
        war_ger = hy_lesart in ("auf", "ab")
        lo_s, hi_s = (45, 55) if war_ger else (40, 60)
        if pb < lo_s or pb > hi_s:
            bricht = pb > 55 if war_ger else pb > hi_s
            res = nah["seite"] == "res"
            lesart = "auf" if (res if bricht else not res) else "ab"
        elif war_ger:
            lesart = hy_lesart
        hy_status, hy_lesart = status, lesart

        # --- Episoden-Dedup wie live: 1x je gruener, gerichteter Episode ---
        if status != "gruen" or lesart == "unklar":
            arm = None
            continue
        d_les = 1 if lesart == "auf" else -1
        kennung = (d_les, nah["level"])
        if arm == kennung:
            continue
        arm = kennung
        n_ep += 1

        if i < frei_ab:              # Slot belegt — Episode zaehlt, kein Trade
            continue

        # --- Richtung je Modus ---
        erlaubt = s30[i]
        if modus == "karte":
            d = d_les
        elif modus == "gegen":
            d = -d_les
        elif modus == "veto":
            if not erlaubt or erlaubt != d_les:
                continue
            d = d_les
        else:                        # 'htf'
            if not erlaubt:
                continue
            d = erlaubt

        # --- Fill bei BERUEHRUNG, dann kanonischer Exit ---
        lvl = nah["level"]
        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, a5f, H, L, C, fill + 1, LIVE, ret_bar=True)
        Rs.append(R - SP[i] / a5f)
        frei_ab = xb + 1
    return Rs, n_ep


def ci(Rs, n=3000, 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):
    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")), "u": u, "o": o}


def zeile(name, k):
    if not k:
        return f"  {name:<34}  —"
    kib = "     —      " if k["u"] is None else f"[{k['u']:+.3f} … {k['o']:+.3f}]"
    null = "" if k["u"] is None else ("" if (k["u"] > 0 or k["o"] < 0) else "  ⚠ KI enthaelt 0")
    return (f"  {name:<34}  n={k['n']:>4}  WR {k['wr']:>4.1f} %  OeR {k['oer']:>+.3f}  "
            f"PF {k['pf']:>5.2f}  SR {k['sr']:>+8.1f}  {kib}{null}")


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]
    mid = len(C) // 2
    print(f"\n  {sym} · {len(C)} M5-Bars · Exit kanonisch · Echtkosten · "
          f"Fill bei Beruehrung · EIN Slot")
    print(f"  Regel vorab fixiert: OeR>0 & PF>1 & KI ohne 0 in BEIDEN Haelften, "
          f"n>=100, schlaegt Kontrolle C, Nachbarn kippen nicht\n")

    print("  ── HAUPTFRAGE: die Karte automatisch handeln (gap<=0,50) ──────")
    erg = {}
    for modus, name in (("karte", "A) Karte (Lesart)"),
                        ("veto", "B) Lesart + Veto-konform"),
                        ("htf", "C) KONTROLLE HTF-Richtung"),
                        ("gegen", "D) KONTROLLE Gegenrichtung")):
        r1, e1 = lauf(H, L, C, SP, 60, mid, 0.50, modus)
        r2, e2 = lauf(H, L, C, SP, mid, len(C), 0.50, modus)
        erg[modus] = (kenn(r1), kenn(r2))
        print(zeile(name + " H1", erg[modus][0]))
        print(zeile(name + " H2", erg[modus][1]))
        if modus == "karte":
            print(f"       (gruene gerichtete Episoden: H1 {e1} · H2 {e2})")
        print()

    print("  ── NACHBAR-SCHWELLEN (Bedingung 5) ────────────────────────────")
    for g in (0.40, 0.50, 0.60):
        r1, _ = lauf(H, L, C, SP, 60, mid, g, "karte")
        r2, _ = lauf(H, L, C, SP, mid, len(C), g, "karte")
        k1, k2 = kenn(r1), kenn(r2)
        print(zeile(f"gap<={g:.2f}  H1", k1))
        print(zeile(f"gap<={g:.2f}  H2", k2))

    print("\n  ── URTEIL ─────────────────────────────────────────────────────")
    a1, a2 = erg["karte"]; c1, c2 = erg["htf"]
    pruef = []
    for nr, txt, ok in (
        (1, "OeR>0 und PF>1 in beiden Haelften",
         bool(a1 and a2 and a1["oer"] > 0 and a2["oer"] > 0
              and a1["pf"] > 1 and a2["pf"] > 1)),
        (2, "KI schliesst die Null in beiden aus",
         bool(a1 and a2 and a1["u"] is not None and a2["u"] is not None
              and (a1["u"] > 0 or a1["o"] < 0) and (a2["u"] > 0 or a2["o"] < 0))),
        (3, "n >= 100 je Haelfte",
         bool(a1 and a2 and a1["n"] >= 100 and a2["n"] >= 100)),
        (4, "schlaegt Kontrolle C in beiden",
         bool(a1 and a2 and c1 and c2
              and a1["oer"] > c1["oer"] and a2["oer"] > c2["oer"])),
    ):
        pruef.append(ok)
        print(f"    ({nr}) {txt:<42} {'✅' if ok else '❌'}")
    print(f"\n  ==> {'ALLE Bedingungen erfuellt' if all(pruef) else 'REGEL VERFEHLT — nicht bauen'}"
          f"   (Bedingung 5 aus der Nachbar-Tabelle ablesen)")
    return 0


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