pydvl.value.stopping ¶

Stopping criteria for value computations.

This module provides a basic set of stopping criteria, like MaxUpdates, MaxTime, or HistoryDeviation among others. These can behave in different ways depending on the context. For example, MaxUpdates limits the number of updates to values, which depending on the algorithm may mean a different number of utility evaluations or imply other computations like solving a linear or quadratic program.

Stopping criteria are callables that are evaluated on a ValuationResult and return a Status object. They can be combined using boolean operators.

How convergence is determined¶

Most stopping criteria keep track of the convergence of each index separately but make global decisions based on the overall convergence of some fraction of all indices. For example, if we have a stopping criterion that checks whether the standard error of 90% of values is below a threshold, then methods will keep updating all indices until 90% of them have converged, irrespective of the quality of the individual estimates, and without freezing updates for indices along the way as values individually attain low standard error.

This has some practical implications, because some values do tend to converge sooner than others. For example, assume we use the criterion AbsoluteStandardError(0.02) | MaxUpdates(1000). Then values close to 0 might be marked as "converged" rather quickly because they fulfill the first criterion, say after 20 iterations, despite being poor estimates. Because other indices take much longer to have low standard error and the criterion is a global check, the "converged" ones keep being updated and end up being good estimates. In this case, this has been beneficial, but one might not wish for converged values to be updated, if one is sure that the criterion is adequate for individual values.

Semi-value methods include a parameter skip_converged that allows to skip the computation of values that have converged. The way to avoid doing this too early is to use a more stringent check, e.g. AbsoluteStandardError(1e-3) | MaxUpdates(1000). With skip_converged=True this check can still take less time than the first one, despite requiring more iterations for some indices.

Choosing a stopping criterion¶

The choice of a stopping criterion greatly depends on the algorithm and the context. A safe bet is to combine a MaxUpdates or a MaxTime with a HistoryDeviation or an AbsoluteStandardError. The former will ensure that the computation does not run for too long, while the latter will try to achieve results that are stable enough. Note however that if the threshold is too strict, one will always end up running until a maximum number of iterations or time. Also keep in mind that different values converge at different times, so you might want to use tight thresholds and skip_converged as described above for semi-values.

Example

from pydvl.value import AbsoluteStandardError, MaxUpdates, compute_banzhaf_semivalues

utility = ...  # some utility object
criterion = AbsoluteStandardError(threshold=1e-3, burn_in=32) | MaxUpdates(1000)
values = compute_banzhaf_semivalues(
    utility,
    criterion,
    skip_converged=True,  # skip values that have converged (CAREFUL!)
)

This will compute the Banzhaf semivalues for utility until either the absolute standard error is below 1e-3 or 1000 updates have been performed. The burn_in parameter is used to discard the first 32 updates from the computation of the standard error. The skip_converged parameter is used to avoid computing more marginals for indices that have converged, which is useful if AbsoluteStandardError is met before MaxUpdates for some indices.

Warning

Be careful not to reuse the same stopping criterion for different computations. The object has state and will not be reset between calls to value computation methods. If you need to reuse the same criterion, you should create a new instance.

Creating stopping criteria¶

The easiest way is to declare a function implementing the interface StoppingCriterionCallable and wrap it with make_criterion(). This creates a StoppingCriterion object that can be composed with other stopping criteria.

Alternatively, and in particular if reporting of completion is required, one can inherit from this class and implement the abstract methods _check and completion.

Combining stopping criteria¶

Objects of type StoppingCriterion can be combined with the binary operators & (and), and | (or), following the truth tables of Status. The unary operator ~ (not) is also supported. See StoppingCriterion for details on how these operations affect the behavior of the stopping criteria.

References¶

Ghorbani, A., Zou, J., 2019. Data Shapley: Equitable Valuation of Data for Machine Learning. In: Proceedings of the 36th International Conference on Machine Learning, PMLR, pp. 2242–2251. ↩
Wang, J.T. and Jia, R., 2023. Data Banzhaf: A Robust Data Valuation Framework for Machine Learning. In: Proceedings of The 26th International Conference on Artificial Intelligence and Statistics, pp. 6388-6421. ↩

StoppingCriterionCallable ¶

Bases: Protocol

Signature for a stopping criterion

StoppingCriterion ¶

StoppingCriterion(modify_result: bool = True)

Bases: ABC

A composable callable object to determine whether a computation must stop.

A StoppingCriterion is a callable taking a ValuationResult and returning a Status. It also keeps track of individual convergence of values with converged, and reports the overall completion of the computation with completion.

Instances of StoppingCriterion can be composed with the binary operators & (and), and | (or), following the truth tables of Status. The unary operator ~ (not) is also supported. These boolean operations act according to the following rules:

