Decision Making

online_cp.decision.UtilityFunction

Utility function bundled with its decision space.

Parameters:

Name	Type	Description	Default
fn	callable	A function (x, y, d) -> float that returns the utility of taking decision d when features are x and outcome is y.	required
decisions	sequence	The set of available decisions (the decision space D).	required

Examples:

>>> utility = UtilityFunction(lambda x, y, d: -(y - d)**2,
...                           decisions=[0.0, 0.5, 1.0, 1.5, 2.0])
>>> utility(None, 1.0, 1.0)
0.0

Source code in src/online_cp/decision.py

class UtilityFunction:
    """Utility function bundled with its decision space.

    Parameters
    ----------
    fn : callable
        A function ``(x, y, d) -> float`` that returns the utility of
        taking decision *d* when features are *x* and outcome is *y*.
    decisions : sequence
        The set of available decisions (the decision space D).

    Examples
    --------
    >>> utility = UtilityFunction(lambda x, y, d: -(y - d)**2,
    ...                           decisions=[0.0, 0.5, 1.0, 1.5, 2.0])
    >>> utility(None, 1.0, 1.0)
    0.0
    """

    def __init__(self, fn: Callable[..., float], decisions: Sequence[Any]) -> None:
        self.fn = fn
        self.decisions = list(decisions)

    def __call__(self, x: Any, y: Any, d: Any) -> float:
        return self.fn(x, y, d)

online_cp.decision.cps_expected_utilities(cpd, utility: UtilityFunction, x: Any, tau: float = 0.5) -> dict[Any, float]

Compute expected utility under a CPD for each decision (Vovk & Bendtsen 2018).

For each decision d in utility.decisions, computes:

.. math::

E_\tau[U(x, \cdot, d)] = \sum_j U(x, C_j, d) \, \Delta Q_\tau(C_j)

where :math:\Delta Q_\tau(C_j) is the probability mass at critical point :math:C_j under randomisation parameter :math:\tau.

Parameters:

Name	Type	Description	Default
cpd	ConformalPredictiveDistributionFunction	A conformal predictive distribution object with attributes Y (sorted critical points), L, and U arrays.	required
utility	UtilityFunction	Utility function with decision space.	required
x	any	Features of the test object (passed to utility).	required
tau	float	Randomisation parameter in [0, 1].	0.5

Returns:

Type	Description
dict	Mapping from each decision to its expected utility (float).

Source code in src/online_cp/decision.py

def cps_expected_utilities(
    cpd,
    utility: UtilityFunction,
    x: Any,
    tau: float = 0.5,
) -> dict[Any, float]:
    """Compute expected utility under a CPD for each decision (Vovk & Bendtsen 2018).

    For each decision *d* in ``utility.decisions``, computes:

    .. math::

        E_\\tau[U(x, \\cdot, d)] = \\sum_j U(x, C_j, d) \\, \\Delta Q_\\tau(C_j)

    where :math:`\\Delta Q_\\tau(C_j)` is the probability mass at critical
    point :math:`C_j` under randomisation parameter :math:`\\tau`.

    Parameters
    ----------
    cpd : ConformalPredictiveDistributionFunction
        A conformal predictive distribution object with attributes ``Y``
        (sorted critical points), ``L``, and ``U`` arrays.
    utility : UtilityFunction
        Utility function with decision space.
    x : any
        Features of the test object (passed to utility).
    tau : float, default 0.5
        Randomisation parameter in [0, 1].

    Returns
    -------
    dict
        Mapping from each decision to its expected utility (float).
    """
    delta_Q = _cpd_masses(cpd, tau)
    # Pre-compute Y values where mass is non-zero AND finite
    finite_mask = np.isfinite(cpd.Y)
    mask = (delta_Q != 0) & finite_mask
    Y_nz = cpd.Y[mask]
    dQ_nz = delta_Q[mask]

    result = {}
    for d in utility.decisions:
        utilities = np.array([utility(x, y, d) for y in Y_nz])
        result[d] = float(utilities @ dQ_nz)
    return result

online_cp.decision.cps_decision(cpd, utility: UtilityFunction, x: Any, tau: float = 0.5) -> Any

Select optimal decision by maximising expected utility under a label-CPD.

This is the naive approach: a single CPD is trained on raw labels, and expected utility is computed by weighting U(x, C_j, d) by the CPD masses. Practical for large or continuous decision spaces.

For the exact Vovk & Bendtsen (2018) Algorithm 1 (one CPD per decision, trained on utility-transformed labels), use :class:ConformalPredictiveDecisionMaker.

Parameters:

Name	Type	Description	Default
cpd	ConformalPredictiveDistributionFunction	Conformal predictive distribution.	required
utility	UtilityFunction	Utility function with decision space.	required
x	any	Features of the test object.	required
tau	float	Randomisation parameter.	0.5

Returns:

Type	Description
any	The optimal decision.

