AH-Oil-Trader/core/stichprobe.py

"""Stichproben-Hygiene: Entkopplung und Konfidenzintervalle für Live-Auswertungen.

⚠⚠ WARUM ES DIESES MODUL GIBT (2026-08-07). An EINEM Tag ist derselbe Fehler
DREIMAL passiert — zweimal in gebautem Code, einmal in einer Auswertung von Hand:

  1. Die Fälligkeitsbedingung `pbreak_accuracy_v2` zählte **800 ROHE**
     Vorhersagen. Roh waren es 878, entkoppelt **87** — sie überschätzte die
     Beweismenge um rund das Zehnfache und meldete „fällig", obwohl das
     95-%-Intervall der AUC von 0,434 bis 0,674 reichte und damit weder den
     Zufall noch die Messlatte ausschloss.
  2. `analyze_divergence.sec_d0` mittelte `AVG(p_break)` über ALLE Rohzeilen —
     derselbe Fehler, fest eingebaut, mit einer 8-Pp-Alarmschwelle darauf.
  3. Und ich selbst habe daraufhin eine „Basisraten-Lücke" von 14,1 Pp
     untersucht (elf Hypothesen, alle widerlegt), die es gar nicht gab: 53 % der
     Lücke waren Clusterung, 34 % Zeitraum, der Rest Rauschen.

**Die Zeilen in `pbreak_predictions` und `recommendations` sind NICHT
unabhängig.** Dieselbe Position pendelt vielfach um dasselbe Level; dieselbe
Marktlage erzeugt dutzende Empfehlungszeilen. Wer daraus eine Rate berichtet,
berichtet die Anzahl der Messungen, nicht die Menge der Evidenz.

**REGEL: vor jeder berichteten Rate `entkoppeln()`, und jede Aussage über einen
Unterschied mit `anteil_ci()` belegen.** Ein Punktschätzer ohne Intervall ist auf
diesen Daten keine Aussage.

⚠ Verwandt, aber NICHT dasselbe: `backtest_cost_gate.ci95` (Bootstrap für
Mittelwerte von R-Werten). Dort ist die Grenze die Überlappung der
Halte-Fenster; hier ist es die Wiederholung derselben Marktlage.
"""
from __future__ import annotations

import random


def entkoppeln(zeilen, schluessel, /):
    """Behält je Schlüssel nur die ERSTE Zeile — macht aus einer geclusterten
    Stichprobe eine näherungsweise unabhängige.

    `schluessel` bekommt eine Zeile und gibt den Gruppierungswert zurück, üblich
    ist `(Level auf 2 Stellen, Stunde)`. Die Reihenfolge bleibt erhalten.

    ⚠ Das ist eine NÄHERUNG, keine Garantie für Unabhängigkeit — zwei
    Berührungen desselben Levels in aufeinanderfolgenden Stunden hängen weiter
    zusammen. Sie ist aber um Größenordnungen besser als roh zu zählen.
    """
    gesehen = set()
    out = []
    for z in zeilen:
        k = schluessel(z)
        if k in gesehen:
            continue
        gesehen.add(k)
        out.append(z)
    return out


def anteil_ci(treffer, n: int, wiederholungen: int = 4000, seed: int = 42):
    """95-%-Bootstrap-Intervall für einen Anteil. `treffer` = Anzahl oder Liste
    von 0/1. Gibt `(anteil, lo, hi)` in 0…1 zurück, bei n < 10 `(anteil, None,
    None)` — ein Intervall aus einer Handvoll Punkten wäre Schaufensterdekoration.
    """
    if isinstance(treffer, (list, tuple)):
        werte = [1 if x else 0 for x in treffer]
        n = len(werte)
    else:
        werte = [1] * int(treffer) + [0] * (int(n) - int(treffer))
    if not werte:
        return 0.0, None, None
    anteil = sum(werte) / len(werte)
    if len(werte) < 10:
        return anteil, None, None
    rnd = random.Random(seed)
    boot = []
    m = len(werte)
    for _ in range(wiederholungen):
        boot.append(sum(werte[rnd.randrange(m)] for _ in range(m)) / m)
    boot.sort()
    return anteil, boot[int(0.025 * wiederholungen)], boot[int(0.975 * wiederholungen)]


