AH-Oil-Trader/tests/test_stichprobe.py

"""Tests für `core/stichprobe.py` — die Stichproben-Hygiene, ohne die am
2026-08-07 DREIMAL derselbe Fehler passiert ist (Fälligkeitsbedingung,
Divergenz-Wächter, und meine eigene Auswertung).

⚠ Wichtigster Test unten: `test_faengt_den_fehler_vom_07_08` spielt die reale
Konstellation nach — eine geclusterte Stichprobe, deren rohe Rate weit von der
entkoppelten abweicht. Ohne ihn wäre nicht belegt, dass das Modul den Fall
überhaupt fängt.
"""
import pytest

from core.stichprobe import (anteil_ci, entkoppeln, schluessel_level_stunde,
                             unterscheidbar)


def test_entkoppeln_behaelt_erste_je_gruppe():
    zeilen = [(3600, 75.0), (3601, 75.0), (3602, 75.004), (7200, 75.0), (7201, 80.0)]
    ent = entkoppeln(zeilen, schluessel_level_stunde(1, 0))
    # 75,00 in Stunde 1 (auch 75,004 rundet dorthin), 75,00 in Stunde 2, 80,00
    assert len(ent) == 3
    assert ent[0] == (3600, 75.0)          # die ERSTE bleibt, nicht die letzte
    assert [z[0] for z in ent] == [3600, 7200, 7201]   # Reihenfolge erhalten


def test_entkoppeln_leer_und_ohne_duplikate():
    assert entkoppeln([], schluessel_level_stunde(1, 0)) == []
    z = [(0, 1.0), (3600, 2.0), (7200, 3.0)]
    assert entkoppeln(z, schluessel_level_stunde(1, 0)) == z


def test_feldzugriff_tupel_dict_und_row():
    """⚠ Der erste Einbau scheiterte an genau diesem Punkt: ein sqlite3-Cursor
    liefert ohne `row_factory` TUPEL, nicht dicts."""
    k_idx = schluessel_level_stunde(1, 0)
    k_name = schluessel_level_stunde("level", "ts")
    assert k_idx((3600, 75.0)) == (75.0, 1)
    assert k_name({"ts": 3600, "level": 75.0}) == (75.0, 1)


def test_anteil_ci_grundfaelle():
    a, lo, hi = anteil_ci([1] * 50 + [0] * 50, 100)
    assert a == pytest.approx(0.5)
    assert lo < 0.5 < hi
    # zu wenige Punkte -> KEIN Intervall vortaeuschen
    a, lo, hi = anteil_ci([1, 0, 1], 3)
    assert lo is None and hi is None
    # Zaehl-Variante muss dasselbe liefern wie die Listen-Variante
    assert anteil_ci(50, 100)[0] == pytest.approx(anteil_ci([1] * 50 + [0] * 50, 100)[0])


def test_anteil_ci_reproduzierbar():
    """Fester `seed` — sonst schwankt eine Alarmschwelle von Lauf zu Lauf."""
    assert anteil_ci([1] * 30 + [0] * 70, 100) == anteil_ci([1] * 30 + [0] * 70, 100)


def test_unterscheidbar():
    y = [1] * 45 + [0] * 55                     # 45 %
    assert unterscheidbar(y, 100, 0.44) is False    # nah dran -> nicht belegbar
    assert unterscheidbar(y, 100, 0.05) is True     # weit weg -> belegbar
    assert unterscheidbar([1, 0], 2, 0.5) is None   # zu wenig -> keine Aussage


def test_faengt_den_fehler_vom_07_08():
    """Die reale Konstellation: 10 Level, je einmal 'break' und danach viele
    Wiederholungen derselben Lage. Roh sieht die Rate ganz anders aus als
    entkoppelt — genau das hat am 07.08. eine Untersuchung mit elf widerlegten
    Hypothesen ausgeloest."""
    zeilen = []
    for i in range(10):
        ts = 3600 * i
        zeilen.append((ts, 75.0 + i, 1))              # erster Touch: Durchbruch
        zeilen.extend((ts + 60 + j, 75.0 + i, 1) for j in range(19))  # 19 Wiederholungen
    for i in range(10, 30):
        zeilen.append((3600 * i, 75.0 + i, 0))        # 20 einzelne Abpraller
    roh = sum(z[2] for z in zeilen) / len(zeilen)
    ent = entkoppeln(zeilen, schluessel_level_stunde(1, 0))
    entr = sum(z[2] for z in ent) / len(ent)
    assert roh > 0.75, "Aufbau falsch — roh muesste stark ueberhoeht sein"
    assert entr == pytest.approx(1 / 3, abs=0.02)
    # Die Verzerrung muss GROSS sein, sonst prueft der Test nichts
    assert roh - entr > 0.3
    # ...und die rohe Rate muss ausserhalb des Intervalls der entkoppelten liegen
    assert unterscheidbar([z[2] for z in ent], len(ent), roh) is True


# ── Block-Bootstrap fuer abhaengige Beobachtungen (2026-08-20) ──────────────
import random as _rnd

from core.stichprobe import autokorrelation, block_ci95


def _ar1(n, rho, seed):
    """AR(1)-Reihe mit Mittelwert 0 - so sehen ueberlappende Beobachtungen aus."""
    r = _rnd.Random(seed)
    out, v = [], 0.0
    for _ in range(n):
        v = rho * v + r.gauss(0, 1) * (1 - rho ** 2) ** 0.5
        out.append(v)
    return out


def test_autokorrelation_erkennt_unabhaengig_und_geclustert():
    assert abs(autokorrelation(_ar1(3000, 0.0, 1))) < 0.10
    assert autokorrelation(_ar1(3000, 0.9, 2)) > 0.7


def test_block_ci_ist_BREITER_als_das_naive_bei_abhaengigen_daten():
    """Der ganze Zweck: bei Clusterung muss das Intervall breiter werden.

    ⚠ Genau das war die dokumentierte Luecke - `ci95` unterstellt unabhaengige
    Ziehungen und ist bei ueberlappenden Beobachtungen zu optimistisch.
    """
    x = _ar1(2000, 0.9, 3)
    lo_b, hi_b, bl = block_ci95(x)
    assert bl > 2
    naiv = 2 * 1.96 * (sum((v - sum(x) / len(x)) ** 2 for v in x)
                       / len(x)) ** 0.5 / len(x) ** 0.5
    assert (hi_b - lo_b) > 1.5 * naiv, (
        "Block-KI ist nicht breiter als das naive - die Abhaengigkeit wird "
        "nicht erfasst")


def test_block_ci_deckt_den_wahren_mittelwert_meistens():
    """Kalibrierung: bei AR(1) mit Mittel 0 muss die 0 meistens drin liegen.

    ⚠ Bewusst 'meistens' und nicht 'immer': gemessen deckt der Block-Bootstrap
    bei rho=0,9 rund 90 % (das naive Verfahren nur 40 %). Er ist eine
    Verbesserung, keine Garantie - und der Test haelt genau das fest.
    """
    treffer = sum(1 for s in range(40)
                  if (lambda r: r[0] <= 0 <= r[1])(block_ci95(_ar1(800, 0.9, s),
                                                              wiederholungen=400)))
    assert treffer >= 30, f"nur {treffer}/40 Ueberdeckung"


def test_block_ci_lehnt_zu_kleine_stichproben_ab():
    lo, hi, bl = block_ci95([1.0, 2.0, 3.0])
    assert lo is None and hi is None and bl == 0