pydvl.value.shapley

This package holds all routines for the computation of Shapley Data value. Users will want to use compute_shapley_values or compute_semivalues as interfaces to most methods defined in the modules.

Please refer to the guide on data valuation for an overview of all methods.

ValueItem dataclass

ValueItem(
    index: IndexT,
    name: NameT,
    value: float,
    variance: Optional[float],
    count: Optional[int],
)

Bases: Generic[IndexT, NameT]

The result of a value computation for one datum.

ValueItems can be compared with the usual operators, forming a total order. Comparisons take only the value into account.

Todo

Maybe have a mode of comparing similar to np.isclose, or taking the variance into account.

ATTRIBUTE	DESCRIPTION
index	Index of the sample with this value in the original Dataset TYPE: IndexT
name	Name of the sample if it was provided. Otherwise, str(index) TYPE: NameT
value	The value TYPE: float
variance	Variance of the value if it was computed with an approximate method TYPE: Optional[float]
count	Number of updates for this value TYPE: Optional[int]

stderr property

stderr: Optional[float]

Standard error of the value.

ValuationResult

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

Bases: Sequence, Iterable[ValueItem[IndexT, NameT]], Generic[IndexT, NameT]

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.

Alternatively, indexing can be data-based, i.e. using the indices in the original dataset. This is the case for the methods 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. 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().

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: 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: Optional[NDArray[IndexT]] DEFAULT: None
variances	An optional array of variances in the computation of each value. TYPE: Optional[NDArray[float64]] DEFAULT: None
counts	An optional array with the number of updates for each value. Defaults to an array of ones. TYPE: Optional[NDArray[int_]] DEFAULT: None
data_names	Names for the data points. Defaults to index numbers if not set. TYPE: Optional[Sequence[NameT] | NDArray[NameT]] 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 by ascending value. See above how this affects usage as an iterable or sequence. TYPE: bool DEFAULT: False
extra_values	Any 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.

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

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

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

    # Yuk...
    if data_names is None:
        if indices is not None:
            self._names = np.copy(indices)
        else:
            self._names = np.arange(len(self._values), dtype=np.int_)
    elif not isinstance(data_names, np.ndarray):
        self._names = np.array(data_names)
    else:
        self._names = data_names.copy()
    if len(np.unique(self._names)) != len(self._names):
        raise ValueError("Data names must be unique")

    if indices is None:
        indices = np.arange(len(self._values), dtype=np.int_)
    self._indices = indices
    self._positions = {idx: pos for pos, idx in enumerate(indices)}

    self._sort_positions: NDArray[np.int_] = np.arange(
        len(self._values), dtype=np.int_
    )
    if sort:
        self.sort()

values property

values: NDArray[float64]

The values, possibly sorted.

variances property

variances: NDArray[float64]

The variances, possibly sorted.

stderr property

stderr: NDArray[float64]

The raw standard errors, 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.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/value/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.value.result.ValuationResult.values],
    [ValuationResult.variances][pydvl.value.result.ValuationResult.variances],
    [ValuationResult.counts][pydvl.value.result.ValuationResult.counts],
    [ValuationResult.indices][pydvl.value.result.ValuationResult.indices]
    and [ValuationResult.names][pydvl.value.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.value.result.ValueItem].
    """
    keymap = {
        "index": "_indices",
        "value": "_values",
        "variance": "_variances",
        "name": "_names",
    }
    self._sort_positions = np.argsort(getattr(self, keymap[key]))
    if reverse:
        self._sort_positions = self._sort_positions[::-1]
    self._sort_order = reverse

__getattr__

__getattr__(attr: str) -> Any

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

Source code in src/pydvl/value/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[IndexT, NameT]]

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

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

def __iter__(self) -> Iterator[ValueItem[IndexT, NameT]]:
    """Iterate over the results returning [ValueItem][pydvl.value.result.ValueItem] objects.
    To sort in place before iteration, use [sort()][pydvl.value.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[IndexT, NameT],
) -> ValuationResult[IndexT, NameT]

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/value/result.py

def __add__(
    self, other: ValuationResult[IndexT, NameT]
) -> ValuationResult[IndexT, NameT]:
    """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.value.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=object)
        other_names: NDArray = np.empty_like(indices, dtype=object)
        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(idx: int, new_value: float) -> ValuationResult[IndexT, NameT]

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

PARAMETER	DESCRIPTION
idx	Data index of the value to update. TYPE: int
new_value	New value to add to the result. TYPE: float

RETURNS	DESCRIPTION
ValuationResult[IndexT, NameT]	A reference to the same, modified result.

RAISES	DESCRIPTION
IndexError	If the index is not found.

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

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

    Args:
        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[idx]
    except KeyError:
        raise IndexError(f"Index {idx} not found in ValuationResult")
    val, var = running_moments(
        self._values[pos],
        self._variances[pos],
        self._counts[pos],
        new_value,
        unbiased=False,
    )
    self[pos] = ValueItem(
        index=cast(IndexT, idx),  # FIXME
        name=self._names[pos],
        value=val,
        variance=var,
        count=self._counts[pos] + 1,
    )
    return self

scale

scale(factor: float, indices: Optional[NDArray[IndexT]] = None)

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

PARAMETER	DESCRIPTION
factor	Factor to scale by. TYPE: float
indices	Indices to scale. If None, all values are scaled. TYPE: Optional[NDArray[IndexT]] DEFAULT: None

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

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

    Args:
        factor: Factor to scale by.
        indices: Indices to scale. If None, all values are scaled.
    """
    self._values[self._sort_positions[indices]] *= factor
    self._variances[self._sort_positions[indices]] *= factor**2

get

get(idx: Integral) -> ValueItem

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

RAISES	DESCRIPTION
IndexError	If the index is not found.

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

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

    Raises:
         IndexError: If the index is not found.
    """
    try:
        pos = self._positions[idx]
    except KeyError:
        raise IndexError(f"Index {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: Optional[str] = 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: Optional[str] 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 two columns, one for the values, with name given as explained in column, and another with standard errors for approximate algorithms. The latter will be named column+'_stderr'.

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

def to_dataframe(
    self, column: Optional[str] = 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 two columns, one for the values, with name
            given as explained in `column`, and another with standard errors for
            approximate algorithms. The latter will be named `column+'_stderr'`.
    """
    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 + "_stderr"] = self.stderr[self._sort_positions]
    df[column + "_updates"] = self.counts[self._sort_positions]
    # HACK for compatibility with updated support code in the notebooks
    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: Optional[float] = None,
    seed: Optional[Seed] = None,
    **kwargs: 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: Optional[float] DEFAULT: None
kwargs	Any Additional options to pass to the constructor of ValuationResult. Use to override status, names, etc. TYPE: 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/value/result.py

@classmethod
def from_random(
    cls,
    size: int,
    total: Optional[float] = None,
    seed: Optional[Seed] = None,
    **kwargs: Any,
) -> "ValuationResult":
    """Creates a [ValuationResult][pydvl.value.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).
        kwargs: Any Additional options to pass to the constructor of
            [ValuationResult][pydvl.value.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: Optional[Sequence[IndexT] | NDArray[IndexT]] = None,
    data_names: Optional[Sequence[NameT] | NDArray[NameT]] = 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: Optional[Sequence[IndexT] | NDArray[IndexT]] DEFAULT: None
data_names	Optional sequences or array of names for the data points. Defaults to index numbers if not set. TYPE: Optional[Sequence[NameT] | NDArray[NameT]] 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/value/result.py

@classmethod
def empty(
    cls,
    algorithm: str = "",
    indices: Optional[Sequence[IndexT] | NDArray[IndexT]] = None,
    data_names: Optional[Sequence[NameT] | NDArray[NameT]] = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.value.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: Optional[Sequence[IndexT] | NDArray[IndexT]] = None,
    data_names: Optional[Sequence[NameT] | NDArray[NameT]] = 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: Optional[Sequence[IndexT] | NDArray[IndexT]] 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: Optional[Sequence[NameT] | NDArray[NameT]] 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/value/result.py

@classmethod
def zeros(
    cls,
    algorithm: str = "",
    indices: Optional[Sequence[IndexT] | NDArray[IndexT]] = None,
    data_names: Optional[Sequence[NameT] | NDArray[NameT]] = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.value.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.
    """
    if indices is None:
        indices = np.arange(n_samples, dtype=np.int_)
    else:
        indices = np.array(indices, dtype=np.int_)

    if data_names is None:
        data_names = np.array(indices)
    else:
        data_names = np.array(data_names)

    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_),
    )

