code/repro/src/cctm_core.py

"""Source-equivalent CPU implementation of the released CCTM recurrences.

The pinned author code implements the ECDF as a fresh linear scan and the
online rank as another linear scan.  That is transparent but makes the released
20,000-step, 1,100-trial null experiment needlessly expensive.  This module
keeps the scalar mathematical updates unchanged while using binary search for
the fixed ECDF and a Fenwick tree for growing-reference ranks.
"""

from __future__ import annotations

from dataclasses import dataclass
from math import log, sqrt
from typing import Iterable

import numpy as np


@dataclass(frozen=True)
class MartingaleRun:
    """Per-test-point p-values, wealth, and first Ville-threshold crossing."""

    p_values: np.ndarray
    wealth: np.ndarray
    first_crossing: int


class _Fenwick:
    """Integer prefix sums with zero-based public indexes."""

    def __init__(self, size: int) -> None:
        self._tree = np.zeros(size + 1, dtype=np.int64)

    def add(self, index: int, value: int = 1) -> None:
        i = index + 1
        while i < len(self._tree):
            self._tree[i] += value
            i += i & -i

    def prefix(self, stop: int) -> int:
        """Return the sum over the half-open interval [0, stop)."""
        total = 0
        i = stop
        while i:
            total += int(self._tree[i])
            i -= i & -i
        return total


def dkw_band(n: int, delta: float) -> float:
    """Match ``upstream.utils.compute_dkw_band`` exactly."""
    if n <= 0:
        raise ValueError("n must be positive")
    if not 0.0 < delta < 1.0:
        raise ValueError("delta must lie in (0, 1)")
    return sqrt(log(2.0 / delta) / (2.0 * n))


def _update_ons(eta: float, a: float, u: float, eps: float, D: float, smooth: float) -> tuple[float, float]:
    """One exact scalar transition from ``upstream.optimization.ONS.step``."""
    if smooth <= 0.0:
        v = (u - 0.5 - np.sign(eta) * eps) / (0.5 + eps)
        z = v / (1.0 + eta * v)
    else:
        root = sqrt(eta * eta + smooth * smooth)
        denominator = 0.5 + (1.0 + smooth) * eps
        v_dot_eta = (eta * (u - 0.5) - root * eps) / denominator
        derivative = (u - 0.5 - eta * eps / root) / denominator
        z = derivative / (1.0 + v_dot_eta)
    a_next = a + z * z
    eta_next = eta + (2.0 / (2.0 - log(3.0))) * z / a_next
    return float(np.clip(eta_next, -D, D)), float(a_next)


def _bet(u: float, eta: float, eps: float, smooth: float) -> float:
    """Match ``upstream.utils.betting_function`` for positive smoothing."""
    if smooth <= 0.0:
        return float(1.0 + eta * (u - 0.5 - np.sign(eta) * eps) / (0.5 + eps))
    return float(
        1.0
        + (eta * (u - 0.5) - sqrt(eta * eta + smooth * smooth) * eps)
        / (0.5 + (1.0 + smooth) * eps)
    )


def run_conditional_ctm(
    calibration: Iterable[float],
    stream: Iterable[float],
    *,
    ci_delta: float = 0.1,
    D: float = 0.5,
    clip_C: float = 0.1,
    smooth: float = 1e-6,
    warmup: int = 0,
    test_alpha: float = 0.05,
) -> MartingaleRun:
    """Run the fixed-reference CCTM, exactly preserving author recurrences."""
    calibration_array = np.asarray(list(calibration), dtype=np.float64)
    stream_array = np.asarray(list(stream), dtype=np.float64)
    if calibration_array.size == 0:
        raise ValueError("calibration must be non-empty")
    if not np.isfinite(calibration_array).all() or not np.isfinite(stream_array).all():
        raise ValueError("inputs must be finite")
    if not 0 <= warmup <= stream_array.size:
        raise ValueError("warmup must be within the stream")
    if D <= 0.0 or not 0.0 < test_alpha < 1.0:
        raise ValueError("invalid D or test_alpha")

    sorted_calibration = np.sort(calibration_array)
    eps = 0.0 if ci_delta >= 1.0 else dkw_band(sorted_calibration.size, ci_delta)
    p_values = np.searchsorted(sorted_calibration, stream_array, side="right") / sorted_calibration.size
    wealth = np.ones(stream_array.size, dtype=np.float64)
    eta, a, current = 0.0, 1.0, 1.0
    crossing = -1
    threshold = 1.0 / test_alpha

    for t, u in enumerate(p_values):
        if t >= warmup:
            eta_for_bet = 0.0 if abs(eta) < clip_C else eta
            bet = _bet(float(u), eta_for_bet, eps, smooth)
            if not bet > 0.0:
                raise ArithmeticError("non-positive betting factor")
            current = min(current * bet, 1e250)
            wealth[t] = current
            if crossing == -1 and current >= threshold:
                crossing = t
        eta, a = _update_ons(eta, a, float(u), eps, D, smooth)

    return MartingaleRun(p_values=p_values, wealth=wealth, first_crossing=crossing)


