K-Nearest Neighbours

You move to a new city and want to know if a neighbourhood is safe. You do not build a statistical model of crime rates. You ask the 5 people who live closest to that neighbourhood what they think. If 4 out of 5 say it is safe, you conclude it is safe. That is KNN — ask your nearest neighbours and take a vote.

The key questions are: how do you measure "closeness"? How many neighbours k should you ask? And what happens when the neighbourhood is crowded in some dimensions but empty in others — the curse of dimensionality. This module answers all three.

import numpy as np

# ── KNN from scratch — every step visible ─────────────────────────────
class KNNScratch:
    """
    K-Nearest Neighbours from scratch.
    Training = just storing the data.
    Prediction = find k nearest, aggregate their labels.
    """

    def __init__(self, k: int = 5, task: str = 'regression'):
        self.k      = k
        self.task   = task   # 'regression' or 'classification'
        self.X_train = None
        self.y_train = None

    def fit(self, X: np.ndarray, y: np.ndarray):
        """
        'Training' KNN = just storing the data.
        Nothing is computed. Nothing is learned.
        """
        self.X_train = X.copy()
        self.y_train = y.copy()
        return self

    def _euclidean_distance(self, a: np.ndarray, b: np.ndarray) -> float:
        """Distance between two points: sqrt(sum of squared differences)."""
        return np.sqrt(np.sum((a - b) ** 2))

    def _predict_one(self, x: np.ndarray):
        """Predict for a single query point."""
        # Step 1: compute distance to every training point
        distances = np.array([
            self._euclidean_distance(x, x_train)
            for x_train in self.X_train
        ])

        # Step 2: find indices of k nearest neighbours
        k_nearest_indices = np.argsort(distances)[:self.k]

        # Step 3: get their labels
        k_nearest_labels = self.y_train[k_nearest_indices]

        # Step 4: aggregate
        if self.task == 'regression':
            return k_nearest_labels.mean()
        else:
            # majority vote
            values, counts = np.unique(k_nearest_labels, return_counts=True)
            return values[np.argmax(counts)]

    def predict(self, X: np.ndarray) -> np.ndarray:
        return np.array([self._predict_one(x) for x in X])


# ── Test on Swiggy delivery data ───────────────────────────────────────
np.random.seed(42)
n = 500
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep      = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
delivery  = (8.6 + 7.3*distance + 0.8*prep + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)

X = np.column_stack([distance, traffic, prep])
y = delivery

# Scale — CRITICAL for KNN
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
sc = StandardScaler()
X_train_sc = sc.fit_transform(X_train)
X_test_sc  = sc.transform(X_test)

# From-scratch KNN
knn_scratch = KNNScratch(k=5, task='regression')
knn_scratch.fit(X_train_sc, y_train)
y_pred_scratch = knn_scratch.predict(X_test_sc[:10])  # only 10 for speed

# Verify with sklearn
from sklearn.neighbors import KNeighborsRegressor
from sklearn.metrics import mean_absolute_error
knn_sk = KNeighborsRegressor(n_neighbors=5)
knn_sk.fit(X_train_sc, y_train)
y_pred_sk = knn_sk.predict(X_test_sc[:10])

print("From-scratch vs sklearn predictions (first 10):")
for s, sk, true in zip(y_pred_scratch, y_pred_sk, y_test[:10]):
    print(f"  scratch={s:.1f}  sklearn={sk:.1f}  true={true:.1f}")

# Full test MAE
y_pred_full = knn_sk.predict(X_test_sc)
print(f"
Full test MAE (k=5): {mean_absolute_error(y_test, y_pred_full):.4f} min")

import numpy as np
from sklearn.neighbors import KNeighborsClassifier, KNeighborsRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import mean_absolute_error

np.random.seed(42)
n = 2000
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep      = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
delivery  = (8.6 + 7.3*distance + 0.8*prep + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)

X = np.column_stack([distance, traffic, prep])
y = delivery

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)
sc = StandardScaler()
X_tr_sc = sc.fit_transform(X_tr)
X_te_sc = sc.transform(X_te)

# ── Compare distance metrics ───────────────────────────────────────────
print(f"{'Metric':<20} {'p':<5} {'Test MAE'}")
print("─" * 35)
for metric, p in [
    ('euclidean (L2)', 2),
    ('manhattan (L1)', 1),
    ('minkowski p=3',  3),
    ('minkowski p=4',  4),
]:
    knn = KNeighborsRegressor(n_neighbors=5, metric='minkowski', p=p)
    knn.fit(X_tr_sc, y_tr)
    mae = mean_absolute_error(y_te, knn.predict(X_te_sc))
    print(f"  {metric:<18} {p:<5} {mae:.4f}")

