AH-Oil-Trader/backtest_htf_vola.py

#!/usr/bin/env python3
"""Kostet der M30-Gegen-Trend-Filter in HOHER VOLATILITÄT Geld? (2026-08-04)

Anlass: Am 04.08. fiel WTI um −4,12 $ (davon −4,16 $ in EINER Stunde). Der Bot
stand zu 93 % auf WARTEN; `htf_counter` blockte dabei **160-mal** — also genau
die Richtung, in die der Markt lief. Die M30-EMA läuft nach und steht in einem
schnellen Absturz noch „auf".

Der Filter ist über die GESAMTstichprobe belegt (Ø-Edge ×2, `backtest_htf_filter.py`).
Die Frage ist enger: **trägt er auch im oberen Vola-Terzil, oder kommt sein Nutzen
allein aus ruhigen Phasen?**

Aufbau: gegatete Live-Population, aber der HTF-Filter wird NICHT in `_build`
angewandt (htf_trend=0) — stattdessen wird je Signal festgehalten, ob der Filter
es geblockt HÄTTE. Die geblockten Signale werden mit dem kanonischen Exit
simuliert. Auswertung nach ATR-Terzil × 2 Halbjahre.

⚠ Regel VORAB: Der Filter wird nur dann im Hoch-Vola-Bereich gelockert, wenn die
GEBLOCKTEN Signale dort in BEIDEN Hälften ØR>0 UND PF>1 liefern — und das
mittlere Terzil das nicht widerlegt. Ein einzelnes positives Feld zählt nicht.
⚠ Hohe Beweislast: 21 Signal-Eingriffe sind hier bereits gescheitert, und dieser
würde ein GEMESSEN POSITIVES Gate schwächen.

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

import MetaTrader5 as mt5

from core.analysis import calc_trend_angle
from core.exit_model import LIVE, simulate
from core.wave_rec import (WaveRecommender, _ANGLE_LR, _EMA_FAST, _EMA_SLOW,
                           _HTF_DEADBAND, _ema_series)

_ATRMIN = 0.06
_PIVK = 3
_PIV_LOOK = 300


class _TU:
    def snapshot(self):
        return {}


def _atr_series(H, L, C, p=14):
    t = [0.0]
    for i in range(1, len(C)):
        t.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    return [(sum(t[max(1, i - p + 1):i + 1]) / max(1, len(t[max(1, i - p + 1):i + 1])))
            if i else None for i in range(len(C))]


def _htf_sign(C, i, n, atr):
    lo = max(0, i - n * (_EMA_SLOW + 5))
    sub = C[lo:i + 1:n]
    if len(sub) < _EMA_SLOW + 2 or not atr:
        return 0
    d = (_ema_series(sub, _EMA_FAST)[-1] - _ema_series(sub, _EMA_SLOW)[-1]) / atr
    return 0 if abs(d) < _HTF_DEADBAND else (1 if d > 0 else -1)


def st(Rs):
    if not Rs:
        return None
    n = len(Rs); w = sum(1 for x in Rs if x > 0); s = sum(Rs)
    up = sum(x for x in Rs if x > 0); dn = -sum(x for x in Rs if x < 0)
    return dict(n=n, wr=100 * w / n, oR=s / n, pf=(up / dn if dn > 0 else 9.99), sum=s)


def zeile(lbl, s):
    if not s:
        return f"      {lbl:<24} — (keine Trades)"
    return (f"      {lbl:<24} n={s['n']:>5} WR={s['wr']:>3.0f}% ØR={s['oR']:+.3f} "
            f"PF={s['pf']:>4.2f} ΣR={s['sum']:>+7.0f}")


def main():
    n_req = 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_req)
    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]; SP = [float(b["spread"]) * point for b in bars]
    N = len(C); mid = N // 2
    A = _atr_series(H, L, C)
    EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW)

    print("Vorberechnung ...", flush=True)
    AD = [90.0] * N
    for i in range(_ANGLE_LR + 2, N):
        AD[i] = calc_trend_angle(C[i - (_ANGLE_LR + 1):i + 1], _ANGLE_LR)
    HTF = [0] * N; H1 = [0] * N
    for i in range(N):
        if A[i]:
            HTF[i] = _htf_sign(C, i, 6, A[i]); H1[i] = _htf_sign(C, i, 12, A[i])
    PH = [j for j in range(_PIVK, N - _PIVK) if H[j] == max(H[j - _PIVK:j + _PIVK + 1])]
    PL = [j for j in range(_PIVK, N - _PIVK) if L[j] == min(L[j - _PIVK:j + _PIVK + 1])]

    gueltig = sorted(A[i] for i in range(200, N) if A[i] and A[i] >= _ATRMIN)
    t1 = gueltig[len(gueltig) // 3]; t2 = gueltig[2 * len(gueltig) // 3]
    print(f"ATR-Terzil-Grenzen: niedrig < {t1:.3f} ≤ mittel < {t2:.3f} ≤ hoch\n")

    w = WaveRecommender(_TU(), mt5.TIMEFRAME_M5)
    w.set_reversal_enabled(False); w.set_entry_room(0.6)
    w.set_dead_hours([]); w.set_breakout_k(0.3)

    # Signale OHNE HTF-Filter erzeugen (htf_trend=0), Blockade separat feststellen
    geblockt = {}; durchgelassen = {}
    pend = None
    i = _PIV_LOOK + _EMA_SLOW + 5
    while i < N - LIVE.max_hold - 2:
        atr = A[i]
        if not atr or atr < _ATRMIN:
            i += 1; continue
        w._pb_feats = {"atr": max(atr, 0.12)}
        w._pb_levels = {"ph": [C[j] for j in PH if i - _PIV_LOOK <= j <= i - _PIVK],
                        "pl": [C[j] for j in PL if i - _PIV_LOOK <= j <= i - _PIVK]}
        rec, _ = w._build(EF[i], ES[i], C[i], atr, "M5", 5,
                          htf_trend=0, h1_trend=H1[i], angle=AD[i])   # Filter AUS
        sig = rec.get("signal")
        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 * 0.3 * atr,
                        "invalid": C[i] - d * 0.3 * atr, "confirmed": False, "bar": i}
            if not pend["confirmed"]:
                broke = (C[i] >= pend["level"]) if d > 0 else (C[i] <= pend["level"])
                if broke:
                    pend["confirmed"] = True
                else:
                    if ((C[i] <= pend["invalid"]) if d > 0 else (C[i] >= pend["invalid"])) \
                            or (i - pend["bar"]) > 12:
                        pend = {"dir": d, "level": C[i] + d * 0.3 * atr,
                                "invalid": C[i] - d * 0.3 * atr,
                                "confirmed": False, "bar": i}
                    i += 1; continue
            rec = w._room_gate(rec, C[i])
            if rec.get("signal") in ("LONG", "SHORT"):
                r, xj = simulate(C[i], d, atr, H, L, C, i + 1, LIVE,
                                 timestop=True, use_tp=True, ret_bar=True)
                R = r - (SP[i] if SP[i] > 0 else 0.0225) / atr
                terz = "niedrig" if atr < t1 else ("mittel" if atr < t2 else "hoch")
                haelfte = "H1" if i < mid else "H2"
                # Hätte der M30-Filter das Signal geblockt?
                ziel = geblockt if (HTF[i] != 0 and HTF[i] != d) else durchgelassen
                ziel.setdefault((terz, haelfte), []).append(R)
                pend = None
                i = xj + 3
                continue
        elif sig == "WARTEN":
            pend = None
        i += 1

    print("=" * 92)
    print(f"  M30-FILTER nach VOLA-REGIME — {sym} M5 ({N} Bars, kanonischer Exit, Echtkosten)")
    print("  GEBLOCKT = Signale, die der Filter live verwirft. Ist ihr ØR positiv,")
    print("  kostet der Filter in diesem Regime Geld.")
    print("=" * 92)
    for terz in ("niedrig", "mittel", "hoch"):
        print(f"\n  ATR-Terzil {terz.upper()}")
        for h in ("H1", "H2"):
            print(zeile(f"GEBLOCKT {h}", st(geblockt.get((terz, h)))))
        for h in ("H1", "H2"):
            print(zeile(f"durchgelassen {h}", st(durchgelassen.get((terz, h)))))
        g1 = st(geblockt.get((terz, "H1"))); g2 = st(geblockt.get((terz, "H2")))
        if g1 and g2:
            ok = g1["oR"] > 0 and g2["oR"] > 0 and g1["pf"] > 1 and g2["pf"] > 1
            print(f"      → geblockte Signale beidhälftig positiv? "
                  f"{'JA — Filter kostet hier' if ok else 'nein — Filter zu Recht'}")
    print("\n  Regel (vorab): Lockerung nur, wenn die GEBLOCKTEN Signale im hohen Terzil")
    print("  in BEIDEN Hälften ØR>0 und PF>1 liefern und das mittlere Terzil nicht")
    print("  widerspricht. Hohe Beweislast: 21 Signal-Eingriffe sind hier gescheitert.")
    print()


if __name__ == "__main__":
    main()