Source code for pytyche.calibrate.artifact

"""Calibration artifact — the v0.2 minimal contract.

Consumed by ``posterior.apply_calibration(calibration)``.  This module owns
only the minimal frozen shape: the layered correction payload plus the
regime metadata it was fitted on.  The ``from_sweep(name_or_path)``
registry constructor and the shipped artifacts are the calibration
track's scope, NOT part of this module.  There is no "skip" constructor:
the no-calibration choice is spelled ``None`` wherever a calibration is
accepted.
"""

from __future__ import annotations

import dataclasses
from typing import TYPE_CHECKING, Literal

from pytyche.calibrate.layered import LayeredCalibrationCorrection

if TYPE_CHECKING:
    from pytyche.contracts import ObservedExperimentData




@dataclasses.dataclass(frozen=True)
class Calibration:
    """SBC-fitted calibration correction plus the regime it applies to.

    Level: cross-cutting (sweep × experiment).

    Fields:
        correction: The layered R(p) + scale-family correction payload
            fitted on an SBC sweep.
        metric: Canonical metric name the sweep was fitted on (e.g.
            ``"revenue_per_visitor"``).
        n_treatments: The K of the fitted sweep — variant count INCLUDING
            control.
        pooling: Pooling mode of the fitted sweep.  NOT checked by
            :meth:`applies_to` (observed data carries no pooling);
            ``apply_calibration`` checks it against hurdle posteriors'
            ``pooling`` field (binary/continuous have no pooling concept).
    """

    correction: LayeredCalibrationCorrection
    metric: str
    n_treatments: int
    pooling: Literal["joint", "independent"]



    def applies_to(self, observed: ObservedExperimentData) -> bool:
        """Whether this artifact's fitted regime matches ``observed``.

        ``True`` iff ``observed.metric`` equals :attr:`metric` and
        ``len(observed.variants)`` equals :attr:`n_treatments`.
        """
        return (
            observed.metric == self.metric
            and len(observed.variants) == self.n_treatments
        )