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