def unterscheidbar(treffer, n: int, referenz: float, **kw) -> bool | None:
    """Ist der beobachtete Anteil von `referenz` UNTERSCHEIDBAR?

    `True` = ja (Referenz liegt ausserhalb des Intervalls) · `False` = nein ·
    `None` = zu wenige Daten für eine Aussage.

    ⚠ `False` heisst **nicht** „gleich", sondern „nicht belegbar verschieden" —
    genau diese Unterscheidung ging heute verloren, als aus einem Punktabstand
    von 14 Pp ein „Fehler" wurde.
    """
    _, lo, hi = anteil_ci(treffer, n, **kw)
    if lo is None:
        return None
    return not (lo <= referenz <= hi)


def _feld(z, name):
    """Holt ein Feld aus dict, sqlite3.Row ODER einem einfachen Tupel.

    ⚠ Die Tupel-Variante ist der Normalfall: ein sqlite3-Cursor liefert ohne
    `row_factory` Tupel, und genau daran ist der erste Einbau gescheitert
    (`TypeError: tuple indices must be integers`). Deshalb sind Zahlen als
    Feldangabe ausdrücklich erlaubt.
    """
    if isinstance(name, int):
        return z[name]
    if isinstance(z, dict):
        return z.get(name)
    try:
        return z[name]
    except (TypeError, IndexError, KeyError):
        return None