# ── Effect of not scaling — the most common mistake ───────────────────
print("
Effect of not scaling on distance-based algorithms:")
knn_unscaled = KNeighborsRegressor(n_neighbors=5)
knn_unscaled.fit(X_tr, y_tr)   # no scaling
mae_unscaled = mean_absolute_error(y_te, knn_unscaled.predict(X_te))

knn_scaled = KNeighborsRegressor(n_neighbors=5)
knn_scaled.fit(X_tr_sc, y_tr)
mae_scaled = mean_absolute_error(y_te, knn_scaled.predict(X_te_sc))

print(f"  Without scaling: MAE = {mae_unscaled:.4f}  ← dominated by large-scale features")
print(f"  With scaling:    MAE = {mae_scaled:.4f}  ← all features contribute equally")

# ── Weighted KNN — closer neighbours vote more ────────────────────────
# weights='uniform': all k neighbours vote equally
# weights='distance': closer neighbours have more influence (1/distance weighting)
print("
Uniform vs distance-weighted KNN:")
for weights in ['uniform', 'distance']:
    knn = KNeighborsRegressor(n_neighbors=5, weights=weights)
    knn.fit(X_tr_sc, y_tr)
    mae = mean_absolute_error(y_te, knn.predict(X_te_sc))
    print(f"  weights='{weights}': MAE = {mae:.4f}")

import numpy as np
from sklearn.neighbors import KNeighborsRegressor, KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score, train_test_split
from sklearn.metrics import mean_absolute_error
import warnings
warnings.filterwarnings('ignore')

np.random.seed(42)
n = 2000
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep      = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
delivery  = (8.6 + 7.3*distance + 0.8*prep + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)

X = np.column_stack([distance, traffic, prep])
y = delivery

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, random_state=42)
sc = StandardScaler()
X_tr_sc = sc.fit_transform(X_tr)
X_te_sc = sc.transform(X_te)

# ── k sweep — find optimal k ──────────────────────────────────────────
print(f"{'k':<6} {'Train MAE':<12} {'CV MAE (5-fold)':<18} {'Test MAE'}")
print("─" * 54)

best_k, best_cv = None, float('inf')
for k in [1, 3, 5, 7, 10, 15, 20, 30, 50, 100]:
    knn = KNeighborsRegressor(n_neighbors=k, weights='distance')
    knn.fit(X_tr_sc, y_tr)

    train_mae = mean_absolute_error(y_tr, knn.predict(X_tr_sc))
    cv_scores = cross_val_score(knn, X_tr_sc, y_tr,
                                 cv=5, scoring='neg_mean_absolute_error')
    cv_mae    = -cv_scores.mean()
    test_mae  = mean_absolute_error(y_te, knn.predict(X_te_sc))

    flag = ' ← best CV' if cv_mae < best_cv else ''
    if cv_mae < best_cv:
        best_cv, best_k = cv_mae, k

    overfit = ' ← overfit' if train_mae < cv_mae * 0.7 else ''
    print(f"  {k:<4}  {train_mae:<12.4f} {cv_mae:<16.4f}  {test_mae:.4f}{flag}{overfit}")

print(f"
Best k by CV: {best_k}  (CV MAE = {best_cv:.4f})")

# ── Rule of thumb: k = sqrt(n_train) as starting point ────────────────
k_rule_of_thumb = int(np.sqrt(len(X_tr)))
print(f"
Rule of thumb: k = sqrt(n_train) = sqrt({len(X_tr)}) = {k_rule_of_thumb}")
print("Always use an odd k for binary classification to avoid ties.")
print("Tune k with cross-validation — do not rely on the rule of thumb alone.")

Imagine searching for the nearest person to you on a street (1D). Easy — look left and right. Now on a field (2D). Harder, but doable. Now in a building (3D). Now in a 100-dimensional space where each dimension is one feature of a customer profile.

In 100 dimensions, the "nearest" customer might actually be almost as far away as the furthest customer. The ratio of nearest-to-furthest distance approaches 1 as dimensions grow. All distances become equally large and equally meaningless.

In d dimensions, to maintain the same neighbourhood density you need exponentially more data. To have 10 neighbours within a distance of 0.1 in 1D, you might need ~100 points. In 10D, you need ~10 billion points for the same density. In practice, your data becomes infinitely sparse.

import numpy as np
from sklearn.neighbors import KNeighborsRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score
import warnings
warnings.filterwarnings('ignore')

np.random.seed(42)

# ── Demonstrate curse of dimensionality ───────────────────────────────
# Add progressively more noise dimensions — watch KNN performance degrade

n = 1000
# Core signal: 3 features with real predictive power
X_core = np.column_stack([
    np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15),
    np.random.randint(1, 11, n).astype(float),
    np.abs(np.random.normal(15, 5, n)).clip(5, 35),
])
y = (8.6 + 7.3*X_core[:,0] + 0.8*X_core[:,2] + 1.5*X_core[:,1]
     + np.random.normal(0, 4, n)).clip(10, 120)

print("KNN performance as dimensions increase (3 real + noise):")
print(f"{'Dimensions':<14} {'CV MAE':>10} {'Distance ratio':>16}")
print("─" * 44)

for n_noise in [0, 2, 5, 10, 20, 50, 100]:
    X_noise = np.random.randn(n, n_noise) if n_noise > 0 else np.empty((n, 0))
    X_full  = np.hstack([X_core, X_noise])
    total_d = X_full.shape[1]

    sc   = StandardScaler()
    X_sc = sc.fit_transform(X_full)

    # KNN CV MAE
    knn    = KNeighborsRegressor(n_neighbors=10)
    scores = cross_val_score(knn, X_sc, y, cv=5, scoring='neg_mean_absolute_error')
    cv_mae = -scores.mean()

    # Distance ratio: nearest / farthest — approaches 1 as dims increase
    from sklearn.metrics.pairwise import euclidean_distances
    sample_dists = euclidean_distances(X_sc[:100])
    np.fill_diagonal(sample_dists, np.inf)
    nearest  = sample_dists.min(axis=1).mean()
    np.fill_diagonal(sample_dists, 0)
    farthest = sample_dists.max(axis=1).mean()
    ratio    = nearest / farthest

    print(f"  {total_d:<12}d  {cv_mae:>10.4f} {ratio:>14.4f}")

print("
As dimensions grow: MAE degrades, distance ratio → 1 (all equidistant)")

# ── Fix: reduce dimensions before KNN ─────────────────────────────────
from sklearn.decomposition import PCA
from sklearn.pipeline import Pipeline

X_high = np.hstack([X_core, np.random.randn(n, 47)])  # 50 total dimensions
sc = StandardScaler()
X_high_sc = sc.fit_transform(X_high)

print("
PCA dimensionality reduction before KNN:")
for n_components in [50, 20, 10, 5, 3]:
    pca  = PCA(n_components=n_components)
    X_pca = pca.fit_transform(X_high_sc)
    knn   = KNeighborsRegressor(n_neighbors=10)
    score = cross_val_score(knn, X_pca, y, cv=5,
                             scoring='neg_mean_absolute_error')
    print(f"  PCA to {n_components:2d} dims: CV MAE = {-score.mean():.4f}  "
          f"({pca.explained_variance_ratio_.sum()*100:.0f}% variance retained)")

import numpy as np
from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.metrics import classification_report, roc_auc_score
from sklearn.pipeline import Pipeline
import warnings
warnings.filterwarnings('ignore')

np.random.seed(42)
n = 3000

# Late delivery classification: is_late = delivery > 45 min
distance  = np.abs(np.random.normal(4.0, 2.0, n)).clip(0.5, 15)
traffic   = np.random.randint(1, 11, n).astype(float)
prep      = np.abs(np.random.normal(15, 5, n)).clip(5, 35)
value     = np.abs(np.random.normal(350, 150, n)).clip(50, 1200)
delivery  = (8.6 + 7.3*distance + 0.8*prep + 1.5*traffic
             + np.random.normal(0, 4, n)).clip(10, 120)
y_cls     = (delivery > 45).astype(int)

X = np.column_stack([distance, traffic, prep, value])
X_tr, X_te, y_tr, y_te = train_test_split(
    X, y_cls, test_size=0.2, stratify=y_cls, random_state=42
)

# ── Pipeline: scale → KNN ─────────────────────────────────────────────
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('knn',    KNeighborsClassifier(n_neighbors=11, weights='distance')),
])
pipe.fit(X_tr, y_tr)

y_pred  = pipe.predict(X_te)
y_proba = pipe.predict_proba(X_te)[:, 1]

print(f"KNN Classifier (k=11):")
print(f"  Accuracy: {pipe.score(X_te, y_te):.4f}")
print(f"  ROC-AUC:  {roc_auc_score(y_te, y_proba):.4f}")
print(classification_report(y_te, y_pred, target_names=['On-time', 'Late']))

# ── Probability = fraction of neighbours in each class ────────────────
# Inspect probabilities for a few examples
print("Sample predictions with neighbourhood breakdown:")
knn_fitted = pipe.named_steps['knn']
sc_fitted  = pipe.named_steps['scaler']
X_te_sc    = sc_fitted.transform(X_te)

# Get the actual k nearest neighbours for first 3 test points
distances_arr, indices = knn_fitted.kneighbors(X_te_sc[:3])
for i in range(3):
    neighbour_labels = y_tr[indices[i]]
    late_count  = neighbour_labels.sum()
    total       = len(neighbour_labels)
    p_late      = y_proba[i]
    actual      = 'LATE' if y_te[i] == 1 else 'ON-TIME'
    predicted   = 'LATE' if y_pred[i] == 1 else 'ON-TIME'
    print(f"  Sample {i+1}: {late_count}/{total} neighbours are late "
          f"→ P(late)={p_late:.3f}  predicted={predicted}  actual={actual}")

# ── GridSearch over k and weights ─────────────────────────────────────
param_grid = {
    'knn__n_neighbors': [3, 5, 7, 11, 15, 21, 31],
    'knn__weights':     ['uniform', 'distance'],
    'knn__metric':      ['euclidean', 'manhattan'],
}
grid = GridSearchCV(pipe, param_grid, cv=5, scoring='roc_auc', n_jobs=-1)
grid.fit(X_tr, y_tr)
print(f"
Best params: {grid.best_params_}")
print(f"Best CV AUC: {grid.best_score_:.4f}")

import numpy as np
from sklearn.neighbors import KNeighborsRegressor, BallTree, KDTree
from sklearn.preprocessing import StandardScaler
import time

np.random.seed(42)
n_train = 10_000
n_query = 1000

X_train = np.random.randn(n_train, 5)
y_train = np.random.randn(n_train)
X_query = np.random.randn(n_query, 5)

sc = StandardScaler()
X_tr_sc = sc.fit_transform(X_train)
X_qu_sc = sc.transform(X_query)

# ── algorithm parameter controls the search structure ─────────────────
# 'brute':    compute all distances — O(n) per query, exact, slow for large n
# 'ball_tree': partition space into nested hyperspheres — faster for low-dim
# 'kd_tree':  partition space into axis-aligned boxes — fastest for very low-dim
# 'auto':     sklearn picks the best based on data characteristics

print("KNN algorithm comparison (10,000 training points, 1,000 queries):")
print(f"{'Algorithm':<15} {'Fit time':>12} {'Query time':>12} {'Speed vs brute':>16}")
print("─" * 58)

brute_time = None
for algo in ['brute', 'kd_tree', 'ball_tree', 'auto']:
    knn = KNeighborsRegressor(n_neighbors=10, algorithm=algo)

    t0 = time.time()
    knn.fit(X_tr_sc, y_train)
    fit_time = time.time() - t0

    t0 = time.time()
    knn.predict(X_qu_sc)
    query_time = time.time() - t0

    if algo == 'brute':
        brute_time = query_time
    speedup = brute_time / query_time if brute_time else 1.0
    print(f"  {algo:<13}  {fit_time:>10.4f}s  {query_time:>10.4f}s  {speedup:>13.1f}×")

# ── leaf_size: BallTree/KDTree trade-off ─────────────────────────────
# Smaller leaf_size → deeper tree → faster query but more memory
# Larger leaf_size → shallower tree → less memory but more brute-force at leaves
print("
leaf_size effect on BallTree query time:")
for leaf in [5, 10, 30, 50, 100]:
    knn = KNeighborsRegressor(n_neighbors=10, algorithm='ball_tree', leaf_size=leaf)
    knn.fit(X_tr_sc, y_train)
    t0 = time.time()
    knn.predict(X_qu_sc)
    qt = time.time() - t0
    print(f"  leaf_size={leaf:<5}: {qt:.4f}s")