The results of check() are combined with the operator. See Status for the truth tables.
The results of converged are combined with the operator (returning another boolean array).
The completion method returns the min, max, or the complement to 1 of the completions of the operands, for AND, OR and NOT respectively. This is required for cases where one of the criteria does not keep track of the convergence of single values, e.g. MaxUpdates, because completion by default returns the mean of the boolean convergence array.

Subclassing¶

Subclassing this class requires implementing a check() method that returns a Status object based on a given ValuationResult. This method should update the attribute _converged, which is a boolean array indicating whether the value for each index has converged. When this does not make sense for a particular stopping criterion, completion should be overridden to provide an overall completion value, since its default implementation attempts to compute the mean of _converged.

PARAMETER	DESCRIPTION
`modify_result`	If `True` the status of the input ValuationResult is modified in place after the call. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/value/stopping.py

def __init__(self, modify_result: bool = True):
    self.modify_result = modify_result
    self._converged = np.full(0, False)

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

completion ¶

completion() -> float

Returns a value between 0 and 1 indicating the completion of the computation.

Source code in src/pydvl/value/stopping.py

def completion(self) -> float:
    """Returns a value between 0 and 1 indicating the completion of the
    computation.
    """
    if self.converged.size == 0:
        return 0.0
    return float(np.mean(self.converged).item())

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

AbsoluteStandardError ¶

AbsoluteStandardError(
    threshold: float,
    fraction: float = 1.0,
    burn_in: int = 4,
    modify_result: bool = True,
)

Bases: StoppingCriterion

Determine convergence based on the standard error of the values.

If \(s_i\) is the standard error for datum \(i\), then this criterion returns Converged if \(s_i < \epsilon\) for all \(i\) and a threshold value \(\epsilon \gt 0\).

PARAMETER	DESCRIPTION
`threshold`	A value is considered to have converged if the standard error is below this threshold. A way of choosing it is to pick some percentage of the range of the values. For Shapley values this is the difference between the maximum and minimum of the utility function (to see this substitute the maximum and minimum values of the utility into the marginal contribution formula). TYPE: `float`
`fraction`	The fraction of values that must have converged for the criterion to return Converged. TYPE: `float` DEFAULT: `1.0`
`burn_in`	The number of iterations to ignore before checking for convergence. This is required because computations typically start with zero variance, as a result of using zeros(). The default is set to an arbitrary minimum which is usually enough but may need to be increased. TYPE: `int` DEFAULT: `4`

Source code in src/pydvl/value/stopping.py

def __init__(
    self,
    threshold: float,
    fraction: float = 1.0,
    burn_in: int = 4,
    modify_result: bool = True,
):
    super().__init__(modify_result=modify_result)
    self.threshold = threshold
    self.fraction = fraction
    self.burn_in = burn_in

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

completion ¶

completion() -> float

Returns a value between 0 and 1 indicating the completion of the computation.

Source code in src/pydvl/value/stopping.py

def completion(self) -> float:
    """Returns a value between 0 and 1 indicating the completion of the
    computation.
    """
    if self.converged.size == 0:
        return 0.0
    return float(np.mean(self.converged).item())

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxChecks ¶

MaxChecks(n_checks: Optional[int], modify_result: bool = True)

Bases: StoppingCriterion

Terminate as soon as the number of checks exceeds the threshold.

A "check" is one call to the criterion.

PARAMETER	DESCRIPTION
`n_checks`	Threshold: if `None`, no _check is performed, effectively creating a (never) stopping criterion that always returns `Pending`. TYPE: `Optional[int]`

Source code in src/pydvl/value/stopping.py