def schluessel_level_stunde(level_feld="level", ts_feld="ts"):
    """Der übliche Schlüssel für `pbreak_predictions`: dasselbe Level in derselben
    Stunde zählt einmal.

    Feldangaben dürfen Namen (dict/Row) ODER Spaltennummern (Tupel) sein.
    """
    def f(z):
        lvl = _feld(z, level_feld)
        ts = _feld(z, ts_feld)
        return (round(float(lvl or 0), 2), int(ts or 0) // 3600)
    return f


def paarweise(basis: dict, variante: dict, wiederholungen: int = 3000,
              seed: int = 7):
    """GEPAARTER Vergleich zweier Exit-/Regel-Varianten auf DENSELBEN Faellen.

    ⚠⚠ WARUM DAS NOETIG IST (2026-08-14, real passiert). Beim `trail_start`-Sweep
    sahen SEQUENTIELL vier Werte in beiden Haelften besser aus (Δ bis +0,101).
    Gepaart kippte in H2 das Vorzeichen. Ursache: eine sequentielle Sim mit EINEM
    Positions-Slot nimmt bei anderem Exit ANDERE Trades — der scheinbare Gewinn
    war ein **Selektionseffekt**, kein Exit-Effekt. Dieselbe Falle steckte am
    31.07. in `backtest_exit_combo.py` („verschiedene Trade-Folgen sind nicht
    vergleichbar").

    ⚠ Und der gepaarte Test ist nicht nur richtiger, sondern **trennschaerfer**:
    die Einzel-Intervalle messen die Streuung der TRADES, das gepaarte die des
    UNTERSCHIEDS — und die ist um ein Vielfaches kleiner.

    `basis` / `variante`: {schluessel: ergebnis}. Der Schluessel identifiziert den
    Fall (z. B. der Einstiegs-Bar-Index). Verglichen wird nur die Schnittmenge.

    Rueckgabe: (mittlere Differenz, lo, hi, n) — bei n < 20 (None, None, None, n).
    """
    import random as _r
    gemeinsam = [k for k in basis if k in variante]
    dif = [variante[k] - basis[k] for k in gemeinsam]
    n = len(dif)
    if n < 20:
        return None, None, None, n
    m = sum(dif) / n
    r = _r.Random(seed)
    bs = sorted(sum(dif[r.randrange(n)] for _ in range(n)) / n
                for _ in range(wiederholungen))
    return m, bs[int(.025 * wiederholungen)], bs[int(.975 * wiederholungen)], n


def zeig_paarweise(name: str, basis: dict, variante: dict) -> bool:
    """Druckt den gepaarten Vergleich und sagt, ob er die Null ausschliesst."""
    m, lo, hi, n = paarweise(basis, variante)
    if m is None:
        print(f"    {name:<28} zu wenige gemeinsame Faelle (n={n})")
        return False
    ok = lo > 0 or hi < 0
    print(f"    {name:<28} Δ je Fall {m:>+.4f}  KI [{lo:+.4f} … {hi:+.4f}]  "
          f"n={n}  {'✅ ohne 0' if ok else '⚠ enthaelt 0'}")
    return ok


def autokorrelation(x, lag: int = 1) -> float:
    """Autokorrelation zum gegebenen Lag. 0 = unabhaengig, ->1 = stark geclustert."""
    n = len(x)
    if n <= lag + 1:
        return 0.0
    m = sum(x) / n
    nen = sum((v - m) ** 2 for v in x)
    if nen <= 0:
        return 0.0
    zae = sum((x[i] - m) * (x[i + lag] - m) for i in range(n - lag))
    return max(-1.0, min(1.0, zae / nen))


def block_ci95(x, blocklen: int | None = None, wiederholungen: int = 2000,
               seed: int = 7):
    """95-%-Intervall fuer den Mittelwert bei ABHAENGIGEN Beobachtungen.

    ⚠⚠ WOZU — die Luecke stand seit dem 2026-08-07 woertlich in
    `backtest_cost_gate.ci95`: *"Fuer ueberlappende Beobachtungen ist er ZU
    OPTIMISTISCH ... dafuer braeuchte es einen Block-Bootstrap — `ci95` ist
    dafuer NICHT gedacht."* Der Behelf war bis jetzt `entkoppeln()`, und der
    kostet teuer: auf `pbreak_predictions` bleiben von **4.702 Zeilen nur 552**
    (88 % Datenverlust) — und er VERSCHIEBT die Schaetzung, weil er
    systematisch die ERSTE Beruehrung je Level+Stunde nimmt (0,399 gegen 0,427).

    Der Moving-Block-Bootstrap zieht stattdessen zusammenhaengende BLOECKE und
    behaelt damit die Abhaengigkeit INNERHALB eines Blocks. Auf denselben Daten:
        naiv            n=4702   Breite 0,028   <- 3x zu eng
        entkoppelt      n= 552   Breite 0,083   <- 88 % weg, Schaetzung verschoben
        Block L=30      n=4702   Breite 0,081   <- gleiche Breite, ALLE Daten

    ✅ KALIBRIERUNG GEMESSEN, nicht behauptet (AR(1)-Reihen, wahres Mittel
    bekannt, 200 Laeufe je Zelle — Ueberdeckung soll 95 % sein):
        rho    naiv    L=10   L=30   L=60
        0,00   94 %    94 %   91 %   92 %
        0,50   76 %    92 %   91 %   91 %
        0,90   40 %    80 %   89 %   90 %
        0,97   18 %    55 %   74 %   83 %
    ⚠ EHRLICH ZU DEN GRENZEN: (a) bei rho=0 kostet es ~3 Punkte Ueberdeckung
    gegenueber dem naiven Verfahren — bei UNABHAENGIGEN Daten bleibt `ci95` die
    bessere Wahl; (b) bei sehr starker Abhaengigkeit (rho=0,97) deckt auch der
    Block-Bootstrap nur 83 %, ist also weiter zu optimistisch, nur eben viel
    weniger. Er ist eine Verbesserung, keine Garantie.

    `blocklen=None` waehlt automatisch aus der gemessenen Lag-1-Autokorrelation
    (mehr Abhaengigkeit -> laengere Bloecke). Rueckgabe (lo, hi, blocklen) bzw.
    (None, None, 0) bei zu kleiner Stichprobe.
    """
    import random as _r
    x = [float(v) for v in x]
    n = len(x)
    if n < 40:
        return None, None, 0
    if blocklen is None:
        rho = abs(autokorrelation(x))
        # rho 0 -> ~n^(1/3); stark abhaengig -> deutlich laenger. Gedeckelt auf
        # n//5, damit noch mindestens 5 Bloecke gezogen werden.
        basis = max(2, int(round(n ** (1 / 3))))
        blocklen = max(2, min(n // 5, int(round(basis * (1.0 + 6.0 * rho)))))
    if n < 2 * blocklen:
        return None, None, 0
    r = _r.Random(seed)
    k = -(-n // blocklen)
    mittel = []
    for _ in range(wiederholungen):
        s = []
        for _ in range(k):
            i = r.randrange(n - blocklen + 1)
            s.extend(x[i:i + blocklen])
        mittel.append(sum(s[:n]) / n)
    mittel.sort()
    return (mittel[int(0.025 * wiederholungen)],
            mittel[int(0.975 * wiederholungen)], blocklen)