"""
Weighted K-Nearest Neighbors
=============================
From: Alpaydın, E. (2020). Introduction to Machine Learning (4th ed.), Chapter 8.
Extension of standard KNN with distance-weighted voting:
y_hat = argmax_c Σ_{i ∈ kNN(x)} w_i · I(y_i = c)
where w_i = 1 / d(x, x_i)^p (inverse distance weighting)
Also implements "Condensed Nearest Neighbor" (Alpaydın, 1997):
Voting over Multiple Condensed Nearest Neighbors.
Artificial Intelligence Review, 11, 115-132.
"""
from typing import Optional
import numpy as np
from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.preprocessing import LabelEncoder
from sklearn.utils.multiclass import check_classification_targets
from sklearn.utils.validation import check_array, check_is_fitted, check_X_y
from neural_trees._validation import check_predict_input
[docs]
class WeightedKNN(ClassifierMixin, BaseEstimator):
"""
Distance-Weighted K-Nearest Neighbors Classifier.
Parameters
----------
k : int, default=5
Number of neighbors.
weight_power : float, default=2.0
Power for inverse-distance weighting. Set to 0 for uniform weights.
metric : str, default="euclidean"
Distance metric: "euclidean" or "manhattan".
condense : bool, default=False
If True, apply condensing: keep only a subset of training samples that
correctly classifies all the others (Hart's CNN).
n_condensed_sets : int, default=1
How many condensed subsets to build and vote over when `condense=True`.
Condensing is order dependent: which samples end up as prototypes
depends on the order they were visited in, and a single pass throws
away information that a different order would have kept. Alpaydin
(1997) builds several subsets from different orderings and combines
their votes, which is where some of the accuracy a single subset gives
away comes back. 5-fold accuracy averaged over 5 seeds, by number of
subsets, against keeping every sample::
1 3 5 9 all
Iris 0.917 0.939 0.937 0.937 0.956
Wine 0.947 0.955 0.964 0.971 0.966
Breast Canc. 0.951 0.963 0.966 0.968 0.966
Voting beats a single subset everywhere. It reaches the uncondensed
classifier on Wine and Breast Cancer while storing roughly a sixth of
the data, and closes about half the gap on Iris without closing it.
Ignored when `condense=False`.
random_state : int or None, default=None
Seed for the orderings used to build the condensed subsets.
References
----------
Alpaydın, E. (1997). Voting over Multiple Condensed Nearest Neighbors.
Artificial Intelligence Review, 11, 115-132.
"""
def __init__(
self,
k: int = 5,
weight_power: float = 2.0,
metric: str = "euclidean",
condense: bool = False,
n_condensed_sets: int = 1,
random_state: Optional[int] = None,
):
self.k = k
self.weight_power = weight_power
self.metric = metric
self.condense = condense
self.n_condensed_sets = n_condensed_sets
self.random_state = random_state
def _distance(self, x1: np.ndarray, x2: np.ndarray) -> np.ndarray:
if self.metric == "euclidean":
return np.sqrt(((x1[:, None] - x2[None, :]) ** 2).sum(axis=-1))
elif self.metric == "manhattan":
return np.abs(x1[:, None] - x2[None, :]).sum(axis=-1)
raise ValueError(f"Unknown metric: {self.metric}")
def _condense(self, X: np.ndarray, y: np.ndarray):
"""
Condensed Nearest Neighbor (Hart, 1968).
Sweep the training set repeatedly, moving into the store any sample the
current store misclassifies under the 1-nearest-neighbor rule, until a
full sweep adds nothing. The result is *consistent*: the store
classifies every training sample correctly.
A single sweep, which is what this used to do, stops before that
property holds, because samples seen early are judged against a store
that later grows.
"""
store_idx = [0]
remaining = list(range(1, len(X)))
changed = True
while changed:
changed = False
still_remaining = []
for i in remaining:
dists = self._distance(X[i][None, :], X[store_idx])[0]
if y[store_idx[int(np.argmin(dists))]] != y[i]:
store_idx.append(i)
changed = True
else:
still_remaining.append(i)
remaining = still_remaining
return X[store_idx], y[store_idx]
def fit(self, X, y) -> "WeightedKNN":
if self.metric not in ("euclidean", "manhattan"):
raise ValueError(
f"metric must be 'euclidean' or 'manhattan', got {self.metric!r}"
)
if isinstance(self.k, bool) or not isinstance(self.k, int) or self.k < 1:
raise ValueError(f"k must be a positive integer, got {self.k!r}")
X, y = check_X_y(X, y)
check_classification_targets(y)
self.le_ = LabelEncoder()
y_enc = self.le_.fit_transform(y)
self.classes_ = self.le_.classes_
self.n_features_in_ = X.shape[1]
if isinstance(self.n_condensed_sets, bool) or not isinstance(
self.n_condensed_sets, int
) or self.n_condensed_sets < 1:
raise ValueError(
"n_condensed_sets must be a positive integer, got "
f"{self.n_condensed_sets!r}"
)
if self.condense:
rng = np.random.RandomState(self.random_state)
self.stores_ = []
for i in range(self.n_condensed_sets):
# The first subset uses the data as given, so a single set
# reproduces the previous behaviour exactly.
order = np.arange(len(X)) if i == 0 else rng.permutation(len(X))
store_X, store_y = self._condense(X[order], y_enc[order])
self.stores_.append((store_X, store_y))
else:
self.stores_ = [(X, y_enc)]
# The first store is the one an unvoted classifier would use, and the
# attribute is part of the public surface.
self.X_train_, self.y_train_ = self.stores_[0]
return self
[docs]
def predict_proba(self, X) -> np.ndarray:
"""
Class probabilities, averaged over the condensed subsets.
Each subset votes with its own distance-weighted neighbours, and the
votes are averaged. With `n_condensed_sets=1` this is a plain weighted
KNN over a single store.
"""
check_is_fitted(self)
X = check_predict_input(self, X)
votes = [self._vote(X, store_X, store_y) for store_X, store_y in self.stores_]
return np.mean(votes, axis=0)
def _vote(self, X: np.ndarray, store_X: np.ndarray, store_y: np.ndarray) -> np.ndarray:
dists = self._distance(X, store_X) # (n_test, n_store)
k = min(self.k, len(store_X))
n_classes = len(self.classes_)
probs = np.zeros((len(X), n_classes))
for i, row in enumerate(dists):
nn_idx = np.argsort(row)[:k]
nn_dists = row[nn_idx]
if self.weight_power == 0:
weights = np.ones(k)
elif nn_dists[0] == 0:
# Exact matches carry the whole vote. Falling back to uniform
# weights here, as this used to, let k - 1 unrelated neighbors
# outvote a sample identical to the query.
weights = (nn_dists == 0).astype(float)
else:
weights = 1.0 / (nn_dists ** self.weight_power + 1e-10)
for j, idx in enumerate(nn_idx):
probs[i, store_y[idx]] += weights[j]
probs /= probs.sum(axis=1, keepdims=True)
return probs
def predict(self, X) -> np.ndarray:
check_is_fitted(self)
return self.le_.inverse_transform(self.predict_proba(X).argmax(axis=1))