"""The printed MDE holds its stated power on resampled REAL A/A rows (QA R2-1).

Cells are drawn from the committed identical-binary rows

(tests/fixtures/devloop_real_rows). For each trial the printed MDE X is taken

from a null draw; a fresh draw with a true +X% shift in one cell of the family

must then be confirmed. The stated power (cellstats.MDE_POWER, 80%) must hold.

Mann-Whitney p uses the normal approximation for n >= 10 here purely for test

speed (it closely tracks the exact/Monte-Carlo p at n = 20).

"""

from __future__ import annotations

import gzip

import json

import math

import random

import sys

import unittest

from pathlib import Path

from unittest import mock

ROOT = Path(__file__).resolve().parents[1]

sys.path.insert(0, str(ROOT / "scripts"))

from oakbench import cellstats as C  # noqa: E402

FIXTURES = ROOT / "tests" / "fixtures" / "devloop_real_rows"

EXACT = C.permutation_p_value

def fast_p(baseline, target, **kwargs):

    if len(baseline) < 10:

        return EXACT(baseline, target, **kwargs)

    n1, n2 = len(baseline), len(target)

    sd = math.sqrt(n1 * n2 * (n1 + n2 + 1) / 12)

    z = (abs(C._mann_whitney_u(baseline, target) - n1 * n2 / 2) - 0.5) / sd

    return max(0.0, min(1.0, math.erfc(z / math.sqrt(2))))

def real_pool() -> dict[tuple[str, str], list[float]]:

    pool: dict[tuple[str, str], list[float]] = {}

    for path in FIXTURES.glob("*.jsonl.gz"):

        with gzip.open(path, "rt") as handle:

            for line in handle:

                row = json.loads(line)

                if (row.get("skipped") or row.get("returncode") or not isinstance(row.get("elapsed_ms"), (int, float))

                        or row["subject"] not in ("oak_installed", "oak_installed_aa", "oak_local")

                        or row["operation"].endswith((".add", ".commit")) or row["operation"].startswith("remote")):

                    continue

                pool.setdefault((f"{path.name.split('-')[1]}:{row['scenario']}", row["operation"]), []).append(row["elapsed_ms"])

    return {key: values for key, values in pool.items() if len(values) >= 30}

POOL = real_pool()

KEYS = sorted(POOL)

def draw(rng, n, cells, shift=None, shifted=None):

    rows = []

    for key in cells:

        for run in range(n):

            for subject in ("oak_installed", "oak_installed_aa", "oak_local"):

                value = rng.choice(POOL[key])

                if shift and subject == "oak_local" and key == shifted:

                    value *= 1 + shift / 100

                rows.append({"subject": subject, "scenario": key[0], "operation": key[1], "run": run,

                             "elapsed_ms": value, "returncode": 0})

    return rows

def classify(rows, cells=None):

    floors = C.cell_floors(rows, "oak_installed", "oak_installed_aa", lambda r: r["elapsed_ms"], samples=400)

    return C.classify_cells(rows, "oak_local", "oak_installed", floors, cells=cells, bootstrap_samples=400, seed=3)

@mock.patch.object(C, "permutation_p_value", fast_p)

class StatedPowerTest(unittest.TestCase):

    def test_pool_is_real_and_large_enough(self):

        self.assertGreaterEqual(len(KEYS), 20)

    def test_confirmation_stage_mde_is_detected_at_stated_power(self):

        rng = random.Random(20260927)

        hits = trials = 0

        for _ in range(30):

            cells = rng.sample(KEYS, 10)

            null = classify(draw(rng, 20, cells))[cells[0]]

            if null.get("mde_pct") is None:

                continue

            trials += 1

            shifted = classify(draw(rng, 20, cells, shift=null["mde_pct"], shifted=cells[0]))[cells[0]]

            hits += shifted["verdict"] == "regression"

        self.assertGreaterEqual(trials, 25)

        self.assertGreaterEqual(hits / trials, C.MDE_POWER, f"{hits}/{trials}")

    def test_screen_stage_mde_is_detected_by_the_two_stage_pipeline(self):

        rng = random.Random(7)

        hits = trials = 0

        for _ in range(20):

            cells = rng.sample(KEYS, 14)

            null = classify(draw(rng, 5, cells))[cells[0]]

            if null.get("mde_pct") is None:

                continue

            trials += 1

            x = null["mde_pct"]

            flagged = C.screen_candidates(classify(draw(rng, 5, cells, shift=x, shifted=cells[0])), target_mde_pct=20.0)

            if cells[0] not in flagged:

                continue

            confirmed = classify(draw(rng, 20, flagged, shift=x, shifted=cells[0]), cells=flagged)

            hits += confirmed[cells[0]]["verdict"] == "regression"

        self.assertGreaterEqual(trials, 15)

        self.assertGreaterEqual(hits / trials, C.MDE_POWER, f"{hits}/{trials}")

    def test_old_half_power_point_is_not_what_is_printed(self):

        item = {"floor_pct": 8.0, "ci_low_pct": -5.0, "ci_high_pct": 5.0}

        self.assertEqual(C.minimum_detectable_effect_pct(item), 8.0 + C.MDE_POWER_MULTIPLIER * 5.0)

        self.assertGreater(C.minimum_detectable_effect_pct(item), max(8.0, 5.0))

if __name__ == "__main__":

    unittest.main()