StoppingCriterion

StoppingCriterion(modify_result: bool = True)

Bases: ABC

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

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

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

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

Subclassing

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

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

completion

completion() -> float

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

AbsoluteStandardError

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

Bases: StoppingCriterion

Determine convergence based on the standard error of the values.

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

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

completion

completion() -> float

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

MaxChecks

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

Bases: StoppingCriterion

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

A "check" is one call to the criterion.

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

MaxUpdates

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

Bases: StoppingCriterion

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

Note

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

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

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

MinUpdates

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

Bases: StoppingCriterion

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

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

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

MaxTime

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

Bases: StoppingCriterion

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

Checks the elapsed time since construction

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

HistoryDeviation

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

Bases: StoppingCriterion

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

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

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

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

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

Warning

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

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

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

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

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

converged property

converged: NDArray[bool_]

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

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

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

completion

completion() -> float

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

RankCorrelation

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

Bases: StoppingCriterion

A check for stability of Spearman correlation between checks.

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

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

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

Added in 0.9.0

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

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

converged property

converged: NDArray[bool_]

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

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

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

__call__

__call__(result: ValuationResult) -> Status

Calls check(), maybe updating the result.

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

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

ClasswiseScorer

ClasswiseScorer(
    scoring: Union[str, ScorerCallable] = "accuracy",
    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,
    initial_label: Optional[int] = None,
    name: Optional[str] = None,
)

Bases: Scorer

A Scorer designed for evaluation in classification problems. Its value is computed from an in-class and an out-of-class "inner score" (Schoch et al., 2022) ¹. Let \(S\) be the training set and \(D\) be the valuation set. For each label \(c\), \(D\) is factorized into two disjoint sets: \(D_c\) for in-class instances and \(D_{-c}\) for out-of-class instances. The score combines an in-class metric of performance, adjusted by a discounted out-of-class metric. These inner scores must be provided upon construction or default to accuracy. They are combined into:

\[ u(S_{y_i}) = f(a_S(D_{y_i}))\ g(a_S(D_{-y_i})), \]

where \(f\) and \(g\) are continuous, monotonic functions. For a detailed explanation, refer to section four of (Schoch et al., 2022) ¹.

Warning

Metrics must support multiple class labels if you intend to apply them to a multi-class problem. For instance, the metric 'accuracy' supports multiple classes, but the metric 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 Scorer. TYPE: Union[str, ScorerCallable] DEFAULT: 'accuracy'
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
initial_label	Set initial label (for the first iteration) TYPE: Optional[int] DEFAULT: None
name	Name of the scorer. If not provided, the name of the inner scoring function will be prefixed by classwise. TYPE: Optional[str] DEFAULT: None

New in version 0.7.1

Source code in src/pydvl/value/shapley/classwise.py

