Skip to content

pydvl.valuation.methods

BetaShapleyValuation 

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

Bases: SemivalueValuation

Computes Beta-Shapley values.

PARAMETER DESCRIPTION
utility

Object to compute utilities.

TYPE: UtilityBase

sampler

Sampling scheme to use.

TYPE: IndexSampler

is_done

Stopping criterion to use.

TYPE: StoppingCriterion

skip_converged

Whether to skip converged indices. Convergence is determined by the stopping criterion's converged array.

TYPE: bool DEFAULT: False

alpha

The alpha parameter of the Beta distribution.

TYPE: float

beta

The beta parameter of the Beta distribution.

TYPE: float

progress

Whether to show a progress bar. If a dictionary, it is passed to tqdm as keyword arguments, and the progress bar is displayed.

TYPE: bool DEFAULT: False

Source code in src/pydvl/valuation/methods/beta_shapley.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    alpha: float,
    beta: float,
    progress: bool = False,
    skip_converged: bool = False,
):
    super().__init__(
        utility, sampler, is_done, skip_converged=skip_converged, progress=progress
    )

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

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

ClasswiseShapleyValuation 

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

Bases: Valuation

Class to compute Class-wise Shapley values.

PARAMETER DESCRIPTION
utility

Class-wise utility object with model and class-wise scoring function.

TYPE: ClasswiseModelUtility

sampler

Class-wise sampling scheme to use.

TYPE: ClasswiseSampler

is_done

Stopping criterion to use.

TYPE: StoppingCriterion

progress

Whether to show a progress bar.

TYPE: dict[str, Any] | bool DEFAULT: False

normalize_values

Whether to normalize values after valuation.

TYPE: bool DEFAULT: True

