Merge pull request #250 from appliedAI-Initiative/feature/ghorbani-stopping-step2

Generalised stopping criteria, truncation policies and improved data structures

Author: mdbenito

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Generalised stopping criteria for valuation algorithms. Improved classes
+  `ValuationResult` and `Status` with more operations. Some minor issues fixed.
+  [PR #252](https://github.com/appliedAI-Initiative/pyDVL/pull/250)
 - Fixed a bug whereby `compute_shapley_values` would only spawn one process when
   using `n_jobs=-1` and Monte Carlo methods.
   [PR #270](https://github.com/appliedAI-Initiative/pyDVL/pull/270)

build_scripts/update_docs.py

@@ -24,9 +24,6 @@ def module_template(module_qualname: str):
    :undoc-members:
    
    ----
-   
-   Module members
-   ==============
  
 .. footbibliography::
 

docs/30-data-valuation.rst

@@ -174,7 +174,7 @@ definitions, but other methods are typically preferable.
    values = naive_loo(utility)
 
 The return value of all valuation functions is an object of type
-:class:`~pydvl.value.results.ValuationResult`. This can be iterated over,
+:class:`~pydvl.value.result.ValuationResult`. This can be iterated over,
 indexed with integers, slices and Iterables, as well as converted to a
 `pandas DataFrame <https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html>`_.
 
@@ -217,11 +217,11 @@ v_u(x_i) = \frac{1}{n} \sum_{S \subseteq D \setminus \{x_i\}}
    values = compute_shapley_values(utility, mode="combinatorial_exact")
    df = values.to_dataframe(column='value')
 
-We convert the return value to a
+We can convert the return value to a
 `pandas DataFrame <https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html>`_
 and name the column with the results as `value`. Please refer to the
 documentation in :mod:`pydvl.value.shapley` and
-:class:`~pydvl.value.results.ValuationResult` for more information.
+:class:`~pydvl.value.result.ValuationResult` for more information.
 
 Monte Carlo Combinatorial Shapley
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -240,12 +240,19 @@ same pattern:
    model = ...
    data = Dataset(...)
    utility = Utility(model, data)
-   values = compute_shapley_values(utility, mode="combinatorial_montecarlo")
+   values = compute_shapley_values(
+       utility, mode="combinatorial_montecarlo", done=MaxUpdates(1000)
+   )
    df = values.to_dataframe(column='cmc')
 
 The DataFrames returned by most Monte Carlo methods will contain approximate
 standard errors as an additional column, in this case named `cmc_stderr`.
 
+Note the usage of the object :class:`~pydvl.value.stopping.MaxUpdates` as the
+stop condition. This is an instance of a
+:class:`~pydvl.value.stopping.StoppingCriterion`. Other examples are
+:class:`~pydvl.value.stopping.MaxTime` and :class:`~pydvl.value.stopping.StandardError`.
+
 
 Owen sampling
 ^^^^^^^^^^^^^
@@ -281,6 +288,10 @@ sampling, and its variant *Antithetic Owen Sampling* in the documentation for th
 function doing the work behind the scenes:
 :func:`~pydvl.value.shapley.montecarlo.owen_sampling_shapley`.
 
+Note that in this case we do not pass a
+:class:`~pydvl.value.stopping.StoppingCriterion` to the function, but instead
+the number of iterations and the maximum number of samples to use in the
+integration.
 
 Permutation Shapley
 ^^^^^^^^^^^^^^^^^^^
@@ -309,7 +320,7 @@ efficient enough to be useful in some applications.
    data = Dataset(...)
    utility = Utility(model, data)
    values = compute_shapley_values(
-       u=utility, mode="truncated_montecarlo", n_iterations=100
+       u=utility, mode="truncated_montecarlo", done=MaxUpdates(1000)
    )
 
 

requirements-dev.txt

@@ -1,14 +1,14 @@
 black[jupyter] == 22.10.0
-isort == 5.12
+isort == 5.12.0
 jupyter
 mypy == 0.982
-nbconvert
+nbconvert>=7.2.9
 nbstripout == 0.6.1
 bump2version
-pre-commit == 2.20.0
-pytest
+pre-commit==3.0.4
+pytest==7.2.1
 pytest-cov
-pytest-docker
+pytest-docker==0.12.0
 pytest-mock
 pytest-timeout
 ray[default] >= 0.8

src/pydvl/reporting/scores.py

@@ -1,43 +1,18 @@
-from collections import OrderedDict
-from operator import itemgetter
-from typing import Dict, Iterable, Mapping, Sequence, TypeVar, Union
+from typing import Dict, Iterable, Union
 
 import numpy as np
 from numpy.typing import NDArray
 
 from pydvl.utils import Utility, maybe_progress
-from pydvl.value.results import ValuationResult
+from pydvl.value.result import ValuationResult
 
-__all__ = [
-    "sort_values",
-    "sort_values_array",
-    "sort_values_history",
-    "compute_removal_score",
-]
-
-KT = TypeVar("KT")
-VT = TypeVar("VT")
-
-
-def sort_values_array(values: np.ndarray) -> Dict[int, "NDArray"]:
-    vals = np.mean(values, axis=1)
-    return OrderedDict(sorted(enumerate(vals), key=itemgetter(1)))
-
-
-def sort_values_history(values: Mapping[KT, Sequence[VT]]) -> Dict[KT, Sequence[VT]]:
-    """Sorts a dict of sample_id: [values] by the last item in each list."""
-    return OrderedDict(sorted(values.items(), key=itemgetter(1, -1)))
-
-
-def sort_values(values: Mapping[KT, VT]) -> Dict[KT, VT]:
-    """Sorts a dict of sample_id: value_float by value."""
-    return OrderedDict(sorted(values.items(), key=itemgetter(1)))
+__all__ = ["compute_removal_score"]
 
 
 def compute_removal_score(
     u: Utility,
     values: ValuationResult,
-    percentages: Union["NDArray", Iterable[float]],
+    percentages: Union[NDArray[np.float_], Iterable[float]],
     *,
     remove_best: bool = False,
     progress: bool = False,
@@ -66,11 +41,7 @@ def compute_removal_score(
     # We sort in descending order if we want to remove the best values
     values.sort(reverse=remove_best)
 
-    for pct in maybe_progress(
-        percentages,
-        display=progress,
-        desc="Removal Scores",
-    ):
+    for pct in maybe_progress(percentages, display=progress, desc="Removal Scores"):
         n_removal = int(pct * len(u.data))
         indices = values.indices[n_removal:]
         score = u(indices)

src/pydvl/utils/caching.py

@@ -271,7 +271,7 @@ def __call__(self, *args, **kwargs) -> T:
                     ):
                         new_value = fun(*args, **kwargs)
                         new_avg, new_var = running_moments(
-                            value, variance, cast(float, new_value), int(count)
+                            value, variance, int(count), cast(float, new_value)
                         )
                         result_dict["value"] = new_avg
                         result_dict["count"] = count + 1

src/pydvl/utils/dataset.py

@@ -408,7 +408,7 @@ def from_sklearn(
         train_size: float = 0.8,
         random_state: Optional[int] = None,
         stratify_by_target: bool = False,
-        data_groups: Optional[List] = None,
+        data_groups: Optional[Sequence] = None,
     ) -> "GroupedDataset":
         """Constructs a :class:`GroupedDataset` object from an sklearn bunch as returned by the
         `load_*` functions in `sklearn toy datasets
@@ -444,7 +444,7 @@ def from_arrays(
         train_size: float = 0.8,
         random_state: Optional[int] = None,
         stratify_by_target: bool = False,
-        data_groups: Optional[List] = None,
+        data_groups: Optional[Sequence] = None,
     ) -> "Dataset":
         """.. versionadded:: 0.4.0
 

src/pydvl/utils/numeric.py

@@ -69,20 +69,22 @@ def num_samples_permutation_hoeffding(eps: float, delta: float, u_range: float)
 
 
 def random_powerset(
-    s: NDArray[T], max_subsets: Optional[int] = None, q: float = 0.5
+    s: NDArray[T], n_samples: Optional[int] = None, q: float = 0.5
 ) -> Generator[NDArray[T], None, None]:
     """Samples subsets from the power set of the argument, without
     pre-generating all subsets and in no order.
 
     See `powerset()` if you wish to deterministically generate all subsets.
 
-    To generate subsets, `len(s)` Bernoulli draws with probability `q` are drawn.
+    To generate subsets, `len(s)` Bernoulli draws with probability `q` are
+    drawn.
     The default value of `q = 0.5` provides a uniform distribution over the
     power set of `s`. Other choices can be used e.g. to implement
-    :func:`Owen sampling <pydvl.value.shapley.montecarlo.owen_sampling_shapley>`.
+    :func:`Owen sampling
+    <pydvl.value.shapley.montecarlo.owen_sampling_shapley>`.
 
     :param s: set to sample from
-    :param max_subsets: if set, stop the generator after this many steps.
+    :param n_samples: if set, stop the generator after this many steps.
         Defaults to `np.iinfo(np.int32).max`
     :param q: Sampling probability for elements. The default 0.5 yields a
         uniform distribution over the power set of s.
@@ -99,9 +101,9 @@ def random_powerset(
 
     rng = np.random.default_rng()
     total = 1
-    if max_subsets is None:
-        max_subsets = np.iinfo(np.int32).max
-    while total <= max_subsets:
+    if n_samples is None:
+        n_samples = np.iinfo(np.int32).max
+    while total <= n_samples:
         selection = rng.uniform(size=len(s)) > q
         subset = s[selection]
         yield subset
@@ -228,8 +230,8 @@ def linear_regression_analytical_derivative_d_x_d_theta(
 def running_moments(
     previous_avg: FloatOrArray,
     previous_variance: FloatOrArray,
-    new_value: FloatOrArray,
     count: IntOrArray,
+    new_value: FloatOrArray,
 ) -> Tuple:  # [FloatOrArray, FloatOrArray]:
     """Uses Welford's algorithm to calculate the running average and variance of
      a set of numbers.
@@ -248,9 +250,9 @@ def running_moments(
 
     :param previous_avg: average value at previous step
     :param previous_variance: variance at previous step
-    :param new_value: new value in the series of numbers
     :param count: number of points seen so far
-    :return: new_average, new_variance, calculated with the new number
+    :param new_value: new value in the series of numbers
+    :return: new_average, new_variance, calculated with the new count
     """
     # broadcasted operations seem not to be supported by mypy, so we ignore the type
     new_average = (new_value + count * previous_avg) / (count + 1)  # type: ignore

src/pydvl/utils/parallel/actor.py

@@ -1,10 +1,12 @@
 import abc
 import inspect
 import logging
-from typing import Any, Dict, Optional, Type, Union, cast
+from time import sleep
+from typing import Generic, List, Optional, Type, TypeVar, cast
 
-from pydvl.utils.config import ParallelConfig
-from pydvl.utils.parallel.backend import RayParallelBackend, init_parallel_backend
+from ..config import ParallelConfig
+from ..status import Status
+from .backend import RayParallelBackend, init_parallel_backend
 
 __all__ = ["RayActorWrapper", "Coordinator", "Worker"]
 
@@ -65,50 +67,45 @@ def wrapper(
                 setattr(self, name, remote_caller(name))
 
 
-class Coordinator(abc.ABC):
+Result = TypeVar("Result")  # Avoids circular import with ValuationResult
+
+
+class Coordinator(Generic[Result], abc.ABC):
     """The coordinator has two main tasks: aggregating the results of the
     workers and terminating the process once a certain accuracy or total
     number of iterations is reached.
-
-    :param progress: Whether to display a progress bar
     """
 
-    def __init__(self, *, progress: Optional[bool] = True):
-        self.progress = progress
-        # For each worker: values, stddev, num_iterations
-        self.workers_results: Dict[int, Dict[str, float]] = dict()
-        self._total_iterations = 0
-        self._is_done = False
+    _status: Status
 
-    def add_results(self, worker_id: int, results: Dict[str, Union[float, int]]):
+    def __init__(self):
+        self.worker_results: List[Result] = []
+        self._status = Status.Pending
+
+    def add_results(self, results: Result):
         """Used by workers to report their results. Stores the results directly
-        into the `worker_status` dictionary.
+        into :attr:`worker_results`
 
-        :param worker_id: id of the worker
-        :param results: results of worker calculations
+        :param results: results of worker's calculations
         """
-        self.workers_results[worker_id] = results
+        self.worker_results.append(results)
 
     # this should be a @property, but with it ray.get messes up
     def is_done(self) -> bool:
         """Used by workers to check whether to terminate their process.
 
-        :return: `True` if workers must terminate, `False` otherwise.
+        :return: ``True`` if workers must terminate, ``False`` otherwise.
         """
-        return self._is_done
+        return bool(self._status)
 
     @abc.abstractmethod
-    def get_results(self) -> Any:
+    def accumulate(self) -> Result:
         """Aggregates the results of the different workers."""
         raise NotImplementedError()
 
     @abc.abstractmethod
-    def check_done(self) -> bool:
-        """Checks whether the accuracy of the calculation or the total number
-        of iterations have crossed the set thresholds.
-
-        If so, it sets the `is_done` label to `True`.
-        """
+    def check_convergence(self) -> bool:
+        """Evaluates the convergence criteria on the aggregated results."""
         raise NotImplementedError()
 
 
@@ -117,25 +114,22 @@ class Worker(abc.ABC):
 
     def __init__(
         self,
-        coordinator: "Coordinator",
+        coordinator: Coordinator,
         worker_id: int,
         *,
-        progress: bool = False,
         update_period: int = 30,
     ):
         """A worker
 
         :param coordinator: worker results will be pushed to this coordinator
         :param worker_id: id used for reporting through maybe_progress
-        :param progress: set to True to report progress, else False
         :param update_period: interval in seconds between different updates
             to and from the coordinator
         """
         super().__init__()
         self.worker_id = worker_id
         self.coordinator = coordinator
         self.update_period = update_period
-        self.progress = progress
 
     def run(self, *args, **kwargs):
         """Runs the worker."""

src/pydvl/utils/parallel/map_reduce.py

@@ -1,5 +1,4 @@
 import inspect
-from collections.abc import Iterable
 from functools import singledispatch, update_wrapper
 from itertools import accumulate, repeat
 from typing import (
@@ -70,6 +69,7 @@ def _(v: np.ndarray, *, timeout: Optional[float] = None) -> NDArray:
     return v
 
 
+# Careful to use list as hint. The dispatch does not work with typing generics
 @_get_value.register
 def _(v: list, *, timeout: Optional[float] = None) -> List[Any]:
     return [_get_value(x, timeout=timeout) for x in v]