def __init__(self, n_checks: Optional[int], modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    if n_checks is not None and n_checks < 1:
        raise ValueError("n_iterations must be at least 1 or None")
    self.n_checks = n_checks
    self._count = 0

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxUpdates ¶

MaxUpdates(n_updates: Optional[int], modify_result: bool = True)

Bases: StoppingCriterion

Terminate if any number of value updates exceeds or equals the given threshold.

Note

If you want to ensure that all values have been updated, you probably want MinUpdates instead.

This checks the counts field of a ValuationResult, i.e. the number of times that each index has been updated. For powerset samplers, the maximum of this number coincides with the maximum number of subsets sampled. For permutation samplers, it coincides with the number of permutations sampled.

PARAMETER	DESCRIPTION
`n_updates`	Threshold: if `None`, no _check is performed, effectively creating a (never) stopping criterion that always returns `Pending`. TYPE: `Optional[int]`

Source code in src/pydvl/value/stopping.py

def __init__(self, n_updates: Optional[int], modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    if n_updates is not None and n_updates < 1:
        raise ValueError("n_updates must be at least 1 or None")
    self.n_updates = n_updates
    self.last_max = 0

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MinUpdates ¶

MinUpdates(n_updates: Optional[int], modify_result: bool = True)

Bases: StoppingCriterion

Terminate as soon as all value updates exceed or equal the given threshold.

This checks the counts field of a ValuationResult, i.e. the number of times that each index has been updated. For powerset samplers, the minimum of this number is a lower bound for the number of subsets sampled. For permutation samplers, it lower-bounds the amount of permutations sampled.

PARAMETER	DESCRIPTION
`n_updates`	Threshold: if `None`, no _check is performed, effectively creating a (never) stopping criterion that always returns `Pending`. TYPE: `Optional[int]`

Source code in src/pydvl/value/stopping.py

def __init__(self, n_updates: Optional[int], modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    self.n_updates = n_updates
    self.last_min = 0

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxTime ¶

MaxTime(seconds: Optional[float], modify_result: bool = True)

Bases: StoppingCriterion

Terminate if the computation time exceeds the given number of seconds.

Checks the elapsed time since construction

PARAMETER	DESCRIPTION
`seconds`	Threshold: The computation is terminated if the elapsed time between object construction and a _check exceeds this value. If `None`, no _check is performed, effectively creating a (never) stopping criterion that always returns `Pending`. TYPE: `Optional[float]`

Source code in src/pydvl/value/stopping.py

def __init__(self, seconds: Optional[float], modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    self.max_seconds = seconds or np.inf
    if self.max_seconds <= 0:
        raise ValueError("Number of seconds for MaxTime must be positive or None")
    self.start = time()

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

HistoryDeviation ¶

HistoryDeviation(
    n_steps: int,
    rtol: float,
    pin_converged: bool = True,
    modify_result: bool = True,
)

Bases: StoppingCriterion

A simple check for relative distance to a previous step in the computation.

The method used by (Ghorbani and Zou, 2019)¹ computes the relative distances between the current values \(v_i^t\) and the values at the previous checkpoint \(v_i^{t-\tau}\). If the sum is below a given threshold, the computation is terminated.

\[\sum_{i=1}^n \frac{\left| v_i^t - v_i^{t-\tau} \right|}{v_i^t} < \epsilon.\]

When the denominator is zero, the summand is set to the value of \(v_i^{ t-\tau}\).

This implementation is slightly generalised to allow for different number of updates to individual indices, as happens with powerset samplers instead of permutations. Every subset of indices that is found to converge can be pinned to that state. Once all indices have converged the method has converged.

Warning

This criterion is meant for the reproduction of the results in the paper, but we do not recommend using it in practice.

PARAMETER	DESCRIPTION
`n_steps`	Checkpoint values every so many updates and use these saved values to compare. TYPE: `int`
`rtol`	Relative tolerance for convergence (\(\epsilon\) in the formula). TYPE: `float`
`pin_converged`	If `True`, once an index has converged, it is pinned TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/value/stopping.py

def __init__(
    self,
    n_steps: int,
    rtol: float,
    pin_converged: bool = True,
    modify_result: bool = True,
):
    super().__init__(modify_result=modify_result)
    if n_steps < 1:
        raise ValueError("n_steps must be at least 1")
    if rtol <= 0 or rtol >= 1:
        raise ValueError("rtol must be in (0, 1)")

    self.n_steps = n_steps
    self.rtol = rtol
    self.update_op = np.logical_or if pin_converged else np.logical_and
    self._memory = None  # type: ignore

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

completion ¶

completion() -> float

Returns a value between 0 and 1 indicating the completion of the computation.

Source code in src/pydvl/value/stopping.py

def completion(self) -> float:
    """Returns a value between 0 and 1 indicating the completion of the
    computation.
    """
    if self.converged.size == 0:
        return 0.0
    return float(np.mean(self.converged).item())

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

RankCorrelation ¶

RankCorrelation(rtol: float, burn_in: int, modify_result: bool = True)

Bases: StoppingCriterion

A check for stability of Spearman correlation between checks.

When the change in rank correlation between two successive iterations is below a given threshold, the computation is terminated. The criterion computes the Spearman correlation between two successive iterations. The Spearman correlation uses the ordering indices of the given values and correlates them. This means it focuses on the order of the elements instead of their exact values. If the order stops changing (meaning the Banzhaf semivalues estimates converge), the criterion stops the algorithm.

This criterion is used in (Wang et. al.)².

PARAMETER	DESCRIPTION
`rtol`	Relative tolerance for convergence (\(\epsilon\) in the formula) TYPE: `float`
`modify_result`	If `True`, the status of the input ValuationResult is modified in place after the call. TYPE: `bool` DEFAULT: `True`
`burn_in`	The minimum number of iterations before checking for convergence. This is required because the first correlation is meaningless. TYPE: `int`

Added in 0.9.0

Source code in src/pydvl/value/stopping.py

def __init__(
    self,
    rtol: float,
    burn_in: int,
    modify_result: bool = True,
):
    super().__init__(modify_result=modify_result)
    if rtol <= 0 or rtol >= 1:
        raise ValueError("rtol must be in (0, 1)")
    self.rtol = rtol
    self.burn_in = burn_in
    self._memory: NDArray[np.float_] | None = None
    self._corr = 0.0
    self._completion = 0.0
    self._iterations = 0

converged `property` ¶

converged: NDArray[bool_]

Returns a boolean array indicating whether the values have converged for each data point.

Inheriting classes must set the _converged attribute in their check().

RETURNS	DESCRIPTION
`NDArray[bool_]`	A boolean array indicating whether the values have converged for
`NDArray[bool_]`	each data point.

call ¶

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

Source code in src/pydvl/value/stopping.py

def __call__(self, result: ValuationResult) -> Status:
    """Calls `check()`, maybe updating the result."""
    if len(result) == 0:
        logger.warning(
            "At least one iteration finished but no results where generated. "
            "Please check that your scorer and utility return valid numbers."
        )
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

make_criterion ¶

make_criterion(
    fun: StoppingCriterionCallable,
    converged: Callable[[], NDArray[bool_]] | None = None,
    completion: Callable[[], float] | None = None,
    name: str | None = None,
) -> Type[StoppingCriterion]

Create a new StoppingCriterion from a function. Use this to enable simpler functions to be composed with bitwise operators

PARAMETER	DESCRIPTION
`fun`	The callable to wrap. TYPE: `StoppingCriterionCallable`
`converged`	A callable that returns a boolean array indicating what values have converged. TYPE: `Callable[[], NDArray[bool_]] \| None` DEFAULT: `None`
`completion`	A callable that returns a value between 0 and 1 indicating the rate of completion of the computation. If not provided, the fraction of converged values is used. TYPE: `Callable[[], float] \| None` DEFAULT: `None`
`name`	The name of the new criterion. If `None`, the `__name__` of the function is used. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Type[StoppingCriterion]`	A new subclass of StoppingCriterion.

Source code in src/pydvl/value/stopping.py

def make_criterion(
    fun: StoppingCriterionCallable,
    converged: Callable[[], NDArray[np.bool_]] | None = None,
    completion: Callable[[], float] | None = None,
    name: str | None = None,
) -> Type[StoppingCriterion]:
    """Create a new [StoppingCriterion][pydvl.value.stopping.StoppingCriterion] from a function.
    Use this to enable simpler functions to be composed with bitwise operators

    Args:
        fun: The callable to wrap.
        converged: A callable that returns a boolean array indicating what
            values have converged.
        completion: A callable that returns a value between 0 and 1 indicating
            the rate of completion of the computation. If not provided, the fraction
            of converged values is used.
        name: The name of the new criterion. If `None`, the `__name__` of
            the function is used.

    Returns:
        A new subclass of [StoppingCriterion][pydvl.value.stopping.StoppingCriterion].
    """

    class WrappedCriterion(StoppingCriterion):
        def __init__(self, modify_result: bool = True):
            super().__init__(modify_result=modify_result)
            self._name = name or getattr(fun, "__name__", "WrappedCriterion")

        def _check(self, result: ValuationResult) -> Status:
            return fun(result)

        @property
        def converged(self) -> NDArray[np.bool_]:
            if converged is None:
                return super().converged
            return converged()

        def __str__(self):
            return self._name

        def completion(self) -> float:
            if completion is None:
                return super().completion()
            return completion()

    return WrappedCriterion

pydvl.value.stopping ¶

How convergence is determined¶

Choosing a stopping criterion¶

Creating stopping criteria¶

Combining stopping criteria¶

References¶

StoppingCriterionCallable ¶

StoppingCriterion ¶

Subclassing¶

converged property ¶

completion ¶

__call__ ¶

AbsoluteStandardError ¶

converged property ¶

completion ¶

__call__ ¶

MaxChecks ¶

converged property ¶

__call__ ¶

MaxUpdates ¶

converged property ¶

__call__ ¶

MinUpdates ¶

converged property ¶

__call__ ¶

MaxTime ¶

converged property ¶

__call__ ¶

HistoryDeviation ¶

converged property ¶

completion ¶

__call__ ¶

RankCorrelation ¶

converged property ¶

__call__ ¶

make_criterion ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶

converged `property` ¶

call ¶