Weighted KNN

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import numpy as np
 
def inverse_distance_weights(distances: np.ndarray, 
                               power: float = 2.0,
                               epsilon: float = 1e-10) -> np.ndarray:
    """
    Compute inverse distance weights with numerical stability.
    
    Parameters:
    -----------
    distances : array of distances to k neighbors
    power : exponent for distance (higher = more weight to close neighbors)
    epsilon : small constant to prevent division by zero
    
    Returns:
    --------
    weights : normalized weights summing to 1
    """
    # Handle exact matches (distance = 0)
    exact_match = distances < epsilon
    
    if np.any(exact_match):
        # If any distance is essentially zero, that neighbor gets all weight
        weights = exact_match.astype(float)
    else:
        # Standard inverse distance weighting
        weights = 1.0 / (distances ** power)
    
    # Normalize to sum to 1
    return weights / weights.sum()
 
 
def shepard_weights(distances: np.ndarray,
                     power: float = 2.0) -> np.ndarray:
    """
    Shepard's interpolation weights - alternative stable formulation.
    
    Uses the formula: w_i = (d_max - d_i) / d_i
    This naturally handles the boundary case more gracefully.
    """
    d_max = distances.max()
    
    # Avoid division by zero
    safe_distances = np.maximum(distances, 1e-10)
    
    weights = ((d_max - distances) / safe_distances) ** power
    
    return weights / weights.sum()
 
 
# Demonstration
distances = np.array([0.1, 0.5, 0.8, 1.2, 1.5])
print("Distances:", distances)
print("IDW weights (p=2):", inverse_distance_weights(distances, power=2))
print("Shepard weights:", shepard_weights(distances, power=2))
 
