pydvl.valuation ¶

This module collects methods for data valuation mostly based on marginal utility computation, approximations thereof or other game-theoretic methods.

Additionally, it includes subset sampling schemes, dataset utilitites and utility learning techniques.

RawData `dataclass` ¶

RawData(x: NDArray, y: NDArray)

A view on a dataset's raw data. This is not a copy.

GroupedDataset ¶

GroupedDataset(
    x: NDArray,
    y: NDArray,
    data_groups: Sequence[int] | NDArray[int_],
    feature_names: Sequence[str] | NDArray[str_] | None = None,
    target_names: Sequence[str] | NDArray[str_] | None = None,
    data_names: Sequence[str] | NDArray[str_] | None = None,
    group_names: Sequence[str] | NDArray[str_] | None = None,
    description: str | None = None,
    **kwargs: Any,
)

Bases: Dataset

Used for calculating values of subsets of the data considered as logical units. For instance, one can group by value of a categorical feature, by bin into which a continuous feature falls, or by label.

PARAMETER	DESCRIPTION
`x`	training data TYPE: `NDArray`
`y`	labels of training data TYPE: `NDArray`
`data_groups`	Sequence of the same length as `x_train` containing a group id for each training data point. Data points with the same id will then be grouped by this object and considered as one for effects of valuation. Group ids are assumed to be zero-based consecutive integers TYPE: `Sequence[int] \| NDArray[int_]`
`feature_names`	names of the covariates' features. TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`target_names`	names of the labels or targets y TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`data_names`	names of the data points. For example, if the dataset is a time series, each entry can be a timestamp. TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`group_names`	names of the groups. If not provided, the numerical group ids from `data_groups` will be used. TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`description`	A textual description of the dataset TYPE: `str \| None` DEFAULT: `None`
`kwargs`	Additional keyword arguments to pass to the Dataset constructor. TYPE: `Any` DEFAULT: `{}`

Changed in version 0.6.0

Added group_names and forwarding of kwargs

Changed in version 0.10.0

No longer holds split data, but only x, y and group information. Added methods to retrieve indices for groups and vice versa.

Source code in src/pydvl/valuation/dataset.py

def __init__(
    self,
    x: NDArray,
    y: NDArray,
    data_groups: Sequence[int] | NDArray[np.int_],
    feature_names: Sequence[str] | NDArray[np.str_] | None = None,
    target_names: Sequence[str] | NDArray[np.str_] | None = None,
    data_names: Sequence[str] | NDArray[np.str_] | None = None,
    group_names: Sequence[str] | NDArray[np.str_] | None = None,
    description: str | None = None,
    **kwargs: Any,
):
    """Class for grouping datasets.

    Used for calculating values of subsets of the data considered as logical units.
    For instance, one can group by value of a categorical feature, by bin into which
    a continuous feature falls, or by label.

    Args:
        x: training data
        y: labels of training data
        data_groups: Sequence of the same length as `x_train` containing
            a group id for each training data point. Data points with the same
            id will then be grouped by this object and considered as one for
            effects of valuation. Group ids are assumed to be zero-based consecutive
            integers
        feature_names: names of the covariates' features.
        target_names: names of the labels or targets y
        data_names: names of the data points. For example, if the dataset is a
            time series, each entry can be a timestamp.
        group_names: names of the groups. If not provided, the numerical group ids
            from `data_groups` will be used.
        description: A textual description of the dataset
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor.

    !!! tip "Changed in version 0.6.0"
        Added `group_names` and forwarding of `kwargs`

    !!! tip "Changed in version 0.10.0"
        No longer holds split data, but only x, y and group information. Added
            methods to retrieve indices for groups and vice versa.
    """
    super().__init__(
        x=x,
        y=y,
        feature_names=feature_names,
        target_names=target_names,
        data_names=data_names,
        description=description,
        **kwargs,
    )

    if len(data_groups) != len(x):
        raise ValueError(
            f"data_groups and x must have the same length."
            f"Instead got {len(data_groups)=} and {len(x)=}"
        )

    # data index -> abstract index (group id)
    try:
        self.data_to_group: NDArray[np.int_] = np.array(data_groups, dtype=int)
    except ValueError as e:
        raise ValueError(
            "data_groups must be a mapping from integer data indices to integer group ids"
        ) from e
    # abstract index (group id) -> data index
    self.group_to_data: OrderedDict[int, list[int]] = OrderedDict(
        {k: [] for k in set(data_groups)}
    )
    for data_idx, group_idx in enumerate(self.data_to_group):
        self.group_to_data[group_idx].append(data_idx)  # type: ignore
    self._indices = np.array(list(self.group_to_data.keys()), dtype=np.int_)
    self._group_names = (
        np.array(group_names, dtype=np.str_)
        if group_names is not None
        else np.array(list(self.group_to_data.keys()), dtype=np.str_)
    )
    if len(self._group_names) != len(self.group_to_data):
        raise ValueError(
            f"The number of group names ({len(self._group_names)}) "
            f"does not match the number of groups ({len(self.group_to_data)})"
        )

n_features `property` ¶

n_features: int

Returns the number of dimensions of a sample.

indices `property` ¶

indices

Indices of the groups.

names `property` ¶

names: NDArray[str_]

Names of the groups.

feature ¶

feature(name: str) -> tuple[slice, int]

Returns a slice for the feature with the given name.

Source code in src/pydvl/valuation/dataset.py

def feature(self, name: str) -> tuple[slice, int]:
    """Returns a slice for the feature with the given name."""
    try:
        return np.index_exp[:, self.feature_names.index(name)]  # type: ignore
    except ValueError:
        raise ValueError(f"Feature {name} is not in {self.feature_names}")

data ¶

data(
    indices: int | slice | Sequence[int] | NDArray[int_] | None = None,
) -> RawData

Returns the data and labels of all samples in the given groups.

PARAMETER	DESCRIPTION
`indices`	group indices whose elements to return. If `None`, all data from all groups are returned. TYPE: `int \| slice \| Sequence[int] \| NDArray[int_] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`RawData`	Tuple of training data `x` and labels `y`.

Source code in src/pydvl/valuation/dataset.py

def data(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> RawData:
    """Returns the data and labels of all samples in the given groups.

    Args:
        indices: group indices whose elements to return. If `None`,
            all data from all groups are returned.

    Returns:
        Tuple of training data `x` and labels `y`.
    """
    return super().data(self.data_indices(indices))

data_indices ¶

data_indices(
    indices: int | slice | Sequence[int] | NDArray[int_] | None = None,
) -> NDArray[int_]

Returns the indices of the samples in the given groups.

PARAMETER	DESCRIPTION
`indices`	group indices whose elements to return. If `None`, all indices from all groups are returned. TYPE: `int \| slice \| Sequence[int] \| NDArray[int_] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[int_]`	Indices of the samples in the given groups.

Source code in src/pydvl/valuation/dataset.py

def data_indices(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> NDArray[np.int_]:
    """Returns the indices of the samples in the given groups.

    Args:
        indices: group indices whose elements to return. If `None`,
            all indices from all groups are returned.

    Returns:
        Indices of the samples in the given groups.
    """
    if indices is None:
        indices = self._indices
    if isinstance(indices, slice):
        indices = range(*indices.indices(len(self.group_to_data)))
    return np.concatenate([self.group_to_data[i] for i in indices], dtype=np.int_)  # type: ignore

logical_indices ¶

logical_indices(indices: Sequence[int] | None = None) -> NDArray[int_]

Returns the group indices for the given data indices.

PARAMETER	DESCRIPTION
`indices`	indices of the data points in the data array. If `None`, the group indices for all data points are returned. TYPE: `Sequence[int] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[int_]`	Group indices for the given data indices.

Source code in src/pydvl/valuation/dataset.py

def logical_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns the group indices for the given data indices.

    Args:
        indices: indices of the data points in the data array. If `None`,
            the group indices for all data points are returned.

    Returns:
        Group indices for the given data indices.
    """
    if indices is None:
        return self.data_to_group
    return self.data_to_group[indices]

from_sklearn `classmethod` ¶

from_sklearn(
    data: Bunch,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs,
) -> tuple[GroupedDataset, GroupedDataset]

from_sklearn(
    data: Bunch,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs,
) -> tuple[GroupedDataset, GroupedDataset]

from_sklearn(
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: dict[str, Any],
) -> tuple[GroupedDataset, GroupedDataset]

Constructs a GroupedDataset object, and an ungrouped Dataset object from a sklearn.utils.Bunch as returned by the load_* functions in scikit-learn toy datasets and groups it.

Example

>>> from sklearn.datasets import load_iris
>>> from pydvl.valuation.dataset import GroupedDataset
>>> iris = load_iris()
>>> data_groups = iris.test_data[:, 0] // 0.5
>>> train, test = GroupedDataset.from_sklearn(iris, data_groups=data_groups)

PARAMETER	DESCRIPTION
`data`	scikit-learn Bunch object. The following attributes are supported: - `data`: covariates. - `target`: target variables (labels). - `feature_names` (optional): the feature names. - `target_names` (optional): the target names. - `DESCR` (optional): a description. TYPE: `Bunch`
`train_size`	size of the training dataset. Used in `train_test_split` float values represent the fraction of the dataset to include in the training split and should be in (0,1). An integer value sets the absolute number of training samples. TYPE: `int \| float` DEFAULT: `0.8`
`random_state`	seed for train / test split. TYPE: `int \| None` DEFAULT: `None`
`stratify_by_target`	If `True`, data is split in a stratified fashion, using the target variable as labels. Read more in sklearn's user guide. TYPE: `bool` DEFAULT: `False`
`data_groups`	an array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset. TYPE: `Sequence[int] \| None` DEFAULT: `None`
`kwargs`	Additional keyword arguments to pass to the Dataset constructor. TYPE: `dict[str, Any]` DEFAULT: `{}`

RETURNS	DESCRIPTION
`tuple[GroupedDataset, GroupedDataset]`	Datasets with the selected sklearn data

Changed in version 0.10.0

Returns a tuple of two GroupedDataset objects.

Source code in src/pydvl/valuation/dataset.py

@classmethod
def from_sklearn(
    cls,
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: dict[str, Any],
) -> tuple[GroupedDataset, GroupedDataset]:
    """Constructs a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object, and an
    ungrouped [Dataset][pydvl.valuation.dataset.Dataset] object from a
    [sklearn.utils.Bunch][sklearn.utils.Bunch] as returned by the `load_*` functions in
    [scikit-learn toy datasets](https://scikit-learn.org/stable/datasets/toy_dataset.html) and groups
    it.

    ??? Example
        ```pycon
        >>> from sklearn.datasets import load_iris
        >>> from pydvl.valuation.dataset import GroupedDataset
        >>> iris = load_iris()
        >>> data_groups = iris.test_data[:, 0] // 0.5
        >>> train, test = GroupedDataset.from_sklearn(iris, data_groups=data_groups)
        ```

    Args:
        data: scikit-learn Bunch object. The following attributes are supported:
            - `data`: covariates.
            - `target`: target variables (labels).
            - `feature_names` (**optional**): the feature names.
            - `target_names` (**optional**): the target names.
            - `DESCR` (**optional**): a description.
        train_size: size of the training dataset. Used in `train_test_split`
            float values represent the fraction of the dataset to include in the
            training split and should be in (0,1). An integer value sets the
            absolute number of training samples.
        random_state: seed for train / test split.
        stratify_by_target: If `True`, data is split in a stratified
            fashion, using the target variable as labels. Read more in
            [sklearn's user guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        data_groups: an array holding the group index or name for each
            data point. The length of this array must be equal to the number of
            data points in the dataset.
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor.

    Returns:
        Datasets with the selected sklearn data

    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two [GroupedDataset][pydvl.valuation.dataset.GroupedDataset]
            objects.
    """

    return cls.from_arrays(
        X=data.data,
        y=data.target,
        train_size=train_size,
        random_state=random_state,
        stratify_by_target=stratify_by_target,
        data_groups=data_groups,
        **kwargs,
    )

from_arrays `classmethod` ¶

from_arrays(
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs,
) -> tuple[GroupedDataset, GroupedDataset]

from_arrays(
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs,
) -> tuple[GroupedDataset, GroupedDataset]

from_arrays(
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: Any,
) -> tuple[GroupedDataset, GroupedDataset]

Constructs a GroupedDataset object, and an ungrouped Dataset object from X and y numpy arrays as returned by the make_* functions in scikit-learn generated datasets.

Example

>>> from sklearn.datasets import make_classification
>>> from pydvl.valuation.dataset import GroupedDataset
>>> X, y = make_classification(
...     n_samples=100,
...     n_features=4,
...     n_informative=2,
...     n_redundant=0,
...     random_state=0,
...     shuffle=False
... )
>>> data_groups = X[:, 0] // 0.5
>>> train, test = GroupedDataset.from_arrays(X, y, data_groups=data_groups)

PARAMETER	DESCRIPTION
`X`	array of shape (n_samples, n_features) TYPE: `NDArray`
`y`	array of shape (n_samples,) TYPE: `NDArray`
`train_size`	size of the training dataset. Used in `train_test_split`. TYPE: `float` DEFAULT: `0.8`
`random_state`	seed for train / test split. TYPE: `int \| None` DEFAULT: `None`
`stratify_by_target`	If `True`, data is split in a stratified fashion, using the y variable as labels. Read more in sklearn's user guide. TYPE: `bool` DEFAULT: `False`
`data_groups`	an array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset. TYPE: `Sequence[int] \| None` DEFAULT: `None`
`kwargs`	Additional keyword arguments that will be passed to the GroupedDataset constructor. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`tuple[GroupedDataset, GroupedDataset]`	Dataset with the passed X and y arrays split across training and test sets.

New in version 0.4.0

Changed in version 0.6.0

Added kwargs to pass to the GroupedDataset constructor.

Changed in version 0.10.0

Returns a tuple of two GroupedDataset objects.

Source code in src/pydvl/valuation/dataset.py

@classmethod
def from_arrays(
    cls,
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: Any,
) -> tuple[GroupedDataset, GroupedDataset]:
    """Constructs a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object,
    and an ungrouped [Dataset][pydvl.valuation.dataset.Dataset] object from X and y
    numpy arrays as returned by the `make_*` functions in
    [scikit-learn generated datasets](https://scikit-learn.org/stable/datasets/sample_generators.html).

    ??? Example
        ```pycon
        >>> from sklearn.datasets import make_classification
        >>> from pydvl.valuation.dataset import GroupedDataset
        >>> X, y = make_classification(
        ...     n_samples=100,
        ...     n_features=4,
        ...     n_informative=2,
        ...     n_redundant=0,
        ...     random_state=0,
        ...     shuffle=False
        ... )
        >>> data_groups = X[:, 0] // 0.5
        >>> train, test = GroupedDataset.from_arrays(X, y, data_groups=data_groups)
        ```

    Args:
        X: array of shape (n_samples, n_features)
        y: array of shape (n_samples,)
        train_size: size of the training dataset. Used in `train_test_split`.
        random_state: seed for train / test split.
        stratify_by_target: If `True`, data is split in a stratified
            fashion, using the y variable as labels. Read more in
            [sklearn's user guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        data_groups: an array holding the group index or name for each data
            point. The length of this array must be equal to the number of
            data points in the dataset.
        kwargs: Additional keyword arguments that will be passed to the
            [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] constructor.

    Returns:
        Dataset with the passed X and y arrays split across training and
            test sets.

    !!! tip "New in version 0.4.0"

    !!! tip "Changed in version 0.6.0"
        Added kwargs to pass to the
            [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] constructor.

    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two
            [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] objects.
    """

    if data_groups is None:
        raise ValueError(
            "data_groups must be provided when constructing a GroupedDataset"
        )
    x_train, x_test, y_train, y_test, groups_train, groups_test = train_test_split(
        X,
        y,
        data_groups,
        train_size=train_size,
        random_state=random_state,
        stratify=y if stratify_by_target else None,
    )
    training_set = cls(x=x_train, y=y_train, data_groups=groups_train, **kwargs)
    test_set = cls(x=x_test, y=y_test, data_groups=groups_test, **kwargs)
    return training_set, test_set

from_dataset `classmethod` ¶

from_dataset(
    data: Dataset,
    data_groups: Sequence[int] | NDArray[int_],
    group_names: Sequence[str] | NDArray[str_] | None = None,
    **kwargs: Any,
) -> GroupedDataset

Creates a GroupedDataset object from a Dataset object and a mapping of data groups.

Example

>>> import numpy as np
>>> from pydvl.valuation.dataset import Dataset, GroupedDataset
>>> train, test = Dataset.from_arrays(
...     X=np.asarray([[1, 2], [3, 4], [5, 6], [7, 8]]),
...     y=np.asarray([0, 1, 0, 1]),
... )
>>> grouped_train = GroupedDataset.from_dataset(train, data_groups=[0, 0, 1, 1])

PARAMETER	DESCRIPTION
`data`	The original data. TYPE: `Dataset`
`data_groups`	An array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset. TYPE: `Sequence[int] \| NDArray[int_]`
`group_names`	Names of the groups. If not provided, the numerical group ids from `data_groups` will be used. TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`kwargs`	Additional arguments to be passed to the GroupedDataset constructor. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`GroupedDataset`	A GroupedDataset with the initial Dataset grouped by `data_groups`.

Source code in src/pydvl/valuation/dataset.py

@classmethod
def from_dataset(
    cls,
    data: Dataset,
    data_groups: Sequence[int] | NDArray[np.int_],
    group_names: Sequence[str] | NDArray[np.str_] | None = None,
    **kwargs: Any,
) -> GroupedDataset:
    """Creates a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object from a
    [Dataset][pydvl.valuation.dataset.Dataset] object and a mapping of data groups.

    ??? Example
        ```pycon
        >>> import numpy as np
        >>> from pydvl.valuation.dataset import Dataset, GroupedDataset
        >>> train, test = Dataset.from_arrays(
        ...     X=np.asarray([[1, 2], [3, 4], [5, 6], [7, 8]]),
        ...     y=np.asarray([0, 1, 0, 1]),
        ... )
        >>> grouped_train = GroupedDataset.from_dataset(train, data_groups=[0, 0, 1, 1])
        ```

    Args:
        data: The original data.
        data_groups: An array holding the group index or name for each data
            point. The length of this array must be equal to the number of
            data points in the dataset.
        group_names: Names of the groups. If not provided, the numerical group ids
            from `data_groups` will be used.
        kwargs: Additional arguments to be passed to the
            [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] constructor.

    Returns:
        A [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] with the initial
            [Dataset][pydvl.valuation.dataset.Dataset] grouped by `data_groups`.
    """
    return cls(
        x=data._x,
        y=data._y,
        data_groups=data_groups,
        feature_names=data.feature_names,
        target_names=data.target_names,
        description=data.description,
        group_names=group_names,
        **kwargs,
    )

BetaShapleyValuation ¶

BetaShapleyValuation(
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    alpha: float,
    beta: float,
    progress: bool = False,
    skip_converged: bool = False,
)

Bases: SemivalueValuation

Computes Beta-Shapley values.

PARAMETER	DESCRIPTION
`utility`	Object to compute utilities. TYPE: `UtilityBase`
`sampler`	Sampling scheme to use. TYPE: `IndexSampler`
`is_done`	Stopping criterion to use. TYPE: `StoppingCriterion`
`skip_converged`	Whether to skip converged indices. Convergence is determined by the stopping criterion's `converged` array. TYPE: `bool` DEFAULT: `False`
`alpha`	The alpha parameter of the Beta distribution. TYPE: `float`
`beta`	The beta parameter of the Beta distribution. TYPE: `float`
`progress`	Whether to show a progress bar. If a dictionary, it is passed to `tqdm` as keyword arguments, and the progress bar is displayed. TYPE: `bool` DEFAULT: `False`

Source code in src/pydvl/valuation/methods/beta_shapley.py

def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    alpha: float,
    beta: float,
    progress: bool = False,
    skip_converged: bool = False,
):
    super().__init__(
        utility, sampler, is_done, skip_converged=skip_converged, progress=progress
    )

    self.alpha = alpha
    self.beta = beta
    self.const = sp.special.beta(alpha, beta)
    self.log_const = sp.special.betaln(alpha, beta)

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

ClasswiseShapleyValuation ¶

ClasswiseShapleyValuation(
    utility: ClasswiseModelUtility,
    sampler: ClasswiseSampler,
    is_done: StoppingCriterion,
    progress: dict[str, Any] | bool = False,
    *,
    normalize_values: bool = True,
)

Bases: Valuation

Class to compute Class-wise Shapley values.

PARAMETER	DESCRIPTION
`utility`	Class-wise utility object with model and class-wise scoring function. TYPE: `ClasswiseModelUtility`
`sampler`	Class-wise sampling scheme to use. TYPE: `ClasswiseSampler`
`is_done`	Stopping criterion to use. TYPE: `StoppingCriterion`
`progress`	Whether to show a progress bar. TYPE: `dict[str, Any] \| bool` DEFAULT: `False`
`normalize_values`	Whether to normalize values after valuation. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/methods/classwise_shapley.py

def __init__(
    self,
    utility: ClasswiseModelUtility,
    sampler: ClasswiseSampler,
    is_done: StoppingCriterion,
    progress: dict[str, Any] | bool = False,
    *,
    normalize_values: bool = True,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.labels: NDArray | None = None
    if not isinstance(utility.scorer, ClasswiseSupervisedScorer):
        raise ValueError("scorer must be an instance of ClasswiseSupervisedScorer")
    self.scorer: ClasswiseSupervisedScorer = utility.scorer
    self.is_done = is_done
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    else:
        self.tqdm_args.update(progress if isinstance(progress, dict) else {})
    self.normalize_values = normalize_values

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

DataBanzhafValuation ¶

DataBanzhafValuation(
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
)

Bases: SemivalueValuation

Computes Banzhaf values.

Source code in src/pydvl/valuation/methods/semivalue.py

def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

BaggingModel ¶

Bases: Protocol

Any model with the attributes n_estimators and max_samples is considered a bagging model.

fit ¶

fit(x: NDArray, y: NDArray | None)

Fit the model to the data

PARAMETER	DESCRIPTION
`x`	Independent variables TYPE: `NDArray`
`y`	Dependent variable TYPE: `NDArray \| None`

Source code in src/pydvl/utils/types.py

def fit(self, x: NDArray, y: NDArray | None):
    """Fit the model to the data

    Args:
        x: Independent variables
        y: Dependent variable
    """
    pass

predict ¶

predict(x: NDArray) -> NDArray

Compute predictions for the input

PARAMETER	DESCRIPTION
`x`	Independent variables for which to compute predictions TYPE: `NDArray`

RETURNS	DESCRIPTION
`NDArray`	Predictions for the input

Source code in src/pydvl/utils/types.py

def predict(self, x: NDArray) -> NDArray:
    """Compute predictions for the input

    Args:
        x: Independent variables for which to compute predictions

    Returns:
        Predictions for the input
    """
    pass

Valuation ¶

Valuation()

Bases: ABC

Source code in src/pydvl/valuation/base.py

def __init__(self) -> None:
    self.result: ValuationResult | None = None

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

ValuationResult ¶

ValuationResult(
    *,
    values: Sequence[float64] | NDArray[float64],
    variances: Sequence[float64] | NDArray[float64] | None = None,
    counts: Sequence[int_] | NDArray[int_] | None = None,
    indices: Sequence[IndexT] | NDArray[IndexT] | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    algorithm: str = "",
    status: Status = Pending,
    sort: bool | None = None,
    **extra_values: Any,
)

Bases: Sequence, Iterable[ValueItem]

Objects of this class hold the results of valuation algorithms.

These include indices in the original Dataset, any data names (e.g. group names in GroupedDataset), the values themselves, and variance of the computation in the case of Monte Carlo methods. ValuationResults can be iterated over like any Sequence: iter(valuation_result) returns a generator of ValueItem in the order in which the object is sorted.

Indexing¶

Indexing can be position-based, when accessing any of the attributes values, variances, counts and indices, as well as when iterating over the object, or using the item access operator, both getter and setter. The "position" is either the original sequence in which the data was passed to the constructor, or the sequence in which the object is sorted, see below. One can retrieve the position for a given data index using the method positions().

Some methods use data indices instead. This is the case for get() and update().

Sorting¶

Results can be sorted in-place with sort(), or alternatively using python's standard sorted() and reversed() Note that sorting values affects how iterators and the object itself as Sequence behave: values[0] returns a ValueItem with the highest or lowest ranking point if this object is sorted by descending or ascending value, respectively.the methods If unsorted, values[0] returns the ValueItem at position 0, which has data index indices[0] in the Dataset.

The same applies to direct indexing of the ValuationResult: the index is positional, according to the sorting. It does not refer to the "data index". To sort according to data index, use sort() with key="index".

In order to access ValueItem objects by their data index, use get(), or use positions() to convert data indices to positions.

Converting back and forth from data indices and positions

data_indices = result.indices[result.positions(data_indices)] is a noop.

Operating on results¶

Results can be added to each other with the + operator. Means and variances are correctly updated, using the counts attribute.

Results can also be updated with new values using update(). Means and variances are updated accordingly using the Welford algorithm.

Empty objects behave in a special way, see empty().

PARAMETER	DESCRIPTION
`values`	An array of values. If omitted, defaults to an empty array or to an array of zeros if `indices` are given. TYPE: `Sequence[float64] \| NDArray[float64]`
`indices`	An optional array of indices in the original dataset. If omitted, defaults to `np.arange(len(values))`. Warning: It is common to pass the indices of a Dataset here. Attention must be paid in a parallel context to copy them to the local process. Just do `indices=np.copy(data.indices)`. TYPE: `Sequence[IndexT] \| NDArray[IndexT] \| None` DEFAULT: `None`
`variances`	An optional array of variances of the marginals from which the values are computed. TYPE: `Sequence[float64] \| NDArray[float64] \| None` DEFAULT: `None`
`counts`	An optional array with the number of updates for each value. Defaults to an array of ones. TYPE: `Sequence[int_] \| NDArray[int_] \| None` DEFAULT: `None`
`data_names`	Names for the data points. Defaults to index numbers if not set. TYPE: `Sequence[NameT] \| NDArray[NameT] \| None` DEFAULT: `None`
`algorithm`	The method used. TYPE: `str` DEFAULT: `''`
`status`	The end status of the algorithm. TYPE: `Status` DEFAULT: `Pending`
`sort`	Whether to sort the indices. Defaults to `None` for no sorting. Set to `True` for ascending order by value, `False` for descending. See above how sorting affects usage as an iterable or sequence. TYPE: `bool \| None` DEFAULT: `None`
`extra_values`	Additional values that can be passed as keyword arguments. This can contain, for example, the least core value. TYPE: `Any` DEFAULT: `{}`

RAISES	DESCRIPTION
`ValueError`	If input arrays have mismatching lengths.

Changed in 0.10.0

Changed the behaviour of the sort argument.

Source code in src/pydvl/valuation/result.py

def __init__(
    self,
    *,
    values: Sequence[np.float64] | NDArray[np.float64],
    variances: Sequence[np.float64] | NDArray[np.float64] | None = None,
    counts: Sequence[np.int_] | NDArray[np.int_] | None = None,
    indices: Sequence[IndexT] | NDArray[IndexT] | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    algorithm: str = "",
    status: Status = Status.Pending,
    sort: bool | None = None,
    **extra_values: Any,
):
    if variances is not None and len(variances) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and variances ({len(variances)}) do not match"
        )
    if data_names is not None and len(data_names) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and data_names ({len(data_names)}) do not match"
        )
    if indices is not None and len(indices) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and indices ({len(indices)}) do not match"
        )

    self._algorithm = algorithm
    self._status = Status(status)  # Just in case we are given a string
    self._values = np.asarray(values, dtype=np.float64)
    self._variances = (
        np.zeros_like(values) if variances is None else np.asarray(variances)
    )
    self._counts = (
        np.ones_like(values, dtype=int) if counts is None else np.asarray(counts)
    )
    self._sort_order = None
    self._extra_values = extra_values or {}

    # Internal indices -> data indices
    self._indices = self._create_indices_array(indices, len(self._values))
    self._names = self._create_names_array(data_names, self._indices)

    # Data indices -> Internal indices
    self._positions = {idx: pos for pos, idx in enumerate(self._indices)}

    # Sorted indices -> Internal indices
    self._sort_positions = np.arange(len(self._values), dtype=np.int_)
    if sort is not None:
        self.sort(reverse=not sort)

values `property` ¶

values: NDArray[float64]

The values, possibly sorted.

variances `property` ¶

variances: NDArray[float64]

Variances of the marginals from which values were computed, possibly sorted.

Note that this is not the variance of the value estimate, but the sample variance of the marginals used to compute it.

stderr `property` ¶

stderr: NDArray[float64]

Standard errors of the value estimates, possibly sorted.

counts `property` ¶

counts: NDArray[int_]

The raw counts, possibly sorted.

indices `property` ¶

indices: NDArray[IndexT]

The indices for the values, possibly sorted.

If the object is unsorted, then these are the same as declared at construction or np.arange(len(values)) if none were passed.

names `property` ¶

names: NDArray[NameT]

The names for the values, possibly sorted. If the object is unsorted, then these are the same as declared at construction or np.arange(len(values)) if none were passed.

sort ¶

sort(
    reverse: bool = False,
    key: Literal["value", "variance", "index", "name"] = "value",
) -> None

Sorts the indices in place by key.

Once sorted, iteration over the results, and indexing of all the properties ValuationResult.values, ValuationResult.variances, ValuationResult.stderr, ValuationResult.counts, ValuationResult.indices and ValuationResult.names will follow the same order.

PARAMETER	DESCRIPTION
`reverse`	Whether to sort in descending order by value. TYPE: `bool` DEFAULT: `False`
`key`	The key to sort by. Defaults to ValueItem.value. TYPE: `Literal['value', 'variance', 'index', 'name']` DEFAULT: `'value'`

Source code in src/pydvl/valuation/result.py

def sort(
    self,
    reverse: bool = False,
    # Need a "Comparable" type here
    key: Literal["value", "variance", "index", "name"] = "value",
) -> None:
    """Sorts the indices in place by `key`.

    Once sorted, iteration over the results, and indexing of all the
    properties
    [ValuationResult.values][pydvl.valuation.result.ValuationResult.values],
    [ValuationResult.variances][pydvl.valuation.result.ValuationResult.variances],
    [ValuationResult.stderr][pydvl.valuation.result.ValuationResult.stderr],
    [ValuationResult.counts][pydvl.valuation.result.ValuationResult.counts],
    [ValuationResult.indices][pydvl.valuation.result.ValuationResult.indices]
    and [ValuationResult.names][pydvl.valuation.result.ValuationResult.names]
    will follow the same order.

    Args:
        reverse: Whether to sort in descending order by value.
        key: The key to sort by. Defaults to
            [ValueItem.value][pydvl.valuation.result.ValueItem].
    """
    keymap = {
        "index": "_indices",
        "value": "_values",
        "variance": "_variances",
        "name": "_names",
        "stderr": "stderr",
    }
    self._sort_positions = np.argsort(getattr(self, keymap[key])).astype(int)
    if reverse:
        self._sort_positions = self._sort_positions[::-1]
    self._sort_order = not reverse

positions ¶

positions(data_indices: IndexSetT | list[IndexT]) -> IndexSetT

Return the location (indices) within the ValuationResult for the given data indices.

Sorting is taken into account. This operation is the inverse of indexing the indices property:

np.all(v.indices[v.positions(data_indices)] == data_indices) == True

Source code in src/pydvl/valuation/result.py

def positions(self, data_indices: IndexSetT | list[IndexT]) -> IndexSetT:
    """Return the location (indices) within the `ValuationResult` for the given
    data indices.

    Sorting is taken into account. This operation is the inverse of indexing the
    [indices][pydvl.valuation.result.ValuationResult.indices] property:

    ```python
    np.all(v.indices[v.positions(data_indices)] == data_indices) == True
    ```
    """
    indices = [self._positions[idx] for idx in data_indices]
    return self._sort_positions[indices]

copy ¶

copy() -> ValuationResult

Returns a copy of the object.

Source code in src/pydvl/valuation/result.py

def copy(self) -> ValuationResult:
    """Returns a copy of the object."""
    return ValuationResult(
        values=self._values.copy(),
        variances=self._variances.copy(),
        counts=self._counts.copy(),
        indices=self._indices.copy(),
        data_names=self._names.copy(),
        algorithm=self._algorithm,
        status=self._status,
        sort=self._sort_order,
        **self._extra_values,
    )

getattr ¶

__getattr__(attr: str) -> Any

Allows access to extra values as if they were properties of the instance.

Source code in src/pydvl/valuation/result.py

def __getattr__(self, attr: str) -> Any:
    """Allows access to extra values as if they were properties of the instance."""
    # This is here to avoid a RecursionError when copying or pickling the object
    if attr == "_extra_values":
        raise AttributeError()
    try:
        return self._extra_values[attr]
    except KeyError as e:
        raise AttributeError(
            f"{self.__class__.__name__} object has no attribute {attr}"
        ) from e

iter ¶

__iter__() -> Iterator[ValueItem]

Iterate over the results returning ValueItem objects. To sort in place before iteration, use sort().

Source code in src/pydvl/valuation/result.py

def __iter__(self) -> Iterator[ValueItem]:
    """Iterate over the results returning [ValueItem][pydvl.valuation.result.ValueItem] objects.
    To sort in place before iteration, use [sort()][pydvl.valuation.result.ValuationResult.sort].
    """
    for pos in self._sort_positions:
        yield ValueItem(
            self._indices[pos],
            self._names[pos],
            self._values[pos],
            self._variances[pos],
            self._counts[pos],
        )

add ¶

__add__(other: ValuationResult) -> ValuationResult

Adds two ValuationResults.

The values must have been computed with the same algorithm. An exception to this is if one argument has empty values, in which case the other argument is returned.

Warning

Abusing this will introduce numerical errors.

Means and standard errors are correctly handled. Statuses are added with bit-wise &, see Status. data_names are taken from the left summand, or if unavailable from the right one. The algorithm string is carried over if both terms have the same one or concatenated.

It is possible to add ValuationResults of different lengths, and with different or overlapping indices. The result will have the union of indices, and the values.

Warning

FIXME: Arbitrary extra_values aren't handled.

Source code in src/pydvl/valuation/result.py

def __add__(self, other: ValuationResult) -> ValuationResult:
    """Adds two ValuationResults.

    The values must have been computed with the same algorithm. An exception
    to this is if one argument has empty values, in which case the other
    argument is returned.

    !!! Warning
        Abusing this will introduce numerical errors.

    Means and standard errors are correctly handled. Statuses are added with
    bit-wise `&`, see [Status][pydvl.valuation.result.Status].
    `data_names` are taken from the left summand, or if unavailable from
    the right one. The `algorithm` string is carried over if both terms
    have the same one or concatenated.

    It is possible to add ValuationResults of different lengths, and with
    different or overlapping indices. The result will have the union of
    indices, and the values.

    !!! Warning
        FIXME: Arbitrary `extra_values` aren't handled.

    """
    # empty results
    if len(self.values) == 0:
        return other
    if len(other.values) == 0:
        return self

    self._check_compatible(other)

    indices = np.union1d(self._indices, other._indices).astype(self._indices.dtype)
    this_pos = np.searchsorted(indices, self._indices)
    other_pos = np.searchsorted(indices, other._indices)

    n: NDArray[np.int_] = np.zeros_like(indices, dtype=int)
    m: NDArray[np.int_] = np.zeros_like(indices, dtype=int)
    xn: NDArray[np.int_] = np.zeros_like(indices, dtype=float)
    xm: NDArray[np.int_] = np.zeros_like(indices, dtype=float)
    vn: NDArray[np.int_] = np.zeros_like(indices, dtype=float)
    vm: NDArray[np.int_] = np.zeros_like(indices, dtype=float)

    n[this_pos] = self._counts
    xn[this_pos] = self._values
    vn[this_pos] = self._variances
    m[other_pos] = other._counts
    xm[other_pos] = other._values
    vm[other_pos] = other._variances

    # np.maximum(1, n + m) covers case n = m = 0.
    n_m_sum = np.maximum(1, n + m)

    # Sample mean of n+m samples from two means of n and m samples
    xnm = (n * xn + m * xm) / n_m_sum

    # Sample variance of n+m samples from two sample variances of n and m samples
    vnm = (n * (vn + xn**2) + m * (vm + xm**2)) / n_m_sum - xnm**2

    if np.any(vnm < 0):
        if np.any(vnm < -1e-6):
            logger.warning(
                "Numerical error in variance computation. "
                f"Negative sample variances clipped to 0 in {vnm}"
            )
        vnm[np.where(vnm < 0)] = 0

    # Merging of names:
    # If an index has the same name in both results, it must be the same.
    # If an index has a name in one result but not the other, the name is
    # taken from the result with the name.
    if self._names.dtype != other._names.dtype:
        if np.can_cast(other._names.dtype, self._names.dtype, casting="safe"):
            logger.warning(
                f"Casting ValuationResult.names from {other._names.dtype} to {self._names.dtype}"
            )
            other._names = other._names.astype(self._names.dtype)
        else:
            raise TypeError(
                f"Cannot cast ValuationResult.names from "
                f"{other._names.dtype} to {self._names.dtype}"
            )

    both_pos = np.intersect1d(this_pos, other_pos)

    if len(both_pos) > 0:
        this_names: NDArray = np.empty_like(indices, dtype=np.str_)
        other_names: NDArray = np.empty_like(indices, dtype=np.str_)
        this_names[this_pos] = self._names
        other_names[other_pos] = other._names

        this_shared_names = np.take(this_names, both_pos)
        other_shared_names = np.take(other_names, both_pos)

        if np.any(this_shared_names != other_shared_names):
            raise ValueError("Mismatching names in ValuationResults")

    names = np.empty_like(indices, dtype=self._names.dtype)
    names[this_pos] = self._names
    names[other_pos] = other._names

    return ValuationResult(
        algorithm=self.algorithm or other.algorithm or "",
        status=self.status & other.status,
        indices=indices,
        values=xnm,
        variances=vnm,
        counts=n + m,
        data_names=names,
        # FIXME: What to do with extra_values? This is not commutative:
        # extra_values=self._extra_values.update(other._extra_values),
    )

update ¶

update(data_idx: int | IndexT, new_value: float) -> ValuationResult

Updates the result in place with a new value, using running mean and variance.

The variance computation uses Bessel's correction for sample estimates of the variance.

PARAMETER	DESCRIPTION
`data_idx`	Data index of the value to update. TYPE: `int \| IndexT`
`new_value`	New value to add to the result. TYPE: `float`

RETURNS	DESCRIPTION
`ValuationResult`	A reference to the same, modified result.

RAISES	DESCRIPTION
`IndexError`	If the index is not found.

Source code in src/pydvl/valuation/result.py

def update(self, data_idx: int | IndexT, new_value: float) -> ValuationResult:
    """Updates the result in place with a new value, using running mean
    and variance.

    The variance computation uses Bessel's correction for sample estimates of the
    variance.

    Args:
        data_idx: Data index of the value to update.
        new_value: New value to add to the result.

    Returns:
        A reference to the same, modified result.

    Raises:
        IndexError: If the index is not found.
    """
    try:
        pos = self._positions[data_idx]
    except KeyError:
        raise IndexError(f"Index {data_idx} not found in ValuationResult")
    val, var = running_moments(
        self._values[pos], self._variances[pos], self._counts[pos], new_value
    )
    self._values[pos] = val
    self._counts[pos] += 1
    self._variances[pos] = var
    return self

scale ¶

scale(factor: float, data_indices: NDArray[IndexT] | None = None)

Scales the values and variances of the result by a coefficient.

PARAMETER	DESCRIPTION
`factor`	Factor to scale by. TYPE: `float`
`data_indices`	Data indices to scale. If `None`, all values are scaled. TYPE: `NDArray[IndexT] \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/result.py