def __init__(
    self,
    scoring: Union[str, ScorerCallable] = "accuracy",
    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,
    initial_label: Optional[int] = None,
    name: Optional[str] = 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,
        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 = initial_label

estimate_in_class_and_out_of_class_score

estimate_in_class_and_out_of_class_score(
    model: SupervisedModel,
    x_test: NDArray[float64],
    y_test: NDArray[int_],
    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: SupervisedModel
x_test	Array containing the features of the classification problem. TYPE: NDArray[float64]
y_test	Array containing the labels of the classification problem. TYPE: NDArray[int_]
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/value/shapley/classwise.py

def estimate_in_class_and_out_of_class_score(
    self,
    model: SupervisedModel,
    x_test: NDArray[np.float64],
    y_test: NDArray[np.int_],
    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.
        x_test: Array containing the features of the classification problem.
        y_test: Array containing the labels of the classification problem.
        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.
    """
    scorer = self._scorer
    label_set_match = y_test == self.label
    label_set = np.where(label_set_match)[0]
    num_classes = len(np.unique(y_test))

    if len(label_set) == 0:
        return 0, 1 / (num_classes - 1)

    complement_label_set = np.where(~label_set_match)[0]
    in_class_score = scorer(model, x_test[label_set], y_test[label_set])
    out_of_class_score = scorer(
        model, x_test[complement_label_set], y_test[complement_label_set]
    )

    if rescale_scores:
        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

OwenAlgorithm

Bases: Enum

Choices for the Owen sampling method.

ATTRIBUTE	DESCRIPTION
Standard	Use q ∈ [0, 1]
Antithetic	Use q ∈ [0, 0.5] and correlated samples

TruncationPolicy

TruncationPolicy()

Bases: ABC

A policy for deciding whether to stop computing marginals in a permutation.

Statistics are kept on the 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/value/shapley/truncated.py

def __init__(self) -> None:
    self.n_calls: int = 0
    self.n_truncations: int = 0

reset abstractmethod

reset(u: Optional[Utility] = None)

Reset the policy to a state ready for a new permutation.

Source code in src/pydvl/value/shapley/truncated.py

@abc.abstractmethod
def reset(self, u: Optional[Utility] = None):
    """Reset the policy to a state ready for a new permutation."""
    ...

__call__

__call__(idx: int, score: float) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
idx	Position in the permutation currently being computed. TYPE: int
score	Last utility computed. TYPE: float

RETURNS	DESCRIPTION
bool	True if the computation should be interrupted.

Source code in src/pydvl/value/shapley/truncated.py

def __call__(self, idx: int, score: float) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the permutation currently being computed.
        score: Last utility computed.

    Returns:
        `True` if the computation should be interrupted.
    """
    ret = self._check(idx, score)
    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/value/shapley/truncated.py

def __init__(self) -> None:
    self.n_calls: int = 0
    self.n_truncations: int = 0

__call__

__call__(idx: int, score: float) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
idx	Position in the permutation currently being computed. TYPE: int
score	Last utility computed. TYPE: float

RETURNS	DESCRIPTION
bool	True if the computation should be interrupted.

Source code in src/pydvl/value/shapley/truncated.py

def __call__(self, idx: int, score: float) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the permutation currently being computed.
        score: Last utility computed.

    Returns:
        `True` if the computation should be interrupted.
    """
    ret = self._check(idx, score)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

FixedTruncation

FixedTruncation(u: Utility, fraction: float)

Bases: TruncationPolicy

Break a permutation after computing a fixed number of marginals.

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
u	Utility object with model, data, and scoring function TYPE: Utility
fraction	Fraction of marginals in a permutation to compute before stopping (e.g. 0.5 to compute half of the marginals). TYPE: float

Source code in src/pydvl/value/shapley/truncated.py

def __init__(self, u: Utility, fraction: float):
    super().__init__()
    if fraction <= 0 or fraction > 1:
        raise ValueError("fraction must be in (0, 1]")
    self.max_marginals = len(u.data) * fraction
    self.count = 0

__call__

__call__(idx: int, score: float) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
idx	Position in the permutation currently being computed. TYPE: int
score	Last utility computed. TYPE: float

RETURNS	DESCRIPTION
bool	True if the computation should be interrupted.

Source code in src/pydvl/value/shapley/truncated.py

def __call__(self, idx: int, score: float) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the permutation currently being computed.
        score: Last utility computed.

    Returns:
        `True` if the computation should be interrupted.
    """
    ret = self._check(idx, score)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

RelativeTruncation

RelativeTruncation(u: Utility, rtol: float)

Bases: TruncationPolicy

Break a permutation if the marginal utility is too low.

This is called "performance tolerance" in (Ghorbani and Zou, 2019)¹.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
rtol	Relative tolerance. The permutation is broken if the last computed utility is less than total_utility * rtol. TYPE: float

Source code in src/pydvl/value/shapley/truncated.py

def __init__(self, u: Utility, rtol: float):
    super().__init__()
    self.rtol = rtol
    logger.info("Computing total utility for permutation truncation.")
    self.total_utility = self.reset(u)
    self._u = u

__call__

__call__(idx: int, score: float) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
idx	Position in the permutation currently being computed. TYPE: int
score	Last utility computed. TYPE: float

RETURNS	DESCRIPTION
bool	True if the computation should be interrupted.

Source code in src/pydvl/value/shapley/truncated.py

def __call__(self, idx: int, score: float) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the permutation currently being computed.
        score: Last utility computed.

    Returns:
        `True` if the computation should be interrupted.
    """
    ret = self._check(idx, score)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

BootstrapTruncation

BootstrapTruncation(u: Utility, n_samples: int, sigmas: float = 1)

Bases: TruncationPolicy

Break a permutation if the last computed utility is close to the total utility, measured as a multiple of the standard deviation of the utilities.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
n_samples	Number of bootstrap samples to use to compute the variance of the utilities. TYPE: int
sigmas	Number of standard deviations to use as a threshold. TYPE: float DEFAULT: 1

Source code in src/pydvl/value/shapley/truncated.py

def __init__(self, u: Utility, n_samples: int, sigmas: float = 1):
    super().__init__()
    self.n_samples = n_samples
    logger.info("Computing total utility for permutation truncation.")
    self.total_utility = u(u.data.indices)
    self.count: int = 0
    self.variance: float = 0
    self.mean: float = 0
    self.sigmas: float = sigmas

__call__

__call__(idx: int, score: float) -> bool

Check whether the computation should be interrupted.

PARAMETER	DESCRIPTION
idx	Position in the permutation currently being computed. TYPE: int
score	Last utility computed. TYPE: float

RETURNS	DESCRIPTION
bool	True if the computation should be interrupted.

Source code in src/pydvl/value/shapley/truncated.py

def __call__(self, idx: int, score: float) -> bool:
    """Check whether the computation should be interrupted.

    Args:
        idx: Position in the permutation currently being computed.
        score: Last utility computed.

    Returns:
        `True` if the computation should be interrupted.
    """
    ret = self._check(idx, score)
    self.n_calls += 1
    self.n_truncations += 1 if ret else 0
    return ret

ShapleyMode

Bases: str, Enum

Supported algorithms for the computation of Shapley values.

Todo

Make algorithms register themselves here.

make_criterion

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

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

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

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

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

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

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

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

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

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

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

        def __str__(self):
            return self._name

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

    return WrappedCriterion

compute_classwise_shapley_values

compute_classwise_shapley_values(
    u: Utility,
    *,
    done: StoppingCriterion,
    truncation: TruncationPolicy,
    done_sample_complements: Optional[StoppingCriterion] = None,
    normalize_values: bool = True,
    use_default_scorer_value: bool = True,
    min_elements_per_label: int = 1,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult

Computes an approximate Class-wise Shapley value by sampling independent permutations of the index set for each label and index sets sampled from the powerset of the complement (with respect to the currently evaluated label), approximating the sum:

\[ v_u(i) = \frac{1}{K} \sum_k \frac{1}{L} \sum_l [u(\sigma^{(l)}_{:i} \cup \{i\} | S^{(k)} ) − u( \sigma^{(l)}_{:i} | S^{(k)})], \]

where \(\sigma_{:i}\) denotes the set of indices in permutation sigma before the position where \(i\) appears and \(S\) is a subset of the index set of all other labels (see [the main documentation][#intro-to-cw-shapley] for details).

PARAMETER	DESCRIPTION
u	Utility object containing model, data, and scoring function. The scorer must be of type ClasswiseScorer. TYPE: Utility
done	Function that checks whether the computation needs to stop. TYPE: StoppingCriterion
truncation	Callable function that decides whether to interrupt processing a permutation and set subsequent marginals to zero. TYPE: TruncationPolicy
done_sample_complements	Function checking whether computation needs to stop. Otherwise, it will resample conditional sets until the stopping criterion is met. TYPE: Optional[StoppingCriterion] DEFAULT: None
normalize_values	Indicates whether to normalize the values by the variation in each class times their in-class accuracy. TYPE: bool DEFAULT: True
done_sample_complements	Number of times to resample the complement set for each permutation. TYPE: Optional[StoppingCriterion] DEFAULT: None
use_default_scorer_value	The first set of indices is the sampled complement set. Unless not otherwise specified, the default scorer value is used for this. If it is set to false, the base score is calculated from the utility. TYPE: bool DEFAULT: True
min_elements_per_label	The minimum number of elements for each opposite label. TYPE: int DEFAULT: 1
n_jobs	Number of parallel jobs to run. TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display a progress bar. TYPE: bool DEFAULT: False
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None

RETURNS	DESCRIPTION
ValuationResult	ValuationResult object containing computed data values.

New in version 0.7.1

Source code in src/pydvl/value/shapley/classwise.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def compute_classwise_shapley_values(
    u: Utility,
    *,
    done: StoppingCriterion,
    truncation: TruncationPolicy,
    done_sample_complements: Optional[StoppingCriterion] = None,
    normalize_values: bool = True,
    use_default_scorer_value: bool = True,
    min_elements_per_label: int = 1,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult:
    r"""
    Computes an approximate Class-wise Shapley value by sampling independent
    permutations of the index set for each label and index sets sampled from the
    powerset of the complement (with respect to the currently evaluated label),
    approximating the sum:

    $$
    v_u(i) = \frac{1}{K} \sum_k \frac{1}{L} \sum_l
    [u(\sigma^{(l)}_{:i} \cup \{i\} | S^{(k)} ) − u( \sigma^{(l)}_{:i} | S^{(k)})],
    $$

    where $\sigma_{:i}$ denotes the set of indices in permutation sigma before
    the position where $i$ appears and $S$ is a subset of the index set of all
    other labels (see [the main documentation][#intro-to-cw-shapley] for
    details).

    Args:
        u: Utility object containing model, data, and scoring function. The
            scorer must be of type
            [ClasswiseScorer][pydvl.value.shapley.classwise.ClasswiseScorer].
        done: Function that checks whether the computation needs to stop.
        truncation: Callable function that decides whether to interrupt processing a
            permutation and set subsequent marginals to zero.
        done_sample_complements: Function checking whether computation needs to stop.
            Otherwise, it will resample conditional sets until the stopping criterion is
            met.
        normalize_values: Indicates whether to normalize the values by the variation
            in each class times their in-class accuracy.
        done_sample_complements: Number of times to resample the complement set
            for each permutation.
        use_default_scorer_value: The first set of indices is the sampled complement
            set. Unless not otherwise specified, the default scorer value is used for
            this. If it is set to false, the base score is calculated from the utility.
        min_elements_per_label: The minimum number of elements for each opposite
            label.
        n_jobs: Number of parallel jobs to run.
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display a progress bar.
        seed: Either an instance of a numpy random number generator or a seed for it.

    Returns:
        ValuationResult object containing computed data values.

    !!! tip "New in version 0.7.1"
    """
    dim_correct = u.data.y_train.ndim == 1 and u.data.y_test.ndim == 1
    is_integral = all(
        map(
            lambda v: isinstance(v, numbers.Integral), (*u.data.y_train, *u.data.y_test)
        )
    )
    if not dim_correct or not is_integral:
        raise ValueError(
            "The supplied dataset has to be a 1-dimensional classification dataset."
        )

    if not isinstance(u.scorer, ClasswiseScorer):
        raise ValueError(
            "Please set a subclass of ClasswiseScorer object as scorer object of the"
            " utility. See scoring argument of Utility."
        )

    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)
    u_ref = parallel_backend.put(u)
    n_jobs = parallel_backend.effective_n_jobs(n_jobs)
    n_submitted_jobs = 2 * n_jobs

    pbar = tqdm(disable=not progress, position=0, total=100, unit="%")
    algorithm = "classwise_shapley"
    accumulated_result = ValuationResult.zeros(
        algorithm=algorithm, indices=u.data.indices, data_names=u.data.data_names
    )
    terminate_exec = False
    seed_sequence = ensure_seed_sequence(seed)

    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)

    with parallel_backend.executor(max_workers=n_jobs) as executor:
        pending: Set[Future] = set()
        while True:
            completed_futures, pending = wait(
                pending, timeout=60, return_when=FIRST_COMPLETED
            )
            for future in completed_futures:
                accumulated_result += future.result()
                if done(accumulated_result):
                    terminate_exec = True
                    break

            pbar.n = 100 * done.completion()
            pbar.refresh()
            if terminate_exec:
                break

            n_remaining_slots = n_submitted_jobs - len(pending)
            seeds = seed_sequence.spawn(n_remaining_slots)
            for i in range(n_remaining_slots):
                future = executor.submit(
                    _permutation_montecarlo_classwise_shapley_one_step,
                    u_ref,
                    truncation=truncation,
                    done_sample_complements=done_sample_complements,
                    use_default_scorer_value=use_default_scorer_value,
                    min_elements_per_label=min_elements_per_label,
                    algorithm_name=algorithm,
                    seed=seeds[i],
                )
                pending.add(future)

    result = accumulated_result
    if normalize_values:
        result = _normalize_classwise_shapley_values(result, u)

    return result

compute_shapley_values

compute_shapley_values(
    u: Utility,
    *,
    done: StoppingCriterion = MaxChecks(None),
    mode: ShapleyMode = TruncatedMontecarlo,
    n_jobs: int = 1,
    seed: Optional[Seed] = None,
    **kwargs,
) -> ValuationResult

Umbrella method to compute Shapley values with any of the available algorithms.

See Data valuation for an overview.

The following algorithms are available. Note that the exact methods can only work with very small datasets and are thus intended only for testing. Some algorithms also accept additional arguments, please refer to the documentation of each particular method.

• combinatorial_exact: uses the combinatorial implementation of data Shapley. Implemented in combinatorial_exact_shapley().
• combinatorial_montecarlo: uses the approximate Monte Carlo implementation of combinatorial data Shapley. Implemented in combinatorial_montecarlo_shapley().
• permutation_exact: uses the permutation-based implementation of data Shapley. Computation is not parallelized. Implemented in permutation_exact_shapley().
• permutation_montecarlo: uses the approximate Monte Carlo implementation of permutation data Shapley. Accepts a TruncationPolicy to stop computing marginals. Implemented in permutation_montecarlo_shapley().
• owen_sampling: Uses the Owen continuous extension of the utility function to the unit cube. Implemented in owen_sampling_shapley(). This method does not take a StoppingCriterion but instead requires a parameter q_max for the number of subdivisions of the unit interval to use for integration, and another parameter n_samples for the number of subsets to sample for each \(q\).
• owen_halved: Same as 'owen_sampling' but uses correlated samples in the expectation. Implemented in owen_sampling_shapley(). This method requires an additional parameter q_max for the number of subdivisions of the interval [0,0.5] to use for integration, and another parameter n_samples for the number of subsets to sample for each \(q\).
• group_testing: estimates differences of Shapley values and solves a constraint satisfaction problem. High sample complexity, not recommended. Implemented in group_testing_shapley(). This method does not take a StoppingCriterion but instead requires a parameter n_samples for the number of iterations to run.

Additionally, one can use model-specific methods:

• knn: Exact method for K-Nearest neighbour models. Implemented in knn_shapley().

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function. TYPE: Utility
done	Object used to determine when to stop the computation for Monte Carlo methods. The default is to stop after 100 iterations. See the available criteria in stopping. It is possible to combine several of them using boolean operators. Some methods ignore this argument, others require specific subtypes. TYPE: StoppingCriterion DEFAULT: MaxChecks(None)
n_jobs	Number of parallel jobs (available only to some methods) TYPE: int DEFAULT: 1
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None
mode	Choose which shapley algorithm to use. See ShapleyMode for a list of allowed value. TYPE: ShapleyMode DEFAULT: TruncatedMontecarlo

RETURNS	DESCRIPTION
ValuationResult	Object with the results.

Source code in src/pydvl/value/shapley/common.py

def compute_shapley_values(
    u: Utility,
    *,
    done: StoppingCriterion = MaxChecks(None),
    mode: ShapleyMode = ShapleyMode.TruncatedMontecarlo,
    n_jobs: int = 1,
    seed: Optional[Seed] = None,
    **kwargs,
) -> ValuationResult:
    """Umbrella method to compute Shapley values with any of the available
    algorithms.

    See [Data valuation][data-valuation] for an overview.

    The following algorithms are available. Note that the exact methods can only
    work with very small datasets and are thus intended only for testing. Some
    algorithms also accept additional arguments, please refer to the
    documentation of each particular method.

    - `combinatorial_exact`: uses the combinatorial implementation of data
      Shapley. Implemented in
      [combinatorial_exact_shapley()][pydvl.value.shapley.naive.combinatorial_exact_shapley].
    - `combinatorial_montecarlo`:  uses the approximate Monte Carlo
      implementation of combinatorial data Shapley. Implemented in
      [combinatorial_montecarlo_shapley()][pydvl.value.shapley.montecarlo.combinatorial_montecarlo_shapley].
    - `permutation_exact`: uses the permutation-based implementation of data
      Shapley. Computation is **not parallelized**. Implemented in
      [permutation_exact_shapley()][pydvl.value.shapley.naive.permutation_exact_shapley].
    - `permutation_montecarlo`: uses the approximate Monte Carlo
      implementation of permutation data Shapley. Accepts a
      [TruncationPolicy][pydvl.value.shapley.truncated.TruncationPolicy] to stop
      computing marginals. Implemented in
      [permutation_montecarlo_shapley()][pydvl.value.shapley.montecarlo.permutation_montecarlo_shapley].
    - `owen_sampling`: Uses the Owen continuous extension of the utility
      function to the unit cube. Implemented in
      [owen_sampling_shapley()][pydvl.value.shapley.owen.owen_sampling_shapley]. This
      method does not take a [StoppingCriterion][pydvl.value.stopping.StoppingCriterion]
      but instead requires a parameter `q_max` for the number of subdivisions
      of the unit interval to use for integration, and another parameter
      `n_samples` for the number of subsets to sample for each $q$.
    - `owen_halved`: Same as 'owen_sampling' but uses correlated samples in the
      expectation. Implemented in
      [owen_sampling_shapley()][pydvl.value.shapley.owen.owen_sampling_shapley].
      This method  requires an additional parameter `q_max` for the number of
      subdivisions of the interval [0,0.5] to use for integration, and another
      parameter `n_samples` for the number of subsets to sample for each $q$.
    - `group_testing`: estimates differences of Shapley values and solves a
      constraint satisfaction problem. High sample complexity, not recommended.
      Implemented in [group_testing_shapley()][pydvl.value.shapley.gt.group_testing_shapley]. This
      method does not take a [StoppingCriterion][pydvl.value.stopping.StoppingCriterion]
      but instead requires a parameter `n_samples` for the number of
      iterations to run.

    Additionally, one can use model-specific methods:

    - `knn`: Exact method for K-Nearest neighbour models. Implemented in
      [knn_shapley()][pydvl.value.shapley.knn.knn_shapley].

    Args:
        u: [Utility][pydvl.utils.utility.Utility] object with model, data, and
            scoring function.
        done: Object used to determine when to stop the computation for Monte
            Carlo methods. The default is to stop after 100 iterations. See the
            available criteria in [stopping][pydvl.value.stopping]. It is
            possible to combine several of them using boolean operators. Some
            methods ignore this argument, others require specific subtypes.
        n_jobs: Number of parallel jobs (available only to some methods)
        seed: Either an instance of a numpy random number generator or a seed
            for it.
        mode: Choose which shapley algorithm to use. See
            [ShapleyMode][pydvl.value.shapley.ShapleyMode] for a list of allowed
            value.

    Returns:
        Object with the results.

    """
    progress: bool = kwargs.pop("progress", False)

    if mode not in list(ShapleyMode):
        raise ValueError(f"Invalid value encountered in {mode=}")

    if mode in (
        ShapleyMode.PermutationMontecarlo,
        ShapleyMode.ApproShapley,
        ShapleyMode.TruncatedMontecarlo,
    ):
        truncation = kwargs.pop("truncation", NoTruncation())
        return permutation_montecarlo_shapley(  # type: ignore
            u=u,
            done=done,
            truncation=truncation,
            n_jobs=n_jobs,
            seed=seed,
            progress=progress,
            **kwargs,
        )
    elif mode == ShapleyMode.CombinatorialMontecarlo:
        return combinatorial_montecarlo_shapley(  # type: ignore
            u, done=done, n_jobs=n_jobs, seed=seed, progress=progress
        )
    elif mode == ShapleyMode.CombinatorialExact:
        return combinatorial_exact_shapley(u, n_jobs=n_jobs, progress=progress)  # type: ignore
    elif mode == ShapleyMode.PermutationExact:
        return permutation_exact_shapley(u, progress=progress)
    elif mode == ShapleyMode.Owen or mode == ShapleyMode.OwenAntithetic:
        if kwargs.get("n_samples") is None:
            raise ValueError("n_samples cannot be None for Owen methods")
        if kwargs.get("max_q") is None:
            raise ValueError("Owen Sampling requires max_q for the outer integral")

        method = (
            OwenAlgorithm.Standard
            if mode == ShapleyMode.Owen
            else OwenAlgorithm.Antithetic
        )
        return owen_sampling_shapley(  # type: ignore
            u,
            n_samples=int(kwargs.get("n_samples", -1)),
            max_q=int(kwargs.get("max_q", -1)),
            method=method,
            n_jobs=n_jobs,
            seed=seed,
        )
    elif mode == ShapleyMode.KNN:
        return knn_shapley(u, progress=progress)
    elif mode == ShapleyMode.GroupTesting:
        n_samples = kwargs.pop("n_samples")
        if n_samples is None:
            raise ValueError("n_samples cannot be None for Group Testing")
        epsilon = kwargs.pop("epsilon")
        if epsilon is None:
            raise ValueError("Group Testing requires error bound epsilon")
        delta = kwargs.pop("delta", 0.05)
        return group_testing_shapley(  # type: ignore
            u,
            epsilon=float(epsilon),
            delta=delta,
            n_samples=int(n_samples),
            n_jobs=n_jobs,
            progress=progress,
            seed=seed,
            **kwargs,
        )
    else:
        raise ValueError(f"Invalid value encountered in {mode=}")

num_samples_eps_delta

num_samples_eps_delta(
    eps: float, delta: float, n: int, utility_range: float
) -> int

Implements the formula in Theorem 3 of (Jia, R. et al., 2019)¹ 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.

PARAMETER	DESCRIPTION
eps	ε TYPE: float
delta	δ TYPE: float
n	Number of data points TYPE: int
utility_range	Range of the Utility function TYPE: float

Returns: Number of samples from \(2^{[n]}\) guaranteeing ε/√n-correct Shapley pair-wise differences of values with probability 1-δ/(N(N-1)).

New in version 0.4.0

Source code in src/pydvl/value/shapley/gt.py

def num_samples_eps_delta(
    eps: float, delta: float, n: int, utility_range: float
) -> int:
    r"""Implements the formula in Theorem 3 of (Jia, R. et al., 2019)<sup><a href="#jia_efficient_2019">1</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.

    Args:
        eps: ε
        delta: δ
        n: Number of data points
        utility_range: Range of the [Utility][pydvl.utils.utility.Utility] function
    Returns:
        Number of samples from $2^{[n]}$ guaranteeing ε/√n-correct Shapley
            pair-wise differences of values with probability 1-δ/(N(N-1)).

    !!! tip "New in version 0.4.0"

    """
    constants = _constants(n=n, epsilon=eps, delta=delta, utility_range=utility_range)
    return int(constants.T)

group_testing_shapley

group_testing_shapley(
    u: Utility,
    n_samples: int,
    epsilon: float,
    delta: float,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
    **options: dict,
) -> ValuationResult

Implements group testing for approximation of Shapley values as described in (Jia, R. et al., 2019)¹.

Warning

This method is very inefficient. It requires several orders of magnitude more evaluations of the utility than others in montecarlo. It also uses several intermediate objects like the results from the runners and the constraint matrices which can become rather large.

By picking a specific distribution over subsets, the differences in Shapley values can be approximated with a Monte Carlo sum. These are then used to solve for the individual values in a feasibility problem.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
n_samples	Number of tests to perform. Use num_samples_eps_delta to estimate this. TYPE: int
epsilon	From the (ε,δ) sample bound. Use the same as for the estimation of n_iterations. TYPE: float
delta	From the (ε,δ) sample bound. Use the same as for the estimation of n_iterations. TYPE: float
n_jobs	Number of parallel jobs to use. Each worker performs a chunk of all tests (i.e. utility evaluations). TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display progress bars for each job. TYPE: bool DEFAULT: False
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None
options	Additional options to pass to cvxpy.Problem.solve(). E.g. to change the solver (which defaults to cvxpy.SCS) pass solver=cvxpy.CVXOPT. TYPE: dict DEFAULT: {}

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

New in version 0.4.0

Changed in version 0.5.0

Changed the solver to cvxpy instead of scipy's linprog. Added the ability to pass arbitrary options to it.

Changed in version 0.9.0

Deprecated config argument and added a parallel_backend argument to allow users to pass the Parallel Backend instance directly.

Source code in src/pydvl/value/shapley/gt.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def group_testing_shapley(
    u: Utility,
    n_samples: int,
    epsilon: float,
    delta: float,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
    **options: dict,
) -> ValuationResult:
    """Implements group testing for approximation of Shapley values as described
    in (Jia, R. et al., 2019)<sup><a href="#jia_efficient_2019">1</a></sup>.

    !!! Warning
        This method is very inefficient. It requires several orders of magnitude
        more evaluations of the utility than others in
        [montecarlo][pydvl.value.shapley.montecarlo]. It also uses several intermediate
        objects like the results from the runners and the constraint matrices
        which can become rather large.

    By picking a specific distribution over subsets, the differences in Shapley
    values can be approximated with a Monte Carlo sum. These are then used to
    solve for the individual values in a feasibility problem.

    Args:
        u: Utility object with model, data, and scoring function
        n_samples: Number of tests to perform. Use
            [num_samples_eps_delta][pydvl.value.shapley.gt.num_samples_eps_delta]
            to estimate this.
        epsilon: From the (ε,δ) sample bound. Use the same as for the
            estimation of `n_iterations`.
        delta: From the (ε,δ) sample bound. Use the same as for the
            estimation of `n_iterations`.
        n_jobs: Number of parallel jobs to use. Each worker performs a chunk
            of all tests (i.e. utility evaluations).
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display progress bars for each job.
        seed: Either an instance of a numpy random number generator or a seed for it.
        options: Additional options to pass to
            [cvxpy.Problem.solve()](https://www.cvxpy.org/tutorial/advanced/index.html#solve-method-options).
            E.g. to change the solver (which defaults to `cvxpy.SCS`) pass
            `solver=cvxpy.CVXOPT`.

    Returns:
        Object with the data values.

    !!! tip "New in version 0.4.0"

    !!! tip "Changed in version 0.5.0"
        Changed the solver to cvxpy instead of scipy's linprog. Added the ability
        to pass arbitrary options to it.

    !!! tip "Changed in version 0.9.0"
        Deprecated `config` argument and added a `parallel_backend`
        argument to allow users to pass the Parallel Backend instance
        directly.
    """

    n = len(u.data.indices)

    const = _constants(
        n=n,
        epsilon=epsilon,
        delta=delta,
        utility_range=u.score_range.max() - u.score_range.min(),
    )
    T = n_samples
    if T < const.T:
        log.warning(
            f"n_samples of {T} are below the required {const.T} for the "
            f"ε={epsilon:.02f} guarantee at δ={1 - delta:.02f} probability"
        )

    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)

    samples_per_job = max(1, n_samples // parallel_backend.effective_n_jobs(n_jobs))

    def reducer(
        results_it: Iterable[Tuple[NDArray, NDArray]],
    ) -> Tuple[NDArray, NDArray]:
        return np.concatenate(list(x[0] for x in results_it)).astype(
            np.float64
        ), np.concatenate(list(x[1] for x in results_it)).astype(np.int_)

    seed_sequence = ensure_seed_sequence(seed)
    map_reduce_seed_sequence, cvxpy_seed = tuple(seed_sequence.spawn(2))

    map_reduce_job: MapReduceJob[Utility, Tuple[NDArray, NDArray]] = MapReduceJob(
        u,
        map_func=_group_testing_shapley,
        reduce_func=reducer,
        map_kwargs=dict(n_samples=samples_per_job, progress=progress),
        parallel_backend=parallel_backend,
        n_jobs=n_jobs,
    )
    uu, betas = map_reduce_job(seed=map_reduce_seed_sequence)

    # Matrix of estimated differences. See Eqs. (3) and (4) in the paper.
    C = np.zeros(shape=(n, n))
    for i in range(n):
        for j in range(i + 1, n):
            C[i, j] = np.dot(uu, betas[:, i] - betas[:, j])
    C *= const.Z / T
    total_utility = u(u.data.indices)

    ###########################################################################
    # Solution of the constraint problem with cvxpy

    v = cp.Variable(n)
    constraints = [cp.sum(v) == total_utility]
    for i in range(n):
        for j in range(i + 1, n):
            constraints.append(v[i] - v[j] <= epsilon + C[i, j])
            constraints.append(v[j] - v[i] <= epsilon - C[i, j])

    problem = cp.Problem(cp.Minimize(0), constraints)
    solver = options.pop("solver", cp.SCS)
    problem.solve(solver=solver, **options)

    if problem.status != "optimal":
        log.warning(f"cvxpy returned status {problem.status}")
        values = (
            np.nan * np.ones_like(u.data.indices)
            if not hasattr(v.value, "__len__")
            else cast(NDArray[np.float64], v.value)
        )
        status = Status.Failed
    else:
        values = cast(NDArray[np.float64], v.value)
        status = Status.Converged

    return ValuationResult(
        algorithm="group_testing_shapley",
        status=status,
        values=values,
        data_names=u.data.data_names,
        solver_status=problem.status,
    )

knn_shapley

knn_shapley(u: Utility, *, progress: bool = True) -> ValuationResult

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 value computation to sorting of the training points by distance to the test point and applying a recursive formula, thus reducing computation time to \(O(n_test n_train log(n_train)\).

PARAMETER	DESCRIPTION
u	Utility with a KNN model to extract parameters from. The object will not be modified nor used other than to call get_params() TYPE: Utility
progress	Whether to display a progress bar. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

RAISES	DESCRIPTION
TypeError	If the model in the utility is not a sklearn.neighbors.KNeighborsClassifier.

New in version 0.1.0

Source code in src/pydvl/value/shapley/knn.py

def knn_shapley(u: Utility, *, progress: bool = True) -> ValuationResult:
    """Computes exact Shapley values for a KNN classifier.

    This implements the method described in (Jia, R. et al., 2019)<sup><a
    href="#jia_efficient_2019a">1</a></sup>. It exploits the local structure of
    K-Nearest Neighbours to reduce the value computation to sorting of the training
    points by distance to the test point and applying a recursive formula,
    thus reducing computation time to $O(n_test n_train log(n_train)$.

    Args:
        u: Utility with a KNN model to extract parameters from. The object
            will not be modified nor used other than to call [get_params()](
            <https://scikit-learn.org/stable/modules/generated/sklearn.base.BaseEstimator.html#sklearn.base.BaseEstimator.get_params>)
        progress: Whether to display a progress bar.

    Returns:
        Object with the data values.

    Raises:
        TypeError: If the model in the utility is not a
            [sklearn.neighbors.KNeighborsClassifier][].

    !!! tip "New in version 0.1.0"

    """
    if not isinstance(u.model, KNeighborsClassifier):
        raise TypeError("KNN Shapley requires a K-Nearest Neighbours model")

    defaults: Dict[str, Union[int, str]] = {
        "algorithm": "ball_tree" if u.data.dim >= 20 else "kd_tree",
        "metric": "minkowski",
        "p": 2,
    }
    defaults.update(u.model.get_params())
    # HACK: NearestNeighbors doesn't support this. There will be more...
    del defaults["weights"]
    n_neighbors: int = int(defaults["n_neighbors"])
    defaults["n_neighbors"] = len(u.data)  # We want all training points sorted

    assert n_neighbors < len(u.data)
    # assert data.target_dim == 1

    nns = NearestNeighbors(**defaults).fit(u.data.x_train)
    # closest to farthest
    _, indices = nns.kneighbors(u.data.x_test)

    res = np.zeros_like(u.data.indices, dtype=np.float64)
    n = len(u.data)
    yt = u.data.y_train
    iterator = enumerate(zip(u.data.y_test, indices), start=1)
    for j, (y, ii) in tqdm(iterator, disable=not progress):
        values = np.zeros_like(u.data.indices, dtype=np.float64)
        idx = ii[-1]
        values[idx] = int(yt[idx] == y) / n

        for i in range(n - 1, 0, -1):
            prev_idx = idx
            idx = ii[i - 1]
            values[idx] = values[prev_idx] + (
                int(yt[idx] == y) - int(yt[prev_idx] == y)
            ) / max(n_neighbors, i)
        res += values

    return ValuationResult(
        algorithm="knn_shapley",
        status=Status.Converged,
        values=res,
        data_names=u.data.data_names,
    )

permutation_montecarlo_shapley

permutation_montecarlo_shapley(
    u: Utility,
    done: StoppingCriterion,
    *,
    truncation: TruncationPolicy = NoTruncation(),
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult

Computes an approximate Shapley value by sampling independent permutations of the index set, approximating the sum:

\[ v_u(x_i) = \frac{1}{n!} \sum_{\sigma \in \Pi(n)} \tilde{w}( | \sigma_{:i} | )[u(\sigma_{:i} \cup \{i\}) − u(\sigma_{:i})], \]

where \(\sigma_{:i}\) denotes the set of indices in permutation sigma before the position where \(i\) appears (see [[data-valuation]] for details).

This implements the method described in (Ghorbani and Zou, 2019)¹ with a double stopping criterion.

Todo

Think of how to add Robin-Gelman or some other more principled stopping criterion.

Instead of naively implementing the expectation, we sequentially add points to coalitions from a permutation and incrementally compute marginal utilities. We stop computing marginals for a given permutation based on a TruncationPolicy. (Ghorbani and Zou, 2019)¹ mention two policies: one that stops after a certain fraction of marginals are computed, implemented in FixedTruncation, and one that stops if the last computed utility ("score") is close to the total utility using the standard deviation of the utility as a measure of proximity, implemented in BootstrapTruncation.

We keep sampling permutations and updating all shapley values until the StoppingCriterion returns True.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function. TYPE: Utility
done	function checking whether computation must stop. TYPE: StoppingCriterion
truncation	An optional callable which decides whether to interrupt processing a permutation and set all subsequent marginals to zero. Typically used to stop computation when the marginal is small. TYPE: TruncationPolicy DEFAULT: NoTruncation()
n_jobs	number of jobs across which to distribute the computation. TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display a progress bar. TYPE: bool DEFAULT: False
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

Changed in version 0.9.0

Deprecated config argument and added a parallel_backend argument to allow users to pass the Parallel Backend instance directly.

Source code in src/pydvl/value/shapley/montecarlo.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def permutation_montecarlo_shapley(
    u: Utility,
    done: StoppingCriterion,
    *,
    truncation: TruncationPolicy = NoTruncation(),
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult:
    r"""Computes an approximate Shapley value by sampling independent
    permutations of the index set, approximating the sum:

    $$
    v_u(x_i) = \frac{1}{n!} \sum_{\sigma \in \Pi(n)}
    \tilde{w}( | \sigma_{:i} | )[u(\sigma_{:i} \cup \{i\}) − u(\sigma_{:i})],
    $$

    where $\sigma_{:i}$ denotes the set of indices in permutation sigma before
    the position where $i$ appears (see [[data-valuation]] for details).

    This implements the method described in (Ghorbani and Zou, 2019)<sup><a
    href="#ghorbani_data_2019">1</a></sup> with a double stopping criterion.

    !!! Todo
        Think of how to add Robin-Gelman or some other more principled stopping
        criterion.

    Instead of naively implementing the expectation, we sequentially add points
    to coalitions from a permutation and incrementally compute marginal utilities.
    We stop computing marginals for a given permutation based on a
    [TruncationPolicy][pydvl.value.shapley.truncated.TruncationPolicy].
    (Ghorbani and Zou, 2019)<sup><a href="#ghorbani_data_2019">1</a></sup>
    mention two policies: one that stops after a certain
    fraction of marginals are computed, implemented in
    [FixedTruncation][pydvl.value.shapley.truncated.FixedTruncation],
    and one that stops if the last computed utility ("score") is close to the
    total utility using the standard deviation of the utility as a measure of
    proximity, implemented in
    [BootstrapTruncation][pydvl.value.shapley.truncated.BootstrapTruncation].

    We keep sampling permutations and updating all shapley values
    until the [StoppingCriterion][pydvl.value.stopping.StoppingCriterion] returns
    `True`.

    Args:
        u: Utility object with model, data, and scoring function.
        done: function checking whether computation must stop.
        truncation: An optional callable which decides whether to interrupt
            processing a permutation and set all subsequent marginals to
            zero. Typically used to stop computation when the marginal is small.
        n_jobs: number of jobs across which to distribute the computation.
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display a progress bar.
        seed: Either an instance of a numpy random number generator or a seed for it.

    Returns:
        Object with the data values.

    !!! tip "Changed in version 0.9.0"
        Deprecated `config` argument and added a `parallel_backend`
        argument to allow users to pass the Parallel Backend instance
        directly.
    """
    algorithm = "permutation_montecarlo_shapley"

    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)
    u = parallel_backend.put(u)
    max_workers = parallel_backend.effective_n_jobs(n_jobs)
    n_submitted_jobs = 2 * max_workers  # number of jobs in the executor's queue

    seed_sequence = ensure_seed_sequence(seed)
    result = ValuationResult.zeros(
        algorithm=algorithm, indices=u.data.indices, data_names=u.data.data_names
    )

    pbar = tqdm(disable=not progress, total=100, unit="%")

    with parallel_backend.executor(
        max_workers=max_workers, cancel_futures=CancellationPolicy.ALL
    ) as executor:
        pending: set[Future] = set()
        while True:
            pbar.n = 100 * done.completion()
            pbar.refresh()

            completed, pending = wait(pending, timeout=1.0, return_when=FIRST_COMPLETED)
            for future in completed:
                result += future.result()
                # we could check outside the loop, but that means more
                # submissions if the stopping criterion is unstable
                if done(result):
                    return result

            # Ensure that we always have n_submitted_jobs in the queue or running
            n_remaining_slots = n_submitted_jobs - len(pending)
            seeds = seed_sequence.spawn(n_remaining_slots)
            for i in range(n_remaining_slots):
                future = executor.submit(
                    _permutation_montecarlo_one_step,
                    u,
                    truncation,
                    algorithm,
                    seed=seeds[i],
                )
                pending.add(future)

combinatorial_montecarlo_shapley

combinatorial_montecarlo_shapley(
    u: Utility,
    done: StoppingCriterion,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult

Computes an approximate Shapley value using the combinatorial definition:

\[v_u(i) = \frac{1}{n} \sum_{S \subseteq N \setminus \{i\}} \binom{n-1}{ | S | }^{-1} [u(S \cup \{i\}) − u(S)]\]

This consists of randomly sampling subsets of the power set of the training indices in u.data, and computing their marginal utilities. See Data valuation for details.

Note that because sampling is done with replacement, the approximation is poor even for \(2^{m}\) subsets with \(m>n\), even though there are \(2^{n-1}\) subsets for each \(i\). Prefer permutation_montecarlo_shapley().

Parallelization is done by splitting the set of indices across processes and computing the sum over subsets \(S \subseteq N \setminus \{i\}\) separately.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
done	Stopping criterion for the computation. TYPE: StoppingCriterion
n_jobs	number of parallel jobs across which to distribute the computation. Each worker receives a chunk of indices TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display progress bars for each job. TYPE: bool DEFAULT: False
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

Changed in version 0.9.0

Deprecated config argument and added a parallel_backend argument to allow users to pass the Parallel Backend instance directly.

Source code in src/pydvl/value/shapley/montecarlo.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def combinatorial_montecarlo_shapley(
    u: Utility,
    done: StoppingCriterion,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult:
    r"""Computes an approximate Shapley value using the combinatorial
    definition:

    $$v_u(i) = \frac{1}{n} \sum_{S \subseteq N \setminus \{i\}}
    \binom{n-1}{ | S | }^{-1} [u(S \cup \{i\}) − u(S)]$$

    This consists of randomly sampling subsets of the power set of the training
    indices in [u.data][pydvl.utils.utility.Utility], and computing their
    marginal utilities. See [Data valuation][data-valuation] for details.

    Note that because sampling is done with replacement, the approximation is
    poor even for $2^{m}$ subsets with $m>n$, even though there are $2^{n-1}$
    subsets for each $i$. Prefer
    [permutation_montecarlo_shapley()][pydvl.value.shapley.montecarlo.permutation_montecarlo_shapley].

    Parallelization is done by splitting the set of indices across processes and
    computing the sum over subsets $S \subseteq N \setminus \{i\}$ separately.

    Args:
        u: Utility object with model, data, and scoring function
        done: Stopping criterion for the computation.
        n_jobs: number of parallel jobs across which to distribute the
            computation. Each worker receives a chunk of
            [indices][pydvl.utils.dataset.Dataset.indices]
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display progress bars for each job.
        seed: Either an instance of a numpy random number generator or a seed for it.

    Returns:
        Object with the data values.

    !!! tip "Changed in version 0.9.0"
        Deprecated `config` argument and added a `parallel_backend`
        argument to allow users to pass the Parallel Backend instance
        directly.
    """
    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)

    map_reduce_job: MapReduceJob[NDArray, ValuationResult] = MapReduceJob(
        u.data.indices,
        map_func=_combinatorial_montecarlo_shapley,
        reduce_func=lambda results: reduce(operator.add, results),
        map_kwargs=dict(u=u, done=done, progress=progress),
        n_jobs=n_jobs,
        parallel_backend=parallel_backend,
    )
    return map_reduce_job(seed=seed)

permutation_exact_shapley

permutation_exact_shapley(
    u: Utility, *, progress: bool = True
) -> ValuationResult

Computes the exact Shapley value using the formulation with permutations:

\[v_u(x_i) = \frac{1}{n!} \sum_{\sigma \in \Pi(n)} [u(\sigma_{i-1} \cup {i}) − u(\sigma_{i})].\]

See Data valuation for details.

When the length of the training set is > 10 this prints a warning since the computation becomes too expensive. Used mostly for internal testing and simple use cases. Please refer to the Monte Carlo approximations for practical applications.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
progress	Whether to display progress bars for each job. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

Source code in src/pydvl/value/shapley/naive.py

def permutation_exact_shapley(u: Utility, *, progress: bool = True) -> ValuationResult:
    r"""Computes the exact Shapley value using the formulation with permutations:

    $$v_u(x_i) = \frac{1}{n!} \sum_{\sigma \in \Pi(n)} [u(\sigma_{i-1}
    \cup {i}) − u(\sigma_{i})].$$

    See [Data valuation][data-valuation] for details.

    When the length of the training set is > 10 this prints a warning since the
    computation becomes too expensive. Used mostly for internal testing and
    simple use cases. Please refer to the [Monte Carlo
    approximations][pydvl.value.shapley.montecarlo] for practical applications.

    Args:
        u: Utility object with model, data, and scoring function
        progress: Whether to display progress bars for each job.

    Returns:
        Object with the data values.
    """

    n = len(u.data)
    # Note that the cache in utility saves most of the refitting because we
    # use frozenset for the input.
    if n > 10:
        warnings.warn(
            f"Large dataset! Computation requires {n}! calls to utility()",
            RuntimeWarning,
        )

    values = np.zeros(n)
    for p in tqdm(
        permutations(u.data.indices),
        disable=not progress,
        desc="Permutation",
        total=math.factorial(n),
    ):
        for i, idx in enumerate(p):
            values[idx] += u(p[: i + 1]) - u(p[:i])
    values /= math.factorial(n)

    return ValuationResult(
        algorithm="permutation_exact_shapley",
        status=Status.Converged,
        values=values,
        data_names=u.data.data_names,
    )

combinatorial_exact_shapley

combinatorial_exact_shapley(
    u: Utility,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
) -> ValuationResult

Computes the exact Shapley value using the combinatorial definition.

\[v_u(i) = \frac{1}{n} \sum_{S \subseteq N \setminus \{i\}} \binom{n-1}{ | S | }^{-1} [u(S \cup \{i\}) − u(S)].\]

See Data valuation for details.

Note

If the length of the training set is > n_jobs*20 this prints a warning because the computation is very expensive. Used mostly for internal testing and simple use cases. Please refer to the Monte Carlo approximations for practical applications.

PARAMETER	DESCRIPTION
u	Utility object with model, data, and scoring function TYPE: Utility
n_jobs	Number of parallel jobs to use TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display progress bars for each job. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

Changed in version 0.9.0

Deprecated config argument and added a parallel_backend argument to allow users to pass the Parallel Backend instance directly.

Source code in src/pydvl/value/shapley/naive.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def combinatorial_exact_shapley(
    u: Utility,
    *,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
) -> ValuationResult:
    r"""Computes the exact Shapley value using the combinatorial definition.

    $$v_u(i) = \frac{1}{n} \sum_{S \subseteq N \setminus \{i\}}
    \binom{n-1}{ | S | }^{-1} [u(S \cup \{i\}) − u(S)].$$

    See [Data valuation][data-valuation] for details.

    !!! Note
        If the length of the training set is > n_jobs*20 this prints a warning
        because the computation is very expensive. Used mostly for internal
        testing and simple use cases. Please refer to the
        [Monte Carlo][pydvl.value.shapley.montecarlo] approximations for
        practical applications.

    Args:
        u: Utility object with model, data, and scoring function
        n_jobs: Number of parallel jobs to use
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display progress bars for each job.

    Returns:
        Object with the data values.

    !!! tip "Changed in version 0.9.0"
        Deprecated `config` argument and added a `parallel_backend`
        argument to allow users to pass the Parallel Backend instance
        directly.
    """
    # Arbitrary choice, will depend on time required, caching, etc.
    if len(u.data) // n_jobs > 20:
        warnings.warn(
            f"Large dataset! Computation requires 2^{len(u.data)} calls to model.fit()"
        )

    def reduce_fun(results: List[NDArray]) -> NDArray:
        return np.array(results).sum(axis=0)  # type: ignore

    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)

    map_reduce_job: MapReduceJob[NDArray, NDArray] = MapReduceJob(
        u.data.indices,
        map_func=_combinatorial_exact_shapley,
        map_kwargs=dict(u=u, progress=progress),
        reduce_func=reduce_fun,
        n_jobs=n_jobs,
        parallel_backend=parallel_backend,
    )
    values = map_reduce_job()
    return ValuationResult(
        algorithm="combinatorial_exact_shapley",
        status=Status.Converged,
        values=values,
        data_names=u.data.data_names,
    )

owen_sampling_shapley

owen_sampling_shapley(
    u: Utility,
    n_samples: int,
    max_q: int,
    *,
    method: OwenAlgorithm = Standard,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult

Owen sampling of Shapley values as described in (Okhrati and Lipani, 2021)¹.

This function computes a Monte Carlo approximation to

\[v_u(i) = \int_0^1 \mathbb{E}_{S \sim P_q(D_{\backslash \{i\}})} [u(S \cup \{i\}) - u(S)]\]

using one of two methods. The first one, selected with the argument mode = OwenAlgorithm.Standard, approximates the integral with:

\[\hat{v}_u(i) = \frac{1}{Q M} \sum_{j=0}^Q \sum_{m=1}^M [u(S^{(q_j)}_m \cup \{i\}) - u(S^{(q_j)}_m)],\]

where \(q_j = \frac{j}{Q} \in [0,1]\) and the sets \(S^{(q_j)}\) are such that a sample \(x \in S^{(q_j)}\) if a draw from a \(Ber(q_j)\) distribution is 1.

The second method, selected with the argument mode = OwenAlgorithm.Antithetic, uses correlated samples in the inner sum to reduce the variance:

\[\hat{v}_u(i) = \frac{1}{2 Q M} \sum_{j=0}^Q \sum_{m=1}^M [u(S^{(q_j)}_m \cup \{i\}) - u(S^{(q_j)}_m) + u((S^{(q_j)}_m)^c \cup \{i\}) - u((S^{( q_j)}_m)^c)],\]

where now \(q_j = \frac{j}{2Q} \in [0,\frac{1}{2}]\), and \(S^c\) is the complement of \(S\).

Note

The outer integration could be done instead with a quadrature rule.

PARAMETER	DESCRIPTION
u	Utility object holding data, model and scoring function. TYPE: Utility
n_samples	Numer of sets to sample for each value of q TYPE: int
max_q	Number of subdivisions for q ∈ [0,1] (the element sampling probability) used to approximate the outer integral. TYPE: int
method	Selects the algorithm to use, see the description. Either OwenAlgorithm.Full for \(q \in [0,1]\) or OwenAlgorithm.Halved for \(q \in [0,0.5]\) and correlated samples TYPE: OwenAlgorithm DEFAULT: Standard
n_jobs	Number of parallel jobs to use. Each worker receives a chunk of the total of max_q values for q. TYPE: int DEFAULT: 1
parallel_backend	Parallel backend instance to use for parallelizing computations. If None, use JoblibParallelBackend backend. See the Parallel Backends package for available options. TYPE: Optional[ParallelBackend] DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: Optional[ParallelConfig] DEFAULT: None
progress	Whether to display progress bars for each job. TYPE: bool DEFAULT: False
seed	Either an instance of a numpy random number generator or a seed for it. TYPE: Optional[Seed] DEFAULT: None

RETURNS	DESCRIPTION
ValuationResult	Object with the data values.

New in version 0.3.0

Changed in version 0.5.0

Support for parallel computation and enable antithetic sampling.

Changed in version 0.9.0

Deprecated config argument and added a parallel_backend argument to allow users to pass the Parallel Backend instance directly.

Source code in src/pydvl/value/shapley/owen.py

@deprecated(
    target=True,
    args_mapping={"config": "config"},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def owen_sampling_shapley(
    u: Utility,
    n_samples: int,
    max_q: int,
    *,
    method: OwenAlgorithm = OwenAlgorithm.Standard,
    n_jobs: int = 1,
    parallel_backend: Optional[ParallelBackend] = None,
    config: Optional[ParallelConfig] = None,
    progress: bool = False,
    seed: Optional[Seed] = None,
) -> ValuationResult:
    r"""Owen sampling of Shapley values as described in
    (Okhrati and Lipani, 2021)<sup><a href="#okhrati_multilinear_2021">1</a></sup>.

    This function computes a Monte Carlo approximation to

    $$v_u(i) = \int_0^1 \mathbb{E}_{S \sim P_q(D_{\backslash \{i\}})}
    [u(S \cup \{i\}) - u(S)]$$

    using one of two methods. The first one, selected with the argument ``mode =
    OwenAlgorithm.Standard``, approximates the integral with:

    $$\hat{v}_u(i) = \frac{1}{Q M} \sum_{j=0}^Q \sum_{m=1}^M [u(S^{(q_j)}_m
    \cup \{i\}) - u(S^{(q_j)}_m)],$$

    where $q_j = \frac{j}{Q} \in [0,1]$ and the sets $S^{(q_j)}$ are such that a
    sample $x \in S^{(q_j)}$ if a draw from a $Ber(q_j)$ distribution is 1.

    The second method, selected with the argument ``mode =
    OwenAlgorithm.Antithetic``, uses correlated samples in the inner sum to
    reduce the variance:

    $$\hat{v}_u(i) = \frac{1}{2 Q M} \sum_{j=0}^Q \sum_{m=1}^M [u(S^{(q_j)}_m
    \cup \{i\}) - u(S^{(q_j)}_m) + u((S^{(q_j)}_m)^c \cup \{i\}) - u((S^{(
    q_j)}_m)^c)],$$

    where now $q_j = \frac{j}{2Q} \in [0,\frac{1}{2}]$, and $S^c$ is the
    complement of $S$.

    !!! Note
        The outer integration could be done instead with a quadrature rule.

    Args:
        u: [Utility][pydvl.utils.utility.Utility] object holding data, model
            and scoring function.
        n_samples: Numer of sets to sample for each value of q
        max_q: Number of subdivisions for q ∈ [0,1] (the element sampling
            probability) used to approximate the outer integral.
        method: Selects the algorithm to use, see the description. Either
            [OwenAlgorithm.Full][pydvl.value.shapley.owen.OwenAlgorithm] for
            $q \in [0,1]$ or
            [OwenAlgorithm.Halved][pydvl.value.shapley.owen.OwenAlgorithm] for
            $q \in [0,0.5]$ and correlated samples
        n_jobs: Number of parallel jobs to use. Each worker receives a chunk
            of the total of `max_q` values for q.
        parallel_backend: Parallel backend instance to use
            for parallelizing computations. If `None`,
            use [JoblibParallelBackend][pydvl.parallel.backends.JoblibParallelBackend] backend.
            See the [Parallel Backends][pydvl.parallel.backends] package
            for available options.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        progress: Whether to display progress bars for each job.
        seed: Either an instance of a numpy random number generator or a seed for it.

    Returns:
        Object with the data values.

    !!! tip "New in version 0.3.0"

    !!! tip "Changed in version 0.5.0"
        Support for parallel computation and enable antithetic sampling.

    !!! tip "Changed in version 0.9.0"
        Deprecated `config` argument and added a `parallel_backend`
        argument to allow users to pass the Parallel Backend instance
        directly.

    """
    parallel_backend = _maybe_init_parallel_backend(parallel_backend, config)

    map_reduce_job: MapReduceJob[NDArray, ValuationResult] = MapReduceJob(
        u.data.indices,
        map_func=_owen_sampling_shapley,
        reduce_func=lambda results: reduce(operator.add, results),
        map_kwargs=dict(
            u=u,
            method=OwenAlgorithm(method),
            n_samples=n_samples,
            max_q=max_q,
            progress=progress,
        ),
        n_jobs=n_jobs,
        parallel_backend=parallel_backend,
    )

    return map_reduce_job(seed=seed)