AH-Oil-Trader/test_m15_karte_woche.py

#!/usr/bin/env python3
"""M15-Karte über eine ganze Woche — was behauptet sie, und stimmt es?

⚠⚠ DIE FRAGE „WIE VIELE KORREKTE EMPFEHLUNGEN?" MUSS NEU GESTELLT WERDEN.
Seit der Korrektur vom 08.08. gibt die Karte **keine Richtung** mehr aus
(`bias.richtung` ist dauerhaft None). Grund: der erste Entwurf machte aus dem
gemessenen VERBOT („gegen den M30-Trend halbiert den Edge") einen
Richtungsgeber — und erzeugte am 07.08. dreizehn LONG-Episoden an einem Tag,
der 1,02 $ fiel (Trefferquote 0 %).

Prüfbar ist deshalb nur, was die Karte NOCH behauptet:
  (A) das VETO — ist die verbotene Richtung tatsächlich schlechter als die
      erlaubte? Das ist die einzige Richtungsaussage, die überhaupt bleibt.
  (B) das LEVEL — trägt eine Order AM stehenden Level (statt zum Markt)?
  (C) die MECHANIK — wie oft grün/gelb/rot, welches Gate blockt?

⚠ Beide Richtungen werden auf DENSELBEN Episoden gerechnet — nur so misst der
Vergleich das Veto und nicht zwei verschiedene Trade-Folgen.
⚠ EINE WOCHE = EIN REGIME. Das ist ausdrücklich KEIN Edge-Nachweis (die
2-Stichproben-Regel bleibt), sondern eine Mechanik- und Plausibilitätsprüfung.

Aufruf:  python test_m15_karte_woche.py [tage]        (Default 7)
"""
from __future__ import annotations

import sys
from collections import defaultdict

import MetaTrader5 as mt5

from core.exit_model import LIVE, simulate
from core.sr_levels import halten, linien, naechste
from test_m15_karte_tag import _berlin, atr_series, htf_sign, m30_levels

_KOST_MAX = 0.32
_REICH = 0.5        # ⚠ GESETZT, nicht gemessen — der einzige diskriminierende Gate


def main():
    tage = int(sys.argv[1]) if len(sys.argv) > 1 else 7
    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_M15, 0, 6000)
    point = mt5.symbol_info(sym).point
    mt5.shutdown()
    T = [int(b["time"]) for b in bars]
    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)
    S30, S1H = htf_sign(H, L, C, 2), htf_sign(H, L, C, 4)
    LV = m30_levels(H, L, C)

    proTag = defaultdict(list)
    for i, t in enumerate(T):
        proTag[_berlin(t).strftime("%Y-%m-%d")].append(i)
    datteln = sorted(proTag)[-tage:]

    print("=" * 100)
    print(f"  M15-KARTE — {len(datteln)} Handelstage ({datteln[0]} bis {datteln[-1]}), {sym}")
    print("  ⚠ Die Karte gibt KEINE Richtung mehr aus. Geprüft wird das VETO und das LEVEL.")
    print("=" * 100)
    print(f"\n  {'Tag':<12}{'Bars':>5}{'grün':>7}{'gelb':>7}{'rot':>6}"
          f"{'Epis.':>7}{'Bewegung':>11}")

    alle_erlaubt, alle_verboten, alle_epi = [], [], 0
    tagszeilen = []
    for d in datteln:
        idx = proTag[d]
        if len(idx) < 20:
            continue
        st_liste = []
        prev_r = prev_s = None
        for i in idx:
            a = A[i]
            if not a or not LV[i]:
                st_liste.append((i, None, None, None)); continue
            ph, pl = LV[i]
            lv = linien(ph, pl, a)
            r, s = naechste(lv, C[i])
            r = halten(r, prev_r, C[i], a, True)
            s = halten(s, prev_s, C[i], a, False)
            prev_r, prev_s = r, s
            kr = SP[i] / a
            gaps = [(abs(x - C[i]) / a, x) for x in (r, s) if x is not None]
            nah = [g for g in gaps if g[0] <= _REICH]
            zu = "rot" if kr >= _KOST_MAX else ("gruen" if nah else "gelb")
            st_liste.append((i, zu, S30[i], nah[0][1] if nah else None))

        gr = sum(1 for _, z, _, _ in st_liste if z == "gruen")
        ge = sum(1 for _, z, _, _ in st_liste if z == "gelb")
        ro = sum(1 for _, z, _, _ in st_liste if z == "rot")

        # Episoden = erste Bar je zusammenhaengendem Gruen-Block
        epi, vor = [], None
        for i, z, h30, lvl in st_liste:
            if z == "gruen" and vor != "gruen":
                epi.append((i, h30, lvl, A[i], SP[i] / A[i]))
            vor = z
        alle_epi += len(epi)

        # ⚠ BEIDE Richtungen auf DENSELBEN Episoden — das misst das Veto
        for i, h30, lvl, a, kr in epi:
            if not h30:
                continue                      # HTF flach -> kein Veto -> nichts zu pruefen
            erlaubt = 1 if h30 == 1 else -1   # mit dem HTF
            alle_erlaubt.append(simulate(lvl, erlaubt, a, H, L, C, i + 1, LIVE) - kr)
            alle_verboten.append(simulate(lvl, -erlaubt, a, H, L, C, i + 1, LIVE) - kr)

        bew = C[idx[-1]] - C[idx[0]]
        tagszeilen.append((d, len(idx), gr, ge, ro, len(epi), bew))
        print(f"  {d:<12}{len(idx):>5}{gr:>7}{ge:>7}{ro:>6}{len(epi):>7}{bew:>+10.2f} $")

    def kz(Rs, name):
        if not Rs:
            return f"  {name:<34} —"
        n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
        g = sum(x for x in Rs if x > 0); v = -sum(x for x in Rs if x < 0)
        return (f"  {name:<34} n={n:>3} Treffer={100*w/n:>3.0f}% ØR={s/n:+.3f} "
                f"PF={(g/v if v > 0 else 9.99):>4.2f} ΣR={s:>+6.1f}")

    print(f"\n  {'-'*96}")
    print(f"  (A) DAS VETO — beide Richtungen auf denselben {len(alle_erlaubt)} Episoden")
    print(kz(alle_erlaubt, "MIT dem HTF (erlaubt)"))
    print(kz(alle_verboten, "GEGEN den HTF (verboten)"))
    if alle_erlaubt and alle_verboten:
        d = sum(alle_erlaubt) / len(alle_erlaubt) - sum(alle_verboten) / len(alle_verboten)
        print(f"  → Unterschied {d:+.3f} R je Trade zugunsten der ERLAUBTEN Richtung")
        print(f"  {'✅ Veto zeigt in die richtige Richtung' if d > 0 else '❌ Veto zeigt FALSCH'}")
    print(f"\n  Episoden gesamt: {alle_epi} an {len(tagszeilen)} Tagen "
          f"= {alle_epi/max(1,len(tagszeilen)):.1f} je Tag")
    print("\n  ⚠ EINE WOCHE = EIN REGIME. Kein Edge-Nachweis, nur Mechanik.")
    return 0


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