def scale(self, factor: float, data_indices: NDArray[IndexT] | None = None):
    """
    Scales the values and variances of the result by a coefficient.

    Args:
        factor: Factor to scale by.
        data_indices: Data indices to scale. If `None`, all values are scaled.
    """
    if data_indices is None:
        positions = None
    else:
        positions = [self._positions[idx] for idx in data_indices]
    self._values[positions] *= factor
    self._variances[positions] *= factor**2

get ¶

get(data_idx: Integral) -> ValueItem

Retrieves a ValueItem by data index, as opposed to sort index, like the indexing operator.

PARAMETER	DESCRIPTION
`data_idx`	Data index of the value to retrieve. TYPE: `Integral`

RAISES	DESCRIPTION
`IndexError`	If the index is not found.

Source code in src/pydvl/valuation/result.py

def get(self, data_idx: Integral) -> ValueItem:
    """Retrieves a ValueItem by data index, as opposed to sort index, like
    the indexing operator.

    Args:
        data_idx: Data index of the value to retrieve.

    Raises:
         IndexError: If the index is not found.
    """
    try:
        pos = self._positions[data_idx]
    except KeyError:
        raise IndexError(f"Index {data_idx} not found in ValuationResult")

    return ValueItem(
        self._indices[pos],
        self._names[pos],
        self._values[pos],
        self._variances[pos],
        self._counts[pos],
    )

to_dataframe ¶

to_dataframe(column: str | None = None, use_names: bool = False) -> DataFrame

Returns values as a dataframe.

PARAMETER	DESCRIPTION
`column`	Name for the column holding the data value. Defaults to the name of the algorithm used. TYPE: `str \| None` DEFAULT: `None`
`use_names`	Whether to use data names instead of indices for the DataFrame's index. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`DataFrame`	A dataframe with three columns: `name`, `name_variances` and `name_counts`, where `name` is the value of argument `column`.

Source code in src/pydvl/valuation/result.py

def to_dataframe(
    self, column: str | None = None, use_names: bool = False
) -> pd.DataFrame:
    """Returns values as a dataframe.

    Args:
        column: Name for the column holding the data value. Defaults to
            the name of the algorithm used.
        use_names: Whether to use data names instead of indices for the
            DataFrame's index.

    Returns:
        A dataframe with three columns: `name`, `name_variances` and
            `name_counts`, where `name` is the value of argument `column`.
    """
    column = column or self._algorithm
    df = pd.DataFrame(
        self._values[self._sort_positions],
        index=(
            self._names[self._sort_positions]
            if use_names
            else self._indices[self._sort_positions]
        ),
        columns=[column],
    )
    df[column + "_variances"] = self.variances[self._sort_positions]
    df[column + "_counts"] = self.counts[self._sort_positions]
    return df

from_random `classmethod` ¶

from_random(
    size: int,
    total: float | None = None,
    seed: Seed | None = None,
    **kwargs: dict[str, Any],
) -> "ValuationResult"

Creates a ValuationResult object and fills it with an array of random values from a uniform distribution in [-1,1]. The values can be made to sum up to a given total number (doing so will change their range).

PARAMETER	DESCRIPTION
`size`	Number of values to generate TYPE: `int`
`total`	If set, the values are normalized to sum to this number ("efficiency" property of Shapley values). TYPE: `float \| None` DEFAULT: `None`
`seed`	Random seed to use TYPE: `Seed \| None` DEFAULT: `None`
`kwargs`	Additional options to pass to the constructor of ValuationResult. Use to override status, names, etc. TYPE: `dict[str, Any]` DEFAULT: `{}`

RETURNS	DESCRIPTION
`'ValuationResult'`	A valuation result with its status set to
`'ValuationResult'`	Status.Converged by default.

RAISES	DESCRIPTION
`ValueError`	If `size` is less than 1.

Changed in version 0.6.0

Added parameter total. Check for zero size

Source code in src/pydvl/valuation/result.py

@classmethod
def from_random(
    cls,
    size: int,
    total: float | None = None,
    seed: Seed | None = None,
    **kwargs: dict[str, Any],
) -> "ValuationResult":
    """Creates a [ValuationResult][pydvl.valuation.result.ValuationResult] object
    and fills it with an array of random values from a uniform distribution in
    [-1,1]. The values can be made to sum up to a given total number (doing so will
    change their range).

    Args:
        size: Number of values to generate
        total: If set, the values are normalized to sum to this number
            ("efficiency" property of Shapley values).
        seed: Random seed to use
        kwargs: Additional options to pass to the constructor of
            [ValuationResult][pydvl.valuation.result.ValuationResult]. Use to override status, names, etc.

    Returns:
        A valuation result with its status set to
        [Status.Converged][pydvl.utils.status.Status] by default.

    Raises:
         ValueError: If `size` is less than 1.

    !!! tip "Changed in version 0.6.0"
        Added parameter `total`. Check for zero size
    """
    if size < 1:
        raise ValueError("Size must be a positive integer")

    rng = np.random.default_rng(seed)
    values = rng.uniform(low=-1, high=1, size=size)
    if total is not None:
        values *= total / np.sum(values)

    options = dict(values=values, status=Status.Converged, algorithm="random")
    options.update(kwargs)
    return cls(**options)  # type: ignore

empty `classmethod` ¶

empty(
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult

Creates an empty ValuationResult object.

Empty results are characterised by having an empty array of values. When another result is added to an empty one, the empty one is discarded.

PARAMETER	DESCRIPTION
`algorithm`	Name of the algorithm used to compute the values TYPE: `str` DEFAULT: `''`
`indices`	Optional sequence or array of indices. TYPE: `IndexSetT \| None` DEFAULT: `None`
`data_names`	Optional sequences or array of names for the data points. Defaults to index numbers if not set. TYPE: `Sequence[NameT] \| NDArray[NameT] \| None` DEFAULT: `None`
`n_samples`	Number of valuation result entries. TYPE: `int` DEFAULT: `0`

RETURNS	DESCRIPTION
`ValuationResult`	Object with the results.

Source code in src/pydvl/valuation/result.py

@classmethod
def empty(
    cls,
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.valuation.result.ValuationResult] object.

    Empty results are characterised by having an empty array of values. When
    another result is added to an empty one, the empty one is discarded.

    Args:
        algorithm: Name of the algorithm used to compute the values
        indices: Optional sequence or array of indices.
        data_names: Optional sequences or array of names for the data points.
            Defaults to index numbers if not set.
        n_samples: Number of valuation result entries.

    Returns:
        Object with the results.
    """
    if indices is not None or data_names is not None or n_samples != 0:
        return cls.zeros(
            algorithm=algorithm,
            indices=indices,
            data_names=data_names,
            n_samples=n_samples,
        )
    return cls(algorithm=algorithm, status=Status.Pending, values=np.array([]))

zeros `classmethod` ¶

zeros(
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult

Creates an empty ValuationResult object.

Empty results are characterised by having an empty array of values. When another result is added to an empty one, the empty one is ignored.

PARAMETER	DESCRIPTION
`algorithm`	Name of the algorithm used to compute the values TYPE: `str` DEFAULT: `''`
`indices`	Data indices to use. A copy will be made. If not given, the indices will be set to the range `[0, n_samples)`. TYPE: `IndexSetT \| None` DEFAULT: `None`
`data_names`	Data names to use. A copy will be made. If not given, the names will be set to the string representation of the indices. TYPE: `Sequence[NameT] \| NDArray[NameT] \| None` DEFAULT: `None`
`n_samples`	Number of data points whose values are computed. If not given, the length of `indices` will be used. TYPE: `int` DEFAULT: `0`

RETURNS	DESCRIPTION
`ValuationResult`	Object with the results.

Source code in src/pydvl/valuation/result.py

@classmethod
def zeros(
    cls,
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.valuation.result.ValuationResult] object.

    Empty results are characterised by having an empty array of values. When
    another result is added to an empty one, the empty one is ignored.

    Args:
        algorithm: Name of the algorithm used to compute the values
        indices: Data indices to use. A copy will be made. If not given,
            the indices will be set to the range `[0, n_samples)`.
        data_names: Data names to use. A copy will be made. If not given,
            the names will be set to the string representation of the indices.
        n_samples: Number of data points whose values are computed. If
            not given, the length of `indices` will be used.

    Returns:
        Object with the results.
    """
    indices = cls._create_indices_array(indices, n_samples)
    data_names = cls._create_names_array(data_names, indices)

    return cls(
        algorithm=algorithm,
        status=Status.Pending,
        indices=indices,
        data_names=data_names,
        values=np.zeros(len(indices)),
        variances=np.zeros(len(indices)),
        counts=np.zeros(len(indices), dtype=np.int_),
    )

DataOOBValuation ¶

DataOOBValuation(model: BaggingModel, score: PointwiseScore | None = None)

Bases: Valuation

Computes Data Out-Of-Bag values.

This class implements the method described in (Kwon and Zou, 2023)¹.

PARAMETER	DESCRIPTION
`model`	A fitted bagging model. Bagging models in sklearn include [[BaggingClassifier]], [[BaggingRegressor]], [[IsolationForest]], RandomForest, ExtraTrees, or any model which defines an attribute `estimators_` and uses bootstrapped subsamples to compute predictions. TYPE: `BaggingModel`
`score`	A callable for point-wise comparison of true values with the predictions. If `None`, uses point-wise accuracy for classifiers and negative $l_2$ distance for regressors. TYPE: `PointwiseScore \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/methods/data_oob.py

def __init__(
    self,
    model: BaggingModel,
    score: PointwiseScore | None = None,
):
    super().__init__()
    self.model = model
    self.score = score

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

fit ¶

fit(data: Dataset) -> Self

Compute the Data-OOB values.

This requires the bagging model passed upon construction to be fitted.

PARAMETER	DESCRIPTION
`data`	Data for which to compute values TYPE: `Dataset`

RETURNS	DESCRIPTION
`Self`	The fitted object.

Source code in src/pydvl/valuation/methods/data_oob.py

def fit(self, data: Dataset) -> Self:
    """Compute the Data-OOB values.

    This requires the bagging model passed upon construction to be fitted.

    Args:
        data: Data for which to compute values

    Returns:
        The fitted object.
    """
    # TODO: automate str representation for all Valuations
    algorithm_name = f"Data-OOB-{str(self.model)}"
    self.result = ValuationResult.empty(
        algorithm=algorithm_name,
        indices=data.indices,
        data_names=data.names,
    )

    check_is_fitted(
        self.model,
        msg="The bagging model has to be fitted before calling the valuation method.",
    )

    # This should always be present after fitting
    try:
        estimators = self.model.estimators_  # type: ignore
    except AttributeError:
        raise ValueError(
            "The model has to be an sklearn-compatible bagging model, including "
            "BaggingClassifier, BaggingRegressor, IsolationForest, RandomForest*, "
            "and ExtraTrees*"
        )

    if self.score is None:
        self.score = (
            point_wise_accuracy if is_classifier(self.model) else neg_l2_distance
        )

    if hasattr(self.model, "estimators_samples_"):  # Bagging(Classifier|Regressor)
        unsampled_indices = [
            np.setxor1d(data.indices, np.unique(sampled))
            for sampled in self.model.estimators_samples_
        ]
    else:  # RandomForest*, ExtraTrees*, IsolationForest
        n_samples_bootstrap = _get_n_samples_bootstrap(
            len(data), self.model.max_samples
        )
        unsampled_indices = [
            _generate_unsampled_indices(
                est.random_state, len(data.indices), n_samples_bootstrap
            )
            for est in estimators
        ]

    for est, oob_indices in zip(estimators, unsampled_indices):
        subset = data[oob_indices].data()
        score_array = self.score(y_true=subset.y, y_pred=est.predict(subset.x))
        self.result += ValuationResult(
            algorithm=algorithm_name,
            indices=oob_indices,
            names=data[oob_indices].names,
            values=score_array,
            counts=np.ones_like(score_array, dtype=data.indices.dtype),
        )
    return self

DeltaShapleyValuation ¶

DeltaShapleyValuation(
    utility: UtilityBase,
    is_done: StoppingCriterion,
    lower_bound: int,
    upper_bound: int,
    seed: Seed | None = None,
    skip_converged: bool = False,
    progress: bool = False,
)

Bases: SemivalueValuation

Computes $\delta$-Shapley values.

$\delta$-Shapley does not accept custom samplers. Instead, it uses a StratifiedSampler with a lower and upper bound on the size of the sets to sample from.

PARAMETER	DESCRIPTION
`utility`	Object to compute utilities. TYPE: `UtilityBase`
`is_done`	Stopping criterion to use. TYPE: `StoppingCriterion`
`lower_bound`	The lower bound of the size of the subsets to sample from. TYPE: `int`
`upper_bound`	The upper bound of the size of the subsets to sample from. TYPE: `int`
`seed`	The seed for the random number generator used by the sampler. TYPE: `Seed \| None` DEFAULT: `None`
`progress`	Whether to show a progress bar. If a dictionary, it is passed to `tqdm` as keyword arguments, and the progress bar is displayed. TYPE: `bool` DEFAULT: `False`
`skip_converged`	Whether to skip converged indices, as determined by the stopping criterion's `converged` array. TYPE: `bool` DEFAULT: `False`

Source code in src/pydvl/valuation/methods/delta_shapley.py

def __init__(
    self,
    utility: UtilityBase,
    is_done: StoppingCriterion,
    lower_bound: int,
    upper_bound: int,
    seed: Seed | None = None,
    skip_converged: bool = False,
    progress: bool = False,
):
    sampler = StratifiedSampler(
        sample_sizes=ConstantSampleSize(
            1, lower_bound=lower_bound, upper_bound=upper_bound
        ),
        sample_sizes_iteration=RandomSizeIteration,
        index_iteration=RandomIndexIteration,
        seed=seed,
    )
    self.lower_bound = lower_bound
    self.upper_bound = upper_bound
    super().__init__(
        utility, sampler, is_done, progress=progress, skip_converged=skip_converged
    )

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

GroupTestingShapleyValuation ¶

GroupTestingShapleyValuation(
    utility: UtilityBase,
    n_samples: int,
    epsilon: float,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
)

Bases: Valuation

Class to calculate the group-testing approximation to shapley values.

See Data valuation for an overview.

Warning

This method is very inefficient. Potential improvements to the implementation notwithstanding, convergence seems to be very slow (in terms of evaluations of the utility required). We recommend other Monte Carlo methods instead.

PARAMETER	DESCRIPTION
`utility`	Utility object with model, data and scoring function. TYPE: `UtilityBase`
`n_samples`	The number of samples to use. A sample size with theoretical guarantees can be computed using compute_n_samples(). TYPE: `int`
`epsilon`	The error tolerance. TYPE: `float`
`solver_options`	Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see this tutorial. For additional options cvxpy's documentation. TYPE: `dict \| None` DEFAULT: `None`
`progress`	Whether to show a progress bar during the construction of the group-testing problem. TYPE: `bool` DEFAULT: `True`
`seed`	Seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`
`batch_size`	The number of samples to draw in each batch. Can be used to reduce parallelization overhead for fast utilities. Defaults to 1. TYPE: `int` DEFAULT: `1`

Source code in src/pydvl/valuation/methods/gt_shapley.py

def __init__(
    self,
    utility: UtilityBase,
    n_samples: int,
    epsilon: float,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__()

    self._utility = utility
    self._n_samples = n_samples
    self._solver_options = solver_options
    self._progress = progress
    self._sampler = StratifiedSampler(
        index_iteration=NoIndexIteration,
        sample_sizes=GroupTestingSampleSize(),
        sample_sizes_iteration=RandomSizeIteration,
        batch_size=batch_size,
        seed=seed,
    )
    self._epsilon = epsilon

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

fit ¶

fit(data: Dataset) -> Self

Calculate the group-testing valuation on a dataset.

This method has to be called before calling values().

Calculating the least core valuation is a computationally expensive task that can be parallelized. To do so, call the fit() method inside a joblib.parallel_config context manager as follows:

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)

Source code in src/pydvl/valuation/methods/gt_shapley.py

def fit(self, data: Dataset) -> Self:
    """Calculate the group-testing valuation on a dataset.

    This method has to be called before calling `values()`.

    Calculating the least core valuation is a computationally expensive task that
    can be parallelized. To do so, call the `fit()` method inside a
    `joblib.parallel_config` context manager as follows:

    ```python
    from joblib import parallel_config

    with parallel_config(n_jobs=4):
        valuation.fit(data)
    ```

    """
    self._utility = self._utility.with_dataset(data)

    problem = create_group_testing_problem(
        utility=self._utility,
        sampler=self._sampler,
        n_samples=self._n_samples,
        progress=self._progress,
        epsilon=self._epsilon,
    )

    solution = solve_group_testing_problem(
        problem=problem,
        solver_options=self._solver_options,
        algorithm_name=self.algorithm_name,
        data_names=data.names,
    )

    self.result = solution
    return self

Status ¶

Bases: Enum

Status of a computation.

Statuses can be combined using bitwise or (|) and bitwise and (&) to get the status of a combined computation. For example, if we have two computations, one that has converged and one that has failed, then the combined status is Status.Converged | Status.Failed == Status.Converged, but Status.Converged & Status.Failed == Status.Failed.

OR¶

The result of bitwise or-ing two valuation statuses with | is given by the following table:

	P	C	F
P	P	C	P
C	C	C	C
F	P	C	F

where P = Pending, C = Converged, F = Failed.

AND¶

The result of bitwise and-ing two valuation statuses with & is given by the following table:

	P	C	F
P	P	P	F
C	P	C	F
F	F	F	F

where P = Pending, C = Converged, F = Failed.

NOT¶

The result of bitwise negation of a Status with ~ is Failed if the status is Converged, or Converged otherwise:

~P == C, ~C == F, ~F == C

Boolean casting¶

A Status evaluates to True iff it's Converged or Failed:

bool(Status.Pending) == False
bool(Status.Converged) == True
bool(Status.Failed) == True

Warning

These truth values are inconsistent with the usual boolean operations. In particular the XOR of two instances of Status is not the same as the XOR of their boolean values.

KNNShapleyValuation ¶

KNNShapleyValuation(
    model: KNeighborsClassifier,
    test_data: Dataset,
    progress: bool = True,
    clone_before_fit: bool = True,
)

Bases: Valuation

Computes exact Shapley values for a KNN classifier.

This implements the method described in (Jia, R. et al., 2019)¹. It exploits the local structure of K-Nearest Neighbours to reduce the number of calls to the utility function to a constant number per index, thus reducing computation time to $O(n)$.

PARAMETER	DESCRIPTION
`model`	KNeighborsClassifier model to use for valuation TYPE: `KNeighborsClassifier`
`test_data`	Dataset containing test data for valuation TYPE: `Dataset`
`progress`	Whether to display a progress bar. TYPE: `bool` DEFAULT: `True`
`clone_before_fit`	Whether to clone the model before fitting. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/methods/knn_shapley.py

def __init__(
    self,
    model: KNeighborsClassifier,
    test_data: Dataset,
    progress: bool = True,
    clone_before_fit: bool = True,
):
    super().__init__()
    if not isinstance(model, KNeighborsClassifier):
        raise TypeError("KNN Shapley requires a K-Nearest Neighbours model")
    self.model = model
    self.test_data = test_data
    self.progress = progress
    self.clone_before_fit = clone_before_fit

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

fit ¶

fit(data: Dataset) -> Self

Calculate exact shapley values for a KNN model on a dataset.

This fit method bypasses direct evaluations of the utility function and calculates the Shapley values directly.

In contrast to other data valuation models, the runtime increases linearly with the size of the dataset.

Calculating the KNN valuation is a computationally expensive task that can be parallelized. To do so, call the fit() method inside a joblib.parallel_config context manager as follows:

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)