Source code in src/pydvl/valuation/methods/classwise_shapley.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def __init__(
    self,
    utility: ClasswiseModelUtility,
    sampler: ClasswiseSampler,
    is_done: StoppingCriterion,
    progress: dict[str, Any] | bool = False,
    *,
    normalize_values: bool = True,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.labels: NDArray | None = None
    if not isinstance(utility.scorer, ClasswiseSupervisedScorer):
        raise ValueError("scorer must be an instance of ClasswiseSupervisedScorer")
    self.scorer: ClasswiseSupervisedScorer = utility.scorer
    self.is_done = is_done
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    else:
        self.tqdm_args.update(progress if isinstance(progress, dict) else {})
    self.normalize_values = normalize_values

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

DataBanzhafValuation 

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

Bases: SemivalueValuation

Computes Banzhaf values.

Source code in src/pydvl/valuation/methods/semivalue.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

BaggingModel

Bases: Protocol

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

fit 

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

Fit the model to the data

PARAMETER DESCRIPTION
x

Independent variables

TYPE: NDArray

y

Dependent variable

TYPE: NDArray | None

Source code in src/pydvl/utils/types.py 
105
106
107
108
109
110
111
112
def fit(self, x: NDArray, y: NDArray | None):
    """Fit the model to the data

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

predict 

predict(x: NDArray) -> NDArray

Compute predictions for the input

PARAMETER DESCRIPTION
x

Independent variables for which to compute predictions

TYPE: NDArray

RETURNS DESCRIPTION
NDArray

Predictions for the input
Source code in src/pydvl/utils/types.py 
114
115
116
117
118
119
120
121
122
123
def predict(self, x: NDArray) -> NDArray:
    """Compute predictions for the input

    Args:
        x: Independent variables for which to compute predictions

    Returns:
        Predictions for the input
    """
    pass

DataOOBValuation 

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

Bases: Valuation

Computes Data Out-Of-Bag values.

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

PARAMETER DESCRIPTION
model

A fitted bagging model. Bagging models in sklearn include [[BaggingClassifier]], [[BaggingRegressor]], [[IsolationForest]], RandomForest, ExtraTrees, or any model which defines an attribute estimators_ and uses bootstrapped subsamples to compute predictions.

TYPE: BaggingModel

score

A callable for point-wise comparison of true values with the predictions. If None, uses point-wise accuracy for classifiers and negative \(l_2\) distance for regressors.

TYPE: PointwiseScore | None DEFAULT: None

Source code in src/pydvl/valuation/methods/data_oob.py 
72
73
74
75
76
77
78
79
def __init__(
    self,
    model: BaggingModel,
    score: PointwiseScore | None = None,
):
    super().__init__()
    self.model = model
    self.score = score

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

fit 

fit(data: Dataset) -> Self

Compute the Data-OOB values.

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

PARAMETER DESCRIPTION
data

Data for which to compute values

TYPE: Dataset

RETURNS DESCRIPTION
Self

The fitted object.
Source code in src/pydvl/valuation/methods/data_oob.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def fit(self, data: Dataset) -> Self:
    """Compute the Data-OOB values.

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

    Args:
        data: Data for which to compute values

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

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

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

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

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

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

DeltaShapleyValuation 

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

Bases: SemivalueValuation

Computes \(\delta\)-Shapley values.

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

PARAMETER DESCRIPTION
utility

Object to compute utilities.

TYPE: UtilityBase

is_done

Stopping criterion to use.

TYPE: StoppingCriterion

lower_bound

The lower bound of the size of the subsets to sample from.

TYPE: int

upper_bound

The upper bound of the size of the subsets to sample from.

TYPE: int

seed

The seed for the random number generator used by the sampler.

TYPE: Seed | None DEFAULT: None

progress

Whether to show a progress bar. If a dictionary, it is passed to tqdm as keyword arguments, and the progress bar is displayed.

TYPE: bool DEFAULT: False

skip_converged

Whether to skip converged indices, as determined by the stopping criterion's converged array.

TYPE: bool DEFAULT: False

Source code in src/pydvl/valuation/methods/delta_shapley.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(
    self,
    utility: UtilityBase,
    is_done: StoppingCriterion,
    lower_bound: int,
    upper_bound: int,
    seed: Seed | None = None,
    skip_converged: bool = False,
    progress: bool = False,
):
    sampler = StratifiedSampler(
        sample_sizes=ConstantSampleSize(
            1, lower_bound=lower_bound, upper_bound=upper_bound
        ),
        sample_sizes_iteration=RandomSizeIteration,
        index_iteration=RandomIndexIteration,
        seed=seed,
    )
    self.lower_bound = lower_bound
    self.upper_bound = upper_bound
    super().__init__(
        utility, sampler, is_done, progress=progress, skip_converged=skip_converged
    )

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

GroupTestingShapleyValuation 

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

Bases: Valuation

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

See Data valuation for an overview.

Warning

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

PARAMETER DESCRIPTION
utility

Utility object with model, data and scoring function.

TYPE: UtilityBase

n_samples

The number of samples to use. A sample size with theoretical guarantees can be computed using compute_n_samples().

TYPE: int

epsilon

The error tolerance.

TYPE: float

solver_options

Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see this tutorial. For additional options cvxpy's documentation.

TYPE: dict | None DEFAULT: None

progress

Whether to show a progress bar during the construction of the group-testing problem.

TYPE: bool DEFAULT: True

seed

Seed for the random number generator.

TYPE: Seed | None DEFAULT: None

batch_size

The number of samples to draw in each batch. Can be used to reduce parallelization overhead for fast utilities. Defaults to 1.

TYPE: int DEFAULT: 1

Source code in src/pydvl/valuation/methods/gt_shapley.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __init__(
    self,
    utility: UtilityBase,
    n_samples: int,
    epsilon: float,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__()

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

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

fit 

fit(data: Dataset) -> Self

Calculate the group-testing valuation on a dataset.

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

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

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)
Source code in src/pydvl/valuation/methods/gt_shapley.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def fit(self, data: Dataset) -> Self:
    """Calculate the group-testing valuation on a dataset.

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

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

    ```python
    from joblib import parallel_config

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

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

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

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

    self.result = solution
    return self

Valuation 

Valuation()

Bases: ABC

Source code in src/pydvl/valuation/base.py 
16
17
def __init__(self) -> None:
    self.result: ValuationResult | None = None

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

Dataset 

Dataset(
    x: NDArray,
    y: NDArray,
    feature_names: Sequence[str] | NDArray[str_] | None = None,
    target_names: Sequence[str] | NDArray[str_] | None = None,
    data_names: Sequence[str] | NDArray[str_] | None = None,
    description: str | None = None,
    multi_output: bool = False,
)

A convenience class to handle datasets.

It holds a dataset, together with info on feature names, target names, and data names. It is used to pass data around to valuation methods.

The underlying data arrays can be accessed via Dataset.data(), which returns the tuple (X, y) as a read-only RawData object. The data can be accessed by indexing the object directly, e.g. dataset[0] will return the data point corresponding to index 0 in dataset. For this base class, this is the same as dataset.data([0]), which is the first point in the data array, but derived classes can behave differently.

PARAMETER DESCRIPTION
x

training data

TYPE: NDArray

y

labels for training data

TYPE: NDArray

feature_names

names of the features of x data

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

target_names

names of the features of y data

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

data_names

names assigned to data points. For example, if the dataset is a time series, each entry can be a timestamp which can be referenced directly instead of using a row number.

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

description

A textual description of the dataset.

TYPE: str | None DEFAULT: None

multi_output

set to False if labels are scalars, or to True if they are vectors of dimension > 1.

TYPE: bool DEFAULT: False

Changed in version 0.10.0

No longer holds split data, but only x, y.

Changed in version 0.10.0

Slicing now return a new Dataset object, not raw data.

Source code in src/pydvl/valuation/dataset.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def __init__(
    self,
    x: NDArray,
    y: NDArray,
    feature_names: Sequence[str] | NDArray[np.str_] | None = None,
    target_names: Sequence[str] | NDArray[np.str_] | None = None,
    data_names: Sequence[str] | NDArray[np.str_] | None = None,
    description: str | None = None,
    multi_output: bool = False,
):
    self._x, self._y = check_X_y(
        x, y, multi_output=multi_output, estimator="Dataset"
    )

    def make_names(s: str, a: np.ndarray) -> list[str]:
        n = a.shape[1] if len(a.shape) > 1 else 1
        return [f"{s}{i:0{1 + int(np.log10(n))}d}" for i in range(1, n + 1)]

    self.feature_names = (
        list(feature_names) if feature_names is not None else make_names("x", x)
    )
    self.target_names = (
        list(target_names) if target_names is not None else make_names("y", y)
    )

    if len(self._x.shape) > 1:
        if len(self.feature_names) != self._x.shape[-1]:
            raise ValueError("Mismatching number of features and names")
    if len(self._y.shape) > 1:
        if len(self.target_names) != self._y.shape[-1]:
            raise ValueError("Mismatching number of targets and names")

    self.description = description or "No description"
    self._indices = np.arange(len(self._x), dtype=np.int_)
    self._data_names = (
        np.array(data_names, dtype=np.str_)
        if data_names is not None
        else self._indices.astype(np.str_)
    )

indices property 

indices: NDArray[int_]

Index of positions in data.x_train.

Contiguous integers from 0 to len(Dataset).

names property 

names: NDArray[str_]

Names of each individual datapoint.

Used for reporting Shapley values.

n_features property 

n_features: int

Returns the number of dimensions of a sample.

feature 

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

Returns a slice for the feature with the given name.

Source code in src/pydvl/valuation/dataset.py 
185
186
187
188
189
190
def feature(self, name: str) -> tuple[slice, int]:
    """Returns a slice for the feature with the given name."""
    try:
        return np.index_exp[:, self.feature_names.index(name)]  # type: ignore
    except ValueError:
        raise ValueError(f"Feature {name} is not in {self.feature_names}")

data 

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

Given a set of indices, returns the training data that refer to those indices, as a read-only tuple-like structure.

This is used mainly by subclasses of UtilityBase to retrieve subsets of the data from indices.

PARAMETER DESCRIPTION
indices

Optional indices that will be used to select points from the training data. If None, the entire training data will be returned.

TYPE: int | slice | Sequence[int] | NDArray[int_] | None DEFAULT: None

RETURNS DESCRIPTION
RawData

If indices is not None, the selected x and y arrays from the training data. Otherwise, the entire dataset.
Source code in src/pydvl/valuation/dataset.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def data(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> RawData:
    """Given a set of indices, returns the training data that refer to those
    indices, as a read-only tuple-like structure.

    This is used mainly by subclasses of
    [UtilityBase][pydvl.valuation.utility.base.UtilityBase] to retrieve subsets of
    the data from indices.

    Args:
        indices: Optional indices that will be used to select points from
            the training data. If `None`, the entire training data will be
            returned.

    Returns:
        If `indices` is not `None`, the selected x and y arrays from the
            training data. Otherwise, the entire dataset.
    """
    if indices is None:
        return RawData(self._x, self._y)
    return RawData(self._x[indices], self._y[indices])

data_indices 

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

Returns a subset of indices.

This is equivalent to using Dataset.indices[logical_indices] but allows subclasses to define special behaviour, e.g. when indices in Dataset do not match the indices in the data.

For Dataset, this is a simple pass-through.

PARAMETER DESCRIPTION
indices

A set of indices held by this object

TYPE: Sequence[int] | None DEFAULT: None

RETURNS DESCRIPTION
NDArray[int_]

The indices of the data points in the data array.
Source code in src/pydvl/valuation/dataset.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def data_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns a subset of indices.

    This is equivalent to using `Dataset.indices[logical_indices]` but allows
    subclasses to define special behaviour, e.g. when indices in `Dataset` do not
    match the indices in the data.

    For `Dataset`, this is a simple pass-through.

    Args:
        indices: A set of indices held by this object

    Returns:
        The indices of the data points in the data array.
    """
    if indices is None:
        return self._indices
    return self._indices[indices]

logical_indices 

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

Returns the indices in this Dataset for the given indices in the data array.

This is equivalent to using Dataset.indices[data_indices] but allows subclasses to define special behaviour, e.g. when indices in Dataset do not match the indices in the data.

PARAMETER DESCRIPTION
indices

A set of indices in the data array.

TYPE: Sequence[int] | None DEFAULT: None

RETURNS DESCRIPTION
NDArray[int_]

The abstract indices for the given data indices.
Source code in src/pydvl/valuation/dataset.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def logical_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns the indices in this `Dataset` for the given indices in the data array.

    This is equivalent to using `Dataset.indices[data_indices]` but allows
    subclasses to define special behaviour, e.g. when indices in `Dataset` do not
    match the indices in the data.

    Args:
        indices: A set of indices in the data array.

    Returns:
        The abstract indices for the given data indices.
    """
    if indices is None:
        return self._indices
    return self._indices[indices]

from_sklearn classmethod 

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

Constructs two Dataset objects from a sklearn.utils.Bunch, as returned by the load_* functions in scikit-learn toy datasets.

Example 
>>> from pydvl.valuation.dataset import Dataset
>>> from sklearn.datasets import load_boston  # noqa
>>> train, test = Dataset.from_sklearn(load_boston())
PARAMETER DESCRIPTION
data

scikit-learn Bunch object. The following attributes are supported:

  • data: covariates.
  • target: target variables (labels).
  • feature_names (optional): the feature names.
  • target_names (optional): the target names.
  • DESCR (optional): a description.

TYPE: Bunch

train_size

size of the training dataset. Used in train_test_split float values represent the fraction of the dataset to include in the training split and should be in (0,1). An integer value sets the absolute number of training samples.

TYPE: int | float DEFAULT: 0.8

the value is automatically set to the complement of the test size. random_state: seed for train / test split stratify_by_target: If True, data is split in a stratified fashion, using the target variable as labels. Read more in scikit-learn's user guide. kwargs: Additional keyword arguments to pass to the Dataset constructor. Use this to pass e.g. is_multi_output.

RETURNS DESCRIPTION
tuple[Dataset, Dataset]

Object with the sklearn dataset

Changed in version 0.6.0

Added kwargs to pass to the Dataset constructor.

Changed in version 0.10.0

Returns a tuple of two Dataset objects.

Source code in src/pydvl/valuation/dataset.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
@classmethod
def from_sklearn(
    cls,
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs,
) -> tuple[Dataset, Dataset]:
    """Constructs two [Dataset][pydvl.valuation.dataset.Dataset] objects from a
    [sklearn.utils.Bunch][], as returned by the `load_*`
    functions in [scikit-learn toy datasets](https://scikit-learn.org/stable/datasets/toy_dataset.html).

    ??? Example
        ```pycon
        >>> from pydvl.valuation.dataset import Dataset
        >>> from sklearn.datasets import load_boston  # noqa
        >>> train, test = Dataset.from_sklearn(load_boston())
        ```

    Args:
        data: scikit-learn Bunch object. The following attributes are supported:

            - `data`: covariates.
            - `target`: target variables (labels).
            - `feature_names` (**optional**): the feature names.
            - `target_names` (**optional**): the target names.
            - `DESCR` (**optional**): a description.
        train_size: size of the training dataset. Used in `train_test_split`
            float values represent the fraction of the dataset to include in the
            training split and should be in (0,1). An integer value sets the
            absolute number of training samples.
    the value is automatically set to the complement of the test size.
        random_state: seed for train / test split
        stratify_by_target: If `True`, data is split in a stratified
            fashion, using the target variable as labels. Read more in
            [scikit-learn's user guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor. Use this to pass e.g. `is_multi_output`.

    Returns:
        Object with the sklearn dataset

    !!! tip "Changed in version 0.6.0"
        Added kwargs to pass to the [Dataset][pydvl.valuation.dataset.Dataset] constructor.
    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two [Dataset][pydvl.valuation.dataset.Dataset] objects.
    """
    x_train, x_test, y_train, y_test = train_test_split(
        data.data,
        data.target,
        train_size=train_size,
        random_state=random_state,
        stratify=data.target if stratify_by_target else None,
    )
    return (
        cls(
            x_train,
            y_train,
            feature_names=data.get("feature_names"),
            target_names=data.get("target_names"),
            description=data.get("DESCR"),
            **kwargs,
        ),
        cls(
            x_test,
            y_test,
            feature_names=data.get("feature_names"),
            target_names=data.get("target_names"),
            description=data.get("DESCR"),
            **kwargs,
        ),
    )

from_arrays classmethod 

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

Constructs a Dataset object from X and y numpy arrays as returned by the make_* functions in sklearn generated datasets.

Example 
>>> from pydvl.valuation.dataset import Dataset
>>> from sklearn.datasets import make_regression
>>> X, y = make_regression()
>>> dataset = Dataset.from_arrays(X, y)
PARAMETER DESCRIPTION
X

numpy array of shape (n_samples, n_features)

TYPE: NDArray

y

numpy array of shape (n_samples,)

TYPE: NDArray

train_size

size of the training dataset. Used in train_test_split

TYPE: float DEFAULT: 0.8

random_state

seed for train / test split

TYPE: int | None DEFAULT: None

stratify_by_target

If True, data is split in a stratified fashion, using the y variable as labels. Read more in sklearn's user guide.

TYPE: bool DEFAULT: False

kwargs

Additional keyword arguments to pass to the Dataset constructor. Use this to pass e.g. feature_names or target_names.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
tuple[Dataset, Dataset]

Object with the passed X and y arrays split across training and test sets.

New in version 0.4.0

Changed in version 0.6.0

Added kwargs to pass to the Dataset constructor.

Changed in version 0.10.0

Returns a tuple of two Dataset objects.

Source code in src/pydvl/valuation/dataset.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
@classmethod
def from_arrays(
    cls,
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    **kwargs: Any,
) -> tuple[Dataset, Dataset]:
    """Constructs a [Dataset][pydvl.valuation.dataset.Dataset] object from X and y numpy arrays  as
    returned by the `make_*` functions in [sklearn generated datasets](https://scikit-learn.org/stable/datasets/sample_generators.html).

    ??? Example
        ```pycon
        >>> from pydvl.valuation.dataset import Dataset
        >>> from sklearn.datasets import make_regression
        >>> X, y = make_regression()
        >>> dataset = Dataset.from_arrays(X, y)
        ```

    Args:
        X: numpy array of shape (n_samples, n_features)
        y: numpy array of shape (n_samples,)
        train_size: size of the training dataset. Used in `train_test_split`
        random_state: seed for train / test split
        stratify_by_target: If `True`, data is split in a stratified fashion,
            using the y variable as labels. Read more in [sklearn's user
            guide](https://scikit-learn.org/stable/modules/cross_validation.html#stratification).
        kwargs: Additional keyword arguments to pass to the
            [Dataset][pydvl.valuation.dataset.Dataset] constructor. Use this to pass
            e.g. `feature_names` or `target_names`.

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

    !!! tip "New in version 0.4.0"

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

    !!! tip "Changed in version 0.10.0"
        Returns a tuple of two [Dataset][pydvl.valuation.dataset.Dataset] objects.
    """
    x_train, x_test, y_train, y_test = train_test_split(
        X,
        y,
        train_size=train_size,
        random_state=random_state,
        stratify=y if stratify_by_target else None,
    )
    return cls(x_train, y_train, **kwargs), cls(x_test, y_test, **kwargs)

ValuationResult 

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

Bases: Sequence, Iterable[ValueItem]

Objects of this class hold the results of valuation algorithms.

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

Indexing

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

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

Sorting

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

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

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

Converting back and forth from data indices and positions

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

Operating on results

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

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

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

PARAMETER DESCRIPTION
values

An array of values. If omitted, defaults to an empty array or to an array of zeros if indices are given.

TYPE: Sequence[float64] | NDArray[float64]

indices

An optional array of indices in the original dataset. If omitted, defaults to np.arange(len(values)). Warning: It is common to pass the indices of a Dataset here. Attention must be paid in a parallel context to copy them to the local process. Just do indices=np.copy(data.indices).

TYPE: Sequence[IndexT] | NDArray[IndexT] | None DEFAULT: None

variances

An optional array of variances of the marginals from which the values are computed.

TYPE: Sequence[float64] | NDArray[float64] | None DEFAULT: None

counts

An optional array with the number of updates for each value. Defaults to an array of ones.

TYPE: Sequence[int_] | NDArray[int_] | None DEFAULT: None

data_names

Names for the data points. Defaults to index numbers if not set.

TYPE: Sequence[NameT] | NDArray[NameT] | None DEFAULT: None

algorithm

The method used.

TYPE: str DEFAULT: ''

status

The end status of the algorithm.

TYPE: Status DEFAULT: Pending

sort

Whether to sort the indices. Defaults to None for no sorting. Set to True for ascending order by value, False for descending. See above how sorting affects usage as an iterable or sequence.

TYPE: bool | None DEFAULT: None

extra_values

Additional values that can be passed as keyword arguments. This can contain, for example, the least core value.

TYPE: Any DEFAULT: {}

RAISES DESCRIPTION
ValueError

If input arrays have mismatching lengths.
Changed in 0.10.0

Changed the behaviour of the sort argument.

Source code in src/pydvl/valuation/result.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def __init__(
    self,
    *,
    values: Sequence[np.float64] | NDArray[np.float64],
    variances: Sequence[np.float64] | NDArray[np.float64] | None = None,
    counts: Sequence[np.int_] | NDArray[np.int_] | None = None,
    indices: Sequence[IndexT] | NDArray[IndexT] | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    algorithm: str = "",
    status: Status = Status.Pending,
    sort: bool | None = None,
    **extra_values: Any,
):
    if variances is not None and len(variances) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and variances ({len(variances)}) do not match"
        )
    if data_names is not None and len(data_names) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and data_names ({len(data_names)}) do not match"
        )
    if indices is not None and len(indices) != len(values):
        raise ValueError(
            f"Lengths of values ({len(values)}) "
            f"and indices ({len(indices)}) do not match"
        )

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

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

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

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

values property 

values: NDArray[float64]

The values, possibly sorted.

variances property 

variances: NDArray[float64]

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

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

stderr property 

stderr: NDArray[float64]

Standard errors of the value estimates, possibly sorted.

counts property 

counts: NDArray[int_]

The raw counts, possibly sorted.

indices property 

indices: NDArray[IndexT]

The indices for the values, possibly sorted.

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

names property 

names: NDArray[NameT]

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

sort 

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

Sorts the indices in place by key.

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

PARAMETER DESCRIPTION
reverse

Whether to sort in descending order by value.

TYPE: bool DEFAULT: False

key

The key to sort by. Defaults to ValueItem.value.

TYPE: Literal['value', 'variance', 'index', 'name'] DEFAULT: 'value'

Source code in src/pydvl/valuation/result.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def sort(
    self,
    reverse: bool = False,
    # Need a "Comparable" type here
    key: Literal["value", "variance", "index", "name"] = "value",
) -> None:
    """Sorts the indices in place by `key`.

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

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

positions 

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

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

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

np.all(v.indices[v.positions(data_indices)] == data_indices) == True
Source code in src/pydvl/valuation/result.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
def positions(self, data_indices: IndexSetT | list[IndexT]) -> IndexSetT:
    """Return the location (indices) within the `ValuationResult` for the given
    data indices.

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

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

copy 

copy() -> ValuationResult

Returns a copy of the object.

Source code in src/pydvl/valuation/result.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
def copy(self) -> ValuationResult:
    """Returns a copy of the object."""
    return ValuationResult(
        values=self._values.copy(),
        variances=self._variances.copy(),
        counts=self._counts.copy(),
        indices=self._indices.copy(),
        data_names=self._names.copy(),
        algorithm=self._algorithm,
        status=self._status,
        sort=self._sort_order,
        **self._extra_values,
    )

__getattr__ 

__getattr__(attr: str) -> Any

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

Source code in src/pydvl/valuation/result.py 
401
402
403
404
405
406
407
408
409
410
411
def __getattr__(self, attr: str) -> Any:
    """Allows access to extra values as if they were properties of the instance."""
    # This is here to avoid a RecursionError when copying or pickling the object
    if attr == "_extra_values":
        raise AttributeError()
    try:
        return self._extra_values[attr]
    except KeyError as e:
        raise AttributeError(
            f"{self.__class__.__name__} object has no attribute {attr}"
        ) from e

__iter__ 

__iter__() -> Iterator[ValueItem]

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

Source code in src/pydvl/valuation/result.py 
477
478
479
480
481
482
483
484
485
486
487
488
def __iter__(self) -> Iterator[ValueItem]:
    """Iterate over the results returning [ValueItem][pydvl.valuation.result.ValueItem] objects.
    To sort in place before iteration, use [sort()][pydvl.valuation.result.ValuationResult.sort].
    """
    for pos in self._sort_positions:
        yield ValueItem(
            self._indices[pos],
            self._names[pos],
            self._values[pos],
            self._variances[pos],
            self._counts[pos],
        )

__add__ 

__add__(other: ValuationResult) -> ValuationResult

Adds two ValuationResults.

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

Warning

Abusing this will introduce numerical errors.

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

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

Warning

FIXME: Arbitrary extra_values aren't handled.

Source code in src/pydvl/valuation/result.py 
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
def __add__(self, other: ValuationResult) -> ValuationResult:
    """Adds two ValuationResults.

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

    !!! Warning
        Abusing this will introduce numerical errors.

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

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

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

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

    self._check_compatible(other)

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

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

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

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

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

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

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

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

    both_pos = np.intersect1d(this_pos, other_pos)

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

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

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

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

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

update 

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

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

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

PARAMETER DESCRIPTION
data_idx

Data index of the value to update.

TYPE: int | IndexT

new_value

New value to add to the result.

TYPE: float

RETURNS DESCRIPTION
ValuationResult

A reference to the same, modified result.
RAISES DESCRIPTION
IndexError

If the index is not found.
Source code in src/pydvl/valuation/result.py 
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
def update(self, data_idx: int | IndexT, new_value: float) -> ValuationResult:
    """Updates the result in place with a new value, using running mean
    and variance.

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

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

    Returns:
        A reference to the same, modified result.

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

scale 

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

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

PARAMETER DESCRIPTION
factor

Factor to scale by.

TYPE: float

data_indices

Data indices to scale. If None, all values are scaled.

TYPE: NDArray[IndexT] | None DEFAULT: None

Source code in src/pydvl/valuation/result.py 
676
677
678
679
680
681
682
683
684
685
686
687
688
689
def scale(self, factor: float, data_indices: NDArray[IndexT] | None = None):
    """
    Scales the values and variances of the result by a coefficient.

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

get 

get(data_idx: Integral) -> ValueItem

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

PARAMETER DESCRIPTION
data_idx

Data index of the value to retrieve.

TYPE: Integral

RAISES DESCRIPTION
IndexError

If the index is not found.
Source code in src/pydvl/valuation/result.py 
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
def get(self, data_idx: Integral) -> ValueItem:
    """Retrieves a ValueItem by data index, as opposed to sort index, like
    the indexing operator.

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

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

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

to_dataframe 

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

Returns values as a dataframe.

PARAMETER DESCRIPTION
column

Name for the column holding the data value. Defaults to the name of the algorithm used.

TYPE: str | None DEFAULT: None

use_names

Whether to use data names instead of indices for the DataFrame's index.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
DataFrame

A dataframe with three columns: name, name_variances and name_counts, where name is the value of argument column.
Source code in src/pydvl/valuation/result.py 
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
def to_dataframe(
    self, column: str | None = None, use_names: bool = False
) -> pd.DataFrame:
    """Returns values as a dataframe.

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

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

from_random classmethod 

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

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

PARAMETER DESCRIPTION
size

Number of values to generate

TYPE: int

total

If set, the values are normalized to sum to this number ("efficiency" property of Shapley values).

TYPE: float | None DEFAULT: None

seed

Random seed to use

TYPE: Seed | None DEFAULT: None

kwargs

Additional options to pass to the constructor of ValuationResult. Use to override status, names, etc.

TYPE: dict[str, Any] DEFAULT: {}

RETURNS DESCRIPTION
'ValuationResult'

A valuation result with its status set to
'ValuationResult'

Status.Converged by default.
RAISES DESCRIPTION
ValueError

If size is less than 1.

Changed in version 0.6.0

Added parameter total. Check for zero size

Source code in src/pydvl/valuation/result.py 
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
@classmethod
def from_random(
    cls,
    size: int,
    total: float | None = None,
    seed: Seed | None = None,
    **kwargs: dict[str, Any],
) -> "ValuationResult":
    """Creates a [ValuationResult][pydvl.valuation.result.ValuationResult] object
    and fills it with an array of random values from a uniform distribution in
    [-1,1]. The values can be made to sum up to a given total number (doing so will
    change their range).

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

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

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

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

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

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

empty classmethod 

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

Creates an empty ValuationResult object.

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

PARAMETER DESCRIPTION
algorithm

Name of the algorithm used to compute the values

TYPE: str DEFAULT: ''

indices

Optional sequence or array of indices.

TYPE: IndexSetT | None DEFAULT: None

data_names

Optional sequences or array of names for the data points. Defaults to index numbers if not set.

TYPE: Sequence[NameT] | NDArray[NameT] | None DEFAULT: None

n_samples

Number of valuation result entries.

TYPE: int DEFAULT: 0

RETURNS DESCRIPTION
ValuationResult

Object with the results.
Source code in src/pydvl/valuation/result.py 
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
@classmethod
def empty(
    cls,
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.valuation.result.ValuationResult] object.

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

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

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

zeros classmethod 

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

Creates an empty ValuationResult object.

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

PARAMETER DESCRIPTION
algorithm

Name of the algorithm used to compute the values

TYPE: str DEFAULT: ''

indices

Data indices to use. A copy will be made. If not given, the indices will be set to the range [0, n_samples).

TYPE: IndexSetT | None DEFAULT: None

data_names

Data names to use. A copy will be made. If not given, the names will be set to the string representation of the indices.

TYPE: Sequence[NameT] | NDArray[NameT] | None DEFAULT: None

n_samples

Number of data points whose values are computed. If not given, the length of indices will be used.

TYPE: int DEFAULT: 0

RETURNS DESCRIPTION
ValuationResult

Object with the results.
Source code in src/pydvl/valuation/result.py 
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
@classmethod
def zeros(
    cls,
    algorithm: str = "",
    indices: IndexSetT | None = None,
    data_names: Sequence[NameT] | NDArray[NameT] | None = None,
    n_samples: int = 0,
) -> ValuationResult:
    """Creates an empty [ValuationResult][pydvl.valuation.result.ValuationResult] object.

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

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

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

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

Status

Bases: Enum

Status of a computation.

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

OR

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

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

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

AND

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

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

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

NOT

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

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

Boolean casting

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

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

Warning

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

GroupedDataset 

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

Bases: Dataset

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

PARAMETER DESCRIPTION
x

training data

TYPE: NDArray

y

labels of training data

TYPE: NDArray

data_groups

Sequence of the same length as x_train containing a group id for each training data point. Data points with the same id will then be grouped by this object and considered as one for effects of valuation. Group ids are assumed to be zero-based consecutive integers

TYPE: Sequence[int] | NDArray[int_]

feature_names

names of the covariates' features.

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

target_names

names of the labels or targets y

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

data_names

names of the data points. For example, if the dataset is a time series, each entry can be a timestamp.

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

group_names

names of the groups. If not provided, the numerical group ids from data_groups will be used.

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

description

A textual description of the dataset

TYPE: str | None DEFAULT: None

kwargs

Additional keyword arguments to pass to the Dataset constructor.

TYPE: Any DEFAULT: {}

Changed in version 0.6.0

Added group_names and forwarding of kwargs

Changed in version 0.10.0

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

Source code in src/pydvl/valuation/dataset.py 
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def __init__(
    self,
    x: NDArray,
    y: NDArray,
    data_groups: Sequence[int] | NDArray[np.int_],
    feature_names: Sequence[str] | NDArray[np.str_] | None = None,
    target_names: Sequence[str] | NDArray[np.str_] | None = None,
    data_names: Sequence[str] | NDArray[np.str_] | None = None,
    group_names: Sequence[str] | NDArray[np.str_] | None = None,
    description: str | None = None,
    **kwargs: Any,
):
    """Class for grouping datasets.

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

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

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

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

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

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

n_features property 

n_features: int

Returns the number of dimensions of a sample.

indices property 

indices

Indices of the groups.

names property 

names: NDArray[str_]

Names of the groups.

feature 

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

Returns a slice for the feature with the given name.

Source code in src/pydvl/valuation/dataset.py 
185
186
187
188
189
190
def feature(self, name: str) -> tuple[slice, int]:
    """Returns a slice for the feature with the given name."""
    try:
        return np.index_exp[:, self.feature_names.index(name)]  # type: ignore
    except ValueError:
        raise ValueError(f"Feature {name} is not in {self.feature_names}")

data 

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

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

PARAMETER DESCRIPTION
indices

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

TYPE: int | slice | Sequence[int] | NDArray[int_] | None DEFAULT: None

RETURNS DESCRIPTION
RawData

Tuple of training data x and labels y.
Source code in src/pydvl/valuation/dataset.py 
539
540
541
542
543
544
545
546
547
548
549
550
551
def data(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> RawData:
    """Returns the data and labels of all samples in the given groups.

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

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

data_indices 

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

Returns the indices of the samples in the given groups.

PARAMETER DESCRIPTION
indices

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

TYPE: int | slice | Sequence[int] | NDArray[int_] | None DEFAULT: None

RETURNS DESCRIPTION
NDArray[int_]

Indices of the samples in the given groups.
Source code in src/pydvl/valuation/dataset.py 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
def data_indices(
    self, indices: int | slice | Sequence[int] | NDArray[np.int_] | None = None
) -> NDArray[np.int_]:
    """Returns the indices of the samples in the given groups.

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

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

logical_indices 

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

Returns the group indices for the given data indices.

PARAMETER DESCRIPTION
indices

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

TYPE: Sequence[int] | None DEFAULT: None

RETURNS DESCRIPTION
NDArray[int_]

Group indices for the given data indices.
Source code in src/pydvl/valuation/dataset.py 
571
572
573
574
575
576
577
578
579
580
581
582
583
def logical_indices(self, indices: Sequence[int] | None = None) -> NDArray[np.int_]:
    """Returns the group indices for the given data indices.

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

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

from_sklearn classmethod 

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

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

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

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

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

scikit-learn Bunch object. The following attributes are supported: - data: covariates. - target: target variables (labels). - feature_names (optional): the feature names. - target_names (optional): the target names. - DESCR (optional): a description.

TYPE: Bunch

train_size

size of the training dataset. Used in train_test_split float values represent the fraction of the dataset to include in the training split and should be in (0,1). An integer value sets the absolute number of training samples.

TYPE: int | float DEFAULT: 0.8

random_state

seed for train / test split.

TYPE: int | None DEFAULT: None

stratify_by_target

If True, data is split in a stratified fashion, using the target variable as labels. Read more in sklearn's user guide.

TYPE: bool DEFAULT: False

data_groups

an array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset.

TYPE: Sequence[int] | None DEFAULT: None

kwargs

Additional keyword arguments to pass to the Dataset constructor.

TYPE: dict[str, Any] DEFAULT: {}

RETURNS DESCRIPTION
tuple[GroupedDataset, GroupedDataset]

Datasets with the selected sklearn data

Changed in version 0.10.0

Returns a tuple of two GroupedDataset objects.

Source code in src/pydvl/valuation/dataset.py 
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
@classmethod
def from_sklearn(
    cls,
    data: Bunch,
    train_size: int | float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: dict[str, Any],
) -> tuple[GroupedDataset, GroupedDataset]:
    """Constructs a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object, and an
    ungrouped [Dataset][pydvl.valuation.dataset.Dataset] object from a
    [sklearn.utils.Bunch][sklearn.utils.Bunch] as returned by the `load_*` functions in
    [scikit-learn toy datasets](https://scikit-learn.org/stable/datasets/toy_dataset.html) and groups
    it.

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

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

    Returns:
        Datasets with the selected sklearn data

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

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

from_arrays classmethod 

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

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

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

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

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

array of shape (n_samples, n_features)

TYPE: NDArray

y

array of shape (n_samples,)

TYPE: NDArray

train_size

size of the training dataset. Used in train_test_split.

TYPE: float DEFAULT: 0.8

random_state

seed for train / test split.

TYPE: int | None DEFAULT: None

stratify_by_target

If True, data is split in a stratified fashion, using the y variable as labels. Read more in sklearn's user guide.

TYPE: bool DEFAULT: False

data_groups

an array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset.

TYPE: Sequence[int] | None DEFAULT: None

kwargs

Additional keyword arguments that will be passed to the GroupedDataset constructor.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
tuple[GroupedDataset, GroupedDataset]

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

New in version 0.4.0

Changed in version 0.6.0

Added kwargs to pass to the GroupedDataset constructor.

Changed in version 0.10.0

Returns a tuple of two GroupedDataset objects.

Source code in src/pydvl/valuation/dataset.py 
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
@classmethod
def from_arrays(
    cls,
    X: NDArray,
    y: NDArray,
    train_size: float = 0.8,
    random_state: int | None = None,
    stratify_by_target: bool = False,
    data_groups: Sequence[int] | None = None,
    **kwargs: Any,
) -> tuple[GroupedDataset, GroupedDataset]:
    """Constructs a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object,
    and an ungrouped [Dataset][pydvl.valuation.dataset.Dataset] object from X and y
    numpy arrays as returned by the `make_*` functions in
    [scikit-learn generated datasets](https://scikit-learn.org/stable/datasets/sample_generators.html).

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

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

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

    !!! tip "New in version 0.4.0"

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

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

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

from_dataset classmethod 

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

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

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

The original data.

TYPE: Dataset

data_groups

An array holding the group index or name for each data point. The length of this array must be equal to the number of data points in the dataset.

TYPE: Sequence[int] | NDArray[int_]

group_names

Names of the groups. If not provided, the numerical group ids from data_groups will be used.

TYPE: Sequence[str] | NDArray[str_] | None DEFAULT: None

kwargs

Additional arguments to be passed to the GroupedDataset constructor.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
GroupedDataset

A GroupedDataset with the initial Dataset grouped by data_groups.
Source code in src/pydvl/valuation/dataset.py 
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
@classmethod
def from_dataset(
    cls,
    data: Dataset,
    data_groups: Sequence[int] | NDArray[np.int_],
    group_names: Sequence[str] | NDArray[np.str_] | None = None,
    **kwargs: Any,
) -> GroupedDataset:
    """Creates a [GroupedDataset][pydvl.valuation.dataset.GroupedDataset] object from a
    [Dataset][pydvl.valuation.dataset.Dataset] object and a mapping of data groups.

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

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

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

KNNShapleyValuation 

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

Bases: Valuation

Computes exact Shapley values for a KNN classifier.

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

PARAMETER DESCRIPTION
model

KNeighborsClassifier model to use for valuation

TYPE: KNeighborsClassifier

test_data

Dataset containing test data for valuation

TYPE: Dataset

progress

Whether to display a progress bar.

TYPE: bool DEFAULT: True

clone_before_fit

Whether to clone the model before fitting.

TYPE: bool DEFAULT: True

Source code in src/pydvl/valuation/methods/knn_shapley.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    model: KNeighborsClassifier,
    test_data: Dataset,
    progress: bool = True,
    clone_before_fit: bool = True,
):
    super().__init__()
    if not isinstance(model, KNeighborsClassifier):
        raise TypeError("KNN Shapley requires a K-Nearest Neighbours model")
    self.model = model
    self.test_data = test_data
    self.progress = progress
    self.clone_before_fit = clone_before_fit

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

fit 

fit(data: Dataset) -> Self

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

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

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

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

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)
Source code in src/pydvl/valuation/methods/knn_shapley.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def fit(self, data: Dataset) -> Self:
    """Calculate exact shapley values for a KNN model on a dataset.

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

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

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

    ```python
    from joblib import parallel_config

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

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

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

    n_test = len(self.test_data)

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

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

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

    return self

LeastCoreValuation 

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

Bases: Valuation

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

See Data valuation for an overview.

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

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

PARAMETER DESCRIPTION
utility

Utility object with model, data and scoring function.

TYPE: UtilityBase

sampler

The sampler to use for the valuation.

TYPE: PowersetSampler

n_samples

The number of samples to use for the valuation. If None, it will be set to the sample limit of the chosen sampler (for finite samplers) or 1000 * len(data) (for infinite samplers).

TYPE: int | None DEFAULT: None

non_negative_subsidy

If True, the least core subsidy \(e\) is constrained to be non-negative.

TYPE: bool DEFAULT: False

solver_options

Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here.

TYPE: dict | None DEFAULT: None

progress

Whether to show a progress bar during the construction of the least-core problem.

TYPE: bool DEFAULT: True

Source code in src/pydvl/valuation/methods/least_core.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    utility: UtilityBase,
    sampler: PowersetSampler,
    n_samples: int | None = None,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
):
    super().__init__()

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

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

fit 

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

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

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

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)
Source code in src/pydvl/valuation/methods/least_core.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

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

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

    ```python
    from joblib import parallel_config

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

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

    algorithm = str(self._sampler)

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

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

    self.result = solution
    return self

ExactLeastCoreValuation 

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

Bases: LeastCoreValuation

Class to calculate exact least-core values.

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

The definition of the exact least-core valuation is:

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

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

PARAMETER DESCRIPTION
utility

Utility object with model, data and scoring function.

TYPE: UtilityBase

non_negative_subsidy

If True, the least core subsidy \(e\) is constrained to be non-negative.

TYPE: bool DEFAULT: False

solver_options

Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here.

TYPE: dict | None DEFAULT: None

progress

Whether to show a progress bar during the construction of the least-core problem.

TYPE: bool DEFAULT: True

Source code in src/pydvl/valuation/methods/least_core.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def __init__(
    self,
    utility: UtilityBase,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    batch_size: int = 1,
):
    super().__init__(
        utility=utility,
        sampler=DeterministicUniformSampler(
            index_iteration=FiniteNoIndexIteration, batch_size=batch_size
        ),
        n_samples=None,
        non_negative_subsidy=non_negative_subsidy,
        solver_options=solver_options,
        progress=progress,
    )

fit 

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

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

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

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)
Source code in src/pydvl/valuation/methods/least_core.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

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

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

    ```python
    from joblib import parallel_config

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

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

    algorithm = str(self._sampler)

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

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

    self.result = solution
    return self

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

MonteCarloLeastCoreValuation 

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

Bases: LeastCoreValuation

Class to calculate exact least-core values.

Equivalent to calling LeastCoreValuation with a UniformSampler.

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

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

Where:

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

Utility object with model, data and scoring function.

TYPE: UtilityBase

n_samples

The number of samples to use for the valuation. If None, it will be set to 1000 * len(data).

TYPE: int

non_negative_subsidy

If True, the least core subsidy \(e\) is constrained to be non-negative.

TYPE: bool DEFAULT: False

solver_options

Optional dictionary containing a CVXPY solver and options to configure it. For valid values to the "solver" key see here. For additional options see here.

TYPE: dict | None DEFAULT: None

progress

Whether to show a progress bar during the construction of the least-core problem.

TYPE: bool DEFAULT: True

Source code in src/pydvl/valuation/methods/least_core.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def __init__(
    self,
    utility: UtilityBase,
    n_samples: int,
    non_negative_subsidy: bool = False,
    solver_options: dict | None = None,
    progress: bool = True,
    seed: Seed | None = None,
    batch_size: int = 1,
):
    super().__init__(
        utility=utility,
        sampler=UniformSampler(
            index_iteration=NoIndexIteration, seed=seed, batch_size=batch_size
        ),
        n_samples=n_samples,
        non_negative_subsidy=non_negative_subsidy,
        solver_options=solver_options,
        progress=progress,
    )

fit 

fit(data: Dataset) -> Self

Calculate the least core valuation on a dataset.

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

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

from joblib import parallel_config

with parallel_config(n_jobs=4):
    valuation.fit(data)
Source code in src/pydvl/valuation/methods/least_core.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def fit(self, data: Dataset) -> Self:
    """Calculate the least core valuation on a dataset.

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

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

    ```python
    from joblib import parallel_config

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

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

    algorithm = str(self._sampler)

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

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

    self.result = solution
    return self

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

LOOValuation 

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

Bases: SemivalueValuation

Computes LOO values for a dataset.

Source code in src/pydvl/valuation/methods/loo.py 
48
49
50
51
52
53
54
55
56
def __init__(self, utility: UtilityBase, progress: bool = False):
    self.result: ValuationResult | None = None
    super().__init__(
        utility,
        LOOSampler(batch_size=1, index_iteration=FiniteSequentialIndexIteration),
        # LOO is done when every index has been updated once
        MinUpdates(n_updates=1),
        progress=progress,
    )

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

log_coefficient 

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

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

Source code in src/pydvl/valuation/methods/loo.py 
58
59
60
61
62
63
def log_coefficient(self, n: int, k: int) -> float:
    """The LOOSampler returns only complements of {idx}, so the weight is either
    1/n (the probability of a set of size n-1) or 0 if k != n-1. We cancel
    this out here so that the final coefficient is either 1 if k == n-1 or 0
    otherwise."""
    return float(-np.log(max(1, n))) if k == n - 1 else -np.inf

SemivalueValuation 

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

Bases: Valuation

Abstract class to define semi-values.

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

Note

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

PARAMETER DESCRIPTION
utility

Object to compute utilities.

TYPE: UtilityBase

sampler

Sampling scheme to use.

TYPE: IndexSampler

is_done

Stopping criterion to use.

TYPE: StoppingCriterion

skip_converged

Whether to skip converged indices, as determined by the stopping criterion's converged array.

TYPE: bool DEFAULT: False

show_warnings

Whether to show warnings.

TYPE: bool DEFAULT: True

progress

Whether to show a progress bar. If a dictionary, it is passed to tqdm as keyword arguments, and the progress bar is displayed.

TYPE: dict[str, Any] | bool DEFAULT: False

Source code in src/pydvl/valuation/methods/semivalue.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

log_coefficient abstractmethod 

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

The semi-value coefficient in log-space.

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

PARAMETER DESCRIPTION
n

Total number of elements in the set.

TYPE: int

k

Size of the subset for which the coefficient is being computed

TYPE: int

RETURNS DESCRIPTION
float

The natural logarithm of the semi-value coefficient.
Source code in src/pydvl/valuation/methods/semivalue.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@abstractmethod
def log_coefficient(self, n: int, k: int) -> float:
    """The semi-value coefficient in log-space.

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

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

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

ShapleyValuation 

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

Bases: SemivalueValuation

Computes Shapley values.

Source code in src/pydvl/valuation/methods/semivalue.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    utility: UtilityBase,
    sampler: IndexSampler,
    is_done: StoppingCriterion,
    skip_converged: bool = False,
    show_warnings: bool = True,
    progress: dict[str, Any] | bool = False,
):
    super().__init__()
    self.utility = utility
    self.sampler = sampler
    self.is_done = is_done
    self.skip_converged = skip_converged
    self.show_warnings = show_warnings
    self.tqdm_args: dict[str, Any] = {
        "desc": f"{self.__class__.__name__}: {str(is_done)}"
    }
    # HACK: parse additional args for the progress bar if any (we probably want
    #  something better)
    if isinstance(progress, bool):
        self.tqdm_args.update({"disable": not progress})
    elif isinstance(progress, dict):
        self.tqdm_args.update(progress)
    else:
        raise TypeError(f"Invalid type for progress: {type(progress)}")

values 

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

PARAMETER DESCRIPTION
sort

Whether to sort the valuation result by value before returning it.

TYPE: bool DEFAULT: False

Returns: The result of the valuation.

Source code in src/pydvl/valuation/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def values(self, sort: bool = False) -> ValuationResult:
    """Returns a copy of the valuation result.

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

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

    from copy import copy

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

point_wise_accuracy 

point_wise_accuracy(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]

Point-wise accuracy, or 0-1 score between two arrays.

Higher is better.

PARAMETER DESCRIPTION
y_true

Array of true values (e.g. labels)

TYPE: NDArray[T]

y_pred

Array of estimated values (e.g. model predictions)

TYPE: NDArray[T]

RETURNS DESCRIPTION
NDArray[T]

Array with point-wise 0-1 accuracy between labels and model predictions
Source code in src/pydvl/valuation/methods/data_oob.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
def point_wise_accuracy(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]:
    """Point-wise accuracy, or 0-1 score between two arrays.

    Higher is better.

    Args:
        y_true: Array of true values (e.g. labels)
        y_pred: Array of estimated values (e.g. model predictions)

    Returns:
        Array with point-wise 0-1 accuracy between labels and model predictions
    """
    return np.array(y_pred == y_true, dtype=y_pred.dtype)

neg_l2_distance 

neg_l2_distance(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]

Point-wise negative \(l_2\) distance between two arrays.

Higher is better.

PARAMETER DESCRIPTION
y_true

Array of true values (e.g. labels)

TYPE: NDArray[T]

y_pred

Array of estimated values (e.g. model predictions)

TYPE: NDArray[T]

RETURNS DESCRIPTION
NDArray[T]

Array with point-wise negative \(l_2\) distances between labels and model
NDArray[T]

predictions
Source code in src/pydvl/valuation/methods/data_oob.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def neg_l2_distance(y_true: NDArray[T], y_pred: NDArray[T]) -> NDArray[T]:
    r"""Point-wise negative $l_2$ distance between two arrays.

    Higher is better.

    Args:
        y_true: Array of true values (e.g. labels)
        y_pred: Array of estimated values (e.g. model predictions)

    Returns:
        Array with point-wise negative $l_2$ distances between labels and model
        predictions
    """
    return -np.square(np.array(y_pred - y_true), dtype=y_pred.dtype)

compute_n_samples 

compute_n_samples(epsilon: float, delta: float, n_obs: int) -> int

Compute the minimal sample size with epsilon-delta guarantees.

Based on the formula in Theorem 4 of (Jia, R. et al., 2023)2 which gives a lower bound on the number of samples required to obtain an (ε/√n,δ/(N(N-1))-approximation to all pair-wise differences of Shapley values, wrt. \(\ell_2\) norm.

The updated version refines the lower bound of the original paper. Note that the bound is tighter than earlier versions but might still overestimate the number of samples required.

PARAMETER DESCRIPTION
epsilon

The error tolerance.

TYPE: float

delta

The confidence level.

TYPE: float

n_obs

Number of data points.

TYPE: int

RETURNS DESCRIPTION
int

The sample size.
Source code in src/pydvl/valuation/methods/gt_shapley.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def compute_n_samples(epsilon: float, delta: float, n_obs: int) -> int:
    r"""Compute the minimal sample size with epsilon-delta guarantees.

    Based on the formula in Theorem 4 of
    (Jia, R. et al., 2023)<sup><a href="#jia_update_2023">2</a></sup>
    which gives a lower bound on the number of samples required to obtain an
    (ε/√n,δ/(N(N-1))-approximation to all pair-wise differences of Shapley
    values, wrt. $\ell_2$ norm.

    The updated version refines the lower bound of the original paper. Note that the
    bound is tighter than earlier versions but might still overestimate the number of
    samples required.

    Args:
        epsilon: The error tolerance.
        delta: The confidence level.
        n_obs: Number of data points.

    Returns:
        The sample size.

    """
    kk = _create_sample_sizes(n_obs)
    Z = _calculate_z(n_obs)

    q = _create_sampling_probabilities(kk)
    q_tot = (n_obs - 2) / n_obs * q[0] + np.inner(
        q[1:], 1 + 2 * kk[1:] * (kk[1:] - n_obs) / (n_obs * (n_obs - 1))
    )

    def _h(u: float) -> float:
        return cast(float, (1 + u) * np.log(1 + u) - u)

    n_samples = np.log(n_obs * (n_obs - 1) / delta)
    n_samples /= 1 - q_tot
    n_samples /= _h(epsilon / (2 * Z * np.sqrt(n_obs) * (1 - q_tot)))

    return int(n_samples)