# Edge case: exact match
distances_with_match = np.array([0.0, 0.5, 0.8, 1.2, 1.5])
print("
With exact match (d=0):")
print("IDW weights:", inverse_distance_weights(distances_with_match, power=2))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import numpy as np
 
def effective_sample_size(weights: np.ndarray) -> float:
    """
    Compute effective sample size for weighted estimate.
    
    n_eff = 1 / sum(w_i^2) where weights are normalized to sum to 1.
    
    For uniform weights: n_eff = k
    For concentrated weights: n_eff < k
    Extreme case (single weight = 1): n_eff = 1
    """
    # Ensure normalization
    normalized_weights = weights / weights.sum()
    return 1.0 / np.sum(normalized_weights ** 2)
 
 
# Compare effective sample sizes for different weight schemes
k = 10
uniform_weights = np.ones(k) / k
 
# Inverse distance with distances increasing linearly
distances = np.linspace(0.1, 1.0, k)
inv_weights = 1.0 / distances ** 2
inv_weights = inv_weights / inv_weights.sum()
 
# Very concentrated (exponential decay)
exp_weights = np.exp(-3 * distances)
exp_weights = exp_weights / exp_weights.sum()
 
print(f"k = {k} neighbors")
print(f"Uniform weights:  n_eff = {effective_sample_size(uniform_weights):.2f}")
print(f"Inverse-square:   n_eff = {effective_sample_size(inv_weights):.2f}")
print(f"Strong exponential: n_eff = {effective_sample_size(exp_weights):.2f}")
 
# Variance multiplier relative to uniform
print(f"
Variance multiplier (vs uniform k={k}):")
print(f"Inverse-square:   {k * np.sum(inv_weights**2):.2f}x")
print(f"Strong exponential: {k * np.sum(exp_weights**2):.2f}x")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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
126
127
128
129
130
131
132
133
134

import numpy as np
from collections import defaultdict
from typing import Tuple, Dict, Any
 
class WeightedKNNClassifier:
    """
    K-Nearest Neighbors classifier with distance weighting.
    
    Implements multiple weighting schemes and provides both
    hard predictions and probability estimates.
    """
    
    def __init__(self, 
                 k: int = 5, 
                 weight_type: str = 'inverse_squared',
                 power: float = 2.0):
        """
        Parameters
        ----------
        k : number of neighbors
        weight_type : 'uniform', 'inverse', 'inverse_squared', 'gaussian'
        power : power for inverse weighting (ignored for gaussian)
        """
        self.k = k
        self.weight_type = weight_type
        self.power = power
        self.X_train = None
        self.y_train = None
        self.classes_ = None
    
    def fit(self, X: np.ndarray, y: np.ndarray):
        """Store training data and extract classes."""
        self.X_train = np.array(X)
        self.y_train = np.array(y)
        self.classes_ = np.unique(y)
        return self
    
    def _compute_weights(self, distances: np.ndarray) -> np.ndarray:
        """Compute weights from distances based on weight_type."""
        epsilon = 1e-10
        
        if self.weight_type == 'uniform':
            return np.ones_like(distances)
        
        elif self.weight_type == 'inverse':
            # Handle zero distances
            safe_distances = np.maximum(distances, epsilon)
            return 1.0 / safe_distances
        
        elif self.weight_type == 'inverse_squared':
            safe_distances = np.maximum(distances, epsilon)
            return 1.0 / (safe_distances ** self.power)
        
        elif self.weight_type == 'gaussian':
            # Bandwidth = median distance (adaptive)
            sigma = np.median(distances) + epsilon
            return np.exp(-distances**2 / (2 * sigma**2))
        
        else:
            raise ValueError(f"Unknown weight type: {self.weight_type}")
    
    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        """
        Predict class probabilities for each sample in X.
        
        Returns
        -------
        proba : shape (n_samples, n_classes)
            Probability of each class for each sample
        """
        X = np.atleast_2d(X)
        n_samples = X.shape[0]
        n_classes = len(self.classes_)
        probas = np.zeros((n_samples, n_classes))
        
        for i, x_query in enumerate(X):
            # Compute distances to all training points
            distances = np.linalg.norm(self.X_train - x_query, axis=1)
            
            # Find k nearest neighbors
            k_indices = np.argsort(distances)[:self.k]
            k_distances = distances[k_indices]
            k_labels = self.y_train[k_indices]
            
            # Compute weights
            weights = self._compute_weights(k_distances)
            
            # Accumulate weighted votes per class
            class_scores = np.zeros(n_classes)
            for d, label, weight in zip(k_distances, k_labels, weights):
                class_idx = np.where(self.classes_ == label)[0][0]
                class_scores[class_idx] += weight
            
            # Normalize to probabilities
            probas[i] = class_scores / class_scores.sum()
        
        return probas
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Predict class labels for samples in X."""
        probas = self.predict_proba(X)
        return self.classes_[np.argmax(probas, axis=1)]
 
 
# Demonstration: Uniform vs Weighted KNN near decision boundary
np.random.seed(42)
 
# Create training data: two classes with overlap
X_train = np.vstack([
    np.random.randn(50, 2) + [0, 0],   # Class 0
    np.random.randn(50, 2) + [2, 2],   # Class 1
])
y_train = np.array([0] * 50 + [1] * 50)
 
# Query point near decision boundary
X_query = np.array([[1.0, 1.0]])
 
# Compare uniform and weighted predictions
uniform_knn = WeightedKNNClassifier(k=7, weight_type='uniform')
weighted_knn = WeightedKNNClassifier(k=7, weight_type='inverse_squared')
 
uniform_knn.fit(X_train, y_train)
weighted_knn.fit(X_train, y_train)
 
print("Query point:", X_query[0])
print("
Uniform KNN (k=7):")
print(f"  Probabilities: {uniform_knn.predict_proba(X_query)[0]}")
print(f"  Prediction: {uniform_knn.predict(X_query)[0]}")
 
print("
Weighted KNN (k=7, inverse-square):")
print(f"  Probabilities: {weighted_knn.predict_proba(X_query)[0]}")
print(f"  Prediction: {weighted_knn.predict(X_query)[0]}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

import numpy as np
 
def robust_knn_prediction(X_train: np.ndarray, 
                           y_train: np.ndarray,
                           x_query: np.ndarray,
                           k: int,
                           weight_power: float = 2.0) -> tuple:
    """
    Weighted KNN with robust tie-breaking.
    
    Returns (prediction, probabilities, debugging_info)
    """
    distances = np.linalg.norm(X_train - x_query, axis=1)
    
    # Handle distance ties at k-boundary
    sorted_indices = np.argsort(distances)
    kth_distance = distances[sorted_indices[k-1]]
    
    # Include all points at or closer than kth distance
    effective_indices = np.where(distances <= kth_distance + 1e-10)[0]
    effective_k = len(effective_indices)
    
    # Compute weights
    effective_distances = distances[effective_indices]
    safe_distances = np.maximum(effective_distances, 1e-10)
    weights = 1.0 / safe_distances ** weight_power
    normalized_weights = weights / weights.sum()
    
    # Aggregate by class
    effective_labels = y_train[effective_indices]
    classes = np.unique(y_train)
    class_scores = {}
    class_counts = {}
    
    for c in classes:
        mask = effective_labels == c
        class_scores[c] = normalized_weights[mask].sum()
        class_counts[c] = mask.sum()
    
    # Find max score
    max_score = max(class_scores.values())
    top_classes = [c for c, s in class_scores.items() 
                   if np.isclose(s, max_score, rtol=1e-9)]
    
    # Tie-breaking
    if len(top_classes) == 1:
        prediction = top_classes[0]
        tie_broken = False
    else:
        # Tie-break 1: Choose class of closest neighbor
        closest_idx = sorted_indices[0]
        closest_class = y_train[closest_idx]
        if closest_class in top_classes:
            prediction = closest_class
            tie_broken = True
        else:
            # Tie-break 2: Higher count
            top_counts = {c: class_counts[c] for c in top_classes}
            prediction = max(top_counts, key=top_counts.get)
            tie_broken = True
    
    # Probability vector
    total_score = sum(class_scores.values())
    probas = {c: s/total_score for c, s in class_scores.items()}
    
    debug_info = {
        'effective_k': effective_k,
        'requested_k': k,
        'tie_broken': tie_broken,
        'top_classes_before_tiebreak': top_classes,
        'class_scores': class_scores,
    }
    
    return prediction, probas, debug_info
 
 
# Test with potential tie scenario
X_train = np.array([
    [0, 0], [1, 0], [0, 1], [1, 1],  # Corners of unit square
])
y_train = np.array([0, 1, 1, 0])  # Diagonal classes
 
x_query = np.array([0.5, 0.5])  # Center - equidistant to all
 
pred, probs, debug = robust_knn_prediction(X_train, y_train, x_query, k=4)
print("Query at center of square (equidistant to all corners)")
print(f"Prediction: {pred}")
print(f"Probabilities: {probs}")
print(f"Debug info: {debug}")