Source code in src/pydvl/valuation/methods/knn_shapley.py

def fit(self, data: Dataset) -> Self:
    """Calculate exact shapley values for a KNN model on a dataset.

    This fit method bypasses direct evaluations of the utility function and
    calculates the Shapley values directly.

    In contrast to other data valuation models, the runtime increases linearly
    with the size of the dataset.

    Calculating the KNN valuation is a computationally expensive task that
    can be parallelized. To do so, call the `fit()` method inside a
    `joblib.parallel_config` context manager as follows:

    ```python
    from joblib import parallel_config

    with parallel_config(n_jobs=4):
        valuation.fit(data)
    ```
    """

    if isinstance(data, GroupedDataset):
        raise TypeError("GroupedDataset is not supported by KNNShapleyValuation")

    x_train, y_train = data.data()
    if self.clone_before_fit:
        self.model = cast(KNeighborsClassifier, clone(self.model))
    self.model.fit(x_train, y_train)

    n_test = len(self.test_data)

    _, n_jobs = get_active_backend()
    n_jobs = n_jobs or 1  # Handle None if outside a joblib context
    batch_size = (n_test // n_jobs) + (1 if n_test % n_jobs else 0)
    x_test, y_test = self.test_data.data()
    batches = zip(chunked(x_test, batch_size), chunked(y_test, batch_size))

    process = delayed(self._compute_values_for_test_points)
    with Parallel(return_as="generator_unordered") as parallel:
        results = parallel(
            process(self.model, x_test, y_test, y_train)
            for x_test, y_test in batches
        )
        values = np.zeros(len(data))
        # FIXME: this progress bar won't add much since we have n_jobs batches and
        #  they will all take about the same time
        for res in tqdm(results, total=n_jobs, disable=not self.progress):
            values += res
        values /= n_test

    self.result = ValuationResult(
        algorithm="knn_shapley",
        status=Status.Converged,
        values=values,
        data_names=data.names,
    )

    return self

LeastCoreValuation ¶

LeastCoreValuation(
    utility: UtilityBase,
    sampler: PowersetSampler,
    n_samples: int | None = None,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
)

Bases: Valuation

Umbrella class to calculate least-core values with multiple sampling methods.

See Data valuation for an overview.

Different samplers correspond to different least-core methods from the literature. For those, we provide convenience subclasses of LeastCoreValuation. See

Other samplers allow you to create your own method and might yield computational gains over a standard Monte Carlo method.

PARAMETER	DESCRIPTION
`utility`	Utility object with model, data and scoring function. TYPE: `UtilityBase`
`sampler`	The sampler to use for the valuation. TYPE: `PowersetSampler`
`n_samples`	The number of samples to use for the valuation. If None, it will be set to the sample limit of the chosen sampler (for finite samplers) or `1000 * len(data)` (for infinite samplers). TYPE: `int \| None` DEFAULT: `None`
`non_negative_subsidy`	If True, the least core subsidy $e$ is constrained to be non-negative. TYPE: `bool` DEFAULT: `False`
`solver_options`	Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here. TYPE: `dict \| None` DEFAULT: `None`
`progress`	Whether to show a progress bar during the construction of the least-core problem. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/methods/least_core.py

def __init__(
    self,
    utility: UtilityBase,
    sampler: PowersetSampler,
    n_samples: int | None = None,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
):
    super().__init__()

    _check_sampler(sampler)
    self._utility = utility
    self._sampler = sampler
    self._non_negative_subsidy = non_negative_subsidy
    self._solver_options = solver_options
    self._n_samples = n_samples
    self._progress = progress

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

fit ¶

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

This method has to be called before calling values().

Calculating the least core valuation is a computationally expensive task that can be parallelized. To do so, call the fit() method inside a joblib.parallel_config context manager as follows:

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)

Source code in src/pydvl/valuation/methods/least_core.py

def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

    This method has to be called before calling `values()`.

    Calculating the least core valuation is a computationally expensive task that
    can be parallelized. To do so, call the `fit()` method inside a
    `joblib.parallel_config` context manager as follows:

    ```python
    from joblib import parallel_config

    with parallel_config(n_jobs=4):
        valuation.fit(data)
    ```

    """
    self._utility = self._utility.with_dataset(data)
    if self._n_samples is None:
        self._n_samples = _get_default_n_samples(
            sampler=self._sampler, indices=data.indices
        )

    algorithm = str(self._sampler)

    problem = create_least_core_problem(
        u=self._utility,
        sampler=self._sampler,
        n_samples=self._n_samples,
        progress=self._progress,
    )

    solution = lc_solve_problem(
        problem=problem,
        u=self._utility,
        algorithm=algorithm,
        non_negative_subsidy=self._non_negative_subsidy,
        solver_options=self._solver_options,
    )

    self.result = solution
    return self

ExactLeastCoreValuation ¶

ExactLeastCoreValuation(
    utility: UtilityBase,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    batch_size: int = 1,
)

Bases: LeastCoreValuation

Class to calculate exact least-core values.

Equivalent to constructing a LeastCoreValuation with a DeterministicUniformSampler and n_samples=None.

The definition of the exact least-core valuation is:

\[ egin{array}{lll} ext{minimize} & \displaystyle{e} & \ ext{subject to} & \displaystyle\sum_{i\in N} x_{i} = v(N) & \ & \displaystyle\sum_{i\in S} x_{i} + e \geq v(S) &, orall S \subseteq N \ \end{array} \]

Where $N = \{1, 2, \dots, n\}$ are the training set's indices.

PARAMETER	DESCRIPTION
`utility`	Utility object with model, data and scoring function. TYPE: `UtilityBase`
`non_negative_subsidy`	If True, the least core subsidy $e$ is constrained to be non-negative. TYPE: `bool` DEFAULT: `False`
`solver_options`	Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here. TYPE: `dict \| None` DEFAULT: `None`
`progress`	Whether to show a progress bar during the construction of the least-core problem. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/methods/least_core.py

def __init__(
    self,
    utility: UtilityBase,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    batch_size: int = 1,
):
    super().__init__(
        utility=utility,
        sampler=DeterministicUniformSampler(
            index_iteration=FiniteNoIndexIteration, batch_size=batch_size
        ),
        n_samples=None,
        non_negative_subsidy=non_negative_subsidy,
        solver_options=solver_options,
        progress=progress,
    )

fit ¶

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

This method has to be called before calling values().

Calculating the least core valuation is a computationally expensive task that can be parallelized. To do so, call the fit() method inside a joblib.parallel_config context manager as follows:

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)

Source code in src/pydvl/valuation/methods/least_core.py

def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

    This method has to be called before calling `values()`.

    Calculating the least core valuation is a computationally expensive task that
    can be parallelized. To do so, call the `fit()` method inside a
    `joblib.parallel_config` context manager as follows:

    ```python
    from joblib import parallel_config

    with parallel_config(n_jobs=4):
        valuation.fit(data)
    ```

    """
    self._utility = self._utility.with_dataset(data)
    if self._n_samples is None:
        self._n_samples = _get_default_n_samples(
            sampler=self._sampler, indices=data.indices
        )

    algorithm = str(self._sampler)

    problem = create_least_core_problem(
        u=self._utility,
        sampler=self._sampler,
        n_samples=self._n_samples,
        progress=self._progress,
    )

    solution = lc_solve_problem(
        problem=problem,
        u=self._utility,
        algorithm=algorithm,
        non_negative_subsidy=self._non_negative_subsidy,
        solver_options=self._solver_options,
    )

    self.result = solution
    return self

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

MonteCarloLeastCoreValuation ¶

MonteCarloLeastCoreValuation(
    utility: UtilityBase,
    n_samples: int,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
)

Bases: LeastCoreValuation

Class to calculate exact least-core values.

Equivalent to calling LeastCoreValuation with a UniformSampler.

The definition of the Monte Carlo least-core valuation is:

\[ egin{array}{lll} ext{minimize} & \displaystyle{e} & \ ext{subject to} & \displaystyle\sum_{i\in N} x_{i} = v(N) & \ & \displaystyle\sum_{i\in S} x_{i} + e \geq v(S) & , orall S \in \{S_1, S_2, \dots, S_m \overset{\mathrm{iid}}{\sim} U(2^N) \} \end{array} \]

Where:

$U(2^N)$ is the uniform distribution over the powerset of $N$.
$m$ is the number of subsets that will be sampled and whose utility will be computed and used to compute the data values.

PARAMETER	DESCRIPTION
`utility`	Utility object with model, data and scoring function. TYPE: `UtilityBase`
`n_samples`	The number of samples to use for the valuation. If None, it will be set to `1000 * len(data)`. TYPE: `int`
`non_negative_subsidy`	If True, the least core subsidy $e$ is constrained to be non-negative. TYPE: `bool` DEFAULT: `False`
`solver_options`	Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here. TYPE: `dict \| None` DEFAULT: `None`
`progress`	Whether to show a progress bar during the construction of the least-core problem. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/methods/least_core.py

def __init__(
    self,
    utility: UtilityBase,
    n_samples: int,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__(
        utility=utility,
        sampler=UniformSampler(
            index_iteration=NoIndexIteration, seed=seed, batch_size=batch_size
        ),
        n_samples=n_samples,
        non_negative_subsidy=non_negative_subsidy,
        solver_options=solver_options,
        progress=progress,
    )

fit ¶

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

This method has to be called before calling values().

Calculating the least core valuation is a computationally expensive task that can be parallelized. To do so, call the fit() method inside a joblib.parallel_config context manager as follows:

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)

Source code in src/pydvl/valuation/methods/least_core.py

def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

    This method has to be called before calling `values()`.

    Calculating the least core valuation is a computationally expensive task that
    can be parallelized. To do so, call the `fit()` method inside a
    `joblib.parallel_config` context manager as follows:

    ```python
    from joblib import parallel_config

    with parallel_config(n_jobs=4):
        valuation.fit(data)
    ```

    """
    self._utility = self._utility.with_dataset(data)
    if self._n_samples is None:
        self._n_samples = _get_default_n_samples(
            sampler=self._sampler, indices=data.indices
        )

    algorithm = str(self._sampler)

    problem = create_least_core_problem(
        u=self._utility,
        sampler=self._sampler,
        n_samples=self._n_samples,
        progress=self._progress,
    )

    solution = lc_solve_problem(
        problem=problem,
        u=self._utility,
        algorithm=algorithm,
        non_negative_subsidy=self._non_negative_subsidy,
        solver_options=self._solver_options,
    )

    self.result = solution
    return self

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

LOOValuation ¶

LOOValuation(utility: UtilityBase, progress: bool = False)

Bases: SemivalueValuation

Computes LOO values for a dataset.

Source code in src/pydvl/valuation/methods/loo.py

def __init__(self, utility: UtilityBase, progress: bool = False):
    self.result: ValuationResult | None = None
    super().__init__(
        utility,
        LOOSampler(batch_size=1, index_iteration=FiniteSequentialIndexIteration),
        # LOO is done when every index has been updated once
        MinUpdates(n_updates=1),
        progress=progress,
    )

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

log_coefficient ¶

log_coefficient(n: int, k: int) -> float

The LOOSampler returns only complements of {idx}, so the weight is either 1/n (the probability of a set of size n-1) or 0 if k != n-1. We cancel this out here so that the final coefficient is either 1 if k == n-1 or 0 otherwise.

Source code in src/pydvl/valuation/methods/loo.py

def log_coefficient(self, n: int, k: int) -> float:
    """The LOOSampler returns only complements of {idx}, so the weight is either
    1/n (the probability of a set of size n-1) or 0 if k != n-1. We cancel
    this out here so that the final coefficient is either 1 if k == n-1 or 0
    otherwise."""
    return float(-np.log(max(1, n))) if k == n - 1 else -np.inf

SemivalueValuation ¶

SemivalueValuation(
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
)

Bases: Valuation

Abstract class to define semi-values.

Implementations must only provide the log_coefficient() method, corresponding to the semi-value coefficient.

Note

For implementation consistency, we slightly depart from the common definition of semi-values, which includes a factor $1/n$ in the sum over subsets. Instead, we subsume this factor into the coefficient $w(k)$.

PARAMETER	DESCRIPTION
`utility`	Object to compute utilities. TYPE: `UtilityBase`
`sampler`	Sampling scheme to use. TYPE: `IndexSampler`
`is_done`	Stopping criterion to use. TYPE: `StoppingCriterion`
`skip_converged`	Whether to skip converged indices, as determined by the stopping criterion's `converged` array. TYPE: `bool` DEFAULT: `False`
`show_warnings`	Whether to show warnings. TYPE: `bool` DEFAULT: `True`
`progress`	Whether to show a progress bar. If a dictionary, it is passed to `tqdm` as keyword arguments, and the progress bar is displayed. TYPE: `dict[str, Any] \| bool` DEFAULT: `False`

Source code in src/pydvl/valuation/methods/semivalue.py

def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

log_coefficient `abstractmethod` ¶

log_coefficient(n: int, k: int) -> float

The semi-value coefficient in log-space.

The semi-value coefficient is a function of the number of elements in the set, and the size of the subset for which the coefficient is being computed. Because both coefficients and sampler weights can be very large or very small, we perform all computations in log-space to avoid numerical issues.

PARAMETER	DESCRIPTION
`n`	Total number of elements in the set. TYPE: `int`
`k`	Size of the subset for which the coefficient is being computed TYPE: `int`

RETURNS	DESCRIPTION
`float`	The natural logarithm of the semi-value coefficient.

Source code in src/pydvl/valuation/methods/semivalue.py

@abstractmethod
def log_coefficient(self, n: int, k: int) -> float:
    """The semi-value coefficient in log-space.

    The semi-value coefficient is a function of the number of elements in the set,
    and the size of the subset for which the coefficient is being computed.
    Because both coefficients and sampler weights can be very large or very small,
    we perform all computations in log-space to avoid numerical issues.

    Args:
        n: Total number of elements in the set.
        k: Size of the subset for which the coefficient is being computed

    Returns:
        The natural logarithm of the semi-value coefficient.
    """
    ...

ShapleyValuation ¶

ShapleyValuation(
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
)

Bases: SemivalueValuation

Computes Shapley values.

Source code in src/pydvl/valuation/methods/semivalue.py

def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values ¶

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

The valuation must have been run with fit() before calling this method.

PARAMETER	DESCRIPTION
`sort`	Whether to sort the valuation result by value before returning it. TYPE: `bool` DEFAULT: `False`

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py

def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

    The valuation must have been run with `fit()` before calling this method.

    Args:
        sort: Whether to sort the valuation result by value before returning it.
    Returns:
        The result of the valuation.
    """
    if not self.is_fitted:
        raise NotFittedException(type(self))
    assert self.result is not None

    from copy import copy

    r = copy(self.result)
    if sort:
        r.sort()
    return r

ResultUpdater ¶

ResultUpdater(result: ValuationResult)

Bases: Protocol[ValueUpdateT]

Protocol for result updaters.

A result updater is a strategy to update a valuation result with a value update.

Source code in src/pydvl/valuation/samplers/base.py

def __init__(self, result: ValuationResult): ...

IndexSampler ¶

IndexSampler(batch_size: int = 1)

Bases: ABC, Generic[ValueUpdateT]

Samplers are custom iterables over batches of subsets of indices.

Calling from_indices(indexset) on a sampler returns a generator over batches of Samples. A Sample is a tuple of the form $(i, S)$, where $i$ is an index of interest, and $S \subset I \setminus \{i\}$ is a subset of the complement of $i$ in $I$.

Note

Samplers are not iterators themselves, so that each call to from_indices(data) e.g. in a new for loop creates a new iterator.

Derived samplers must implement log_weight() and generate(). See the module's documentation for more on these.

Interrupting samplers¶

Calling interrupt() on a sampler will stop the batched generator after the current batch has been yielded.

PARAMETER	DESCRIPTION
`batch_size`	The number of samples to generate per batch. Batches are processed by EvaluationStrategy so that individual valuations in batch are guaranteed to be received in the right sequence. TYPE: `int` DEFAULT: `1`

Example

>>>from pydvl.valuation.samplers import DeterministicUniformSampler
>>>import numpy as np
>>>sampler = DeterministicUniformSampler()
>>>for idx, s in sampler.generate_batches(np.arange(2)):
>>>    print(s, end="")
[][2,][][1,]

    processed by the
    [EvaluationStrategy][pydvl.valuation.samplers.base.EvaluationStrategy]

Source code in src/pydvl/valuation/samplers/base.py

def __init__(self, batch_size: int = 1):
    """
    Args:
        batch_size: The number of samples to generate per batch. Batches are
            processed by the
            [EvaluationStrategy][pydvl.valuation.samplers.base.EvaluationStrategy]
    """
    self._batch_size = batch_size
    self._n_samples = 0
    self._interrupted = False
    self._skip_indices = np.empty(0, dtype=bool)
    self._len: int | None = None

skip_indices `property` `writable` ¶

skip_indices: IndexSetT

Indices being skipped in the sampler. The exact behaviour will be sampler-dependent, so that setting this property is disabled by default.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

sample_limit `abstractmethod` ¶

sample_limit(indices: IndexSetT) -> int | None

Number of samples that can be generated from the indices.

PARAMETER	DESCRIPTION
`indices`	The indices used in the sampler. TYPE: `IndexSetT`

RETURNS	DESCRIPTION
`int \| None`	The maximum number of samples that will be generated, or `None` if the number of samples is infinite. This will depend, among other things, on the type of IndexIteration.

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def sample_limit(self, indices: IndexSetT) -> int | None:
    """Number of samples that can be generated from the indices.

    Args:
        indices: The indices used in the sampler.

    Returns:
        The maximum number of samples that will be generated, or  `None` if the
            number of samples is infinite. This will depend, among other things,
            on the type of [IndexIteration][pydvl.valuation.samplers.IndexIteration].
    """
    ...

generate `abstractmethod` ¶

generate(indices: IndexSetT) -> SampleGenerator

Generates single samples.

IndexSampler.generate_batches() will batch these samples according to the batch size set upon construction.

PARAMETER	DESCRIPTION
`indices`	TYPE: `IndexSetT`

YIELDS	DESCRIPTION
`SampleGenerator`	A tuple (idx, subset) for each sample.

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def generate(self, indices: IndexSetT) -> SampleGenerator:
    """Generates single samples.

    `IndexSampler.generate_batches()` will batch these samples according to the
    batch size set upon construction.

    Args:
        indices:

    Yields:
        A tuple (idx, subset) for each sample.
    """
    ...

log_weight `abstractmethod` ¶

log_weight(n: int, subset_len: int) -> float

Factor by which to multiply Monte Carlo samples, so that the mean converges to the desired expression.

Log-space computation

Because the weight is a probability that can be arbitrarily small, we compute it in log-space for numerical stability.

By the Law of Large Numbers, the sample mean of $f(S_j)$ converges to the expectation under the distribution from which $S_j$ is sampled.

\[ \begin{eqnarray} \frac{1}{m} \sum_{j = 1}^m f (S_j) w (S_j) & \longrightarrow & \underset{S \sim \mathcal{D}_{- i}}{\mathbb{E}} [f (S) w (S)] \\ & & = \sum_{S \subseteq N_{- i}} f (S) w (S) \mathbb{P}_{\mathcal{D}_{- i}} (S) \end{eqnarray}. \]

We add the factor $w(S_j)$ in order to have this expectation coincide with the desired expression, by cancelling out $\mathbb{P} (S)$.

PARAMETER	DESCRIPTION
`n`	The size of the index set. Note that the actual size of the set being sampled will often be n-1, as one index might be removed from the set. See IndexIteration for more. TYPE: `int`
`subset_len`	The size of the subset being sampled TYPE: `int`