Source code in src/online_cp/decision.py

def cps_decision(
    cpd,
    utility: UtilityFunction,
    x: Any,
    tau: float = 0.5,
) -> Any:
    """Select optimal decision by maximising expected utility under a label-CPD.

    This is the *naive* approach: a single CPD is trained on raw labels,
    and expected utility is computed by weighting U(x, C_j, d) by the CPD
    masses. Practical for large or continuous decision spaces.

    For the exact Vovk & Bendtsen (2018) Algorithm 1 (one CPD per decision,
    trained on utility-transformed labels), use
    :class:`ConformalPredictiveDecisionMaker`.

    Parameters
    ----------
    cpd : ConformalPredictiveDistributionFunction
        Conformal predictive distribution.
    utility : UtilityFunction
        Utility function with decision space.
    x : any
        Features of the test object.
    tau : float, default 0.5
        Randomisation parameter.

    Returns
    -------
    any
        The optimal decision.
    """
    exps = cps_expected_utilities(cpd, utility, x, tau)
    return alpha_utility(exps, alpha=0.5)

online_cp.decision.venn_expected_utilities(venn_pred, utility: UtilityFunction, x: Any) -> dict[Any, NDArray[np.floating]]

Compute expected utility under each Venn hypothesis for each decision.

For each decision d and hypothesis v:

.. math::

E_{P^v}[U(x, \cdot, d)] = \sum_j P^v(\text{label}_j) \, U(x, \text{label}_j, d)

Parameters:

Name	Type	Description	Default
venn_pred	VennPrediction	Multiprobability prediction with probs matrix (|Y| × |Y|) and label_space array.	required
utility	UtilityFunction	Utility function with decision space.	required
x	any	Features of the test object (passed to utility).	required

Returns:

Type	Description
dict	Mapping from each decision to an ndarray of shape (|Y|,) giving the expected utility under each hypothesis.

Source code in src/online_cp/decision.py

def venn_expected_utilities(
    venn_pred,
    utility: UtilityFunction,
    x: Any,
) -> dict[Any, NDArray[np.floating]]:
    """Compute expected utility under each Venn hypothesis for each decision.

    For each decision *d* and hypothesis *v*:

    .. math::

        E_{P^v}[U(x, \\cdot, d)] = \\sum_j P^v(\\text{label}_j) \\, U(x, \\text{label}_j, d)

    Parameters
    ----------
    venn_pred : VennPrediction
        Multiprobability prediction with ``probs`` matrix (|Y| × |Y|)
        and ``label_space`` array.
    utility : UtilityFunction
        Utility function with decision space.
    x : any
        Features of the test object (passed to utility).

    Returns
    -------
    dict
        Mapping from each decision to an ndarray of shape (|Y|,) giving
        the expected utility under each hypothesis.
    """
    labels = venn_pred.label_space
    probs = venn_pred.probs  # shape (|Y|, |Y|)

    result = {}
    for d in utility.decisions:
        # Utility vector: U(x, label_j, d) for each label j
        u_vec = np.array([utility(x, y, d) for y in labels])
        # Expected utility under each hypothesis v: probs[v, :] @ u_vec
        result[d] = probs @ u_vec
    return result

online_cp.decision.venn_decision(venn_pred, utility: UtilityFunction, x: Any, criterion: str = 'utility', alpha: float = 0.0) -> Any

Select optimal decision under Venn multiprobability (Venn-PDMS).

Computes expected utilities under each Venn hypothesis, then applies one of two decision criterion families:

• "utility" (α-utility): score(d) = α·upper + (1−α)·lower. α=0 is maximin, α=1 is maximax, α=0.5 is midpoint.
• "regret" (α-regret): minimise the α-weighted regret. α=0 is minimax regret, α=1 is minimin regret.

Parameters:

Name	Type	Description	Default
venn_pred	VennPrediction	Multiprobability prediction.	required
utility	UtilityFunction	Utility function with decision space.	required
x	any	Features of the test object.	required
criterion	str	One of "utility" or "regret".	"utility"
alpha	float	Optimism index in [0, 1]. α=0 is pessimistic (default).	0.0

Returns:

Type	Description
any	The optimal decision.

Source code in src/online_cp/decision.py

def venn_decision(
    venn_pred,
    utility: UtilityFunction,
    x: Any,
    criterion: str = "utility",
    alpha: float = 0.0,
) -> Any:
    """Select optimal decision under Venn multiprobability (Venn-PDMS).

    Computes expected utilities under each Venn hypothesis, then applies
    one of two decision criterion families:

    - ``"utility"`` (α-utility): score(d) = α·upper + (1−α)·lower.
      α=0 is maximin, α=1 is maximax, α=0.5 is midpoint.
    - ``"regret"`` (α-regret): minimise the α-weighted regret.
      α=0 is minimax regret, α=1 is minimin regret.

    Parameters
    ----------
    venn_pred : VennPrediction
        Multiprobability prediction.
    utility : UtilityFunction
        Utility function with decision space.
    x : any
        Features of the test object.
    criterion : str, default "utility"
        One of ``"utility"`` or ``"regret"``.
    alpha : float, default 0.0
        Optimism index in [0, 1]. α=0 is pessimistic (default).

    Returns
    -------
    any
        The optimal decision.
    """
    exps = venn_expected_utilities(venn_pred, utility, x)
    return _apply_criterion(exps, criterion, alpha)

