pydvl.valuation.methods.knn_shapley

This module contains Shapley computations for K-Nearest Neighbours.

Todo

Implement approximate KNN computation for sublinear complexity

References

---

1. Jia, R. et al., 2019. Efficient Task-Specific Data Valuation for Nearest Neighbor Algorithms. In: Proceedings of the VLDB Endowment, Vol. 12, No. 11, pp. 1610–1623. ↩

KNNShapleyValuation

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

Bases: Valuation

Computes exact Shapley values for a KNN classifier.

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

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

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

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

values

values(sort: bool = False) -> ValuationResult

Returns a copy of the valuation result.

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

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

Returns: The result of the valuation.

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

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

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

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

    from copy import copy

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

fit

fit(data: Dataset) -> Self

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

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

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

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

from joblib import parallel_config

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

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

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

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

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

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

    ```python
    from joblib import parallel_config

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

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

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

    n_test = len(self.test_data)

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

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

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

    return self