RETURNS	DESCRIPTION
`float`	The natural logarithm of the probability of sampling a set of the given size, when the index set has size `n`, under the IndexIteration given upon construction.

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def log_weight(self, n: int, subset_len: int) -> float:
    r"""Factor by which to multiply Monte Carlo samples, so that the
    mean converges to the desired expression.

    !!! Info "Log-space computation"
        Because the weight is a probability that can be arbitrarily small, we
        compute it in log-space for numerical stability.

    By the Law of Large Numbers, the sample mean of $f(S_j)$ converges to the
    expectation under the distribution from which $S_j$ is sampled.

    $$
    \begin{eqnarray}
        \frac{1}{m} \sum_{j = 1}^m f (S_j) w (S_j) & \longrightarrow &
            \underset{S \sim \mathcal{D}_{- i}}{\mathbb{E}} [f (S) w (S)] \\
        &  & = \sum_{S \subseteq N_{- i}} f (S) w (S)
            \mathbb{P}_{\mathcal{D}_{- i}} (S)
    \end{eqnarray}.
    $$

    We add the factor $w(S_j)$ in order to have this expectation coincide with the
    desired expression, by cancelling out $\mathbb{P} (S)$.

    Args:
        n: The size of the index set. Note that the actual size of the set being
            sampled will often be n-1, as one index might be removed from the set.
            See [IndexIteration][pydvl.valuation.samplers.IndexIteration] for more.
        subset_len: The size of the subset being sampled

    Returns:
        The natural logarithm of the probability of sampling a set of the given
            size, when the index set has size `n`, under the
            [IndexIteration][pydvl.valuation.samplers.IndexIteration] given upon
            construction.
    """
    ...

make_strategy `abstractmethod` ¶

make_strategy(
    utility: UtilityBase,
    log_coefficient: Callable[[int, int], float] | None = None,
) -> EvaluationStrategy

Returns the strategy for this sampler.

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def make_strategy(
    self,
    utility: UtilityBase,
    log_coefficient: Callable[[int, int], float] | None = None,
) -> EvaluationStrategy:
    """Returns the strategy for this sampler."""
    ...  # return SomeLogEvaluationStrategy(self)

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

EvaluationStrategy ¶

EvaluationStrategy(
    sampler: SamplerT,
    utility: UtilityBase,
    log_coefficient: Callable[[int, int], float] | None = None,
)

Bases: ABC, Generic[SamplerT, ValueUpdateT]

An evaluation strategy for samplers.

Implements the processing strategy for batches returned by an IndexSampler.

Different sampling schemes require different strategies for the evaluation of the utilities. For instance permutations generated by PermutationSampler must be evaluated in sequence to save computation, see PermutationEvaluationStrategy.

This class defines the common interface.

Usage pattern in valuation methods

    def fit(self, data: Dataset):
        self.utility = self.utility.with_dataset(data)
        strategy = self.sampler.strategy(self.utility, self.log_coefficient)
        delayed_batches = Parallel()(
            delayed(strategy.process)(batch=list(batch), is_interrupted=flag)
            for batch in self.sampler
        )
        for batch in delayed_batches:
            for evaluation in batch:
                self.result.update(evaluation.idx, evaluation.update)
            if self.is_done(self.result):
                flag.set()
                break

PARAMETER	DESCRIPTION
`sampler`	Required to set up some strategies. TYPE: `SamplerT`
`utility`	Required to set up some strategies and to process the samples. Since this contains the training data, it is expensive to pickle and send to workers. TYPE: `UtilityBase`
`log_coefficient`	An additional coefficient to multiply marginals with. This depends on the valuation method, hence the delayed setup. TYPE: `Callable[[int, int], float] \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/base.py

def __init__(
    self,
    sampler: SamplerT,
    utility: UtilityBase,
    log_coefficient: Callable[[int, int], float] | None = None,
):
    self.utility = utility
    # Used by the decorator suppress_warnings:
    self.show_warnings = getattr(utility, "show_warnings", False)
    self.n_indices = (
        len(utility.training_data) if utility.training_data is not None else 0
    )

    if log_coefficient is not None:

        def correction_fun(n: int, subset_len: int) -> float:
            return log_coefficient(n, subset_len) - sampler.log_weight(
                n, subset_len
            )

        self.log_correction = correction_fun
    else:
        self.log_correction = lambda n, subset_len: 0.0

process `abstractmethod` ¶

process(
    batch: SampleBatch, is_interrupted: NullaryPredicate
) -> list[ValueUpdateT]

Processes batches of samples using the evaluator, with the strategy required for the sampler.

Warning

This method is intended to be used by the evaluator to process the samples in one batch, which means it might be sent to another process. Be careful with the objects you use here, as they will be pickled and sent over the wire.

PARAMETER	DESCRIPTION
`batch`	A batch of samples to process. TYPE: `SampleBatch`
`is_interrupted`	A predicate that returns True if the processing should be interrupted. TYPE: `NullaryPredicate`

YIELDS	DESCRIPTION
`list[ValueUpdateT]`	Updates to values as tuples (idx, update)

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def process(
    self, batch: SampleBatch, is_interrupted: NullaryPredicate
) -> list[ValueUpdateT]:
    """Processes batches of samples using the evaluator, with the strategy
    required for the sampler.

    !!! Warning
        This method is intended to be used by the evaluator to process the samples
        in one batch, which means it might be sent to another process. Be careful
        with the objects you use here, as they will be pickled and sent over the
        wire.

    Args:
        batch: A batch of samples to process.
        is_interrupted: A predicate that returns True if the processing should be
            interrupted.

    Yields:
        Updates to values as tuples (idx, update)
    """
    ...

ClasswiseSampler ¶

ClasswiseSampler(
    in_class: IndexSampler,
    out_of_class: PowersetSampler,
    *,
    min_elements_per_label: int = 1,
    batch_size: int = 1,
)

Bases: IndexSampler

A sampler that samples elements from a dataset in two steps, based on the labels.

It proceeds by sampling out-of-class indices (training points with a different label to the point of interest), and in-class indices (training points with the same label as the point of interest), in the complement.

Used by the class-wise Shapley valuation method.

PARAMETER	DESCRIPTION
`in_class`	Sampling scheme for elements of a given label. TYPE: `IndexSampler`
`out_of_class`	Sampling scheme for elements of different labels, i.e., the complement set. TYPE: `PowersetSampler`
`min_elements_per_label`	Minimum number of elements per label to sample from the complement set, i.e., out of class elements. TYPE: `int` DEFAULT: `1`

Source code in src/pydvl/valuation/samplers/classwise.py

def __init__(
    self,
    in_class: IndexSampler,
    out_of_class: PowersetSampler,
    *,
    min_elements_per_label: int = 1,
    batch_size: int = 1,
):
    super().__init__(batch_size=batch_size)
    self.in_class = in_class
    self.out_of_class = out_of_class
    self.min_elements_per_label = min_elements_per_label

skip_indices `property` `writable` ¶

skip_indices: IndexSetT

Indices being skipped in the sampler. The exact behaviour will be sampler-dependent, so that setting this property is disabled by default.

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

interrupt ¶

interrupt() -> None

Interrupts the current sampler as well as the passed in samplers

Source code in src/pydvl/valuation/samplers/classwise.py

def interrupt(self) -> None:
    """Interrupts the current sampler as well as the passed in samplers"""
    super().interrupt()
    self.in_class.interrupt()
    self.out_of_class.interrupt()

MSRSampler ¶

MSRSampler(batch_size: int = 1, seed: Seed | None = None)

Bases: StochasticSamplerMixin, IndexSampler[MSRValueUpdate]

Sampler for unweighted Maximum Sample Re-use (MSR) valuation.

The sampling is similar to a UniformSampler but without an outer index. However,the MSR sampler uses a special evaluation strategy and result updater, as returned by the make_strategy() and result_updater() methods, respectively.

Two running means are updated separately for positive and negative updates. The two running means are later combined into a final result.

PARAMETER	DESCRIPTION
`batch_size`	Number of samples to generate in each batch. TYPE: `int` DEFAULT: `1`
`seed`	Seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/msr.py

def __init__(self, batch_size: int = 1, seed: Seed | None = None):
    super().__init__(batch_size=batch_size, seed=seed)

skip_indices `property` `writable` ¶

skip_indices: IndexSetT

Indices being skipped in the sampler. The exact behaviour will be sampler-dependent, so that setting this property is disabled by default.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

log_weight ¶

log_weight(n: int, subset_len: int) -> float

Probability of sampling a set of size k.

In the MSR scheme, the sampling is done from the full power set $2^N$ (each set $S \subseteq N$ with probability $1 / 2^n$), and then for each data point $i$ one partitions the sample into:

* $\mathcal{S}_{\ni i} = \{S \in \mathcal{S}: i \in S\},$ and
* $\mathcal{S}_{\nni i} = \{S \in \mathcal{S}: i \nin S\}.$.

When we condition on the event $i \in S$, the remaining part $S_{- i}$ is uniformly distributed over $2^{N_{- i}}$. In other words, the act of partitioning recovers the uniform distribution on $2^{N_{- i}}$ "for free" because

\[P (S_{- i} = T \mid i \in S) = \frac{1}{2^{n - 1}},\]

for each $T \subseteq N_{- i}$.

PARAMETER	DESCRIPTION
`n`	Size of the index set. TYPE: `int`
`subset_len`	Size of the subset. TYPE: `int`

RETURNS	DESCRIPTION
`float`	The logarithm of the probability of having sampled a set of size `subset_len`.

Source code in src/pydvl/valuation/samplers/msr.py

def log_weight(self, n: int, subset_len: int) -> float:
    r"""Probability of sampling a set of size k.

    In the **MSR scheme**, the sampling is done from the full power set $2^N$ (each
    set $S \subseteq N$ with probability $1 / 2^n$), and then for each data point
    $i$ one partitions the sample into:

        * $\mathcal{S}_{\ni i} = \{S \in \mathcal{S}: i \in S\},$ and
        * $\mathcal{S}_{\nni i} = \{S \in \mathcal{S}: i \nin S\}.$.

    When we condition on the event $i \in S$, the remaining part $S_{- i}$ is
    uniformly distributed over $2^{N_{- i}}$. In other words, the act of
    partitioning recovers the uniform distribution on $2^{N_{- i}}$ "for free"
    because

    $$P (S_{- i} = T \mid i \in S) = \frac{1}{2^{n - 1}},$$

    for each $T \subseteq N_{- i}$.

    Args:
        n: Size of the index set.
        subset_len: Size of the subset.

    Returns:
        The logarithm of the probability of having sampled a set of size
            `subset_len`.
    """
    return float(-(n - 1) * np.log(2)) if n > 0 else 0.0

make_strategy ¶

make_strategy(
    utility: UtilityBase, coefficient: Callable[[int, int], float] | None = None
) -> MSREvaluationStrategy

Returns the strategy for this sampler.

PARAMETER	DESCRIPTION
`utility`	Utility function to evaluate. TYPE: `UtilityBase`
`coefficient`	Coefficient function for the utility function. TYPE: `Callable[[int, int], float] \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/msr.py

def make_strategy(
    self,
    utility: UtilityBase,
    coefficient: Callable[[int, int], float] | None = None,
) -> MSREvaluationStrategy:
    """Returns the strategy for this sampler.

    Args:
        utility: Utility function to evaluate.
        coefficient: Coefficient function for the utility function.
    """
    assert coefficient is not None
    return MSREvaluationStrategy(self, utility, coefficient)

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater

Returns a callable that updates a valuation result with an MSR value update.

MSR updates two running means for positive and negative updates separately. The two running means are later combined into a final result.

PARAMETER	DESCRIPTION
`result`	The valuation result to update with each call of the returned callable. TYPE: `ValuationResult`

Returns: A callable object that updates the valuation result with very MSRValueUpdate.

Source code in src/pydvl/valuation/samplers/msr.py

def result_updater(self, result: ValuationResult) -> ResultUpdater:
    """Returns a callable that updates a valuation result with an MSR value update.

    MSR updates two running means for positive and negative updates separately. The
    two running means are later combined into a final result.

    Args:
        result: The valuation result to update with each call of the returned
            callable.
    Returns:
        A callable object that updates the valuation result with very
            [MSRValueUpdate][pydvl.valuation.samplers.msr.MSRValueUpdate].
    """
    return MSRResultUpdater(result)

OwenStrategy ¶

OwenStrategy(n_samples_outer: int)

Bases: ABC

Base class for strategies for the Owen sampler to sample probability values.

Source code in src/pydvl/valuation/samplers/owen.py

def __init__(self, n_samples_outer: int):
    self.n_samples_outer = n_samples_outer

UniformOwenStrategy ¶

UniformOwenStrategy(n_samples_outer: int, seed: Seed | None = None)

Bases: OwenStrategy

A strategy for OwenSampler to sample probability values uniformly between 0 and $q_ ext{stop}$.

PARAMETER	DESCRIPTION
`n_samples_outer`	The number of probability values $q$ used for the outer loop. Since samples are taken anew for each index, a high number will delay updating new indices and has no effect on the final accuracy if using an infinite index iteration. In general, it only makes sense to change this number if using a finite index iteration. TYPE: `int`
`seed`	The seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/owen.py

def __init__(self, n_samples_outer: int, seed: Seed | None = None):
    super().__init__(n_samples_outer=n_samples_outer)
    self.rng = np.random.default_rng(seed)

GridOwenStrategy ¶

GridOwenStrategy(n_samples_outer: int)

Bases: OwenStrategy

A strategy for OwenSampler to sample probability values on a linear grid.

PARAMETER	DESCRIPTION
`n_samples_outer`	The number of probability values $q$ used for the outer loop. These will be linearly spaced between 0 and $q_ ext{stop}$. TYPE: `int`

Source code in src/pydvl/valuation/samplers/owen.py

def __init__(self, n_samples_outer: int):
    super().__init__(n_samples_outer=n_samples_outer)

OwenSampler ¶

OwenSampler(
    outer_sampling_strategy: OwenStrategy,
    n_samples_inner: int = 2,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
)

Bases: StochasticSamplerMixin, PowersetSampler

A sampler for semi-values using the Owen method.

For each index $i$ we sample n_samples_outer probability values $q_j$ between 0 and 1 and then, for each $j$ we draw n_samples_inner subsets of the complement of the current index where each element is sampled probability $q_j$.

The distribution for the outer sampling can be either uniform or deterministic. The default is deterministic on a grid, which is the original method described in Okhrati and Lipani (2021)¹. This can be achieved by using the GridOwenStrategy strategy.

Alternatively, the distribution can be uniform between 0 and 1. This can be achieved by using the UniformOwenStrategy strategy.

By combining a UniformOwenStrategy with an infinite IndexIteration strategy, this sampler can be used with a stopping criterion to estimate semi-values. This follows more closely the typical usage pattern in PyDVL than the original sampling method described in Okhrati and Lipani (2021)¹.

Example usage

sampler = OwenSampler(
    outer_sampling_strategy=GridOwenStrategy(n_samples_outer=200),
    n_samples_inner=8,
    index_iteration=FiniteSequentialIndexIteration,
)

PARAMETER	DESCRIPTION
`n_samples_inner`	The number of samples drawn for each probability. In the original paper this was fixed to 2 for all experiments. TYPE: `int` DEFAULT: `2`
`batch_size`	The batch size of the sampler. TYPE: `int` DEFAULT: `1`
`index_iteration`	The index iteration strategy, sequential or random, finite or infinite. TYPE: `Type[IndexIteration]` DEFAULT: `FiniteSequentialIndexIteration`
`seed`	The seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/owen.py

def __init__(
    self,
    outer_sampling_strategy: OwenStrategy,
    n_samples_inner: int = 2,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
):
    super().__init__(
        batch_size=batch_size, index_iteration=index_iteration, seed=seed
    )
    self.n_samples_inner = n_samples_inner
    self.sampling_probabilities = outer_sampling_strategy
    self.q_stop = 1.0

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

log_weight ¶

log_weight(n: int, subset_len: int) -> float

For each $q_j, j \in \{1, ..., N\}$ in the outer probabilities, the probability of drawing a subset $S_k$ of size $k$ is:

\[ P (| S_{q_j} | = k) = \binom{n}{k} \ q_j^k (1 - q_j)^{n - k}.\]

So, if each $q_j$ is chosen with equal weight (or more generally with probability $p_j$),then by total probability, the overall probability of obtaining a subset of size $k$ is a mixture of the binomials: $$ P (| S | = k) = \sum_{j = 1}^N p_j \ \binom{n}{k} \ q_j^k (1 - q_j)^{n - k}. $$

In our case $p_j = 1/N$, so that $P(|S|=k) = \frac{1}{N} \sum_{j=1}^N P (| S_{q_j} | = k)$. For large enough $N$ this is

\[ P(|S|=k) \approx \binom{n}{k} \int_0^1 q^k (1 - q)^{n - k} \, dq = \frac{1}{ n+1}, \]

where we computed the integral using the beta function and its expression as products of gamma functions.

Now, given the symmetry wrt. the indices in the sampling procedure, any given set $S$ of size $k$ is equally likely to be drawn. So the probability of a set being of size $k$ must be equally divided by the number of sets of that size, and the weight of a set of size $k$ is:

\[ P(S) = \frac{1}{n+1} \binom{n}{|S|}^{-1}. \]

PARAMETER	DESCRIPTION
`n`	Size of the index set. TYPE: `int`
`subset_len`	Size of the subset. TYPE: `int`

Returns: The logarithm of the weight of a subset of size subset_len.

Source code in src/pydvl/valuation/samplers/owen.py

def log_weight(self, n: int, subset_len: int) -> float:
    r"""For each $q_j, j \in \{1, ..., N\}$ in the outer probabilities, the
    probability of drawing a subset $S_k$ of size $k$ is:

    $$ P (| S_{q_j} | = k) = \binom{n}{k} \  q_j^k  (1 - q_j)^{n - k}.$$

    So, if each $q_j$ is chosen with equal weight (or more generally with
    probability $p_j$),then by total probability, the overall probability of
    obtaining a subset of size $k$ is a mixture of the binomials:
    $$
    P (| S | = k) = \sum_{j = 1}^N p_j \ \binom{n}{k} \ q_j^k  (1 - q_j)^{n - k}.
    $$

    In our case $p_j = 1/N$, so that $P(|S|=k) = \frac{1}{N} \sum_{j=1}^N P (|
    S_{q_j} | = k)$. For large enough $N$ this is

    $$
    P(|S|=k) \approx \binom{n}{k} \int_0^1 q^k (1 - q)^{n - k} \, dq = \frac{1}{
    n+1},
    $$

    where we computed the integral using the beta function and its expression as
    products of gamma functions.

    Now, given the symmetry wrt. the indices in the sampling procedure, any given
    set $S$ of size $k$ is equally likely to be drawn. So the probability of a set
    being of size $k$ must be equally divided by the number of sets of that size,
    and the weight of a set of size $k$ is:

    $$ P(S) = \frac{1}{n+1} \binom{n}{|S|}^{-1}. $$

    Args:
        n: Size of the index set.
        subset_len: Size of the subset.
    Returns:
        The logarithm of the weight of a subset of size `subset_len`.
    """
    m = self._index_iterator_cls.complement_size(n)
    return float(-logcomb(m, subset_len) - np.log(m + 1))

sample_limit ¶

sample_limit(indices: IndexSetT) -> int | None

The number of samples that will be generated by the sampler.

PARAMETER	DESCRIPTION
`indices`	TYPE: `IndexSetT`

RETURNS	DESCRIPTION
`int \| None`	0 if there are no indices, `None` if there's no limit and the number of
`int \| None`	samples otherwise.

Source code in src/pydvl/valuation/samplers/owen.py

def sample_limit(self, indices: IndexSetT) -> int | None:
    """The number of samples that will be generated by the sampler.

    Args:
        indices:

    Returns:
        0 if there are no indices, `None` if there's no limit and the number of
        samples otherwise.
    """
    if len(indices) == 0:
        return 0
    if not self._index_iterator_cls.is_finite():
        return None

    return (
        cast(int, self._index_iterator_cls.length(len(indices)))
        * self.sampling_probabilities.n_samples_outer
        * self.n_samples_inner
    )

AntitheticOwenSampler ¶

AntitheticOwenSampler(
    outer_sampling_strategy: OwenStrategy,
    n_samples_inner: int = 2,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
)

Bases: OwenSampler

A sampler for antithetic Owen shapley values.

For each sample obtained with the method of OwenSampler, a second sample is generated by taking the complement of the first sample.

For the same number of total samples, the antithetic Owen sampler yields usually more precise estimates of shapley values than the regular Owen sampler.

Source code in src/pydvl/valuation/samplers/owen.py

def __init__(
    self,
    outer_sampling_strategy: OwenStrategy,
    n_samples_inner: int = 2,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
):
    super().__init__(
        outer_sampling_strategy=outer_sampling_strategy,
        n_samples_inner=n_samples_inner,
        batch_size=batch_size,
        index_iteration=index_iteration,
        seed=seed,
    )
    self.q_stop = 0.5

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

log_weight ¶

log_weight(n: int, subset_len: int) -> float

For each $q_j, j \in \{1, ..., N\}$ in the outer probabilities, the probability of drawing a subset $S_k$ of size $k$ is:

\[ P (| S_{q_j} | = k) = \binom{n}{k} \ q_j^k (1 - q_j)^{n - k}.\]

So, if each $q_j$ is chosen with equal weight (or more generally with probability $p_j$),then by total probability, the overall probability of obtaining a subset of size $k$ is a mixture of the binomials: $$ P (| S | = k) = \sum_{j = 1}^N p_j \ \binom{n}{k} \ q_j^k (1 - q_j)^{n - k}. $$

In our case $p_j = 1/N$, so that $P(|S|=k) = \frac{1}{N} \sum_{j=1}^N P (| S_{q_j} | = k)$. For large enough $N$ this is

\[ P(|S|=k) \approx \binom{n}{k} \int_0^1 q^k (1 - q)^{n - k} \, dq = \frac{1}{ n+1}, \]

where we computed the integral using the beta function and its expression as products of gamma functions.

Now, given the symmetry wrt. the indices in the sampling procedure, any given set $S$ of size $k$ is equally likely to be drawn. So the probability of a set being of size $k$ must be equally divided by the number of sets of that size, and the weight of a set of size $k$ is:

\[ P(S) = \frac{1}{n+1} \binom{n}{|S|}^{-1}. \]

PARAMETER	DESCRIPTION
`n`	Size of the index set. TYPE: `int`
`subset_len`	Size of the subset. TYPE: `int`

Returns: The logarithm of the weight of a subset of size subset_len.

Source code in src/pydvl/valuation/samplers/owen.py

def log_weight(self, n: int, subset_len: int) -> float:
    r"""For each $q_j, j \in \{1, ..., N\}$ in the outer probabilities, the
    probability of drawing a subset $S_k$ of size $k$ is:

    $$ P (| S_{q_j} | = k) = \binom{n}{k} \  q_j^k  (1 - q_j)^{n - k}.$$

    So, if each $q_j$ is chosen with equal weight (or more generally with
    probability $p_j$),then by total probability, the overall probability of
    obtaining a subset of size $k$ is a mixture of the binomials:
    $$
    P (| S | = k) = \sum_{j = 1}^N p_j \ \binom{n}{k} \ q_j^k  (1 - q_j)^{n - k}.
    $$

    In our case $p_j = 1/N$, so that $P(|S|=k) = \frac{1}{N} \sum_{j=1}^N P (|
    S_{q_j} | = k)$. For large enough $N$ this is

    $$
    P(|S|=k) \approx \binom{n}{k} \int_0^1 q^k (1 - q)^{n - k} \, dq = \frac{1}{
    n+1},
    $$

    where we computed the integral using the beta function and its expression as
    products of gamma functions.

    Now, given the symmetry wrt. the indices in the sampling procedure, any given
    set $S$ of size $k$ is equally likely to be drawn. So the probability of a set
    being of size $k$ must be equally divided by the number of sets of that size,
    and the weight of a set of size $k$ is:

    $$ P(S) = \frac{1}{n+1} \binom{n}{|S|}^{-1}. $$

    Args:
        n: Size of the index set.
        subset_len: Size of the subset.
    Returns:
        The logarithm of the weight of a subset of size `subset_len`.
    """
    m = self._index_iterator_cls.complement_size(n)
    return float(-logcomb(m, subset_len) - np.log(m + 1))

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

PermutationSampler ¶

PermutationSampler(
    truncation: TruncationPolicy | None = None,
    seed: Seed | None = None,
    batch_size: int = 1,
)

Bases: StochasticSamplerMixin, PermutationSamplerBase

Samples permutations of indices.

Batching

Even though this sampler supports batching, it is not recommended to use it since the PermutationEvaluationStrategy processes whole permutations in one go, effectively batching the computation of up to n-1 marginal utilities in one process.

PARAMETER	DESCRIPTION
`truncation`	A policy to stop the permutation early. TYPE: `TruncationPolicy \| None` DEFAULT: `None`
`seed`	Seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/permutation.py