online_cp.decision.alpha_utility(expectations: dict[Any, float | NDArray | tuple], alpha: float = 0.0) -> Any

α-utility criterion: Hurwicz-weighted expected utility.

For each decision d, the score is:

score(d) = α · upper(E[U|d]) + (1 − α) · lower(E[U|d])

Special cases:

• α = 0: maximin (pessimistic — maximise worst-case utility)
• α = 1: maximax (optimistic — maximise best-case utility)
• α = 0.5: midpoint (average of best/worst case)

When expectations are scalars (point case), all α values give the same result (maximize).

Parameters:

Name	Type	Description	Default
expectations	dict	Mapping from decision to expected utility. Values can be: scalars (point), ndarrays (one per hypothesis), or tuples (lo, hi).	required
alpha	float	Optimism index (0 = pessimistic, 1 = optimistic).	0.0

Returns:

Type	Description
any	The optimal decision.

Source code in src/online_cp/decision.py

def alpha_utility(expectations: dict[Any, float | NDArray | tuple], alpha: float = 0.0) -> Any:
    """α-utility criterion: Hurwicz-weighted expected utility.

    For each decision *d*, the score is:

    ``score(d) = α · upper(E[U|d]) + (1 − α) · lower(E[U|d])``

    Special cases:

    - α = 0: maximin (pessimistic — maximise worst-case utility)
    - α = 1: maximax (optimistic — maximise best-case utility)
    - α = 0.5: midpoint (average of best/worst case)

    When expectations are scalars (point case), all α values give the
    same result (maximize).

    Parameters
    ----------
    expectations : dict
        Mapping from decision to expected utility. Values can be:
        scalars (point), ndarrays (one per hypothesis), or tuples (lo, hi).
    alpha : float, default 0.0
        Optimism index (0 = pessimistic, 1 = optimistic).

    Returns
    -------
    any
        The optimal decision.
    """
    def score(d):
        lo = _lower(expectations[d])
        hi = _upper(expectations[d])
        return alpha * hi + (1 - alpha) * lo

    return max(expectations, key=score)

online_cp.decision.alpha_regret(expectations: dict[Any, NDArray | tuple], alpha: float = 0.0) -> Any

α-regret criterion: Hurwicz-weighted regret minimisation.

For each scenario/hypothesis v, the regret of decision d is max_{d'} E_v[U(d')] - E_v[U(d)]. The score for each decision is:

score(d) = α · min_v R_v(d) + (1 − α) · max_v R_v(d)

We select the decision minimising this score.

Special cases:

• α = 0: minimax regret (pessimistic — minimise worst-case regret)
• α = 1: minimin regret (optimistic — minimise best-case regret)

Parameters:

Name	Type	Description	Default
expectations	dict	Mapping from decision to imprecise expectations (ndarray or tuple).	required
alpha	float	Optimism index (0 = pessimistic, 1 = optimistic).	0.0

Returns:

Type	Description
any	The decision minimising the α-regret score.

Source code in src/online_cp/decision.py

def alpha_regret(expectations: dict[Any, NDArray | tuple], alpha: float = 0.0) -> Any:
    """α-regret criterion: Hurwicz-weighted regret minimisation.

    For each scenario/hypothesis *v*, the regret of decision *d* is
    ``max_{d'} E_v[U(d')] - E_v[U(d)]``. The score for each decision is:

    ``score(d) = α · min_v R_v(d) + (1 − α) · max_v R_v(d)``

    We select the decision minimising this score.

    Special cases:

    - α = 0: minimax regret (pessimistic — minimise worst-case regret)
    - α = 1: minimin regret (optimistic — minimise best-case regret)

    Parameters
    ----------
    expectations : dict
        Mapping from decision to imprecise expectations (ndarray or tuple).
    alpha : float, default 0.0
        Optimism index (0 = pessimistic, 1 = optimistic).

    Returns
    -------
    any
        The decision minimising the α-regret score.
    """
    decisions = list(expectations.keys())
    cols = [_as_array(expectations[d]) for d in decisions]
    matrix = np.column_stack(cols)  # shape (n_scenarios, n_decisions)

    best_per_scenario = matrix.max(axis=1)
    regret = best_per_scenario[:, None] - matrix

    score = alpha * regret.min(axis=0) + (1 - alpha) * regret.max(axis=0)

    best_idx = int(np.argmin(score))
    return decisions[best_idx]