def run_standard_ctm(
    calibration: Iterable[float],
    stream: Iterable[float],
    *,
    rng: np.random.RandomState,
    D: float = 0.5,
    clip_C: float = 0.1,
    warmup: int = 0,
    test_alpha: float = 0.05,
) -> MartingaleRun:
    """Run the released growing-reference baseline with exact randomized ranks.

    ``rng`` must be at the same state that the author code has immediately
    before ``ConformalTest.test_exchangeability``.  This permits byte-level
    comparisons in tests while retaining O(log T) rank queries.
    """
    calibration_array = np.asarray(list(calibration), dtype=np.float64)
    stream_array = np.asarray(list(stream), dtype=np.float64)
    if calibration_array.size == 0:
        raise ValueError("calibration must be non-empty")
    if not 0 <= warmup <= stream_array.size:
        raise ValueError("warmup must be within the stream")

    all_values = np.concatenate((calibration_array, stream_array))
    coordinates = np.unique(all_values)
    tree = _Fenwick(coordinates.size)

    def randomized_rank(value: float, seen: int) -> float:
        index = int(np.searchsorted(coordinates, value, side="left"))
        smaller = tree.prefix(index)
        equal = tree.prefix(index + 1) - smaller
        p_value = (smaller + rng.uniform(0.0, 1.0) * (equal + 1)) / (seen + 1)
        tree.add(index)
        return float(p_value)

    seen = 0
    for value in calibration_array:
        randomized_rank(float(value), seen)
        seen += 1

    p_values = np.empty(stream_array.size, dtype=np.float64)
    wealth = np.ones(stream_array.size, dtype=np.float64)
    eta, a, current = 0.0, 1.0, 1.0
    crossing = -1
    threshold = 1.0 / test_alpha
    for t, value in enumerate(stream_array):
        p_value = randomized_rank(float(value), seen)
        seen += 1
        p_values[t] = p_value
        if t < warmup:
            eta, a = _update_ons(eta, a, p_value, 0.0, D, 0.0)
            continue
        eta_for_bet = 0.0 if abs(eta) < clip_C else eta
        bet = 1.0 + eta_for_bet * (p_value - 0.5)
        if not bet > 0.0:
            raise ArithmeticError("non-positive betting factor")
        current = min(current * bet, 1e250)
        wealth[t] = current
        if crossing == -1 and current >= threshold:
            crossing = t
        eta, a = _update_ons(eta, a, p_value, 0.0, D, 0.0)

    return MartingaleRun(p_values=p_values, wealth=wealth, first_crossing=crossing)


def normal_trial(
    seed: int,
    *,
    calibration_size: int,
    test_size: int,
    shift: float = 0.0,
    delay: int = 0,
    ci_delta: float = 0.1,
    warmup: int = 0,
    D: float = 0.5,
    clip_C: float = 0.1,
    smooth: float = 1e-6,
    test_alpha: float = 0.05,
    include_standard: bool = True,
) -> tuple[MartingaleRun, MartingaleRun | None]:
    """Author-compatible normal-stream generator used by the main protocol."""
    if not 0 <= delay <= test_size:
        raise ValueError("delay must lie within the test stream")
    rng = np.random.RandomState(seed)
    calibration = rng.normal(0.0, 1.0, calibration_size)
    stream = rng.normal(0.0, 1.0, test_size)
    stream[delay:] += shift
    conditional = run_conditional_ctm(
        calibration,
        stream,
        ci_delta=ci_delta,
        D=D,
        clip_C=clip_C,
        smooth=smooth,
        warmup=warmup,
        test_alpha=test_alpha,
    )
    standard = (
        run_standard_ctm(
            calibration,
            stream,
            rng=rng,
            D=D,
            clip_C=clip_C,
            warmup=warmup,
            test_alpha=test_alpha,
        )
        if include_standard
        else None
    )
    return conditional, standard