def __init__(
    self,
    truncation: TruncationPolicy | None = None,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__(seed=seed, truncation=truncation, batch_size=batch_size)

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

generate ¶

generate(indices: IndexSetT) -> SampleGenerator

Generates the permutation samples.

PARAMETER	DESCRIPTION
`indices`	The indices to sample from. If empty, no samples are generated. If skip_indices is set, these indices are removed from the set before generating the permutation. TYPE: `IndexSetT`

Source code in src/pydvl/valuation/samplers/permutation.py

def generate(self, indices: IndexSetT) -> SampleGenerator:
    """Generates the permutation samples.

    Args:
        indices: The indices to sample from. If empty, no samples are generated. If
            [skip_indices][pydvl.valuation.samplers.base.IndexSampler.skip_indices]
            is set, these indices are removed from the set before generating the
            permutation.
    """
    if len(indices) == 0:
        return
    while True:
        _indices = np.setdiff1d(indices, self.skip_indices)
        yield Sample(None, self._rng.permutation(_indices))

AntitheticPermutationSampler ¶

AntitheticPermutationSampler(
    truncation: TruncationPolicy | None = None,
    seed: Seed | None = None,
    batch_size: int = 1,
)

Bases: PermutationSampler

Samples permutations like PermutationSampler, but after each permutation, it returns the same permutation in reverse order.

This sampler was suggested in (Mitchell et al. 2022)¹

New in version 0.7.1

Source code in src/pydvl/valuation/samplers/permutation.py

def __init__(
    self,
    truncation: TruncationPolicy | None = None,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__(seed=seed, truncation=truncation, batch_size=batch_size)

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

DeterministicPermutationSampler ¶

DeterministicPermutationSampler(
    *args,
    truncation: TruncationPolicy | None = None,
    batch_size: int = 1,
    **kwargs,
)

Bases: PermutationSamplerBase

Samples all n! permutations of the indices deterministically, and iterates through them, returning sets as required for the permutation-based definition of semi-values.

Source code in src/pydvl/valuation/samplers/permutation.py

def __init__(
    self,
    *args,
    truncation: TruncationPolicy | None = None,
    batch_size: int = 1,
    **kwargs,
):
    super().__init__(batch_size=batch_size)
    self.truncation = truncation or NoTruncation()

skip_indices `property` `writable` ¶

skip_indices: IndexSetT

Indices being skipped in the sampler. The exact behaviour will be sampler-dependent, so that setting this property is disabled by default.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

PermutationEvaluationStrategy ¶

PermutationEvaluationStrategy(
    sampler: PermutationSamplerBase,
    utility: UtilityBase,
    coefficient: Callable[[int, int], float] | None = None,
)

Bases: EvaluationStrategy[PermutationSamplerBase, ValueUpdate]

Computes marginal values for permutation sampling schemes in log-space.

This strategy iterates over permutations from left to right, computing the marginal utility wrt. the previous one at each step to save computation.

Source code in src/pydvl/valuation/samplers/permutation.py

def __init__(
    self,
    sampler: PermutationSamplerBase,
    utility: UtilityBase,
    coefficient: Callable[[int, int], float] | None = None,
):
    super().__init__(sampler, utility, coefficient)
    self.truncation = copy(sampler.truncation)
    self.truncation.reset(utility)  # Perform initial setup (e.g. total_utility)

IndexIteration ¶

IndexIteration(indices: IndexSetT)

Bases: ABC

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: IndexSetT):
    self._indices = indices

length `abstractmethod` `staticmethod` ¶

length(n_indices: int) -> int | None

Returns the length of the iteration over the index set

PARAMETER	DESCRIPTION
`n_indices`	The number of indices in the set. TYPE: `int`

RETURNS	DESCRIPTION
`int \| None`	The length of the iteration. It can be: - a non-negative integer, if the iteration is finite - `None` if the iteration never ends.

Source code in src/pydvl/valuation/samplers/powerset.py

@staticmethod
@abstractmethod
def length(n_indices: int) -> int | None:
    """Returns the length of the iteration over the index set

    Args:
        n_indices: The number of indices in the set.

    Returns:
        The length of the iteration. It can be:
            - a non-negative integer, if the iteration is finite
            - `None` if the iteration never ends.
    """
    ...

complement_size `abstractmethod` `staticmethod` ¶

complement_size(n: int) -> int

Returns the size of complements of sets of size n, with respect to the indices returned by the iteration.

If the iteration returns single indices, then this is n-1, if it returns no indices, then it is n. If it returned tuples, then n-2, etc.

Source code in src/pydvl/valuation/samplers/powerset.py

@staticmethod
@abstractmethod
def complement_size(n: int) -> int:
    """Returns the size of complements of sets of size n, with respect to the
    indices returned by the iteration.

    If the iteration returns single indices, then this is n-1, if it returns no
    indices, then it is n. If it returned tuples, then n-2, etc.
    """
    ...

SequentialIndexIteration ¶

SequentialIndexIteration(indices: IndexSetT)

Bases: InfiniteIterationMixin, IndexIteration

Samples indices sequentially, indefinitely.

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: IndexSetT):
    self._indices = indices

FiniteSequentialIndexIteration ¶

FiniteSequentialIndexIteration(indices: IndexSetT)

Bases: FiniteIterationMixin, SequentialIndexIteration

Samples indices sequentially, once.

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: IndexSetT):
    self._indices = indices

RandomIndexIteration ¶

RandomIndexIteration(indices: NDArray[IndexT], seed: Seed)

Bases: InfiniteIterationMixin, StochasticSamplerMixin, IndexIteration

Samples indices at random, indefinitely

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: NDArray[IndexT], seed: Seed):
    super().__init__(indices, seed=seed)

FiniteRandomIndexIteration ¶

FiniteRandomIndexIteration(indices: NDArray[IndexT], seed: Seed)

Bases: FiniteIterationMixin, RandomIndexIteration

Samples indices at random, once

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: NDArray[IndexT], seed: Seed):
    super().__init__(indices, seed=seed)

NoIndexIteration ¶

NoIndexIteration(indices: IndexSetT)

Bases: InfiniteIterationMixin, IndexIteration

An infinite iteration over no indices.

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: IndexSetT):
    self._indices = indices

FiniteNoIndexIteration ¶

FiniteNoIndexIteration(indices: IndexSetT)

Bases: FiniteIterationMixin, NoIndexIteration

A finite iteration over no indices. The iterator will yield None once and then stop.

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(self, indices: IndexSetT):
    self._indices = indices

length `staticmethod` ¶

length(n_indices: int) -> int | None

Returns 1, as the iteration yields exactly one item (None)

Source code in src/pydvl/valuation/samplers/powerset.py

@staticmethod
def length(n_indices: int) -> int | None:
    """Returns 1, as the iteration yields exactly one item (None)"""
    return 1

PowersetSampler ¶

PowersetSampler(
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = SequentialIndexIteration,
)

Bases: IndexSampler, ABC

An abstract class for samplers which iterate over the powerset of the complement of an index in the training set.

This is done in two nested loops, where the outer loop iterates over the set of indices, and the inner loop iterates over subsets of the complement of the current index. The outer iteration can be either sequential or at random.

    processed together by
    [UtilityEvaluator][pydvl.valuation.utility.evaluator.UtilityEvaluator].
index_iteration: the strategy to use for iterating over indices to update

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(
    self,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = SequentialIndexIteration,
):
    """
    Args:
        batch_size: The number of samples to generate per batch. Batches are
            processed together by
            [UtilityEvaluator][pydvl.valuation.utility.evaluator.UtilityEvaluator].
        index_iteration: the strategy to use for iterating over indices to update
    """
    super().__init__(batch_size)
    self._index_iterator_cls = index_iteration
    self._index_iterator: IndexIteration | None = None

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

sample_limit `abstractmethod` ¶

sample_limit(indices: IndexSetT) -> int | None

Number of samples that can be generated from the indices.

PARAMETER	DESCRIPTION
`indices`	The indices used in the sampler. TYPE: `IndexSetT`

RETURNS	DESCRIPTION
`int \| None`	The maximum number of samples that will be generated, or `None` if the number of samples is infinite. This will depend, among other things, on the type of IndexIteration.

Source code in src/pydvl/valuation/samplers/base.py

@abstractmethod
def sample_limit(self, indices: IndexSetT) -> int | None:
    """Number of samples that can be generated from the indices.

    Args:
        indices: The indices used in the sampler.

    Returns:
        The maximum number of samples that will be generated, or  `None` if the
            number of samples is infinite. This will depend, among other things,
            on the type of [IndexIteration][pydvl.valuation.samplers.IndexIteration].
    """
    ...

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

generate `abstractmethod` ¶

generate(indices: IndexSetT) -> SampleGenerator

Generates samples over the powerset of indices

Each PowersetSampler defines its own way to generate the subsets by implementing this method. The outer loop is handled by the index_iterator. Batching is handled by the generate_batches method.

PARAMETER	DESCRIPTION
`indices`	The set from which to generate samples. TYPE: `IndexSetT`

Source code in src/pydvl/valuation/samplers/powerset.py

@abstractmethod
def generate(self, indices: IndexSetT) -> SampleGenerator:
    """Generates samples over the powerset of `indices`

    Each `PowersetSampler` defines its own way to generate the subsets by
    implementing this method. The outer loop is handled by the `index_iterator`.
    Batching is handled by the `generate_batches` method.

    Args:
        indices: The set from which to generate samples.
    """
    ...

log_weight ¶

log_weight(n: int, subset_len: int) -> float

Correction coming from Monte Carlo integration so that the mean of the marginals converges to the value: the uniform distribution over the powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset.

Source code in src/pydvl/valuation/samplers/powerset.py

def log_weight(self, n: int, subset_len: int) -> float:
    """Correction coming from Monte Carlo integration so that the mean of
    the marginals converges to the value: the uniform distribution over the
    powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset."""
    m = self._index_iterator_cls.complement_size(n)
    return float(-m * np.log(2))

LOOSampler ¶

LOOSampler(
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
)

Bases: PowersetSampler

Leave-One-Out sampler.

In this special case of a powerset sampler, for every index $i$ in the set $S$, the sample $(i, S_{-i})$ is returned.

PARAMETER	DESCRIPTION
`batch_size`	The number of samples to generate per batch. Batches are processed together by each subprocess when working in parallel. TYPE: `int` DEFAULT: `1`
`index_iteration`	the strategy to use for iterating over indices to update. By default, a finite sequential index iteration is used, which is what LOOValuation expects. TYPE: `Type[IndexIteration]` DEFAULT: `FiniteSequentialIndexIteration`
`seed`	The seed for the random number generator used in case the index iteration is random. TYPE: `Seed \| None` DEFAULT: `None`

New in version 0.10.0

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(
    self,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
):
    super().__init__(batch_size, index_iteration)
    if not self._index_iterator_cls.is_proper():
        raise ValueError("LOO samplers require a proper index iteration strategy")
    self._rng = np.random.default_rng(seed)

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

log_weight ¶

log_weight(n: int, subset_len: int) -> float

This sampler returns only sets of size n-1. There are n such sets, so the probability of drawing one is 1/n, or 0 if subset_len != n-1.

Source code in src/pydvl/valuation/samplers/powerset.py

def log_weight(self, n: int, subset_len: int) -> float:
    """This sampler returns only sets of size n-1. There are n such sets, so the
    probability of drawing one is 1/n, or 0 if subset_len != n-1."""
    return float(-np.log(n if subset_len == n - 1 else 0))

DeterministicUniformSampler ¶

DeterministicUniformSampler(
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
)

Bases: PowersetSampler

An iterator to perform uniform deterministic sampling of subsets.

For every index $i$, each subset of the complement indices - {i} is returned.

PARAMETER	DESCRIPTION
`batch_size`	The number of samples to generate per batch. Batches are processed together by each subprocess when working in parallel. TYPE: `int` DEFAULT: `1`
`index_iteration`	the strategy to use for iterating over indices to update. This iteration can be either finite or infinite. TYPE: `Type[IndexIteration]` DEFAULT: `FiniteSequentialIndexIteration`

Example

The code:

from pydvl.valuation.samplers import DeterministicUniformSampler
import numpy as np
sampler = DeterministicUniformSampler()
for idx, s in sampler.generate_batches(np.arange(2)):
    print(f"{idx} - {s}", end=", ")

Should produce the output:

1 - [], 1 - [2], 2 - [], 2 - [1],

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(
    self,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
):
    super().__init__(batch_size=batch_size, index_iteration=index_iteration)

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

log_weight ¶

log_weight(n: int, subset_len: int) -> float

Correction coming from Monte Carlo integration so that the mean of the marginals converges to the value: the uniform distribution over the powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset.

Source code in src/pydvl/valuation/samplers/powerset.py

def log_weight(self, n: int, subset_len: int) -> float:
    """Correction coming from Monte Carlo integration so that the mean of
    the marginals converges to the value: the uniform distribution over the
    powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset."""
    m = self._index_iterator_cls.complement_size(n)
    return float(-m * np.log(2))

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

UniformSampler ¶

UniformSampler(
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = SequentialIndexIteration,
    seed: Seed | None = None,
)

Bases: StochasticSamplerMixin, PowersetSampler

Draws random samples uniformly from the powerset of the index set.

Iterating over every index $i$, either in sequence or at random depending on the value of index_iteration, one subset of the complement indices - {i} is sampled with equal probability $2^{n-1}$.

PARAMETER	DESCRIPTION
`batch_size`	The number of samples to generate per batch. Batches are processed together by each subprocess when working in parallel. TYPE: `int` DEFAULT: `1`
`index_iteration`	the strategy to use for iterating over indices to update. This iteration can be either finite or infinite. TYPE: `Type[IndexIteration]` DEFAULT: `SequentialIndexIteration`
`seed`	The seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

Example

The code

for idx, s in UniformSampler(np.arange(3)):
   print(f"{idx} - {s}", end=", ")

Produces the output:

0 - [1 4], 1 - [2 3], 2 - [0 1 3], 3 - [], 4 - [2], 0 - [1 3 4], 1 - [0 2]
(...)

Source code in src/pydvl/valuation/samplers/powerset.py

def __init__(
    self,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = SequentialIndexIteration,
    seed: Seed | None = None,
):
    super().__init__(
        batch_size=batch_size, index_iteration=index_iteration, seed=seed
    )

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

log_weight ¶

log_weight(n: int, subset_len: int) -> float

Correction coming from Monte Carlo integration so that the mean of the marginals converges to the value: the uniform distribution over the powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset.

Source code in src/pydvl/valuation/samplers/powerset.py

def log_weight(self, n: int, subset_len: int) -> float:
    """Correction coming from Monte Carlo integration so that the mean of
    the marginals converges to the value: the uniform distribution over the
    powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset."""
    m = self._index_iterator_cls.complement_size(n)
    return float(-m * np.log(2))

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

AntitheticSampler ¶

AntitheticSampler(*args, seed: Seed | None = None, **kwargs)

Bases: StochasticSamplerMixin, PowersetSampler

A sampler that draws samples uniformly and their complements.

Works as UniformSampler, but for every tuple $(i,S)$, it subsequently returns $(i,S^c)$, where $S^c$ is the complement of the set $S$ in the set of indices, excluding $i$.

Source code in src/pydvl/valuation/samplers/utils.py

def __init__(self, *args, seed: Seed | None = None, **kwargs):
    super().__init__(*args, **kwargs)
    self._rng = np.random.default_rng(seed)

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

log_weight ¶

log_weight(n: int, subset_len: int) -> float

Correction coming from Monte Carlo integration so that the mean of the marginals converges to the value: the uniform distribution over the powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset.

Source code in src/pydvl/valuation/samplers/powerset.py

def log_weight(self, n: int, subset_len: int) -> float:
    """Correction coming from Monte Carlo integration so that the mean of
    the marginals converges to the value: the uniform distribution over the
    powerset of a set with n-1 elements has mass 1/2^{n-1} over each subset."""
    m = self._index_iterator_cls.complement_size(n)
    return float(-m * np.log(2))

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

SampleSizeStrategy ¶

SampleSizeStrategy(n_samples: int)

Bases: ABC

An object to compute the number of samples to take for a given set size. Based on Wu et al. (2023)¹, Theorem 4.2.

To be used with StratifiedSampler.

Sets the number of sets at size $k$ to be

\[m(k) = m \frac{f(k)}{\sum_{j=0}^{n} f(j)},\]

for some choice of $f.$ Implementations of this base class must override the method fun(). It is provided both the size $k$ and the total number of indices $n$ as arguments.

PARAMETER	DESCRIPTION
`n_samples`	Number of samples for the stratified sampler to generate, per index. If the sampler uses NoIndexIteration, then this will coincide with the total number of samples. TYPE: `int`

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, n_samples: int):
    """Construct a heuristic for the given number of samples.

    Args:
        n_samples: Number of samples for the stratified sampler to generate,
            **per index**. If the sampler uses
            [NoIndexIteration][pydvl.valuation.samplers.NoIndexIteration], then this
            will coincide with the total number of samples.
    """
    self.n_samples = n_samples

fun `abstractmethod` ¶

fun(n_indices: int, subset_len: int) -> float

The function $f$ to use in the heuristic. Args: n_indices: Size of the index set. subset_len: Size of the subset.

Source code in src/pydvl/valuation/samplers/stratified.py

@abstractmethod
def fun(self, n_indices: int, subset_len: int) -> float:
    """The function $f$ to use in the heuristic.
    Args:
        n_indices: Size of the index set.
        subset_len: Size of the subset.
    """
    ...

sample_sizes `cached` ¶

sample_sizes(
    n_indices: int, quantize: bool = True
) -> NDArray[int_] | NDArray[float_]

Precomputes the number of samples to take for each set size, from 0 up to n_indices inclusive.

This method corrects rounding errors taking into account the fractional parts so that the total number of samples is respected, while allocating remainders in a way that follows the relative sizes of the fractional parts.

Note

A naive implementation with e.g.

m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
        for k in range(n)]

would not respect the total number of samples, and would not distribute remainders correctly.

PARAMETER	DESCRIPTION
`n_indices`	number of indices in the index set from which to sample. This is typically `len(dataset) - 1` with the usual index iterations. TYPE: `int`
`quantize`	Whether to perform the remainder distribution. If `False`, the raw floating point values are returned. Useful e.g. for RandomSizeIteration where one needs frequencies. In this case `n_samples` can be 1. TYPE: `bool` DEFAULT: `True`

Returns: The exact (integer) number of samples to take for each set size, if quantize is True. Otherwise, the fractional number of samples.

Source code in src/pydvl/valuation/samplers/stratified.py

@lru_cache
def sample_sizes(
    self, n_indices: int, quantize: bool = True
) -> NDArray[np.int_] | NDArray[np.float_]:
    """Precomputes the number of samples to take for each set size, from 0 up to
    `n_indices` inclusive.

    This method corrects rounding errors taking into account the fractional parts
    so that the total number of samples is respected, while allocating remainders
    in a way that follows the relative sizes of the fractional parts.

    ??? Note
        A naive implementation with e.g.
        ```python
        m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
                for k in range(n)]
        ```
        would not respect the total number of samples, and would not distribute
        remainders correctly.

    Args:
        n_indices: number of indices in the index set from which to sample. This is
            typically `len(dataset) - 1` with the usual index iterations.
        quantize: Whether to perform the remainder distribution. If `False`, the raw
            floating point values are returned. Useful e.g. for
            [RandomSizeIteration][pydvl.valuation.samplers.stratified.RandomSizeIteration]
            where one needs frequencies. In this case `n_samples` can
            be 1.
    Returns:
        The exact (integer) number of samples to take for each set size, if
        `quantize` is `True`. Otherwise, the fractional number of samples.
    """

    # m_k = m * f(k) / sum_j f(j)
    values = np.empty(n_indices + 1, dtype=float)
    s = 0.0

    for k in range(n_indices + 1):
        val = self.fun(n_indices, k)
        values[k] = val
        s += val

    values *= self.n_samples / s
    if not quantize:
        return values

    # Round down and distribute remainder by adjusting the largest fractional parts
    int_values: NDArray[np.int_] = np.floor(values).astype(np.int_)
    remainder = self.n_samples - np.sum(int_values)
    fractional_parts = values - int_values
    fractional_parts_indices = np.argsort(-fractional_parts)[:remainder]
    int_values[fractional_parts_indices] += 1
    return int_values

ConstantSampleSize ¶

ConstantSampleSize(
    n_samples: int, lower_bound: int = 0, upper_bound: int | None = None
)

Bases: SampleSizeStrategy

Use a constant number of samples for each set size between two (optional) bounds. The total number of samples (per index) is respected.

PARAMETER	DESCRIPTION
`n_samples`	Total number of samples to generate per index. TYPE: `int`
`lower_bound`	Lower bound for the set size. If the set size is smaller than this, the probability of sampling is 0. TYPE: `int` DEFAULT: `0`
`upper_bound`	Upper bound for the set size. If the set size is larger than this, the probability of sampling is 0. If `None`, the upper bound is set to the number of indices. TYPE: `int \| None` DEFAULT: `None`

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(
    self,
    n_samples: int,
    lower_bound: int = 0,
    upper_bound: int | None = None,
):
    super().__init__(n_samples)
    self.lower_bound = lower_bound
    self.upper_bound = upper_bound

sample_sizes `cached` ¶

sample_sizes(
    n_indices: int, quantize: bool = True
) -> NDArray[int_] | NDArray[float_]

Precomputes the number of samples to take for each set size, from 0 up to n_indices inclusive.

This method corrects rounding errors taking into account the fractional parts so that the total number of samples is respected, while allocating remainders in a way that follows the relative sizes of the fractional parts.

Note

A naive implementation with e.g.

m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
        for k in range(n)]

would not respect the total number of samples, and would not distribute remainders correctly.

PARAMETER	DESCRIPTION
`n_indices`	number of indices in the index set from which to sample. This is typically `len(dataset) - 1` with the usual index iterations. TYPE: `int`
`quantize`	Whether to perform the remainder distribution. If `False`, the raw floating point values are returned. Useful e.g. for RandomSizeIteration where one needs frequencies. In this case `n_samples` can be 1. TYPE: `bool` DEFAULT: `True`

Returns: The exact (integer) number of samples to take for each set size, if quantize is True. Otherwise, the fractional number of samples.

Source code in src/pydvl/valuation/samplers/stratified.py

@lru_cache
def sample_sizes(
    self, n_indices: int, quantize: bool = True
) -> NDArray[np.int_] | NDArray[np.float_]:
    """Precomputes the number of samples to take for each set size, from 0 up to
    `n_indices` inclusive.

    This method corrects rounding errors taking into account the fractional parts
    so that the total number of samples is respected, while allocating remainders
    in a way that follows the relative sizes of the fractional parts.

    ??? Note
        A naive implementation with e.g.
        ```python
        m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
                for k in range(n)]
        ```
        would not respect the total number of samples, and would not distribute
        remainders correctly.

    Args:
        n_indices: number of indices in the index set from which to sample. This is
            typically `len(dataset) - 1` with the usual index iterations.
        quantize: Whether to perform the remainder distribution. If `False`, the raw
            floating point values are returned. Useful e.g. for
            [RandomSizeIteration][pydvl.valuation.samplers.stratified.RandomSizeIteration]
            where one needs frequencies. In this case `n_samples` can
            be 1.
    Returns:
        The exact (integer) number of samples to take for each set size, if
        `quantize` is `True`. Otherwise, the fractional number of samples.
    """

    # m_k = m * f(k) / sum_j f(j)
    values = np.empty(n_indices + 1, dtype=float)
    s = 0.0

    for k in range(n_indices + 1):
        val = self.fun(n_indices, k)
        values[k] = val
        s += val

    values *= self.n_samples / s
    if not quantize:
        return values

    # Round down and distribute remainder by adjusting the largest fractional parts
    int_values: NDArray[np.int_] = np.floor(values).astype(np.int_)
    remainder = self.n_samples - np.sum(int_values)
    fractional_parts = values - int_values
    fractional_parts_indices = np.argsort(-fractional_parts)[:remainder]
    int_values[fractional_parts_indices] += 1
    return int_values

GroupTestingSampleSize ¶

GroupTestingSampleSize(n_samples: int = 1)

Bases: SampleSizeStrategy

Heuristic choice of samples per set size used for Group Testing.

GroupTestingShapleyValuation uses this strategy for the stratified sampling of samples with which to construct the linear problem it requires.

This heuristic sets the number of sets at size $k$ to be

\[m_k = m \frac{f(k)}{\sum_{j=0}^{n-1} f(j)},\]

for a total number of samples $m$ and:

\[ f(k) = \frac{1}{k} + \frac{1}{n-k}, \text{for} k \in \{1, n-1\}. \]

For GT Shapley, $m=1$ and $m_k$ is interpreted as a probability of sampling size $k.$

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, n_samples: int = 1):
    super().__init__(n_samples)

sample_sizes `cached` ¶

sample_sizes(
    n_indices: int, quantize: bool = True
) -> NDArray[int_] | NDArray[float_]

Precomputes the number of samples to take for each set size, from 0 up to n_indices inclusive.

This method corrects rounding errors taking into account the fractional parts so that the total number of samples is respected, while allocating remainders in a way that follows the relative sizes of the fractional parts.

Note

A naive implementation with e.g.

m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
        for k in range(n)]

would not respect the total number of samples, and would not distribute remainders correctly.

PARAMETER	DESCRIPTION
`n_indices`	number of indices in the index set from which to sample. This is typically `len(dataset) - 1` with the usual index iterations. TYPE: `int`
`quantize`	Whether to perform the remainder distribution. If `False`, the raw floating point values are returned. Useful e.g. for RandomSizeIteration where one needs frequencies. In this case `n_samples` can be 1. TYPE: `bool` DEFAULT: `True`

Returns: The exact (integer) number of samples to take for each set size, if quantize is True. Otherwise, the fractional number of samples.

Source code in src/pydvl/valuation/samplers/stratified.py

@lru_cache
def sample_sizes(
    self, n_indices: int, quantize: bool = True
) -> NDArray[np.int_] | NDArray[np.float_]:
    """Precomputes the number of samples to take for each set size, from 0 up to
    `n_indices` inclusive.

    This method corrects rounding errors taking into account the fractional parts
    so that the total number of samples is respected, while allocating remainders
    in a way that follows the relative sizes of the fractional parts.

    ??? Note
        A naive implementation with e.g.
        ```python
        m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
                for k in range(n)]
        ```
        would not respect the total number of samples, and would not distribute
        remainders correctly.

    Args:
        n_indices: number of indices in the index set from which to sample. This is
            typically `len(dataset) - 1` with the usual index iterations.
        quantize: Whether to perform the remainder distribution. If `False`, the raw
            floating point values are returned. Useful e.g. for
            [RandomSizeIteration][pydvl.valuation.samplers.stratified.RandomSizeIteration]
            where one needs frequencies. In this case `n_samples` can
            be 1.
    Returns:
        The exact (integer) number of samples to take for each set size, if
        `quantize` is `True`. Otherwise, the fractional number of samples.
    """

    # m_k = m * f(k) / sum_j f(j)
    values = np.empty(n_indices + 1, dtype=float)
    s = 0.0

    for k in range(n_indices + 1):
        val = self.fun(n_indices, k)
        values[k] = val
        s += val

    values *= self.n_samples / s
    if not quantize:
        return values

    # Round down and distribute remainder by adjusting the largest fractional parts
    int_values: NDArray[np.int_] = np.floor(values).astype(np.int_)
    remainder = self.n_samples - np.sum(int_values)
    fractional_parts = values - int_values
    fractional_parts_indices = np.argsort(-fractional_parts)[:remainder]
    int_values[fractional_parts_indices] += 1
    return int_values

HarmonicSampleSize ¶

HarmonicSampleSize(n_samples: int)

Bases: SampleSizeStrategy

Heuristic choice of samples per set size for VRDS.

Sets the number of sets at size $k$ to be

\[m_k = m \frac{f(k)}{\sum_{j=0}^{n-1} f(j)},\]

for a total number of samples $m$ and:

\[f(k) = \frac{1}{1+k}.\]

PARAMETER	DESCRIPTION
`n_samples`	Number of samples for the stratified sampler to generate, per index. If the sampler uses NoIndexIteration, then this will coincide with the total number of samples. TYPE: `int`

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, n_samples: int):
    """Construct a heuristic for the given number of samples.

    Args:
        n_samples: Number of samples for the stratified sampler to generate,
            **per index**. If the sampler uses
            [NoIndexIteration][pydvl.valuation.samplers.NoIndexIteration], then this
            will coincide with the total number of samples.
    """
    self.n_samples = n_samples

sample_sizes `cached` ¶

sample_sizes(
    n_indices: int, quantize: bool = True
) -> NDArray[int_] | NDArray[float_]

Precomputes the number of samples to take for each set size, from 0 up to n_indices inclusive.

