Learn

This package learns the data checks' settings (reconstruction margin, knife-edge window prefactor, conformal alpha, support minimum, optionally the decision line, and the calibration scores) from labelled history, with a certified error rate and a provenance record. The grid is tested by fixed-sequence testing by default (procedure= selects "bonferroni" instead), and HistoricalCase.was_acted_on lets the learner warn about selective labels. It also provides the exact binomial statistics behind the certificate and RiskMonitor for watching the live error rate afterwards. Open this page when you call learn_parameters, load a params.json, or monitor learned settings.

Everything below is importable from grounded_decision.learn. Recipe: Learn parameters from your history; the method and its guarantee: Learning settings with a guarantee.

learn

Learn decision parameters from labelled history, with provenance.

See :mod:grounded_decision.learn.calibrate for the method and its guarantee, and :mod:grounded_decision.learn.monitor for watching the live error rate afterwards.

Names are imported on first use (PEP 562), so a module that needs only :mod:~grounded_decision.learn.statistics (the games' regimes) does not import numpy.

Procedure module-attribute

Procedure = Literal['fixed_sequence', 'bonferroni']

How Learn-then-Test controls the chance of certifying a bad setting.

"fixed_sequence" (default): test the settings one at a time in an order fixed on the calibration part, each at the full significance, and stop at the first failure. "bonferroni": test every setting at the significance divided by the grid size. Both are valid; see :func:learn_parameters.

CandidateTest dataclass

CandidateTest(
    index: int,
    reconstruction_margin: float,
    window_prefactor: float,
    decision_line: float | None,
    conformal_alpha: float,
    minimum_support: float,
    acts: int,
    errors: int,
    trials: int,
    p_value: float,
    order: int,
    tested: bool,
    certified: bool,
)

One combination of settings and its exact test on the risk part.

What :func:learning_run reports for every combination of the grid, so a reader (or a chart) can see why the learner chose what it chose.

Attributes:

Name	Type	Description
index	int	Position in the grid order (ParameterGrid.combinations()).
reconstruction_margin	float	The combination's margin.
window_prefactor	float	Its knife-edge window prefactor.
decision_line	float | None	Its decision line (None: each case's own).
conformal_alpha	float	Its conformal alpha.
minimum_support	float	Its minimum support.
acts	int	Risk-part cases these settings would act on.
errors	int	Risk-part acts that were wrong.
trials	int	What the risk measure divides by: the acts (error_rate_among_acts) or every risk-part case (false_act_rate).
p_value	float	Exact p-value for "the error rate is at least max_risk" (1.0 with no trials); small is evidence of safety.
order	int	Position in the testing order (0 is tested first): the calibration-part order for fixed_sequence, the grid order for bonferroni.
tested	bool	Whether the procedure tested it (the fixed sequence stops at its first failure; Bonferroni tests every combination).
certified	bool	Whether its error rate was certified below the target.

CaseArrays dataclass

CaseArrays(
    evidence_strength: NDArray[float64],
    entity_count: NDArray[float64],
    link_density: NDArray[float64],
    own_decision_line: NDArray[float64],
    conformal_p_value: NDArray[float64],
    support: NDArray[float64],
    wrong: NDArray[bool_],
)

Per-case quantities the four data checks need, as numpy arrays.

support is NaN where the evidence has no support measurement; such cases pass the support check at every minimum.

HistoricalCase dataclass

HistoricalCase(
    evidence: EvidenceSnapshot,
    proposed_action: str,
    action_was_correct: bool,
    case_id: str = "",
    was_acted_on: bool | None = None,
)

One past decision whose outcome is now known.

Attributes:

Name	Type	Description
evidence	EvidenceSnapshot	The evidence available when the decision was made. Its label_score must be computed the same way it will be in production (for example 1 - model probability that the proposed action is right).
proposed_action	str	What the policy proposed ("block").
action_was_correct	bool	Whether that action turned out to be right, from the review outcome, chargeback, follow-up, and so on.
case_id	str	Optional identifier, kept for audit only.
was_acted_on	bool | None	Whether the proposed action was actually carried out (True) or held (False); None when unknown. The learner does not use it to choose settings; it uses it to warn when the history holds outcomes only for acted-on decisions (selective labels), because then the certificate describes a filtered population.

from_dict classmethod

from_dict(data: Mapping[str, Any]) -> HistoricalCase

Read one case, e.g. one line of a JSON Lines history export.

Source code in src/grounded_decision/learn/calibrate.py

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> HistoricalCase:
    """Read one case, e.g. one line of a JSON Lines history export."""

    return cls(
        evidence=EvidenceSnapshot.from_dict(data["evidence"]),
        proposed_action=str(data["proposed_action"]),
        action_was_correct=bool(data["action_was_correct"]),
        case_id=str(data.get("case_id", "")),
        was_acted_on=(
            None if data.get("was_acted_on") is None else bool(data["was_acted_on"])
        ),
    )

LearnedParameters dataclass

LearnedParameters(
    status: str,
    reconstruction_margin: float,
    operational_fk_prefactor: float,
    conformal_alpha: float,
    decision_line: float | None,
    calibration_scores: Mapping[str, tuple[float, ...]],
    target: RiskTarget,
    grid: ParameterGrid,
    risk_split: SplitReport,
    holdout: SplitReport,
    configurations_tried: int,
    configurations_certified: int,
    minimum_risk_cases_needed: int,
    dataset_fingerprint: str,
    split_sizes: Mapping[str, int],
    seed: int,
    warnings: tuple[str, ...] = tuple(),
    minimum_support: float = 0.0,
    procedure: Procedure = "fixed_sequence",
)

Settings learned from history, with the record of how.

Use :meth:limits to build the Limits for your Module and :meth:apply to complete each request's evidence (calibration scores and decision line). Persist :meth:to_dict next to your policy and store :attr:fingerprint on every decision record.

minimum_support class-attribute instance-attribute

minimum_support: float = 0.0

Learned Limits.min_support; 0.0 means the support check is off.

procedure class-attribute instance-attribute

procedure: Procedure = 'fixed_sequence'

The multiple-testing procedure that certified the settings.

fingerprint property

fingerprint: str

Short stable hash of the parameters and their provenance.

limits

limits(base: Limits = DEFAULT_LIMITS) -> Limits

Limits carrying the learned check settings.

The learner treats evidence without a support measurement as passing the support check, as Module.decide does when base.require_support is off (the default). Keep it off, or measure support on every case, for the certificate to apply.

Source code in src/grounded_decision/learn/calibrate.py

def limits(self, base: Limits = DEFAULT_LIMITS) -> Limits:
    """``Limits`` carrying the learned check settings.

    The learner treats evidence without a ``support`` measurement as
    passing the support check, as ``Module.decide`` does when
    ``base.require_support`` is off (the default). Keep it off, or
    measure support on every case, for the certificate to apply.
    """

    return replace(
        base,
        reconstruction_margin=self.reconstruction_margin,
        operational_fk_prefactor=self.operational_fk_prefactor,
        conformal_alpha=self.conformal_alpha,
        min_support=self.minimum_support,
    )

apply

apply(
    evidence: EvidenceSnapshot, proposed_action: str
) -> EvidenceSnapshot

Fill in the learned calibration scores and decision line.

The calibration scores are chosen by proposed_action: a case is compared only with past cases where that action was right.

Source code in src/grounded_decision/learn/calibrate.py

def apply(
    self, evidence: EvidenceSnapshot, proposed_action: str
) -> EvidenceSnapshot:
    """Fill in the learned calibration scores and decision line.

    The calibration scores are chosen by ``proposed_action``: a case is
    compared only with past cases where *that* action was right.
    """

    return replace(
        evidence,
        scores=tuple(self.calibration_scores.get(proposed_action, ())),
        p_threshold=(
            evidence.p_threshold
            if self.decision_line is None
            else self.decision_line
        ),
    )

from_dict classmethod

from_dict(data: Mapping[str, Any]) -> LearnedParameters

Read :meth:to_dict output.

Files written before the support check and the procedure choice existed load with minimum_support=0.0 (support check off), a grid of minimum_supports=(0.0,), and procedure="bonferroni" (the only procedure then).

Source code in src/grounded_decision/learn/calibrate.py

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> LearnedParameters:
    """Read :meth:`to_dict` output.

    Files written before the support check and the procedure choice
    existed load with ``minimum_support=0.0`` (support check off), a
    grid of ``minimum_supports=(0.0,)``, and ``procedure="bonferroni"``
    (the only procedure then).
    """

    parameters = data["parameters"]
    provenance = data["provenance"]
    target = provenance["target"]
    grid = provenance["grid"]
    return cls(
        status=str(data["status"]),
        reconstruction_margin=float(parameters["reconstruction_margin"]),
        operational_fk_prefactor=float(parameters["operational_fk_prefactor"]),
        conformal_alpha=float(parameters["conformal_alpha"]),
        decision_line=(
            None
            if parameters["decision_line"] is None
            else float(parameters["decision_line"])
        ),
        calibration_scores={
            str(action): tuple(float(score) for score in scores)
            for action, scores in parameters["calibration_scores"].items()
        },
        target=RiskTarget(
            max_risk=float(target["max_risk"]),
            confidence=float(target["confidence"]),
            measure=RiskMeasure(target["measure"]),
        ),
        grid=ParameterGrid(
            reconstruction_margins=tuple(grid["reconstruction_margins"]),
            window_prefactors=tuple(grid["window_prefactors"]),
            decision_lines=tuple(grid["decision_lines"]),
            conformal_alphas=tuple(grid["conformal_alphas"]),
            minimum_supports=tuple(
                float(value) for value in grid.get("minimum_supports", (0.0,))
            ),
        ),
        risk_split=_split_report_from_dict(provenance["risk_split"]),
        holdout=_split_report_from_dict(provenance["holdout"]),
        configurations_tried=int(provenance["configurations_tried"]),
        configurations_certified=int(provenance["configurations_certified"]),
        minimum_risk_cases_needed=int(provenance["minimum_risk_cases_needed"]),
        dataset_fingerprint=str(provenance["dataset_fingerprint"]),
        split_sizes={str(k): int(v) for k, v in provenance["split_sizes"].items()},
        seed=int(provenance["seed"]),
        warnings=tuple(str(w) for w in provenance["warnings"]),
        minimum_support=float(parameters.get("minimum_support", 0.0)),
        procedure=_procedure(provenance.get("procedure", "bonferroni")),
    )

explain

explain() -> list[str]

Plain-language lines describing what was learned and how far to trust it.

Source code in src/grounded_decision/learn/calibrate.py

def explain(self) -> list[str]:
    """Plain-language lines describing what was learned and how far to trust it."""

    target = self.target
    measure_words = (
        "of the decisions it acts on"
        if target.measure is RiskMeasure.ERROR_RATE_AMONG_ACTS
        else "of all decisions (acted and wrong)"
    )
    split_description = ", ".join(
        f"{name} {size}" for name, size in self.split_sizes.items()
    )
    lines = [
        (
            f"Goal: at most {target.max_risk:.1%} wrong {measure_words}, "
            f"certified with {target.confidence:.0%} confidence."
        ),
        (
            f"History: {sum(self.split_sizes.values())} labelled cases, split "
            f"{split_description} (seed {self.seed}, dataset "
            f"{self.dataset_fingerprint})."
        ),
        (
            f"Tried {self.configurations_tried} combinations of settings; "
            f"{self.configurations_certified} were certified."
        ),
        _procedure_line(self.procedure, target.confidence),
    ]
    if not self.certified:
        lines.append(
            "No combination could be certified, so these settings hold every "
            "decision. Either there is too little history (even with zero "
            f"errors, at least {self.minimum_risk_cases_needed} acted-on cases "
            "are needed in the risk part), or every setting that acts often "
            "enough makes too many mistakes. Fixes: more labelled history, a "
            "better definition of related cases or a better model score, or a "
            "higher acceptable risk."
        )
    else:
        line = (
            "each case's own"
            if self.decision_line is None
            else f"{self.decision_line:g}"
        )
        knife_edge = (
            " (knife-edge check off)" if self.operational_fk_prefactor == 0 else ""
        )
        unusual = " (unusual-case check off)" if self.conformal_alpha == 0 else ""
        support = " (support check off)" if self.minimum_support == 0 else ""
        lines.append(
            f"Chosen: reconstruction margin {self.reconstruction_margin:g} "
            f"(evidence strength must exceed {1 + self.reconstruction_margin:g}), "
            f"window prefactor {self.operational_fk_prefactor:g}{knife_edge}, "
            f"decision line {line}, conformal alpha {self.conformal_alpha:g}{unusual}, "
            f"minimum support {self.minimum_support:g}{support}."
        )
        lines.append(
            f"Certificate: with {target.confidence:.0%} confidence the true error "
            f"rate is at most {target.max_risk:.1%}, provided future cases are "
            "independent draws from the same process as this history."
        )
        lines.append(
            f"Risk part (the data the settings were chosen on, so optimistic): "
            f"acted on {self.risk_split.acts} of {self.risk_split.cases}, "
            f"{self.risk_split.errors} wrong."
        )
        lines.append(
            f"Hold-out (never used for choosing): acted on {self.holdout.acts} of "
            f"{self.holdout.cases} ({self.holdout.act_rate:.1%}), "
            f"{self.holdout.errors} wrong; error rate at most "
            f"{self.holdout.error_rate_upper_bound:.2%}."
        )
    lines += [f"Warning: {warning}" for warning in self.warnings]
    return lines

LearningRun dataclass

LearningRun(
    learned: LearnedParameters,
    candidates: tuple[CandidateTest, ...],
    significance_per_test: float,
    chosen: int | None,
)

The learned settings with every combination's test (:func:learning_run).

Attributes:

Name	Type	Description
learned	LearnedParameters	What :func:learn_parameters returns for the same call.
candidates	tuple[CandidateTest, ...]	Every combination of the grid, in grid order.
significance_per_test	float	The level each p-value was compared with (1 - confidence, divided by the grid size for Bonferroni).
chosen	int | None	The chosen combination's index, or None when nothing was certified.

ParameterGrid dataclass

ParameterGrid(
    reconstruction_margins: tuple[float, ...] = (
        -1.0,
        0.0,
        0.05,
        0.25,
        0.5,
        1.0,
        2.0,
        4.0,
    ),
    window_prefactors: tuple[float, ...] = (
        0.0,
        0.5,
        1.0,
        2.0,
        4.0,
    ),
    decision_lines: tuple[float | None, ...] = (None,),
    conformal_alphas: tuple[float, ...] = (
        0.0,
        0.02,
        0.05,
        0.1,
        0.2,
    ),
    minimum_supports: tuple[float, ...] = (
        0.0,
        0.5,
        0.6,
        0.7,
        0.8,
        0.9,
    ),
)

The candidate settings Learn-then-Test chooses among.

Each tuple lists candidate values for one setting. The defaults include a value that switches each check off (margin -1.0 requires only some agreement, prefactor 0.0 gives an empty window, alpha 0.0 never rejects, minimum support 0.0 always passes), so the learner can find that a check does not help on your data. Larger grids find better settings; with the "bonferroni" procedure they also need more history, because the certificate is divided across every combination.

minimum_supports are candidate Limits.min_support values for the support check. Cases whose evidence has no support measurement pass it at every value (as Module.decide does when require_support is off).

decision_lines are candidate p_threshold values; None keeps each case's own p_threshold. Use :func:decision_line_candidates to propose values from your data.

combinations

combinations() -> list[
    tuple[float, float, float | None, float, float]
]

Every (margin, prefactor, decision_line, alpha, minimum_support).

In a fixed order (the grid order) that ties are broken by.

Source code in src/grounded_decision/learn/calibrate.py

def combinations(
    self,
) -> list[tuple[float, float, float | None, float, float]]:
    """Every ``(margin, prefactor, decision_line, alpha, minimum_support)``.

    In a fixed order (the grid order) that ties are broken by.
    """

    return list(
        itertools.product(
            self.reconstruction_margins,
            self.window_prefactors,
            self.decision_lines,
            self.conformal_alphas,
            self.minimum_supports,
        )
    )

RiskMeasure

Bases: StrEnum

Which error rate the guarantee is about.

ERROR_RATE_AMONG_ACTS class-attribute instance-attribute

ERROR_RATE_AMONG_ACTS = 'error_rate_among_acts'

Of the decisions the module acts on, the fraction that are wrong.

FALSE_ACT_RATE class-attribute instance-attribute

FALSE_ACT_RATE = 'false_act_rate'

Of all decisions, the fraction that act and are wrong.

RiskTarget dataclass

RiskTarget(
    max_risk: float,
    confidence: float = 0.95,
    measure: RiskMeasure = ERROR_RATE_AMONG_ACTS,
)

The risk you are willing to accept. This is the business input.

Attributes:

Name	Type	Description
max_risk	float	Largest acceptable error rate, e.g. 0.02 for "at most 2% of automatic actions are wrong".
confidence	float	How sure the certificate must be, e.g. 0.95.
measure	RiskMeasure	Which error rate max_risk limits.

SplitReport dataclass

SplitReport(
    cases: int,
    acts: int,
    errors: int,
    error_rate_upper_bound: float,
)

How the chosen settings did on one part of the history.

MonitorStatus

Bases: StrEnum

INSUFFICIENT_DATA class-attribute instance-attribute

INSUFFICIENT_DATA = 'insufficient_data'

Too few outcomes in the window to say anything.

WITHIN_TARGET class-attribute instance-attribute

WITHIN_TARGET = 'within_target'

The recent error rate is confidently at or below the target.

UNCERTAIN class-attribute instance-attribute

UNCERTAIN = 'uncertain'

Neither confidently below nor confidently above the target.

ABOVE_TARGET class-attribute instance-attribute

ABOVE_TARGET = 'above_target'

The recent error rate is confidently above the target: re-learn now.

RiskMonitor dataclass

RiskMonitor(target: RiskTarget, window: int = 500)

Rolling check of the realised error rate against a RiskTarget.

Record each decision once its outcome is known. status() reports ABOVE_TARGET when even the optimistic (lower) confidence bound on the recent error rate exceeds the target: that is the signal to stop acting automatically and re-learn.

Repeatedly checking a fixed-confidence bound raises the chance of a false alarm over time; treat a single ABOVE_TARGET as a reason to investigate and re-learn, not as proof.

Attributes:

Name	Type	Description
target	RiskTarget	The target the settings were learned for.
window	int	How many recent relevant outcomes to keep.

BOUNDS_KEPT class-attribute

BOUNDS_KEPT: int = 4096

Most bounds remembered: a full window revisits the same few counts.

record

record(*, acted: bool, action_was_correct: bool) -> None

Add one decision whose outcome is now known.

Source code in src/grounded_decision/learn/monitor.py

def record(self, *, acted: bool, action_was_correct: bool) -> None:
    """Add one decision whose outcome is now known."""

    if self.target.measure is RiskMeasure.ERROR_RATE_AMONG_ACTS:
        if acted:
            self._outcomes.append(not action_was_correct)
    else:
        self._outcomes.append(acted and not action_was_correct)

case_arrays

case_arrays(
    cases: Sequence[HistoricalCase],
    calibration: Mapping[str, tuple[float, ...]],
) -> CaseArrays

Pre-compute what the checks need for cases.

Pair with :func:predicted_acts to evaluate any settings on your own history, for example to compare learned settings with hand-set ones.

Source code in src/grounded_decision/learn/calibrate.py

def case_arrays(
    cases: Sequence[HistoricalCase], calibration: Mapping[str, tuple[float, ...]]
) -> CaseArrays:
    """Pre-compute what the checks need for ``cases``.

    Pair with :func:`predicted_acts` to evaluate any settings on your own
    history, for example to compare learned settings with hand-set ones.
    """
    return CaseArrays(
        evidence_strength=np.array(
            [d_lambda_sq(c.evidence.degree, c.evidence.lambda_hat) for c in cases],
            dtype=float,
        ),
        entity_count=np.array([c.evidence.n for c in cases], dtype=float),
        link_density=np.array([c.evidence.p_hat for c in cases], dtype=float),
        own_decision_line=np.array(
            [c.evidence.p_threshold for c in cases], dtype=float
        ),
        conformal_p_value=_conformal_p_values(cases, calibration),
        support=np.array(
            [
                math.nan if c.evidence.support is None else c.evidence.support
                for c in cases
            ],
            dtype=float,
        ),
        wrong=np.array([not c.action_was_correct for c in cases], dtype=bool),
    )

dataset_fingerprint

dataset_fingerprint(cases: Sequence[HistoricalCase]) -> str

Stable hash identifying exactly which history was learned from.

Source code in src/grounded_decision/learn/calibrate.py

def dataset_fingerprint(cases: Sequence[HistoricalCase]) -> str:
    """Stable hash identifying exactly which history was learned from."""

    digest = hashlib.sha256()
    for case in cases:
        digest.update(json.dumps(case.to_dict(), sort_keys=True).encode())
    return "data@" + digest.hexdigest()[:12]

decision_line_candidates

decision_line_candidates(
    cases: Sequence[HistoricalCase], count: int = 5
) -> tuple[float | None, ...]

Candidate decision lines: quantiles of the observed link density.

Returns None (keep each case's own line) followed by count evenly spaced quantiles of p_hat. Pass the result as ParameterGrid(decision_lines=...).

Source code in src/grounded_decision/learn/calibrate.py

def decision_line_candidates(
    cases: Sequence[HistoricalCase], count: int = 5
) -> tuple[float | None, ...]:
    """Candidate decision lines: quantiles of the observed link density.

    Returns ``None`` (keep each case's own line) followed by ``count``
    evenly spaced quantiles of ``p_hat``. Pass the result as
    ``ParameterGrid(decision_lines=...)``.
    """

    densities = np.array([case.evidence.p_hat for case in cases], dtype=float)
    if densities.size == 0:
        return (None,)
    quantile_levels = np.linspace(0.1, 0.9, count)
    quantiles = sorted(
        {round(float(q), 4) for q in np.quantile(densities, quantile_levels)}
    )
    return (None, *quantiles)

learn_parameters

learn_parameters(
    cases: Sequence[HistoricalCase],
    target: RiskTarget,
    *,
    grid: ParameterGrid | None = None,
    calibration_fraction: float = 0.3,
    holdout_fraction: float = 0.2,
    seed: int = 0,
    procedure: Procedure = "fixed_sequence",
) -> LearnedParameters

Learn check settings from labelled history with a certified error rate.

Parameters:

Name	Type	Description	Default
cases	Sequence[HistoricalCase]	Past decisions with known outcomes. Only cases where the policy proposed an action belong here. Set HistoricalCase.was_acted_on so the learner can warn about selective labels.	required
target	RiskTarget	The error rate you accept and how sure the certificate must be.	required
grid	ParameterGrid | None	Candidate settings; defaults to :class:ParameterGrid.	None
calibration_fraction	float	Share of cases used to build calibration scores (and, for "fixed_sequence", the testing order).	0.3
holdout_fraction	float	Share kept aside to report honest performance. The rest is the risk part used to certify settings.	0.2
seed	int	Seed for the random split, recorded for reproducibility.	0
procedure	Procedure	"fixed_sequence" (default) or "bonferroni".	'fixed_sequence'

Returns:

Type	Description
LearnedParameters	The learned settings and their provenance. Check status: if it
LearnedParameters	is "no_safe_configuration" the settings hold every decision.

Why both procedures are valid. Write delta = 1 - confidence. Fix the calibration part; the calibration scores, and so the true error rate R_k of every combination k, are then fixed numbers, and the risk-part cases are independent draws unrelated to them. For each k the null hypothesis H_k: R_k > max_risk has an exact binomial p-value p_k with P(p_k <= u) <= u whenever H_k is true. A combination is certified when its H_k is rejected; the chosen combination is a certified one, so it suffices that the chance of rejecting any true null (the family-wise error) is at most delta.

• "bonferroni" rejects H_k when p_k <= delta / K (K the grid size). The family-wise error is at most the sum over true nulls of P(p_k <= delta / K), which is at most K * delta / K = delta.
• "fixed_sequence" orders the combinations using the calibration part only (most clearly safe there first, ties strictest first; see _testing_sequence), so the order is independent of the risk part. It rejects H_{(1)}, H_{(2)}, ... in that order while p_{(j)} <= delta and stops at the first failure. Let j0 be the position of the first true null in the order. Rejecting any true null requires rejecting every hypothesis before it, in particular H_{(j0)}, which happens with probability at most delta.

Choosing, among certified combinations, the one that acts most often on the risk part does not affect either argument, which bounds the chance that any certified combination is bad. Fixed-sequence testing pays no division by K, so it certifies more (and looser) combinations when the order puts safe combinations first; it loses power when an unsafe or under-powered combination sits early in the order, which is why the order is estimated from the calibration part rather than taken from act rate alone.

Source code in src/grounded_decision/learn/calibrate.py

def learn_parameters(
    cases: Sequence[HistoricalCase],
    target: RiskTarget,
    *,
    grid: ParameterGrid | None = None,
    calibration_fraction: float = 0.3,
    holdout_fraction: float = 0.2,
    seed: int = 0,
    procedure: Procedure = "fixed_sequence",
) -> LearnedParameters:
    """Learn check settings from labelled history with a certified error rate.

    Args:
        cases: Past decisions with known outcomes. Only cases where the
            policy proposed an action belong here. Set
            ``HistoricalCase.was_acted_on`` so the learner can warn about
            selective labels.
        target: The error rate you accept and how sure the certificate must
            be.
        grid: Candidate settings; defaults to :class:`ParameterGrid`.
        calibration_fraction: Share of cases used to build calibration
            scores (and, for ``"fixed_sequence"``, the testing order).
        holdout_fraction: Share kept aside to report honest performance.
            The rest is the risk part used to certify settings.
        seed: Seed for the random split, recorded for reproducibility.
        procedure: ``"fixed_sequence"`` (default) or ``"bonferroni"``.

    Returns:
        The learned settings and their provenance. Check ``status``: if it
        is ``"no_safe_configuration"`` the settings hold every decision.

    Why both procedures are valid. Write ``delta = 1 - confidence``. Fix
    the calibration part; the calibration scores, and so the true error
    rate ``R_k`` of every combination ``k``, are then fixed numbers, and
    the risk-part cases are independent draws unrelated to them. For each
    ``k`` the null hypothesis ``H_k: R_k > max_risk`` has an exact binomial
    p-value ``p_k`` with ``P(p_k <= u) <= u`` whenever ``H_k`` is true. A
    combination is *certified* when its ``H_k`` is rejected; the chosen
    combination is a certified one, so it suffices that the chance of
    rejecting *any* true null (the family-wise error) is at most
    ``delta``.

    * ``"bonferroni"`` rejects ``H_k`` when ``p_k <= delta / K`` (``K`` the
      grid size). The family-wise error is at most the sum over true nulls
      of ``P(p_k <= delta / K)``, which is at most ``K * delta / K = delta``.
    * ``"fixed_sequence"`` orders the combinations using the calibration
      part only (most clearly safe there first, ties strictest first; see
      ``_testing_sequence``), so the order is independent of the risk
      part. It rejects ``H_{(1)}, H_{(2)}, ...`` in that order while
      ``p_{(j)} <= delta`` and stops at the first failure. Let ``j0`` be
      the position of the first true null in the order. Rejecting any true
      null requires rejecting every hypothesis before it, in particular
      ``H_{(j0)}``, which happens with probability at most ``delta``.

    Choosing, among certified combinations, the one that acts most often
    on the risk part does not affect either argument, which bounds the
    chance that *any* certified combination is bad. Fixed-sequence testing
    pays no division by ``K``, so it certifies more (and looser)
    combinations when the order puts safe combinations first; it loses
    power when an unsafe or under-powered combination sits early in the
    order, which is why the order is estimated from the calibration part
    rather than taken from act rate alone.
    """

    return learning_run(
        cases,
        target,
        grid=grid,
        calibration_fraction=calibration_fraction,
        holdout_fraction=holdout_fraction,
        seed=seed,
        procedure=procedure,
    ).learned

learning_run

learning_run(
    cases: Sequence[HistoricalCase],
    target: RiskTarget,
    *,
    grid: ParameterGrid | None = None,
    calibration_fraction: float = 0.3,
    holdout_fraction: float = 0.2,
    seed: int = 0,
    procedure: Procedure = "fixed_sequence",
) -> LearningRun

:func:learn_parameters, also reporting every combination's test.

Takes the same arguments and makes the same choice (it is the code :func:learn_parameters runs); the extra result is one :class:CandidateTest per combination of the grid, with its counts on the risk part, its exact p-value, its place in the testing order and whether it was tested and certified. Combinations the fixed sequence never reached still get their p-value, marked tested=False.

Source code in src/grounded_decision/learn/calibrate.py

def learning_run(
    cases: Sequence[HistoricalCase],
    target: RiskTarget,
    *,
    grid: ParameterGrid | None = None,
    calibration_fraction: float = 0.3,
    holdout_fraction: float = 0.2,
    seed: int = 0,
    procedure: Procedure = "fixed_sequence",
) -> LearningRun:
    """:func:`learn_parameters`, also reporting every combination's test.

    Takes the same arguments and makes the same choice (it is the code
    :func:`learn_parameters` runs); the extra result is one
    :class:`CandidateTest` per combination of the grid, with its counts on
    the risk part, its exact p-value, its place in the testing order and
    whether it was tested and certified. Combinations the fixed sequence
    never reached still get their p-value, marked ``tested=False``.
    """

    grid = grid or ParameterGrid()
    procedure = _procedure(procedure)
    if not 0.0 < calibration_fraction < 1.0 or not 0.0 <= holdout_fraction < 1.0:
        raise ValueError("fractions must be in [0, 1) and calibration_fraction > 0")
    if calibration_fraction + holdout_fraction >= 1.0:
        raise ValueError("calibration and hold-out fractions leave no risk part")

    order: list[int] = [
        int(position)
        for position in np.random.default_rng(seed).permutation(len(cases))
    ]
    calibration_count = round(calibration_fraction * len(cases))
    holdout_count = round(holdout_fraction * len(cases))
    calibration_cases = [cases[i] for i in order[:calibration_count]]
    holdout_cases = [
        cases[i] for i in order[calibration_count : calibration_count + holdout_count]
    ]
    risk_cases = [cases[i] for i in order[calibration_count + holdout_count :]]

    calibration = _calibration_scores_by_action(calibration_cases)
    risk_arrays = case_arrays(risk_cases, calibration)
    holdout_arrays = case_arrays(holdout_cases, calibration)
    significance = 1.0 - target.confidence

    warnings: list[str] = _selective_label_warnings(cases)
    # With n calibration scores the smallest possible conformal p-value is
    # 1 / (1 + n), so the check can hold a case only if 1 / (1 + n) <= alpha.
    smallest_positive_alpha = min(
        (a for a in grid.conformal_alphas if a > 0), default=None
    )
    if smallest_positive_alpha is not None:
        scores_needed = math.ceil(1.0 / smallest_positive_alpha) - 1
        thin_actions = sorted(
            action
            for action, scores in calibration.items()
            if len(scores) < scores_needed
        )
        if thin_actions:
            warnings.append(
                f"fewer than {scores_needed} calibration scores for "
                + ", ".join(thin_actions)
                + f"; the unusual-case check cannot hold them at alpha "
                f"{smallest_positive_alpha:g}"
            )
    proposed_actions = {case.proposed_action for case in cases}
    missing = sorted(proposed_actions - set(calibration))
    if missing:
        warnings.append(
            "no calibration scores for: "
            + ", ".join(missing)
            + " (the unusual-case check always passes for them)"
        )

    combinations = grid.combinations()
    if procedure == "bonferroni":
        significance_per_test = significance / grid.size
        tests = grid.size
        sequence = list(range(len(combinations)))
        minimum_needed = minimum_trials_to_certify(
            target.max_risk, significance, grid.size
        )
    else:
        significance_per_test = significance
        tests = 1
        sequence = _testing_sequence(
            combinations,
            case_arrays(calibration_cases, calibration),
            len(risk_cases),
            target,
        )
        minimum_needed = minimum_trials_to_certify(target.max_risk, significance, 1)

    best: tuple[tuple[int, int, int], int] | None = None
    certified_count = 0
    first_failure: int | None = None
    # (acts, errors, trials) of every combination on the risk part.
    counts: list[tuple[int, int, int]] = []
    for combination in combinations:
        acts = _acts_under(risk_arrays, combination)
        errors, trials = _risk_counts(acts, risk_arrays.wrong, target.measure)
        counts.append((int(np.count_nonzero(acts)), errors, trials))
    most_trials_any_configuration = max((trials for *_, trials in counts), default=0)
    p_values: dict[int, float] = {}
    tested: set[int] = set()
    certified: set[int] = set()
    for index in sequence:
        act_count, errors, trials = counts[index]
        p_values[index] = p_value_risk_below(errors, trials, target.max_risk)
        tested.add(index)
        # The exact p-value against the exact level (D69): the float p-value
        # is rounded up and the level rounded, so neither decides.
        if not certifies_risk_below(
            errors, trials, target.max_risk, target.confidence, tests
        ):
            if procedure == "fixed_sequence":
                first_failure = index
                break
            continue
        certified_count += 1
        certified.add(index)
        # Prefer more acts; break ties with fewer errors, then grid order.
        rank = (act_count, -errors, -index)
        if best is None or rank > best[0]:
            best = (rank, index)

    if best is None:
        status = "no_safe_configuration"
        combination: Combination = (
            HOLD_EVERYTHING_MARGIN,
            DEFAULT_LIMITS.operational_fk_prefactor,
            None,
            DEFAULT_LIMITS.conformal_alpha,
            DEFAULT_LIMITS.min_support,
        )
        if most_trials_any_configuration < minimum_needed:
            counted = (
                "acted-on cases"
                if target.measure is RiskMeasure.ERROR_RATE_AMONG_ACTS
                else "cases"
            )
            how = (
                f"over {grid.size} combinations"
                if procedure == "bonferroni"
                else "in a fixed sequence"
            )
            warnings.append(
                f"at most {most_trials_any_configuration} {counted} in the risk part "
                f"for any setting; certifying {target.max_risk:.1%} at "
                f"{target.confidence:.0%} {how} needs at "
                f"least {minimum_needed} even with zero errors"
            )
        elif procedure == "fixed_sequence":
            if not sequence:
                warnings.append(
                    "no combination acts often enough on the calibration part to "
                    "be certifiable, so none was tested"
                )
            elif first_failure is not None:
                warnings.append(
                    "the first combination tested (the most clearly safe one on "
                    "the calibration part) failed on the risk part, so testing "
                    "stopped: the target is too strict for this history, or there "
                    "is too little history to show it is met"
                )
    else:
        status = "certified"
        combination = combinations[best[1]]

    margin, prefactor, line, alpha, minimum_support = combination
    risk_acts = _acts_under(risk_arrays, combination)
    holdout_acts = _acts_under(holdout_arrays, combination)
    position = {index: place for place, index in enumerate(sequence)}
    candidates = tuple(
        CandidateTest(
            index=index,
            reconstruction_margin=float(candidate[0]),
            window_prefactor=float(candidate[1]),
            decision_line=None if candidate[2] is None else float(candidate[2]),
            conformal_alpha=float(candidate[3]),
            minimum_support=float(candidate[4]),
            acts=counts[index][0],
            errors=counts[index][1],
            trials=counts[index][2],
            p_value=(
                p_values[index]
                if index in p_values
                else p_value_risk_below(
                    counts[index][1], counts[index][2], target.max_risk
                )
            ),
            order=position.get(index, len(sequence) + index),
            tested=index in tested,
            certified=index in certified,
        )
        for index, candidate in enumerate(combinations)
    )
    learned = LearnedParameters(
        status=status,
        reconstruction_margin=float(margin),
        operational_fk_prefactor=float(prefactor),
        conformal_alpha=float(alpha),
        decision_line=None if line is None else float(line),
        calibration_scores=calibration,
        target=target,
        grid=grid,
        risk_split=_report(risk_acts, risk_arrays.wrong, target),
        holdout=_report(holdout_acts, holdout_arrays.wrong, target),
        configurations_tried=grid.size,
        configurations_certified=certified_count,
        minimum_risk_cases_needed=minimum_needed,
        dataset_fingerprint=dataset_fingerprint(cases),
        split_sizes={
            "calibration": len(calibration_cases),
            "risk": len(risk_cases),
            "holdout": len(holdout_cases),
        },
        seed=seed,
        warnings=tuple(warnings),
        minimum_support=float(minimum_support),
        procedure=procedure,
    )
    return LearningRun(
        learned=learned,
        candidates=candidates,
        significance_per_test=significance_per_test,
        chosen=None if best is None else best[1],
    )

predicted_acts

predicted_acts(
    arrays: CaseArrays,
    *,
    margin: float,
    prefactor: float,
    decision_line: float | None,
    alpha: float,
    minimum_support: float = 0.0,
) -> NDArray[bool_]

Which cases all four data checks would pass under these settings.

Mirrors Module.decide with Limits(reconstruction_margin=margin, operational_fk_prefactor=prefactor, conformal_alpha=alpha, min_support=minimum_support) and require_support off; the certificate and game checks depend only on the policy, not on these settings. minimum_support=0.0 (the default) switches the support check off.

Source code in src/grounded_decision/learn/calibrate.py

def predicted_acts(
    arrays: CaseArrays,
    *,
    margin: float,
    prefactor: float,
    decision_line: float | None,
    alpha: float,
    minimum_support: float = 0.0,
) -> npt.NDArray[np.bool_]:
    """Which cases all four data checks would pass under these settings.

    Mirrors ``Module.decide`` with ``Limits(reconstruction_margin=margin,
    operational_fk_prefactor=prefactor, conformal_alpha=alpha,
    min_support=minimum_support)`` and ``require_support`` off; the
    certificate and game checks depend only on the policy, not on these
    settings. ``minimum_support=0.0`` (the default) switches the support
    check off.
    """

    reconstruction = arrays.evidence_strength > 1.0 + margin
    line = arrays.own_decision_line if decision_line is None else decision_line
    threshold = np.abs(arrays.link_density - line) > _window_half_width(
        arrays.entity_count, prefactor
    )
    conformal = arrays.conformal_p_value > alpha
    support = np.isnan(arrays.support) | (arrays.support >= minimum_support)
    return reconstruction & threshold & conformal & support

binomial_cdf

binomial_cdf(
    successes: int, trials: int, probability: float
) -> float

P(X <= successes) for X ~ Binomial(trials, probability).

Computed exactly and rounded to the nearest float, so it is the same on every platform.

Source code in src/grounded_decision/learn/statistics.py

def binomial_cdf(successes: int, trials: int, probability: float) -> float:
    """``P(X <= successes)`` for ``X ~ Binomial(trials, probability)``.

    Computed exactly and rounded to the nearest float, so it is the same
    on every platform.
    """

    return _binomial_cdf(successes, trials, float(probability))

lower_confidence_bound

lower_confidence_bound(
    errors: int, trials: int, confidence: float
) -> float

One-sided Clopper–Pearson lower bound on an error rate.

With probability at least confidence the true rate is at or above the returned value. Returns 0.0 when there were no errors or trials. The value is the largest :data:BOUND_SIGNIFICANT_DIGITS-digit decimal (as a float) at which the exact upper binomial tail is at most 1 - confidence, so it is at or below the exact bound and the same on every platform.

Source code in src/grounded_decision/learn/statistics.py

def lower_confidence_bound(errors: int, trials: int, confidence: float) -> float:
    """One-sided Clopper–Pearson lower bound on an error rate.

    With probability at least ``confidence`` the true rate is at or above
    the returned value. Returns 0.0 when there were no errors or trials.
    The value is the largest :data:`BOUND_SIGNIFICANT_DIGITS`-digit
    decimal (as a float) at which the exact upper binomial tail is at
    most ``1 - confidence``, so it is at or below the exact bound and the
    same on every platform.
    """

    if trials <= 0 or errors <= 0:
        return 0.0
    return _lower_bound(min(errors, trials), trials, float(confidence))

minimum_trials_to_certify

minimum_trials_to_certify(
    max_risk: float,
    significance: float,
    configurations: int = 1,
) -> int

Fewest error-free decisions that could certify max_risk.

Even a perfect record of n decisions certifies an error rate below max_risk only if (1 - max_risk) ** n <= significance / configurations. Use this to tell a team how much labelled history it needs before learning can succeed.

Source code in src/grounded_decision/learn/statistics.py

def minimum_trials_to_certify(
    max_risk: float, significance: float, configurations: int = 1
) -> int:
    """Fewest error-free decisions that could certify ``max_risk``.

    Even a perfect record of ``n`` decisions certifies an error rate below
    ``max_risk`` only if ``(1 - max_risk) ** n <= significance /
    configurations``. Use this to tell a team how much labelled history
    it needs before learning can succeed.
    """

    if not (0.0 < max_risk < 1.0) or not (0.0 < significance < 1.0):
        raise ValueError("max_risk and significance must be in (0, 1)")
    return math.ceil(
        math.log(significance / max(1, configurations)) / math.log1p(-max_risk)
    )

p_value_risk_below

p_value_risk_below(
    errors: int, trials: int, max_risk: float
) -> float

Exact p-value for "the true error rate is at least max_risk".

A small value is evidence that the error rate is below max_risk. It is the probability of seeing errors or fewer mistakes in trials independent decisions if the error rate were exactly max_risk (the least favourable case of the hypothesis), computed exactly and rounded up to a float, so comparing it with a level never certifies more than the exact value would.

Returns 1.0 when there were no trials: nothing was observed, so nothing can be certified.

Source code in src/grounded_decision/learn/statistics.py

def p_value_risk_below(errors: int, trials: int, max_risk: float) -> float:
    """Exact p-value for "the true error rate is at least ``max_risk``".

    A small value is evidence that the error rate is *below* ``max_risk``.
    It is the probability of seeing ``errors`` or fewer mistakes in
    ``trials`` independent decisions if the error rate were exactly
    ``max_risk`` (the least favourable case of the hypothesis), computed
    exactly and rounded *up* to a float, so comparing it with a level
    never certifies more than the exact value would.

    Returns 1.0 when there were no trials: nothing was observed, so
    nothing can be certified.
    """

    if trials <= 0:
        return 1.0
    return _p_value(errors, trials, float(max_risk))

upper_confidence_bound

upper_confidence_bound(
    errors: int, trials: int, confidence: float
) -> float

One-sided Clopper–Pearson upper bound on an error rate.

With probability at least confidence the true rate is at or below the returned value. Returns 1.0 when there were no trials. The value is the smallest :data:BOUND_SIGNIFICANT_DIGITS-digit decimal (as a float) at which the exact binomial tail is at most 1 - confidence, so it is at or above the exact bound and the same on every platform.

Source code in src/grounded_decision/learn/statistics.py

def upper_confidence_bound(errors: int, trials: int, confidence: float) -> float:
    """One-sided Clopper–Pearson upper bound on an error rate.

    With probability at least ``confidence`` the true rate is at or below
    the returned value. Returns 1.0 when there were no trials. The value
    is the smallest :data:`BOUND_SIGNIFICANT_DIGITS`-digit decimal (as a
    float) at which the exact binomial tail is at most ``1 -
    confidence``, so it is at or above the exact bound and the same on
    every platform.
    """

    if trials <= 0 or errors >= trials:
        return 1.0
    return _upper_bound(max(errors, 0), trials, float(confidence))