AH-Oil-Trader/analyze_cone.py

#!/usr/bin/env python3
"""Wahrscheinlichkeits-Kegel: empirische Quantile fitten UND out-of-sample prüfen.

User-Wunsch 2026-07-31 war ein „Kursvorhersagemodul, das mögliche Verläufe in MT5
einzeichnet". Eingezeichnete PFADE würden eine Präzision suggerieren, die es nicht
gibt — das Projekt hat mit `analyze_verdict_calibration.py` teuer gelernt, dass eine
Zahl ohne Kalibrierungsprüfung in die Irre führt (conf_pct war in H2 invertiert).

STATTDESSEN: ein **Streuungs-Kegel**. Er sagt NICHT, wohin der Kurs geht, sondern
**wie weit** er in 30/60/120 min plausibel gelaufen sein kann. Der entscheidende
Unterschied: das ist ÜBERPRÜFBAR — landen wirklich ~80 % der Ausgänge im 80 %-Band?

METHODE
  r(h) = (C[i+h] − C[i]) / ATR[i]     (ATR-normiert → über Vola-Regime vergleichbar)
  Quantile auf H1 fitten, Abdeckung auf H2 messen (echtes out-of-sample).

Ein Kegel ist nur dann brauchbar, wenn die gemessene Abdeckung nah am Nominalwert
liegt. Weicht sie stark ab, wird er NICHT gebaut — dann ist die Streuung nicht
stabil ATR-skalierbar.
"""
import sys

import MetaTrader5 as mt5

_H = (6, 12, 24)                      # 30 / 60 / 120 min auf M5
_ATRMIN = 0.06
# Nominale Bänder: (unteres Quantil, oberes Quantil, Nennabdeckung)
_BANDS = ((0.25, 0.75, 50), (0.10, 0.90, 80), (0.05, 0.95, 90))


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])))
    out = []
    for i in range(len(C)):
        if not i:
            out.append(None); continue
        w = t[max(1, i - p + 1):i + 1]
        out.append(sum(w) / max(1, len(w)))
    return out


def quantile(sorted_xs, q):
    if not sorted_xs:
        return 0.0
    k = q * (len(sorted_xs) - 1)
    lo = int(k); hi = min(lo + 1, len(sorted_xs) - 1)
    return sorted_xs[lo] + (sorted_xs[hi] - sorted_xs[lo]) * (k - lo)


def collect(C, AT, a, b, h):
    out = []
    for i in range(a, min(b, len(C) - h)):
        atr = AT[i]
        if not atr:
            continue
        out.append((C[i + h] - C[i]) / max(atr, _ATRMIN))
    return out


def main():
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
    mt5.initialize(); sym = None
    for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
        if mt5.symbol_info(c):
            sym = c; break
    bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
    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]
    AT = _atr_series(H, L, C)
    N = len(C); mid = N // 2

    print("=" * 92)
    print(f"  WAHRSCHEINLICHKEITS-KEGEL — {sym} M5 ({N} Bars)")
    print("  Fit auf H1, Abdeckung gemessen auf H2 (out-of-sample). Werte in ×ATR.")
    print("=" * 92)

    fitted = {}
    for h in _H:
        tr = sorted(collect(C, AT, 100, mid, h))
        te = collect(C, AT, mid, N, h)
        print(f"\n  Horizont {h} Bars = {h*5} min   (fit n={len(tr)}, test n={len(te)})")
        print(f"    Median H1 {quantile(tr,0.5):+.3f}  ·  Median H2 "
              f"{sorted(te)[len(te)//2]:+.3f}   (Drift-Kontrolle)")
        row = {}
        for qlo, qhi, nominal in _BANDS:
            lo = quantile(tr, qlo); hi = quantile(tr, qhi)
            cov = 100.0 * sum(1 for x in te if lo <= x <= hi) / len(te)
            flag = "OK" if abs(cov - nominal) <= 5 else ("eng" if cov < nominal else "weit")
            print(f"    {nominal}%-Band  [{lo:+.2f} … {hi:+.2f}]×ATR   "
                  f"-> Abdeckung H2 {cov:>5.1f}%   [{flag}]")
            row[nominal] = (round(lo, 3), round(hi, 3))
        fitted[h] = row

    # ── Final auf ALLEN Bars fitten (wie beim P(break)-Modell: erst out-of-sample
    #    validieren, dann für den Live-Einsatz auf der vollen History fitten)
    print("\n" + "=" * 92)
    print("  FINAL (auf allen Bars gefittet) — zum Einbetten in core/cone.py")
    print("=" * 92)
    final = {}
    for h in _H:
        allx = sorted(collect(C, AT, 100, N, h))
        row = {}
        for qlo, qhi, nominal in _BANDS:
            row[nominal] = (round(quantile(allx, qlo), 3), round(quantile(allx, qhi), 3))
        final[h] = row
    print("_CONE = {")
    for h in _H:
        parts = ", ".join(f"{k}: ({v[0]:+.3f}, {v[1]:+.3f})" for k, v in final[h].items())
        print(f"    {h}: {{{parts}}},")
    print("}")


if __name__ == "__main__":
    main()