This method corrects rounding errors taking into account the fractional parts so that the total number of samples is respected, while allocating remainders in a way that follows the relative sizes of the fractional parts.

Note

A naive implementation with e.g.

m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
        for k in range(n)]

would not respect the total number of samples, and would not distribute remainders correctly.

PARAMETER	DESCRIPTION
`n_indices`	number of indices in the index set from which to sample. This is typically `len(dataset) - 1` with the usual index iterations. TYPE: `int`
`quantize`	Whether to perform the remainder distribution. If `False`, the raw floating point values are returned. Useful e.g. for RandomSizeIteration where one needs frequencies. In this case `n_samples` can be 1. TYPE: `bool` DEFAULT: `True`

Returns: The exact (integer) number of samples to take for each set size, if quantize is True. Otherwise, the fractional number of samples.

Source code in src/pydvl/valuation/samplers/stratified.py

@lru_cache
def sample_sizes(
    self, n_indices: int, quantize: bool = True
) -> NDArray[np.int_] | NDArray[np.float_]:
    """Precomputes the number of samples to take for each set size, from 0 up to
    `n_indices` inclusive.

    This method corrects rounding errors taking into account the fractional parts
    so that the total number of samples is respected, while allocating remainders
    in a way that follows the relative sizes of the fractional parts.

    ??? Note
        A naive implementation with e.g.
        ```python
        m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
                for k in range(n)]
        ```
        would not respect the total number of samples, and would not distribute
        remainders correctly.

    Args:
        n_indices: number of indices in the index set from which to sample. This is
            typically `len(dataset) - 1` with the usual index iterations.
        quantize: Whether to perform the remainder distribution. If `False`, the raw
            floating point values are returned. Useful e.g. for
            [RandomSizeIteration][pydvl.valuation.samplers.stratified.RandomSizeIteration]
            where one needs frequencies. In this case `n_samples` can
            be 1.
    Returns:
        The exact (integer) number of samples to take for each set size, if
        `quantize` is `True`. Otherwise, the fractional number of samples.
    """

    # m_k = m * f(k) / sum_j f(j)
    values = np.empty(n_indices + 1, dtype=float)
    s = 0.0

    for k in range(n_indices + 1):
        val = self.fun(n_indices, k)
        values[k] = val
        s += val

    values *= self.n_samples / s
    if not quantize:
        return values

    # Round down and distribute remainder by adjusting the largest fractional parts
    int_values: NDArray[np.int_] = np.floor(values).astype(np.int_)
    remainder = self.n_samples - np.sum(int_values)
    fractional_parts = values - int_values
    fractional_parts_indices = np.argsort(-fractional_parts)[:remainder]
    int_values[fractional_parts_indices] += 1
    return int_values

PowerLawSampleSize ¶

PowerLawSampleSize(n_samples: int, exponent: float)

Bases: SampleSizeStrategy

Heuristic choice of samples per set size for VRDS.

Sets the number of sets at size $k$ to be

\[m_k = m \frac{f(k)}{\sum_{j=0}^{n-1} f(j)},\]

for a total number of samples $m$ and:

\[f(k) = (1+k)^a, \]

and some exponent $a.$ With $a=1$ one recovers the HarmonicSampleSize heuristic.

PARAMETER	DESCRIPTION
`n_samples`	Total number of samples to generate per index. TYPE: `int`
`exponent`	The exponent to use. Recommended values are between -1 and -0.5. TYPE: `float`

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, n_samples: int, exponent: float):
    super().__init__(n_samples)
    self.exponent = exponent

sample_sizes `cached` ¶

sample_sizes(
    n_indices: int, quantize: bool = True
) -> NDArray[int_] | NDArray[float_]

Precomputes the number of samples to take for each set size, from 0 up to n_indices inclusive.

This method corrects rounding errors taking into account the fractional parts so that the total number of samples is respected, while allocating remainders in a way that follows the relative sizes of the fractional parts.

Note

A naive implementation with e.g.

m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
        for k in range(n)]

would not respect the total number of samples, and would not distribute remainders correctly.

PARAMETER	DESCRIPTION
`n_indices`	number of indices in the index set from which to sample. This is typically `len(dataset) - 1` with the usual index iterations. TYPE: `int`
`quantize`	Whether to perform the remainder distribution. If `False`, the raw floating point values are returned. Useful e.g. for RandomSizeIteration where one needs frequencies. In this case `n_samples` can be 1. TYPE: `bool` DEFAULT: `True`

Returns: The exact (integer) number of samples to take for each set size, if quantize is True. Otherwise, the fractional number of samples.

Source code in src/pydvl/valuation/samplers/stratified.py

@lru_cache
def sample_sizes(
    self, n_indices: int, quantize: bool = True
) -> NDArray[np.int_] | NDArray[np.float_]:
    """Precomputes the number of samples to take for each set size, from 0 up to
    `n_indices` inclusive.

    This method corrects rounding errors taking into account the fractional parts
    so that the total number of samples is respected, while allocating remainders
    in a way that follows the relative sizes of the fractional parts.

    ??? Note
        A naive implementation with e.g.
        ```python
        m_k = [max(1, int(round(m * f(k)/sum(f(j) for j in range(n)), 0)))
                for k in range(n)]
        ```
        would not respect the total number of samples, and would not distribute
        remainders correctly.

    Args:
        n_indices: number of indices in the index set from which to sample. This is
            typically `len(dataset) - 1` with the usual index iterations.
        quantize: Whether to perform the remainder distribution. If `False`, the raw
            floating point values are returned. Useful e.g. for
            [RandomSizeIteration][pydvl.valuation.samplers.stratified.RandomSizeIteration]
            where one needs frequencies. In this case `n_samples` can
            be 1.
    Returns:
        The exact (integer) number of samples to take for each set size, if
        `quantize` is `True`. Otherwise, the fractional number of samples.
    """

    # m_k = m * f(k) / sum_j f(j)
    values = np.empty(n_indices + 1, dtype=float)
    s = 0.0

    for k in range(n_indices + 1):
        val = self.fun(n_indices, k)
        values[k] = val
        s += val

    values *= self.n_samples / s
    if not quantize:
        return values

    # Round down and distribute remainder by adjusting the largest fractional parts
    int_values: NDArray[np.int_] = np.floor(values).astype(np.int_)
    remainder = self.n_samples - np.sum(int_values)
    fractional_parts = values - int_values
    fractional_parts_indices = np.argsort(-fractional_parts)[:remainder]
    int_values[fractional_parts_indices] += 1
    return int_values

SampleSizeIteration ¶

SampleSizeIteration(strategy: SampleSizeStrategy, n_indices: int)

Bases: ABC

Given a strategy and the number of indices, yield tuples (k, count) that the sampler loop will use. Args: strategy: The strategy to use for computing the number of samples to take. n_indices: The number of indices in the index set from which samples are taken.

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, strategy: SampleSizeStrategy, n_indices: int):
    self.strategy = strategy
    self.n_indices = n_indices

DeterministicSizeIteration ¶

DeterministicSizeIteration(strategy: SampleSizeStrategy, n_indices: int)

Bases: SampleSizeIteration

Generates exactly $m_k$ samples for each set size $k$ before moving to the next.

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, strategy: SampleSizeStrategy, n_indices: int):
    self.strategy = strategy
    self.n_indices = n_indices

RandomSizeIteration ¶

RandomSizeIteration(
    strategy: SampleSizeStrategy, n_indices: int, seed: Seed | None = None
)

Bases: SampleSizeIteration

Draws a set size $k$ following the distribution of sizes given by the strategy.

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(
    self, strategy: SampleSizeStrategy, n_indices: int, seed: Seed | None = None
):
    super().__init__(strategy, n_indices)
    self._rng = np.random.default_rng(seed)

RoundRobinIteration ¶

RoundRobinIteration(strategy: SampleSizeStrategy, n_indices: int)

Bases: SampleSizeIteration

Generates one sample for each set size $k$ before moving to the next.

This continues yielding until every size $k$ has been emitted exactly $m_k$ times. For example, if strategy.sample_sizes() == [2, 3, 1] then we want the sequence: (0,1), (1,1), (2,1), (0,1), (1,1), (1,1)

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(self, strategy: SampleSizeStrategy, n_indices: int):
    self.strategy = strategy
    self.n_indices = n_indices

StratifiedSampler ¶

StratifiedSampler(
    sample_sizes: SampleSizeStrategy,
    sample_sizes_iteration: Type[
        SampleSizeIteration
    ] = DeterministicSizeIteration,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
)

Bases: StochasticSamplerMixin, PowersetSampler

A sampler stratified by coalition size with variable number of samples per set size.

Variance Reduced Stratified Sampler (VRDS)¶

Stratified sampling was introduced at least as early as Maleki et al. (2014)³. Wu et al. 2023², introduced heuristics adequate for ML tasks.

Choosing the number of samples per set size¶

The idea of VRDS is to allow per-set-size configuration of the total number of samples in order to reduce the variance coming from the marginal utility evaluations.

It is known (Wu et al. (2023), Theorem 4.2) that a minimum variance estimator of Shapley values samples a number $m_k$ of sets of size $k$ based on the variance of the marginal utility at that set size. However, this quantity is unknown in practice, so the authors propose a simple heuristic. This function (sample_sizes in the arguments) is deterministic, and in particular does not depend on run-time variance estimates, as an adaptive method might do. Section 4 of Wu et al. (2023) shows a good default choice is based on the harmonic function of the set size $k$ (see HarmonicSampleSize).

PARAMETER	DESCRIPTION
`sample_sizes`	An object which returns the number of samples to take for a given set size. If `index_iteration` below is finite, then the sampler will generate exactly as many samples of each size as returned by this object. If the iteration is infinite, then the `sample_sizes` will be used as probabilities of sampling. TYPE: `SampleSizeStrategy`
`sample_sizes_iteration`	How to loop over sample sizes. The main modes are: * deterministically. For every k generate m_k samples before moving to k+1. * stochastically. Sample sizes k according to the distribution given by `sample_sizes`. * round-robin. Iterate over k, and generate 1 sample each time, until reaching m_k. But more can be created by subclassing SampleSizeIteration. TYPE: `Type[SampleSizeIteration]` DEFAULT: `DeterministicSizeIteration`
`batch_size`	The number of samples to generate per batch. Batches are processed together by each subprocess when working in parallel. TYPE: `int` DEFAULT: `1`
`index_iteration`	the strategy to use for iterating over indices to update. Note that anything other than returning index exactly once will break the weight computation. TYPE: `Type[IndexIteration]` DEFAULT: `FiniteSequentialIndexIteration`
`seed`	The seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

New in version 0.10.0

Source code in src/pydvl/valuation/samplers/stratified.py

def __init__(
    self,
    sample_sizes: SampleSizeStrategy,
    sample_sizes_iteration: Type[SampleSizeIteration] = DeterministicSizeIteration,
    batch_size: int = 1,
    index_iteration: Type[IndexIteration] = FiniteSequentialIndexIteration,
    seed: Seed | None = None,
):
    super().__init__(
        batch_size=batch_size, index_iteration=index_iteration, seed=seed
    )
    self.sample_sizes_strategy = sample_sizes
    self.sample_sizes_iteration = sample_sizes_iteration

skip_indices `property` `writable` ¶

skip_indices

Set of indices to skip in the outer loop.

interrupt ¶

interrupt()

Signals the sampler to stop generating samples after the current batch.

Source code in src/pydvl/valuation/samplers/base.py

def interrupt(self):
    """Signals the sampler to stop generating samples after the current batch."""
    self._interrupted = True

len ¶

__len__() -> int

Returns the length of the current sample generation in generate_batches.

RAISES	DESCRIPTION
`TypeError`	if the sampler is infinite or generate_batches has not been called yet.

Source code in src/pydvl/valuation/samplers/base.py

def __len__(self) -> int:
    """Returns the length of the current sample generation in generate_batches.

    Raises:
        `TypeError`: if the sampler is infinite or
            [generate_batches][pydvl.valuation.samplers.IndexSampler.generate_batches]
            has not been called yet.
    """
    if self._len is None:
        raise TypeError(f"This {self.__class__.__name__} has no length")
    return self._len

generate_batches ¶

generate_batches(indices: IndexSetT) -> BatchGenerator

Batches the samples and yields them.

Source code in src/pydvl/valuation/samplers/base.py

def generate_batches(self, indices: IndexSetT) -> BatchGenerator:
    """Batches the samples and yields them."""
    self._len = self.sample_limit(indices)

    # Create an empty generator if the indices are empty: `return` acts like a
    # `break`, and produces an empty generator.
    if len(indices) == 0:
        return

    self._interrupted = False
    self._n_samples = 0
    for batch in chunked(self.generate(indices), self.batch_size):
        self._n_samples += len(batch)
        yield batch
        if self._interrupted:
            break

result_updater ¶

result_updater(result: ValuationResult) -> ResultUpdater[ValueUpdateT]

Returns a callable that updates a valuation result with a value update.

Because we use log-space computation for numerical stability, the default result updater keeps track of several quantities required to maintain accurate running 1st and 2nd moments.

PARAMETER	DESCRIPTION
`result`	The result to update TYPE: `ValuationResult`

Returns: A callable object that updates the result with a value update

Source code in src/pydvl/valuation/samplers/base.py

def result_updater(self, result: ValuationResult) -> ResultUpdater[ValueUpdateT]:
    """Returns a callable that updates a valuation result with a value update.

    Because we use log-space computation for numerical stability, the default result
    updater keeps track of several quantities required to maintain accurate running
    1st and 2nd moments.

    Args:
        result: The result to update
    Returns:
        A callable object that updates the result with a value update
    """
    return LogResultUpdater(result)

index_iterator ¶

index_iterator(indices: IndexSetT) -> Generator[IndexT | None, None, None]

Iterates over indices with the method specified at construction.

Source code in src/pydvl/valuation/samplers/powerset.py

def index_iterator(
    self, indices: IndexSetT
) -> Generator[IndexT | None, None, None]:
    """Iterates over indices with the method specified at construction."""
    try:
        self._index_iterator = self._index_iterator_cls(indices, seed=self._rng)  # type: ignore
    except (AttributeError, TypeError):
        self._index_iterator = self._index_iterator_cls(indices)
    for idx in self._index_iterator:
        if idx not in self.skip_indices:
            yield idx

log_weight ¶

log_weight(n: int, subset_len: int) -> float

The probability of sampling a set of size k is 1/(n choose k) times the probability of choosing size k, which is the number of samples for that size divided by the total number of samples for all sizes:

\[P(S) = \binom{n}{k}^{-1} \ \frac{m_k}{m},\]

where $m_k$ is the number of samples of size $k$ and $m$ is the total number of samples.

PARAMETER	DESCRIPTION
`n`	Size of the index set. TYPE: `int`
`subset_len`	Size of the subset. TYPE: `int`

Returns: The logarithm of the probability of having sampled a set of size subset_len.

Source code in src/pydvl/valuation/samplers/stratified.py

def log_weight(self, n: int, subset_len: int) -> float:
    r"""The probability of sampling a set of size k is 1/(n choose k) times the
    probability of choosing size k, which is the number of samples for that size
    divided by the total number of samples for all sizes:

    $$P(S) = \binom{n}{k}^{-1} \ \frac{m_k}{m},$$

    where $m_k$ is the number of samples of size $k$ and $m$ is the total number
    of samples.

    Args:
        n: Size of the index set.
        subset_len: Size of the subset.
    Returns:
        The logarithm of the probability of having sampled a set of size `subset_len`.
    """

    n = self._index_iterator_cls.complement_size(n)
    # Depending on whether we sample from complements or not, the total number of
    # samples passed to the heuristic has a different interpretation.
    index_iteration_length = self._index_iterator_cls.length(n)  # type: ignore
    if index_iteration_length is None:
        index_iteration_length = 1
    index_iteration_length = max(1, index_iteration_length)

    # Note that we can simplify the quotient
    # $$ \frac{m_k}{m} =
    #    \frac{m \frac{f (k)}{\sum_j f (j)}}{m} = \frac{f(k)}{\sum_j f (j)} $$
    # so that in the weight computation we can use the function $f$ directly from
    # the strategy, or equivalently, call `sample_sizes(n, quantize=False)`.
    # This is useful for the stochastic iteration, where we have frequencies
    # and m is possibly 1, so that quantization would yield a bunch of zeros.
    funs = self.sample_sizes_strategy.sample_sizes(n, quantize=False)
    total = np.sum(funs)

    return float(
        -logcomb(n, subset_len)
        + np.log(index_iteration_length)
        + np.log(funs[subset_len])
        - np.log(total)
    )

TruncationPolicy ¶

TruncationPolicy()

Bases: ABC

A policy for deciding whether to stop computation of a batch of samples

Statistics are kept on the total number of calls and truncations as n_calls and n_truncations respectively.

ATTRIBUTE	DESCRIPTION
`n_calls`	Number of calls to the policy. TYPE: `int`
`n_truncations`	Number of truncations made by the policy. TYPE: `int`

Todo

Because the policy objects are copied to the workers, the statistics are not accessible from the coordinating process. We need to add methods for this.

Source code in src/pydvl/valuation/samplers/truncation.py

def __init__(self) -> None:
    self.n_calls: int = 0
    self.n_truncations: int = 0

reset `abstractmethod` ¶

reset(utility: UtilityBase)

(Re)set the policy to a state ready for a new permutation.

Source code in src/pydvl/valuation/samplers/truncation.py

@abstractmethod
def reset(self, utility: UtilityBase):
    """(Re)set the policy to a state ready for a new permutation."""
    ...

call ¶

__call__(idx: IndexT, score: float, batch_size: int) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
`idx`	Position in the batch currently being computed. TYPE: `IndexT`
`score`	Last utility computed. TYPE: `float`
`batch_size`	Size of the batch being computed. TYPE: `int`

RETURNS	DESCRIPTION
`bool`	`True` if the computation should be interrupted.

Source code in src/pydvl/valuation/samplers/truncation.py

def __call__(self, idx: IndexT, score: float, batch_size: int) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the batch currently being computed.
        score: Last utility computed.
        batch_size: Size of the batch being computed.

    Returns:
        `True` if the computation should be interrupted.
    """

    ret = self._check(idx, score, batch_size)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

NoTruncation ¶

NoTruncation()

Bases: TruncationPolicy

A policy which never interrupts the computation.

Source code in src/pydvl/valuation/samplers/truncation.py

def __init__(self) -> None:
    self.n_calls: int = 0
    self.n_truncations: int = 0

call ¶

__call__(idx: IndexT, score: float, batch_size: int) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
`idx`	Position in the batch currently being computed. TYPE: `IndexT`
`score`	Last utility computed. TYPE: `float`
`batch_size`	Size of the batch being computed. TYPE: `int`

RETURNS	DESCRIPTION
`bool`	`True` if the computation should be interrupted.

Source code in src/pydvl/valuation/samplers/truncation.py

def __call__(self, idx: IndexT, score: float, batch_size: int) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the batch currently being computed.
        score: Last utility computed.
        batch_size: Size of the batch being computed.

    Returns:
        `True` if the computation should be interrupted.
    """

    ret = self._check(idx, score, batch_size)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

FixedTruncation ¶

FixedTruncation(fraction: float)

Bases: TruncationPolicy

Break a computation after a fixed number of updates.

The experiments in Appendix B of (Ghorbani and Zou, 2019)¹ show that when the training set size is large enough, one can simply truncate the iteration over permutations after a fixed number of steps. This happens because beyond a certain number of samples in a training set, the model becomes insensitive to new ones. Alas, this strongly depends on the data distribution and the model and there is no automatic way of estimating this number.

PARAMETER	DESCRIPTION
`fraction`	Fraction of updates in a batch to compute before stopping (e.g. 0.5 to compute half of the marginals in a permutation). TYPE: `float`

Source code in src/pydvl/valuation/samplers/truncation.py

def __init__(self, fraction: float):
    super().__init__()
    if fraction <= 0 or fraction > 1:
        raise ValueError("fraction must be in (0, 1]")
    self.fraction = fraction
    self.count = 0  # within-permutation count

call ¶

__call__(idx: IndexT, score: float, batch_size: int) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
`idx`	Position in the batch currently being computed. TYPE: `IndexT`
`score`	Last utility computed. TYPE: `float`
`batch_size`	Size of the batch being computed. TYPE: `int`

RETURNS	DESCRIPTION
`bool`	`True` if the computation should be interrupted.

Source code in src/pydvl/valuation/samplers/truncation.py

def __call__(self, idx: IndexT, score: float, batch_size: int) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the batch currently being computed.
        score: Last utility computed.
        batch_size: Size of the batch being computed.

    Returns:
        `True` if the computation should be interrupted.
    """

    ret = self._check(idx, score, batch_size)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

RelativeTruncation ¶

RelativeTruncation(rtol: float, burn_in_fraction: float = 0.0)

Bases: TruncationPolicy

Break a computation if the utility is close enough to the total utility.

This is called "performance tolerance" in (Ghorbani and Zou, 2019)¹.

Warning

Initialization and reset() of this policy imply the computation of the total utility for the dataset, which can be expensive!

PARAMETER	DESCRIPTION
`rtol`	Relative tolerance. The permutation is broken if the last computed utility is within this tolerance of the total utility. TYPE: `float`
`burn_in_fraction`	Fraction of samples within a permutation to wait until actually checking. TYPE: `float` DEFAULT: `0.0`

Source code in src/pydvl/valuation/samplers/truncation.py

def __init__(self, rtol: float, burn_in_fraction: float = 0.0):
    super().__init__()
    assert 0 <= burn_in_fraction <= 1
    self.burn_in_fraction = burn_in_fraction
    self.rtol = rtol
    self.total_utility = 0.0
    self.count = 0  # within-permutation count
    self._is_setup = False

DeviationTruncation ¶

DeviationTruncation(sigmas: float, burn_in_fraction: float = 0.0)

Bases: TruncationPolicy

Break a computation if the last computed utility is close to the total utility.

This is essentially the same as RelativeTruncation, but with the tolerance determined by a multiple of the standard deviation of the utilities.

Danger

This policy can break early if the utility function has high variance. This can lead to gross underestimation of values. Use with caution.

Warning

Initialization and reset() of this policy imply the computation of the total utility for the dataset, which can be expensive!

PARAMETER	DESCRIPTION
`burn_in_fraction`	Fraction of samples within a permutation to wait until actually checking. TYPE: `float` DEFAULT: `0.0`
`sigmas`	Number of standard deviations to use as a threshold. TYPE: `float`

Source code in src/pydvl/valuation/samplers/truncation.py

def __init__(self, sigmas: float, burn_in_fraction: float = 0.0):
    super().__init__()
    assert 0 <= burn_in_fraction <= 1

    self.burn_in_fraction = burn_in_fraction
    self.total_utility = 0.0
    self.count = 0  # within-permutation count
    self.variance = 0.0
    self.mean = 0.0
    self.sigmas = sigmas
    self._is_setup = False

Dataset ¶

Dataset(
    x: NDArray,
    y: NDArray,
    feature_names: Sequence[str] | NDArray[str_] | None = None,
    target_names: Sequence[str] | NDArray[str_] | None = None,
    data_names: Sequence[str] | NDArray[str_] | None = None,
    description: str | None = None,
    multi_output: bool = False,
)

A convenience class to handle datasets.

It holds a dataset, together with info on feature names, target names, and data names. It is used to pass data around to valuation methods.

The underlying data arrays can be accessed via Dataset.data(), which returns the tuple (X, y) as a read-only RawData object. The data can be accessed by indexing the object directly, e.g. dataset[0] will return the data point corresponding to index 0 in dataset. For this base class, this is the same as dataset.data([0]), which is the first point in the data array, but derived classes can behave differently.

PARAMETER	DESCRIPTION
`x`	training data TYPE: `NDArray`
`y`	labels for training data TYPE: `NDArray`
`feature_names`	names of the features of x data TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`target_names`	names of the features of y data TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`data_names`	names assigned to data points. For example, if the dataset is a time series, each entry can be a timestamp which can be referenced directly instead of using a row number. TYPE: `Sequence[str] \| NDArray[str_] \| None` DEFAULT: `None`
`description`	A textual description of the dataset. TYPE: `str \| None` DEFAULT: `None`
`multi_output`	set to `False` if labels are scalars, or to `True` if they are vectors of dimension > 1. TYPE: `bool` DEFAULT: `False`

Changed in version 0.10.0

No longer holds split data, but only x, y.

Changed in version 0.10.0

Slicing now return a new Dataset object, not raw data.

Source code in src/pydvl/valuation/dataset.py

def __init__(
    self,
    x: NDArray,
    y: NDArray,
    feature_names: Sequence[str] | NDArray[np.str_] | None = None,
    target_names: Sequence[str] | NDArray[np.str_] | None = None,
    data_names: Sequence[str] | NDArray[np.str_] | None = None,
    description: str | None = None,
    multi_output: bool = False,
):
    self._x, self._y = check_X_y(
        x, y, multi_output=multi_output, estimator="Dataset"
    )

    def make_names(s: str, a: np.ndarray) -> list[str]:
        n = a.shape[1] if len(a.shape) > 1 else 1
        return [f"{s}{i:0{1 + int(np.log10(n))}d}" for i in range(1, n + 1)]

    self.feature_names = (
        list(feature_names) if feature_names is not None else make_names("x", x)
    )
    self.target_names = (
        list(target_names) if target_names is not None else make_names("y", y)
    )

    if len(self._x.shape) > 1:
        if len(self.feature_names) != self._x.shape[-1]:
            raise ValueError("Mismatching number of features and names")
    if len(self._y.shape) > 1:
        if len(self.target_names) != self._y.shape[-1]:
            raise ValueError("Mismatching number of targets and names")

    self.description = description or "No description"
    self._indices = np.arange(len(self._x), dtype=np.int_)
    self._data_names = (
        np.array(data_names, dtype=np.str_)
        if data_names is not None
        else self._indices.astype(np.str_)
    )

indices `property` ¶

indices: NDArray[int_]

Index of positions in data.x_train.

Contiguous integers from 0 to len(Dataset).

names `property` ¶

names: NDArray[str_]

Names of each individual datapoint.

Used for reporting Shapley values.

n_features `property` ¶

n_features: int

Returns the number of dimensions of a sample.

feature ¶

feature(name: str) -> tuple[slice, int]

Returns a slice for the feature with the given name.

Source code in src/pydvl/valuation/dataset.py

def feature(self, name: str) -> tuple[slice, int]:
    """Returns a slice for the feature with the given name."""
    try:
        return np.index_exp[:, self.feature_names.index(name)]  # type: ignore
    except ValueError:
        raise ValueError(f"Feature {name} is not in {self.feature_names}")

data ¶

data(
    indices: int | slice | Sequence[int] | NDArray[int_] | None = None,
) -> RawData

Given a set of indices, returns the training data that refer to those indices, as a read-only tuple-like structure.

This is used mainly by subclasses of UtilityBase to retrieve subsets of the data from indices.

PARAMETER	DESCRIPTION
`indices`	Optional indices that will be used to select points from the training data. If `None`, the entire training data will be returned. TYPE: `int \| slice \| Sequence[int] \| NDArray[int_] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`RawData`	If `indices` is not `None`, the selected x and y arrays from the training data. Otherwise, the entire dataset.

Source code in src/pydvl/valuation/dataset.py

def data(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> RawData:
    """Given a set of indices, returns the training data that refer to those
    indices, as a read-only tuple-like structure.

    This is used mainly by subclasses of
    [UtilityBase][pydvl.valuation.utility.base.UtilityBase] to retrieve subsets of
    the data from indices.

    Args:
        indices: Optional indices that will be used to select points from
            the training data. If `None`, the entire training data will be
            returned.

    Returns:
        If `indices` is not `None`, the selected x and y arrays from the
            training data. Otherwise, the entire dataset.
    """
    if indices is None:
        return RawData(self._x, self._y)
    return RawData(self._x[indices], self._y[indices])

data_indices ¶

data_indices(indices: Sequence[int] | None = None) -> NDArray[int_]

Returns a subset of indices.

This is equivalent to using Dataset.indices[logical_indices] but allows subclasses to define special behaviour, e.g. when indices in Dataset do not match the indices in the data.

For Dataset, this is a simple pass-through.

PARAMETER	DESCRIPTION
`indices`	A set of indices held by this object TYPE: `Sequence[int] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[int_]`	The indices of the data points in the data array.

Source code in src/pydvl/valuation/dataset.py

def data_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns a subset of indices.

    This is equivalent to using `Dataset.indices[logical_indices]` but allows
    subclasses to define special behaviour, e.g. when indices in `Dataset` do not
    match the indices in the data.

    For `Dataset`, this is a simple pass-through.

    Args:
        indices: A set of indices held by this object

    Returns:
        The indices of the data points in the data array.
    """
    if indices is None:
        return self._indices
    return self._indices[indices]

logical_indices ¶

logical_indices(indices: Sequence[int] | None = None) -> NDArray[int_]

Returns the indices in this Dataset for the given indices in the data array.

This is equivalent to using Dataset.indices[data_indices] but allows subclasses to define special behaviour, e.g. when indices in Dataset do not match the indices in the data.

PARAMETER	DESCRIPTION
`indices`	A set of indices in the data array. TYPE: `Sequence[int] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`NDArray[int_]`	The abstract indices for the given data indices.

Source code in src/pydvl/valuation/dataset.py

def logical_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns the indices in this `Dataset` for the given indices in the data array.

    This is equivalent to using `Dataset.indices[data_indices]` but allows
    subclasses to define special behaviour, e.g. when indices in `Dataset` do not
    match the indices in the data.

    Args:
        indices: A set of indices in the data array.

    Returns:
        The abstract indices for the given data indices.
    """
    if indices is None:
        return self._indices
    return self._indices[indices]

from_sklearn `classmethod` ¶

from_sklearn(
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs,
) -> tuple[Dataset, Dataset]

Constructs two Dataset objects from a sklearn.utils.Bunch, as returned by the load_* functions in scikit-learn toy datasets.

Example

>>> from pydvl.valuation.dataset import Dataset
>>> from sklearn.datasets import load_boston  # noqa
>>> train, test = Dataset.from_sklearn(load_boston())

PARAMETER	DESCRIPTION
`data`	scikit-learn Bunch object. The following attributes are supported: `data`: covariates. `target`: target variables (labels). `feature_names` (optional): the feature names. `target_names` (optional): the target names. `DESCR` (optional): a description. TYPE: `Bunch`
`train_size`	size of the training dataset. Used in `train_test_split` float values represent the fraction of the dataset to include in the training split and should be in (0,1). An integer value sets the absolute number of training samples. TYPE: `int \| float` DEFAULT: `0.8`

the value is automatically set to the complement of the test size. random_state: seed for train / test split stratify_by_target: If True, data is split in a stratified fashion, using the target variable as labels. Read more in scikit-learn's user guide. kwargs: Additional keyword arguments to pass to the Dataset constructor. Use this to pass e.g. is_multi_output.

RETURNS	DESCRIPTION
`tuple[Dataset, Dataset]`	Object with the sklearn dataset

Changed in version 0.6.0

Added kwargs to pass to the Dataset constructor.

Changed in version 0.10.0

Returns a tuple of two Dataset objects.

Source code in src/pydvl/valuation/dataset.py

@classmethod
def from_sklearn(
    cls,
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs,
) -> tuple[Dataset, Dataset]:
    """Constructs two [Dataset][pydvl.valuation.dataset.Dataset] objects from a
    [sklearn.utils.Bunch][], as returned by the `load_*`
    functions in [scikit-learn toy datasets](https://scikit-learn.org/stable/datasets/toy_dataset.html).

    ??? Example
        ```pycon
        >>> from pydvl.valuation.dataset import Dataset
        >>> from sklearn.datasets import load_boston  # noqa
        >>> train, test = Dataset.from_sklearn(load_boston())
        ```

    Args:
        data: scikit-learn Bunch object. The following attributes are supported:

            - `data`: covariates.
            - `target`: target variables (labels).
            - `feature_names` (**optional**): the feature names.
            - `target_names` (**optional**): the target names.
            - `DESCR` (**optional**): a description.
        train_size: size of the training dataset. Used in `train_test_split`
            float values represent the fraction of the dataset to include in the
            training split and should be in (0,1). An integer value sets the
            absolute number of training samples.
    the value is automatically set to the complement of the test size.
        random_state: seed for train / test split
        stratify_by_target: If `True`, data is split in a stratified
            fashion, using the target variable as labels. Read more in
            [scikit-learn's user guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor. Use this to pass e.g. `is_multi_output`.

    Returns:
        Object with the sklearn dataset

    !!! tip "Changed in version 0.6.0"
        Added kwargs to pass to the [Dataset][pydvl.valuation.dataset.Dataset] constructor.
    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two [Dataset][pydvl.valuation.dataset.Dataset] objects.
    """
    x_train, x_test, y_train, y_test = train_test_split(
        data.data,
        data.target,
        train_size=train_size,
        random_state=random_state,
        stratify=data.target if stratify_by_target else None,
    )
    return (
        cls(
            x_train,
            y_train,
            feature_names=data.get("feature_names"),
            target_names=data.get("target_names"),
            description=data.get("DESCR"),
            **kwargs,
        ),
        cls(
            x_test,
            y_test,
            feature_names=data.get("feature_names"),
            target_names=data.get("target_names"),
            description=data.get("DESCR"),
            **kwargs,
        ),
    )

from_arrays `classmethod` ¶

from_arrays(
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs: Any,
) -> tuple[Dataset, Dataset]

Constructs a Dataset object from X and y numpy arrays as returned by the make_* functions in sklearn generated datasets.

Example

>>> from pydvl.valuation.dataset import Dataset
>>> from sklearn.datasets import make_regression
>>> X, y = make_regression()
>>> dataset = Dataset.from_arrays(X, y)

PARAMETER	DESCRIPTION
`X`	numpy array of shape (n_samples, n_features) TYPE: `NDArray`
`y`	numpy array of shape (n_samples,) TYPE: `NDArray`
`train_size`	size of the training dataset. Used in `train_test_split` TYPE: `float` DEFAULT: `0.8`
`random_state`	seed for train / test split TYPE: `int \| None` DEFAULT: `None`
`stratify_by_target`	If `True`, data is split in a stratified fashion, using the y variable as labels. Read more in sklearn's user guide. TYPE: `bool` DEFAULT: `False`
`kwargs`	Additional keyword arguments to pass to the Dataset constructor. Use this to pass e.g. `feature_names` or `target_names`. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`tuple[Dataset, Dataset]`	Object with the passed X and y arrays split across training and test sets.

New in version 0.4.0

Changed in version 0.6.0

Added kwargs to pass to the Dataset constructor.

Changed in version 0.10.0

Returns a tuple of two Dataset objects.

Source code in src/pydvl/valuation/dataset.py

@classmethod
def from_arrays(
    cls,
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs: Any,
) -> tuple[Dataset, Dataset]:
    """Constructs a [Dataset][pydvl.valuation.dataset.Dataset] object from X and y numpy arrays  as
    returned by the `make_*` functions in [sklearn generated datasets](https://scikit-learn.org/stable/datasets/sample_generators.html).

    ??? Example
        ```pycon
        >>> from pydvl.valuation.dataset import Dataset
        >>> from sklearn.datasets import make_regression
        >>> X, y = make_regression()
        >>> dataset = Dataset.from_arrays(X, y)
        ```

    Args:
        X: numpy array of shape (n_samples, n_features)
        y: numpy array of shape (n_samples,)
        train_size: size of the training dataset. Used in `train_test_split`
        random_state: seed for train / test split
        stratify_by_target: If `True`, data is split in a stratified fashion,
            using the y variable as labels. Read more in [sklearn's user
            guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor. Use this to pass
            e.g. `feature_names` or `target_names`.

    Returns:
        Object with the passed X and y arrays split across training and test sets.

    !!! tip "New in version 0.4.0"

    !!! tip "Changed in version 0.6.0"
        Added kwargs to pass to the [Dataset][pydvl.valuation.dataset.Dataset] constructor.

    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two [Dataset][pydvl.valuation.dataset.Dataset] objects.
    """
    x_train, x_test, y_train, y_test = train_test_split(
        X,
        y,
        train_size=train_size,
        random_state=random_state,
        stratify=y if stratify_by_target else None,
    )
    return cls(x_train, y_train, **kwargs), cls(x_test, y_test, **kwargs)

Scorer ¶

Bases: ABC

A scoring callable that takes a model and returns a scalar.

Added in version 0.10.0

ABC added

SupervisedScorer ¶

SupervisedScorer(
    scoring: str
    | SupervisedScorerCallable[SupervisedModelT]
    | SupervisedModelT,
    test_data: Dataset,
    default: float,
    range: tuple[float, float] = (-inf, inf),
    name: str | None = None,
)

Bases: Generic[SupervisedModelT], Scorer

A scoring callable that takes a model, data, and labels and returns a scalar.

PARAMETER	DESCRIPTION
`scoring`	Either a string or callable that can be passed to get_scorer. TYPE: `str \| SupervisedScorerCallable[SupervisedModelT] \| SupervisedModelT`
`test_data`	Dataset where the score will be evaluated. TYPE: `Dataset`
`default`	score to be used when a model cannot be fit, e.g. when too little data is passed, or errors arise. TYPE: `float`
`range`	numerical range of the score function. Some Monte Carlo methods can use this to estimate the number of samples required for a certain quality of approximation. If not provided, it can be read from the `scoring` object if it provides it, for instance if it was constructed with compose_score(). TYPE: `tuple[float, float]` DEFAULT: `(-inf, inf)`
`name`	The name of the scorer. If not provided, the name of the function passed will be used. TYPE: `str \| None` DEFAULT: `None`

New in version 0.5.0

Changed in version 0.10.0

This is now SupervisedScorer and holds the test data used to evaluate the model.

Source code in src/pydvl/valuation/scorers/supervised.py

def __init__(
    self,
    scoring: str | SupervisedScorerCallable[SupervisedModelT] | SupervisedModelT,
    test_data: Dataset,
    default: float,
    range: tuple[float, float] = (-np.inf, np.inf),
    name: str | None = None,
):
    super().__init__()
    if isinstance(scoring, SupervisedModel):
        from sklearn.metrics import check_scoring

        self._scorer = check_scoring(scoring)
        if name is None:
            name = f"Default scorer for {scoring.__class__.__name__}"
    elif isinstance(scoring, str):
        self._scorer = get_scorer(scoring)
        if name is None:
            name = scoring
    else:
        self._scorer = scoring
        if name is None:
            name = getattr(scoring, "__name__", "scorer")
    self.test_data = test_data
    self.default = default
    # TODO: auto-fill from known scorers ?
    self.range = np.array(range, dtype=np.float64)
    self.name = name

SupervisedScorerCallable ¶

Bases: Protocol[SupervisedModelT]

Signature for a scorer

ClasswiseSupervisedScorer ¶

ClasswiseSupervisedScorer(
    scoring: str
    | SupervisedScorerCallable[SupervisedModelT]
    | SupervisedModelT,
    test_data: Dataset,
    default: float = 0.0,
    range: tuple[float, float] = (0, 1),
    in_class_discount_fn: Callable[[float], float] = lambda x: x,
    out_of_class_discount_fn: Callable[[float], float] = exp,
    rescale_scores: bool = True,
    name: str | None = None,
)

Bases: SupervisedScorer[SupervisedModelT]

A Scorer designed for evaluation in classification problems.

The final score is the combination of the in-class and out-of-class scores, which are e.g. the accuracy of the trained model over the instances of the test set with the same, and different, labels, respectively. See the module's documentation for more on this.

These two scores are computed with an "inner" scoring function, which must be provided upon construction.

Multi-class support

The inner score must support multiple class labels if you intend to apply them to a multi-class problem. For instance, 'accuracy' supports multiple classes, but f1 does not. For a two-class classification problem, using f1_weighted is essentially equivalent to using accuracy.

PARAMETER	DESCRIPTION
`scoring`	Name of the scoring function or a callable that can be passed to SupervisedScorer. TYPE: `str \| SupervisedScorerCallable[SupervisedModelT] \| SupervisedModelT`
`default`	Score to use when a model fails to provide a number, e.g. when too little was used to train it, or errors arise. TYPE: `float` DEFAULT: `0.0`
`range`	Numerical range of the score function. Some Monte Carlo methods can use this to estimate the number of samples required for a certain quality of approximation. If not provided, it can be read from the `scoring` object if it provides it, for instance if it was constructed with compose_score. TYPE: `tuple[float, float]` DEFAULT: `(0, 1)`
`in_class_discount_fn`	Continuous, monotonic increasing function used to discount the in-class score. TYPE: `Callable[[float], float]` DEFAULT: `lambda x: x`
`out_of_class_discount_fn`	Continuous, monotonic increasing function used to discount the out-of-class score. TYPE: `Callable[[float], float]` DEFAULT: `exp`
`rescale_scores`	If set to True, the scores will be denormalized. This is particularly useful when the inner score function $a_S$ is calculated by an estimator of the form $rac{1}{N} \sum_i x_i$. TYPE: `bool` DEFAULT: `True`
`name`	Name of the scorer. If not provided, the name of the inner scoring function will be prefixed by `classwise`. TYPE: `str \| None` DEFAULT: `None`

New in version 0.7.1

Source code in src/pydvl/valuation/scorers/classwise.py

def __init__(
    self,
    scoring: str | SupervisedScorerCallable[SupervisedModelT] | SupervisedModelT,
    test_data: Dataset,
    default: float = 0.0,
    range: tuple[float, float] = (0, 1),
    in_class_discount_fn: Callable[[float], float] = lambda x: x,
    out_of_class_discount_fn: Callable[[float], float] = np.exp,
    rescale_scores: bool = True,
    name: str | None = None,
):
    disc_score_in_class = in_class_discount_fn(range[1])
    disc_score_out_of_class = out_of_class_discount_fn(range[1])
    transformed_range = (0, disc_score_in_class * disc_score_out_of_class)
    super().__init__(
        scoring=scoring,
        test_data=test_data,
        range=transformed_range,
        default=default,
        name=name or f"classwise {str(scoring)}",
    )
    self._in_class_discount_fn = in_class_discount_fn
    self._out_of_class_discount_fn = out_of_class_discount_fn
    self.label: int | None = None
    self.num_classes = len(np.unique(self.test_data.data().y))
    self.rescale_scores = rescale_scores

compute_in_and_out_of_class_scores ¶

compute_in_and_out_of_class_scores(
    model: SupervisedModelT, rescale_scores: bool = True
) -> tuple[float, float]

Computes in-class and out-of-class scores using the provided inner scoring function. The result is

\[ a_S(D=\{(x_1, y_1), \dots, (x_K, y_K)\}) = \frac{1}{N} \sum_k s(y(x_k), y_k). \]

In this context, for label $c$ calculations are executed twice: once for $D_c$ and once for $D_{-c}$ to determine the in-class and out-of-class scores, respectively. By default, the raw scores are multiplied by $\frac{|D_c|}{|D|}$ and $\frac{|D_{-c}|}{|D|}$, respectively. This is done to ensure that both scores are of the same order of magnitude. This normalization is particularly useful when the inner score function $a_S$ is calculated by an estimator of the form $\frac{1}{N} \sum_i x_i$, e.g. the accuracy.

PARAMETER	DESCRIPTION
`model`	Model used for computing the score on the validation set. TYPE: `SupervisedModelT`
`rescale_scores`	If set to True, the scores will be denormalized. This is particularly useful when the inner score function $a_S$ is calculated by an estimator of the form $\frac{1}{N} \sum_i x_i$. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`tuple[float, float]`	Tuple containing the in-class and out-of-class scores.

Source code in src/pydvl/valuation/scorers/classwise.py

def compute_in_and_out_of_class_scores(
    self, model: SupervisedModelT, rescale_scores: bool = True
) -> tuple[float, float]:
    r"""
    Computes in-class and out-of-class scores using the provided inner
    scoring function. The result is

    $$
    a_S(D=\{(x_1, y_1), \dots, (x_K, y_K)\}) = \frac{1}{N} \sum_k s(y(x_k), y_k).
    $$

    In this context, for label $c$ calculations are executed twice: once for $D_c$
    and once for $D_{-c}$ to determine the in-class and out-of-class scores,
    respectively. By default, the raw scores are multiplied by $\frac{|D_c|}{|D|}$
    and $\frac{|D_{-c}|}{|D|}$, respectively. This is done to ensure that both
    scores are of the same order of magnitude. This normalization is particularly
    useful when the inner score function $a_S$ is calculated by an estimator of the
    form $\frac{1}{N} \sum_i x_i$, e.g. the accuracy.

    Args:
        model: Model used for computing the score on the validation set.
        rescale_scores: If set to True, the scores will be denormalized. This is
            particularly useful when the inner score function $a_S$ is calculated by
            an estimator of the form $\frac{1}{N} \sum_i x_i$.

    Returns:
        Tuple containing the in-class and out-of-class scores.
    """
    if self.label is None:
        raise ValueError(
            "The scorer's label attribute should be set before calling it"
        )

    scorer = self._scorer
    label_set_match = self.test_data.data().y == self.label
    label_set = np.where(label_set_match)[0]

    if len(label_set) == 0:
        return 0, 1 / max(1, self.num_classes - 1)

    complement_label_set = np.where(~label_set_match)[0]
    in_class_score = scorer(model, *self.test_data.data(label_set))
    out_of_class_score = scorer(model, *self.test_data.data(complement_label_set))

    if rescale_scores:
        # TODO: This can lead to NaN values
        #       We should clearly indicate this to users
        _, y_test = self.test_data.data()
        n_in_class = np.count_nonzero(y_test == self.label)
        n_out_of_class = len(y_test) - n_in_class
        in_class_score *= n_in_class / (n_in_class + n_out_of_class)
        out_of_class_score *= n_out_of_class / (n_in_class + n_out_of_class)

    return in_class_score, out_of_class_score

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/valuation/stopping.py

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

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    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$.

Warning

This criterion should be used with care. The standard error is a measure of the uncertainty of the estimate, but it does not guarantee that the estimate is close to the true value. For example, if the utility function is very noisy, the standard error might be very low, but the estimate might be far from the true value. In this case, one might want to use a RankCorrelation instead, which checks whether the rank of the values is stable.

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`
`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/valuation/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

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxChecks ¶

MaxChecks(n_checks: int | None, 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. Note that this might have different interpretations depending on the sampler. For example, MSRSampler performs a single utility evaluation to update all indices, so that's len(training_data) checks for a single training of the model. But it also only changes the counts field of the ValuationResult for about half of the indices, which is what e.g. MaxUpdates checks.

PARAMETER	DESCRIPTION
`n_checks`	Threshold: if `None`, no _check is performed, effectively creating a (never) stopping criterion that always returns `Pending`. TYPE: `int \| None`
`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/valuation/stopping.py

def __init__(self, n_checks: int | None, 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 or np.inf

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxUpdates ¶

MaxUpdates(n_updates: int | None, 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: `int \| None`
`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/valuation/stopping.py

def __init__(self, n_updates: int | None, 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

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

NoStopping ¶

NoStopping(sampler: IndexSampler | None = None, modify_result: bool = True)

Bases: StoppingCriterion

Keep running forever or until sampling stops.

If a sampler instance is passed, and it is a finite sampler, its counter will be used to update completion status.

PARAMETER	DESCRIPTION
`sampler`	A sampler instance to use for completion status. TYPE: `IndexSampler \| None` DEFAULT: `None`
`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/valuation/stopping.py

def __init__(self, sampler: IndexSampler | None = None, modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    self.sampler = sampler

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MinUpdates ¶

MinUpdates(n_updates: int | None, 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: `int \| None`
`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/valuation/stopping.py

def __init__(self, n_updates: int | None, modify_result: bool = True):
    super().__init__(modify_result=modify_result)
    self.n_updates = n_updates
    self.last_min = 0
    self._actual_completion = 0.0

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

MaxTime ¶

MaxTime(seconds: float | None, 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: `float \| None`
`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/valuation/stopping.py

def __init__(self, seconds: float | None, 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()

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    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`	Compare values after so many steps. A step is one evaluation of the criterion, which happens once per batch. 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/valuation/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 rtol <= 0 or rtol >= 1:
        raise ValueError("rtol must be in (0, 1)")

    self.memory = RollingMemory(n_steps + 1, default=np.inf, dtype=np.float64)
    self.rtol = rtol
    self.update_op = np.logical_or if pin_converged else np.logical_and

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    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, fraction: float = 1.0, modify_result: bool = True
)

Bases: StoppingCriterion

A check for stability of Spearman correlation between checks.

Convergence is reached when the change in rank correlation between two successive iterations is below a given threshold.

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

The meaning of successive iterations

Stopping criteria in pyDVL are typically evaluated after each batch of value updates is received. This can imply very different things, depending on the configuration of the samplers. For this reason, RankCorrelation keeps itself track of the number of updates that each index has seen, and only checks for correlation changes when a given fraction of all indices has been updated more than burn_in times and once since last time the criterion was checked.

PARAMETER	DESCRIPTION
`rtol`	Relative tolerance for convergence ($\epsilon$ in the formula) TYPE: `float`
`burn_in`	The minimum number of updates an index must have seen before checking for convergence. This is required because the first correlation checks are usually meaningless. TYPE: `int`
`fraction`	The fraction of values that must have been updated between two correlation checks. This is to avoid comparing two results where only one value has been updated, which would have almost perfect rank correlation. TYPE: `float` DEFAULT: `1.0`
`modify_result`	If `True`, the status of the input ValuationResult is modified in place after the call. TYPE: `bool` DEFAULT: `True`

Added in 0.9.0

Changed in 0.10.0

The behaviour of the burn_in parameter was changed to look at value updates. The parameter fraction was added.

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

def __init__(
    self,
    rtol: float,
    burn_in: int,
    fraction: float = 1.0,
    modify_result: bool = True,
):
    super().__init__(modify_result=modify_result)
    if rtol < 0 or rtol > 1:
        raise ValueError("rtol must be in [0, 1]")
    if fraction <= 0 or fraction > 1:
        raise ValueError("fraction must be in (0, 1]")

    self.rtol = rtol
    self.burn_in = burn_in
    self.fraction = fraction
    self.memory = RollingMemory(n_steps=2, default=np.nan, dtype=np.float64)
    self.count_memory = RollingMemory(n_steps=2, default=0, dtype=np.int_)
    self._corr = np.nan
    self._completion = 0.0

count `property` ¶

count: int

The number of times that the criterion has been checked.

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/valuation/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."
        )
    self._count += 1
    if self._converged.size == 0:
        self._converged = np.full_like(result.indices, False, dtype=bool)
    status = self._check(result)
    if self.modify_result:  # FIXME: this is not nice
        result._status = status
    return status

ModelUtility ¶

ModelUtility(
    model: ModelT,
    scorer: Scorer,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
)

Bases: UtilityBase[SampleT], Generic[SampleT, ModelT]

Convenience wrapper with configurable memoization of the utility.

An instance of ModelUtility holds the tuple of model, and scoring function which determines the value of data points. This is used for the computation of all game-theoretic values like Shapley values and the Least Core.

ModelUtility expects the model to fulfill at least the BaseModel interface, i.e. to have a fit() method

When calling the utility, the model will be cloned if it is a Scikit-Learn model, otherwise a copy is created using copy.deepcopy

Since evaluating the scoring function requires retraining the model and that can be time-consuming, this class wraps it and caches the results of each execution. Caching is available both locally and across nodes, but must always be enabled for your project first, because most stochastic methods do not benefit much from it. See the documentation and the module documentation.

ATTRIBUTE	DESCRIPTION
`model`	The supervised model. TYPE: `ModelT`
`scorer`	A scoring function. If None, the `score()` method of the model will be used. See score for ways to create and compose scorers, in particular how to set default values and ranges. TYPE: `Scorer`

PARAMETER	DESCRIPTION
`model`	Any supervised model. Typical choices can be found in the sci-kit learn documentation. TYPE: `ModelT`
`scorer`	A scoring object. If None, the `score()` method of the model will be used. See scorers for ways to create and compose scorers, in particular how to set default values and ranges. For convenience, a string can be passed, which will be used to construct a SupervisedScorer. TYPE: `Scorer`
`catch_errors`	set to `True` to catch the errors when `fit()` fails. This could happen in several steps of the pipeline, e.g. when too little training data is passed, which happens often during Shapley value calculations. When this happens, the scorer's default value is returned as a score and computation continues. TYPE: `bool` DEFAULT: `True`
`show_warnings`	Set to `False` to suppress warnings thrown by `fit()`. TYPE: `bool` DEFAULT: `False`
`cache_backend`	Optional instance of CacheBackend used to memoize results to avoid duplicate computation. Note however, that for most stochastic methods, cache hits are rare, making the memory expense of caching not worth it (YMMV). TYPE: `CacheBackend \| None` DEFAULT: `None`
`cached_func_options`	Optional configuration object for cached utility evaluation. TYPE: `CachedFuncConfig \| None` DEFAULT: `None`
`clone_before_fit`	If `True`, the model will be cloned before calling `fit()`. TYPE: `bool` DEFAULT: `True`

Example

>>> from pydvl.valuation.utility import ModelUtility, DataUtilityLearning
>>> from pydvl.valuation.dataset import Dataset
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from sklearn.datasets import load_iris
>>> train, test = Dataset.from_sklearn(load_iris(), random_state=16)
>>> u = ModelUtility(LogisticRegression(random_state=16), Scorer("accuracy"))
>>> u(Sample(subset=dataset.indices))
0.9

With caching enabled:

>>> from pydvl.valuation.utility import ModelUtility, DataUtilityLearning
>>> from pydvl.valuation.dataset import Dataset
>>> from pydvl.utils.caching.memory import InMemoryCacheBackend
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from sklearn.datasets import load_iris
>>> train, test = Dataset.from_sklearn(load_iris(), random_state=16)
>>> cache_backend = InMemoryCacheBackend()
>>> u = ModelUtility(LogisticRegression(random_state=16), Scorer("accuracy"), cache_backend=cache_backend)
>>> u(Sample(subset=train.indices))
0.9

Source code in src/pydvl/valuation/utility/modelutility.py

def __init__(
    self,
    model: ModelT,
    scorer: Scorer,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
):
    self.clone_before_fit = clone_before_fit
    self.model = self._maybe_clone_model(model, clone_before_fit)
    self.scorer = scorer
    self.catch_errors = catch_errors
    self.show_warnings = show_warnings
    self.cache = cache_backend
    if cached_func_options is None:
        cached_func_options = CachedFuncConfig()
    # TODO: Find a better way to do this.
    if cached_func_options.hash_prefix is None:
        # FIX: This does not handle reusing the same across runs.
        cached_func_options.hash_prefix = str(hash((model, scorer)))
    self.cached_func_options = cached_func_options
    self._initialize_utility_wrapper()

training_data `property` ¶

training_data: Dataset | None

Retrieves the training data used by this utility.

This property is read-only. In order to set it, use with_dataset().

cache_stats `property` ¶

cache_stats: CacheStats | None

Cache statistics are gathered when cache is enabled. See CacheStats for all fields returned.

with_dataset ¶

with_dataset(data: Dataset, copy: bool = True) -> Self

Returns the utility, or a copy of it, with the given dataset. Args: data: The dataset to use for utility fitting (training data) copy: Whether to copy the utility object or not. Valuation methods should always make copies to avoid unexpected side effects. Returns: The utility object.

Source code in src/pydvl/valuation/utility/base.py

def with_dataset(self, data: Dataset, copy: bool = True) -> Self:
    """Returns the utility, or a copy of it, with the given dataset.
    Args:
        data: The dataset to use for utility fitting (training data)
        copy: Whether to copy the utility object or not. Valuation methods should
            always make copies to avoid unexpected side effects.
    Returns:
        The utility object.
    """
    utility = cp.copy(self) if copy else self
    utility._training_data = data
    return utility

call ¶

__call__(sample: SampleT | None) -> float

PARAMETER	DESCRIPTION
`sample`	contains a subset of valid indices for the `x_train` attribute of Dataset. TYPE: `SampleT \| None`

Source code in src/pydvl/valuation/utility/modelutility.py

def __call__(self, sample: SampleT | None) -> float:
    """
    Args:
        sample: contains a subset of valid indices for the
            `x_train` attribute of [Dataset][pydvl.utils.dataset.Dataset].
    """
    if sample is None or len(sample.subset) == 0:
        return self.scorer.default

    return cast(float, self._utility_wrapper(sample))

ClasswiseModelUtility ¶

ClasswiseModelUtility(
    model: SupervisedModel,
    scorer: ClasswiseSupervisedScorer,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
)

Bases: ModelUtility[ClasswiseSample, SupervisedModel]

ModelUtility class that is specific to classwise shapley valuation.

It expects a classwise scorer and a classification task.

PARAMETER	DESCRIPTION
`model`	Any supervised model. Typical choices can be found in the sci-kit learn documentation. TYPE: `SupervisedModel`
`scorer`	A class-wise scoring object. TYPE: `ClasswiseSupervisedScorer`
`catch_errors`	set to `True` to catch the errors when `fit()` fails. This could happen in several steps of the pipeline, e.g. when too little training data is passed, which happens often during Shapley value calculations. When this happens, the scorer's default value is returned as a score and computation continues. TYPE: `bool` DEFAULT: `True`
`show_warnings`	Set to `False` to suppress warnings thrown by `fit()`. TYPE: `bool` DEFAULT: `False`
`cache_backend`	Optional instance of CacheBackend used to wrap the _utility method of the Utility instance. By default, this is set to None and that means that the utility evaluations will not be cached. TYPE: `CacheBackend \| None` DEFAULT: `None`
`cached_func_options`	Optional configuration object for cached utility evaluation. TYPE: `CachedFuncConfig \| None` DEFAULT: `None`
`clone_before_fit`	If `True`, the model will be cloned before calling `fit()`. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/utility/classwise.py

def __init__(
    self,
    model: SupervisedModel,
    scorer: ClasswiseSupervisedScorer,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
):
    super().__init__(
        model,
        scorer,
        catch_errors=catch_errors,
        show_warnings=show_warnings,
        cache_backend=cache_backend,
        cached_func_options=cached_func_options,
        clone_before_fit=clone_before_fit,
    )
    if not isinstance(self.scorer, ClasswiseSupervisedScorer):
        raise ValueError("Scorer must be an instance of ClasswiseSupervisedScorer")
    self.scorer: ClasswiseSupervisedScorer

training_data `property` ¶

training_data: Dataset | None

Retrieves the training data used by this utility.

This property is read-only. In order to set it, use with_dataset().

cache_stats `property` ¶

cache_stats: CacheStats | None

Cache statistics are gathered when cache is enabled. See CacheStats for all fields returned.

with_dataset ¶

with_dataset(data: Dataset, copy: bool = True) -> Self

Returns the utility, or a copy of it, with the given dataset. Args: data: The dataset to use for utility fitting (training data) copy: Whether to copy the utility object or not. Valuation methods should always make copies to avoid unexpected side effects. Returns: The utility object.

Source code in src/pydvl/valuation/utility/base.py

def with_dataset(self, data: Dataset, copy: bool = True) -> Self:
    """Returns the utility, or a copy of it, with the given dataset.
    Args:
        data: The dataset to use for utility fitting (training data)
        copy: Whether to copy the utility object or not. Valuation methods should
            always make copies to avoid unexpected side effects.
    Returns:
        The utility object.
    """
    utility = cp.copy(self) if copy else self
    utility._training_data = data
    return utility

call ¶

__call__(sample: SampleT | None) -> float

PARAMETER	DESCRIPTION
`sample`	contains a subset of valid indices for the `x_train` attribute of Dataset. TYPE: `SampleT \| None`

Source code in src/pydvl/valuation/utility/modelutility.py

def __call__(self, sample: SampleT | None) -> float:
    """
    Args:
        sample: contains a subset of valid indices for the
            `x_train` attribute of [Dataset][pydvl.utils.dataset.Dataset].
    """
    if sample is None or len(sample.subset) == 0:
        return self.scorer.default

    return cast(float, self._utility_wrapper(sample))

KNNClassifierUtility ¶

KNNClassifierUtility(
    model: KNeighborsClassifier,
    test_data: Dataset,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
)

Bases: ModelUtility[Sample, KNeighborsClassifier]

Utility object for KNN Classifiers.

The utility function is the model's predicted probability for the true class.

Uses of this utility

Although this class can be used in conjunction with any semi-value method and sampler, when computing Shapley values, it is recommended to use the dedicated class KNNShapleyValuation, because it implements a more efficient algorithm for computing Shapley values which runs in O(n log n) time for each test point.

PARAMETER	DESCRIPTION
`model`	A KNN classifier model. TYPE: `KNeighborsClassifier`
`test_data`	The test data to evaluate the model on. TYPE: `Dataset`
`catch_errors`	set to `True` to catch the errors when `fit()` fails. This could happen in several steps of the pipeline, e.g. when too little training data is passed, which happens often during Shapley value calculations. When this happens, the scorer's default value is returned as a score and computation continues. TYPE: `bool` DEFAULT: `True`
`show_warnings`	Set to `False` to suppress warnings thrown by `fit()`. TYPE: `bool` DEFAULT: `False`
`cache_backend`	Optional instance of [CacheBackend][ pydvl.utils.caching.base.CacheBackend] used to wrap the _utility method of the Utility instance. By default, this is set to None and that means that the utility evaluations will not be cached. TYPE: `CacheBackend \| None` DEFAULT: `None`
`cached_func_options`	Optional configuration object for cached utility evaluation. TYPE: `CachedFuncConfig \| None` DEFAULT: `None`
`clone_before_fit`	If `True`, the model will be cloned before calling `fit()` in utility evaluations. TYPE: `bool` DEFAULT: `True`

Source code in src/pydvl/valuation/utility/knn.py

def __init__(
    self,
    model: KNeighborsClassifier,
    test_data: Dataset,
    *,
    catch_errors: bool = True,
    show_warnings: bool = False,
    cache_backend: CacheBackend | None = None,
    cached_func_options: CachedFuncConfig | None = None,
    clone_before_fit: bool = True,
):
    self.test_data = test_data
    self.sorted_neighbors: NDArray[np.int_] | None = None
    dummy_scorer = _DummyScorer()

    super().__init__(
        model=model,
        scorer=dummy_scorer,  # not applicable
        catch_errors=catch_errors,
        show_warnings=show_warnings,
        cache_backend=cache_backend,
        cached_func_options=cached_func_options,
        clone_before_fit=clone_before_fit,
    )

training_data `property` ¶

training_data: Dataset | None

Retrieves the training data used by this utility.

This property is read-only. In order to set it, use with_dataset().

cache_stats `property` ¶

cache_stats: CacheStats | None

Cache statistics are gathered when cache is enabled. See CacheStats for all fields returned.

call ¶

__call__(sample: SampleT | None) -> float

PARAMETER	DESCRIPTION
`sample`	contains a subset of valid indices for the `x_train` attribute of Dataset. TYPE: `SampleT \| None`

Source code in src/pydvl/valuation/utility/modelutility.py

def __call__(self, sample: SampleT | None) -> float:
    """
    Args:
        sample: contains a subset of valid indices for the
            `x_train` attribute of [Dataset][pydvl.utils.dataset.Dataset].
    """
    if sample is None or len(sample.subset) == 0:
        return self.scorer.default

    return cast(float, self._utility_wrapper(sample))

with_dataset ¶

with_dataset(data: Dataset, copy: bool = True) -> Self

Return the utility, or a copy of it, with the given dataset and the model fitted on it.

PARAMETER	DESCRIPTION
`data`	The dataset to use. TYPE: `Dataset`
`copy`	Whether to copy the utility object or not. Additionally, if `True` then the model is also cloned. If `False`, the model is only cloned if `clone_before_fit` is `True`. TYPE: `bool` DEFAULT: `True`

Returns: The utility object.

Source code in src/pydvl/valuation/utility/knn.py

def with_dataset(self, data: Dataset, copy: bool = True) -> Self:
    """Return the utility, or a copy of it, with the given dataset and the model
    fitted on it.

    Args:
        data: The dataset to use.
        copy: Whether to copy the utility object or not. Additionally, if `True`
            then the model is also cloned. If `False`, the model is only cloned if
            `clone_before_fit` is `True`.
    Returns:
        The utility object.
    """
    utility: Self = super().with_dataset(data, copy)
    if copy or self.clone_before_fit:
        utility.model = self._maybe_clone_model(self.model, do_clone=True)
    utility.model.fit(*data.data())
    return utility

UtilityModel ¶

Bases: ABC

Interface for utility models.

A utility model predicts the value of a utility function given a sample. The model is trained on a collection of samples and their respective utility values. These tuples are called Utility Samples.

Utility models:

are fitted on dictionaries of Sample -> utility value
predict: Collection[samples] -> NDArray[utility values]

IndicatorUtilityModel ¶

IndicatorUtilityModel(predictor: SupervisedModel, n_data: int)

Bases: UtilityModel

A simple wrapper for arbitrary predictors.

Uses 1-hot encoding of the indices as input for the model, as done in Wang et al., (2022)¹.

Source code in src/pydvl/valuation/utility/learning.py

def __init__(self, predictor: SupervisedModel, n_data: int):
    self.n_data = n_data
    self.predictor = predictor

DataUtilityLearning ¶

DataUtilityLearning(
    utility: UtilityBase,
    training_budget: int,
    model: UtilityModel,
    show_warnings: bool = True,
)

Bases: UtilityBase[SampleT]

This object wraps any class derived from UtilityBase and delegates calls to it, up until a given budget (number of iterations). Every tuple of input and output (a so-called utility sample) is stored. Once the budget is exhausted, DataUtilityLearning fits the given model to the utility samples. Subsequent calls will use the learned model to predict the utility instead of delegating.

PARAMETER	DESCRIPTION
`utility`	The utility to learn. Typically, this will be a ModelUtility object encapsulating a machine learning model which requires fitting on each evaluation of the utility. TYPE: `UtilityBase`
`training_budget`	Number of utility samples to collect before fitting the given model. TYPE: `int`
`model`	A supervised regression model TYPE: `UtilityModel`

Example

from pydvl.valuation import Dataset, DataUtilityLearning, ModelUtility,             Sample, SupervisedScorer
from sklearn.linear_model import LinearRegression, LogisticRegression
from sklearn.datasets import load_iris

train, test = Dataset.from_sklearn(load_iris())
scorer = SupervisedScorer("accuracy", test, 0, (0,1))
utility = ModelUtility(LinearRegression(), scorer)
utility_model = IndicatorUtilityModel(LinearRegression(), len(train))
dul = DataUtilityLearning(utility, 3, utility_model)
# First 3 calls will be computed normally
for i in range(3):
    _ = dul(Sample(0, np.array([])))
# Subsequent calls will be computed using the fitted utility_model
dul(Sample(0, np.array([1, 2, 3])))

Source code in src/pydvl/valuation/utility/learning.py

def __init__(
    self,
    utility: UtilityBase,
    training_budget: int,
    model: UtilityModel,
    show_warnings: bool = True,
) -> None:
    self.utility = utility
    self.training_budget = training_budget
    self.model = model
    self.n_predictions = 0
    self.show_warnings = show_warnings
    self._is_fitted = False
    self._utility_samples: dict[Sample, float] = {}

training_data `property` ¶

training_data: Dataset | None

Retrieves the training data used by this utility.

This property is read-only. In order to set it, use with_dataset().

with_dataset ¶

with_dataset(data: Dataset, copy: bool = True) -> Self

Returns the utility, or a copy of it, with the given dataset. Args: data: The dataset to use for utility fitting (training data) copy: Whether to copy the utility object or not. Valuation methods should always make copies to avoid unexpected side effects. Returns: The utility object.

Source code in src/pydvl/valuation/utility/base.py

def with_dataset(self, data: Dataset, copy: bool = True) -> Self:
    """Returns the utility, or a copy of it, with the given dataset.
    Args:
        data: The dataset to use for utility fitting (training data)
        copy: Whether to copy the utility object or not. Valuation methods should
            always make copies to avoid unexpected side effects.
    Returns:
        The utility object.
    """
    utility = cp.copy(self) if copy else self
    utility._training_data = data
    return utility

point_wise_accuracy ¶

point_wise_accuracy(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]

Point-wise accuracy, or 0-1 score between two arrays.

Higher is better.

PARAMETER	DESCRIPTION
`y_true`	Array of true values (e.g. labels) TYPE: `NDArray[T]`
`y_pred`	Array of estimated values (e.g. model predictions) TYPE: `NDArray[T]`

RETURNS	DESCRIPTION
`NDArray[T]`	Array with point-wise 0-1 accuracy between labels and model predictions

Source code in src/pydvl/valuation/methods/data_oob.py

def point_wise_accuracy(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]:
    """Point-wise accuracy, or 0-1 score between two arrays.

    Higher is better.

    Args:
        y_true: Array of true values (e.g. labels)
        y_pred: Array of estimated values (e.g. model predictions)

    Returns:
        Array with point-wise 0-1 accuracy between labels and model predictions
    """
    return np.array(y_pred == y_true, dtype=y_pred.dtype)

neg_l2_distance ¶

neg_l2_distance(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]

Point-wise negative $l_2$ distance between two arrays.

Higher is better.

PARAMETER	DESCRIPTION
`y_true`	Array of true values (e.g. labels) TYPE: `NDArray[T]`
`y_pred`	Array of estimated values (e.g. model predictions) TYPE: `NDArray[T]`

RETURNS	DESCRIPTION
`NDArray[T]`	Array with point-wise negative $l_2$ distances between labels and model
`NDArray[T]`	predictions

Source code in src/pydvl/valuation/methods/data_oob.py

def neg_l2_distance(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]:
    r"""Point-wise negative $l_2$ distance between two arrays.

    Higher is better.

    Args:
        y_true: Array of true values (e.g. labels)
        y_pred: Array of estimated values (e.g. model predictions)

    Returns:
        Array with point-wise negative $l_2$ distances between labels and model
        predictions
    """
    return -np.square(np.array(y_pred - y_true), dtype=y_pred.dtype)

compute_n_samples ¶

compute_n_samples(epsilon: float, delta: float, n_obs: int) -> int

Compute the minimal sample size with epsilon-delta guarantees.

Based on the formula in Theorem 4 of (Jia, R. et al., 2023)² which gives a lower bound on the number of samples required to obtain an (ε/√n,δ/(N(N-1))-approximation to all pair-wise differences of Shapley values, wrt. $\ell_2$ norm.

The updated version refines the lower bound of the original paper. Note that the bound is tighter than earlier versions but might still overestimate the number of samples required.

PARAMETER	DESCRIPTION
`epsilon`	The error tolerance. TYPE: `float`
`delta`	The confidence level. TYPE: `float`
`n_obs`	Number of data points. TYPE: `int`

RETURNS	DESCRIPTION
`int`	The sample size.

Source code in src/pydvl/valuation/methods/gt_shapley.py

def compute_n_samples(epsilon: float, delta: float, n_obs: int) -> int:
    r"""Compute the minimal sample size with epsilon-delta guarantees.

    Based on the formula in Theorem 4 of
    (Jia, R. et al., 2023)<sup><a href="#jia_update_2023">2</a></sup>
    which gives a lower bound on the number of samples required to obtain an
    (ε/√n,δ/(N(N-1))-approximation to all pair-wise differences of Shapley
    values, wrt. $\ell_2$ norm.

    The updated version refines the lower bound of the original paper. Note that the
    bound is tighter than earlier versions but might still overestimate the number of
    samples required.

    Args:
        epsilon: The error tolerance.
        delta: The confidence level.
        n_obs: Number of data points.

    Returns:
        The sample size.

    """
    kk = _create_sample_sizes(n_obs)
    Z = _calculate_z(n_obs)

    q = _create_sampling_probabilities(kk)
    q_tot = (n_obs - 2) / n_obs * q[0] + np.inner(
        q[1:], 1 + 2 * kk[1:] * (kk[1:] - n_obs) / (n_obs * (n_obs - 1))
    )

    def _h(u: float) -> float:
        return cast(float, (1 + u) * np.log(1 + u) - u)

    n_samples = np.log(n_obs * (n_obs - 1) / delta)
    n_samples /= 1 - q_tot
    n_samples /= _h(epsilon / (2 * Z * np.sqrt(n_obs) * (1 - q_tot)))

    return int(n_samples)

get_unique_labels ¶

get_unique_labels(array: NDArray) -> NDArray

Returns unique labels in a categorical dataset.

PARAMETER	DESCRIPTION
`array`	The input array to find unique labels from. It should be of categorical types such as Object, String, Unicode, Unsigned integer, Signed integer, or Boolean. TYPE: `NDArray`

RETURNS	DESCRIPTION
`NDArray`	An array of unique labels.

RAISES	DESCRIPTION
`ValueError`	If the input array is not of a categorical type.

Source code in src/pydvl/valuation/samplers/classwise.py

def get_unique_labels(array: NDArray) -> NDArray:
    """Returns unique labels in a categorical dataset.

    Args:
        array: The input array to find unique labels from. It should be of
               categorical types such as Object, String, Unicode, Unsigned
               integer, Signed integer, or Boolean.

    Returns:
        An array of unique labels.

    Raises:
        ValueError: If the input array is not of a categorical type.
    """
    # Object, String, Unicode, Unsigned integer, Signed integer, boolean
    if array.dtype.kind in "OSUiub":
        return cast(NDArray, np.unique(array))
    raise ValueError(
        f"Input array has an unsupported data type for categorical labels: {array.dtype}. "
        "Expected types: Object, String, Unicode, Unsigned integer, Signed integer, or Boolean."
    )

compose_score ¶

compose_score(
    scorer: SupervisedScorer,
    transformation: Callable[[float], float],
    name: str,
) -> SupervisedScorer

Composes a scoring function with an arbitrary scalar transformation.

Useful to squash unbounded scores into ranges manageable by data valuation methods.

Example

sigmoid = lambda x: 1/(1+np.exp(-x))
compose_score(Scorer("r2"), sigmoid, range=(0,1), name="squashed r2")

PARAMETER	DESCRIPTION
`scorer`	The object to be composed. TYPE: `SupervisedScorer`
`transformation`	A scalar transformation TYPE: `Callable[[float], float]`
`name`	A string representation for the composition, for `str()`. TYPE: `str`

RETURNS	DESCRIPTION
`SupervisedScorer`	The composite SupervisedScorer.

Source code in src/pydvl/valuation/scorers/utils.py

def compose_score(
    scorer: SupervisedScorer,
    transformation: Callable[[float], float],
    name: str,
) -> SupervisedScorer:
    """Composes a scoring function with an arbitrary scalar transformation.

    Useful to squash unbounded scores into ranges manageable by data valuation
    methods.

    ??? Example
        ```python
        sigmoid = lambda x: 1/(1+np.exp(-x))
        compose_score(Scorer("r2"), sigmoid, range=(0,1), name="squashed r2")
        ```

    Args:
        scorer: The object to be composed.
        transformation: A scalar transformation
        name: A string representation for the composition, for `str()`.

    Returns:
        The composite [SupervisedScorer][pydvl.valuation.scorers.SupervisedScorer].
    """

    class CompositeSupervisedScorer(SupervisedScorer[SupervisedModelT]):
        def __call__(self, model: SupervisedModelT) -> float:
            raw = super().__call__(model)
            return transformation(raw)

    new_scorer = CompositeSupervisedScorer(
        scoring=scorer._scorer,
        test_data=scorer.test_data,
        default=transformation(scorer.default),
        range=(
            transformation(scorer.range[0].item()),
            transformation(scorer.range[1].item()),
        ),
        name=name,
    )
    return new_scorer

PARAMETER	DESCRIPTION
`n_samples_outer`	The number of probability values \(q\) used for the outer loop. Since samples are taken anew for each index, a high number will delay updating new indices and has no effect on the final accuracy if using an infinite index iteration. In general, it only makes sense to change this number if using a finite index iteration. TYPE: `int`
`seed`	The seed for the random number generator. TYPE: `Seed \| None` DEFAULT: `None`

PARAMETER	DESCRIPTION
`rtol`	Relative tolerance for convergence (\(\epsilon\) in the formula) TYPE: `float`
`burn_in`	The minimum number of updates an index must have seen before checking for convergence. This is required because the first correlation checks are usually meaningless. TYPE: `int`
`fraction`	The fraction of values that must have been updated between two correlation checks. This is to avoid comparing two results where only one value has been updated, which would have almost perfect rank correlation. TYPE: `float` DEFAULT: `1.0`
`modify_result`	If `True`, the status of the input ValuationResult is modified in place after the call. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`NDArray[T]`	Array with point-wise negative \(l_2\) distances between labels and model
`NDArray[T]`	predictions

pydvl.valuation ¶

RawData dataclass ¶

GroupedDataset ¶

n_features property ¶

indices property ¶

names property ¶

feature ¶

data ¶

data_indices ¶

logical_indices ¶

from_sklearn classmethod ¶

from_arrays classmethod ¶

from_dataset classmethod ¶

BetaShapleyValuation ¶

values ¶

ClasswiseShapleyValuation ¶

values ¶

DataBanzhafValuation ¶

values ¶

BaggingModel ¶

fit ¶

predict ¶

Valuation ¶

values ¶

ValuationResult ¶

Indexing¶

Sorting¶

Operating on results¶

values property ¶

variances property ¶

stderr property ¶

counts property ¶

indices property ¶

names property ¶

sort ¶

positions ¶

copy ¶

__getattr__ ¶

__iter__ ¶

__add__ ¶

update ¶

scale ¶

get ¶

to_dataframe ¶

from_random classmethod ¶

empty classmethod ¶

zeros classmethod ¶

DataOOBValuation ¶

values ¶

fit ¶

DeltaShapleyValuation ¶

values ¶

GroupTestingShapleyValuation ¶

values ¶

fit ¶

Status ¶

OR¶

AND¶

NOT¶

Boolean casting¶

KNNShapleyValuation ¶

values ¶

fit ¶

LeastCoreValuation ¶

values ¶

fit ¶

ExactLeastCoreValuation ¶

fit ¶

values ¶

MonteCarloLeastCoreValuation ¶

fit ¶

values ¶

LOOValuation ¶

values ¶

log_coefficient ¶

SemivalueValuation ¶

values ¶

log_coefficient abstractmethod ¶

ShapleyValuation ¶

values ¶

RawData `dataclass` ¶

n_features `property` ¶

indices `property` ¶

names `property` ¶

from_sklearn `classmethod` ¶

from_arrays `classmethod` ¶

from_dataset `classmethod` ¶

values `property` ¶

variances `property` ¶

stderr `property` ¶

counts `property` ¶

indices `property` ¶

names `property` ¶

getattr ¶

iter ¶

add ¶

from_random `classmethod` ¶

empty `classmethod` ¶

zeros `classmethod` ¶

log_coefficient `abstractmethod` ¶

skip_indices `property` `writable` ¶

len ¶

sample_limit `abstractmethod` ¶

generate `abstractmethod` ¶

log_weight `abstractmethod` ¶

make_strategy `abstractmethod` ¶

process `abstractmethod` ¶

skip_indices `property` `writable` ¶

len ¶

skip_indices `property` `writable` ¶

len ¶

skip_indices `property` `writable` ¶

len ¶

skip_indices `property` `writable` ¶

len ¶

len ¶

len ¶

skip_indices `property` `writable` ¶

len ¶

length `abstractmethod` `staticmethod` ¶

complement_size `abstractmethod` `staticmethod` ¶