KNN for Recommendation Systems

The Story That Explains KNN Recommenders

KNN is one of the oldest and most intuitive algorithms in machine learning, yet it remains a competitive baseline for recommendation systems to this day. Its beauty is its transparency: every recommendation can be explained by pointing to real people (or real items) that are measurably close to the query. No black boxes, no latent factors, no matrices to factorise. Just distance, neighbours, and votes.

KNN Recommender — How It Works, Step by Step

Strip away the jargon and a KNN recommender has exactly four moving parts. Understand each one and you understand the whole system.

Similar Users vs Similar Items — Choosing Your Space

Distance Metrics — The Engine of KNN

The choice of distance metric is the single most consequential hyperparameter in KNN. It decides what "similar" means. Different metrics capture fundamentally different notions of proximity. Choose wrong and your recommender will surface items nobody wants.

📊 Worked Example — All Four Main Metrics on Real Data

Distance Metrics — From Formula to Python

Before using a library, build each metric from scratch. Understanding the algebra makes the behaviour of each metric intuitive and the choice of metric deliberate.

import numpy as np
import pandas as pd

# ── Sample user rating vectors ────────────────────────────────
# NaN = not rated; we'll handle these per metric
users = {
    'Alice': np.array([5, 4, 3, np.nan, 2]),
    'Bob':   np.array([4, 3, 2, 4,    np.nan]),
    'Carol': np.array([1, 2, 4, 5,    3]),
    'Dave':  np.array([5, 5, 4, np.nan, 1]),
}


def cosine_sim(a: np.ndarray, b: np.ndarray) -> float:
    """Cosine similarity — treats NaN as 0."""
    a = np.nan_to_num(a)
    b = np.nan_to_num(b)
    denom = np.linalg.norm(a) * np.linalg.norm(b)
    return 0.0 if denom == 0 else float(np.dot(a, b) / denom)


def pearson_sim(a: np.ndarray, b: np.ndarray) -> float:
    """Pearson correlation — co-rated items only, mean-centred."""
    mask = ~np.isnan(a) & ~np.isnan(b)
    if mask.sum() < 2:
        return 0.0
    ac, bc = a[mask] - a[mask].mean(), b[mask] - b[mask].mean()
    denom = np.linalg.norm(ac) * np.linalg.norm(bc)
    return 0.0 if denom == 0 else float(np.dot(ac, bc) / denom)


def euclidean_sim(a: np.ndarray, b: np.ndarray) -> float:
    """Euclidean similarity = 1 / (1 + distance) — co-rated only."""
    mask = ~np.isnan(a) & ~np.isnan(b)
    if mask.sum() == 0:
        return 0.0
    d = np.linalg.norm(a[mask] - b[mask])
    return float(1 / (1 + d))


def msd_sim(a: np.ndarray, b: np.ndarray) -> float:
    """Mean Squared Difference similarity — co-rated only."""
    mask = ~np.isnan(a) & ~np.isnan(b)
    if mask.sum() == 0:
        return 0.0
    msd = np.mean((a[mask] - b[mask]) ** 2)
    return float(1 / (1 + msd))


# ── Compute all pairwise similarities for Alice ───────────────
target = 'Alice'
metrics = {
    'Cosine':    cosine_sim,
    'Pearson':   pearson_sim,
    'Euclidean': euclidean_sim,
    'MSD':       msd_sim,
}

rows = []
for other, vec in users.items():
    if other == target:
        continue
    row = {'User': other}
    for name, fn in metrics.items():
        row[name] = round(fn(users[target], vec), 4)
    rows.append(row)

print(pd.DataFrame(rows).set_index('User').to_string())

The K Hyperparameter — Finding the Right Neighbourhood Size

🎬 Project 1 — Movie Recommender with User-KNN from Scratch

Step 1 — Rating Matrix

import numpy as np
import pandas as pd
from itertools import combinations

# ── 7 users × 8 movies ───────────────────────────────────────
movies = ['Inception', 'Interstellar', 'Dune', 'The Matrix',
          'Parasite', 'Joker', 'Tenet', 'Oppenheimer']

data = {
    'Alice':   [5,    4,    5,    4,    np.nan, 2,    np.nan, 5   ],
    'Bob':     [4,    5,    4,    5,    2,     np.nan, 4,    4   ],
    'Carol':   [np.nan,2,   1,    2,    5,     5,    1,    np.nan],
    'Dave':    [5,    4,    np.nan,4,   1,     2,    5,    5   ],
    'Eve':     [3,    np.nan,3,   3,    4,     4,    3,    np.nan],
    'Frank':   [np.nan,5,   4,    np.nan,2,    1,    np.nan, 4   ],
    'Grace':   [4,    4,    5,    3,    np.nan, 2,    4,    5   ],
}

# Target user for all recommendations
TARGET = 'Alice'

ratings = pd.DataFrame(data, index=movies).T
n_total = ratings.size
n_rated = ratings.notna().sum().sum()

print(f"Matrix: {ratings.shape[0]} users × {ratings.shape[1]} movies")
print(f"Rated: {n_rated}/{n_total}  Sparsity: {1-n_rated/n_total:.1%}")
print("\nUser means:")
print(ratings.mean(axis=1).round(2).to_string())

Step 2 — Full User Similarity Matrix (Pearson)

def pearson(u: pd.Series, v: pd.Series) -> float:
    """Pearson correlation over co-rated items only."""
    mask = u.notna() & v.notna()
    if mask.sum() < 2:
        return np.nan
    uc = u[mask] - u[mask].mean()
    vc = v[mask] - v[mask].mean()
    denom = np.linalg.norm(uc) * np.linalg.norm(vc)
    return 0.0 if denom == 0 else float(np.dot(uc, vc) / denom)


users = ratings.index.tolist()
sim_df = pd.DataFrame(np.eye(len(users)), index=users, columns=users)

for u, v in combinations(users, 2):
    s = pearson(ratings.loc[u], ratings.loc[v])
    sim_df.loc[u, v] = sim_df.loc[v, u] = s

print("=== User-User Similarity Matrix (Pearson) ===")
print(sim_df.round(3).to_string())

Step 3 — KNN Prediction Engine

def predict_user_knn(target_user: str, target_movie: str,
                      ratings: pd.DataFrame, sim_df: pd.DataFrame,
                      k: int = 4) -> float:
    """
    Mean-centred weighted average prediction.
    pred(u,i) = r̄_u + Σ[sim(u,v)·(r_vi − r̄_v)] / Σ|sim(u,v)|
    Only neighbours who have rated target_movie contribute.
    """
    other_users = [u for u in ratings.index if u != target_user]

    # Only users who rated this movie
    candidates = [u for u in other_users
                  if pd.notna(ratings.loc[u, target_movie])]

    if not candidates:
        return ratings[target_movie].mean()

    # Sort by similarity, take top-k
    sims = sim_df.loc[target_user, candidates].dropna()
    top_k = sims.nlargest(k)

    user_mean = ratings.loc[target_user].mean()
    num, den  = 0.0, 0.0

    for nb, sim in top_k.items():
        nb_mean   = ratings.loc[nb].mean()
        nb_rating = ratings.loc[nb, target_movie]
        num += sim * (nb_rating - nb_mean)
        den += abs(sim)

    if den == 0:
        return user_mean

    pred = user_mean + num / den
    return float(np.clip(pred, 1, 5))


# ── Top-N recommendations for Alice ──────────────────────────
K = 4
unrated = [m for m in movies if pd.isna(ratings.loc[TARGET, m])]

results = []
for movie in unrated:
    pred  = predict_user_knn(TARGET, movie, ratings, sim_df, k=K)
    top_k = sim_df.loc[TARGET, :].drop(TARGET).nlargest(K)
    results.append({'Movie': movie,
                     'Predicted': round(pred, 2),
                     'Top Neighbour': top_k.index[0],
                     'Top Sim': round(top_k.iloc[0], 3)})

recs_df = pd.DataFrame(results).sort_values('Predicted', ascending=False)
print(f"=== Top Recommendations for {TARGET} (K={K}, Pearson) ===")
print(recs_df.to_string(index=False))

Step 4 — RMSE Evaluation Across Metrics

from sklearn.model_selection import train_test_split

def evaluate_metric(ratings: pd.DataFrame, metric_fn,
                      k: int = 4, test_frac: float = 0.2,
                      seed: int = 42) -> dict:
    """Hold-out RMSE for a given similarity function."""
    known = [(u, m, ratings.loc[u, m])
             for u in ratings.index
             for m in ratings.columns
             if pd.notna(ratings.loc[u, m])]

    train, test = train_test_split(known, test_size=test_frac, random_state=seed)

    # Mask test entries
    train_r = ratings.copy()
    for u, m, _ in test:
        train_r.loc[u, m] = np.nan

    # Rebuild similarity matrix
    ulist = ratings.index.tolist()
    sm    = pd.DataFrame(np.eye(len(ulist)), index=ulist, columns=ulist)
    for u, v in combinations(ulist, 2):
        s = metric_fn(train_r.loc[u], train_r.loc[v])
        sm.loc[u, v] = sm.loc[v, u] = s if not np.isnan(s) else 0

    actuals, preds = [], []
    for u, m, actual in test:
        pred = predict_user_knn(u, m, train_r, sm, k)
        actuals.append(actual)
        preds.append(pred)

    rmse = np.sqrt(np.mean((np.array(actuals) - np.array(preds)) ** 2))
    mae  = np.mean(np.abs(np.array(actuals) - np.array(preds)))
    return {'RMSE': round(rmse, 4), 'MAE': round(mae, 4), 'n_test': len(test)}


# Define wrapper functions for each metric
def pearson_s(a, b):  return pearson(a, b)

def cosine_s(a: pd.Series, b: pd.Series) -> float:
    av, bv = np.nan_to_num(a.values), np.nan_to_num(b.values)
    d = np.linalg.norm(av) * np.linalg.norm(bv)
    return 0.0 if d == 0 else float(np.dot(av, bv) / d)

def msd_s(a: pd.Series, b: pd.Series) -> float:
    mask = a.notna() & b.notna()
    if mask.sum() == 0: return 0.0
    return float(1 / (1 + np.mean((a[mask].values-b[mask].values)**2)))

metrics_map = {
    'Pearson': pearson_s,
    'Cosine':  cosine_s,
    'MSD':     msd_s,
}

print("=== Metric Comparison (K=4) ===")
for name, fn in metrics_map.items():
    res = evaluate_metric(ratings, fn, k=4)
    print(f"  {name:10s}  RMSE={res['RMSE']}  MAE={res['MAE']}  n_test={res['n_test']}")

🛒 Project 2 — Book Recommender with Item-KNN & Scikit-Surprise

Install

pip install scikit-surprise pandas numpy

Step 1 — Load & Filter Book-Crossings

import pandas as pd
import numpy as np
from surprise import Dataset, Reader, KNNWithMeans
from surprise.model_selection import cross_validate, GridSearchCV

# ── Download from: https://www.kaggle.com/datasets/syedjaferk/book-crossing-dataset
# File: BX-Book-Ratings.csv ───────────────────────────────────
ratings_raw = pd.read_csv(
    'BX-Book-Ratings.csv',
    sep=';', encoding='latin-1', on_bad_lines='skip',
    names=['user_id', 'isbn', 'rating'], header=0
)

# Keep only explicit ratings (1–10); 0 = implicit
df = ratings_raw[ratings_raw['rating'] > 0].copy()
print(f"Explicit ratings: {len(df):,}")

# Filter: users with ≥10 ratings, books with ≥20 ratings
user_counts = df['user_id'].value_counts()
book_counts = df['isbn'].value_counts()
df = df[df['user_id'].isin(user_counts[user_counts >= 10].index)]
df = df[df['isbn'].isin(book_counts[book_counts >= 20].index)]
df = df.drop_duplicates(['user_id', 'isbn'])

n_users = df['user_id'].nunique()
n_books = df['isbn'].nunique()
sparsity = 1 - len(df) / (n_users * n_books)

print(f"After filtering — Rows: {len(df):,}")
print(f"Users: {n_users:,}  Books: {n_books:,}  Sparsity: {sparsity:.2%}")
print(f"Rating distribution:\n{df['rating'].value_counts().sort_index()}")

Step 2 — User-KNN vs Item-KNN Baseline (5-Fold CV)

# Build Surprise dataset
reader  = Reader(rating_scale=(1, 10))
dataset = Dataset.load_from_df(df[['user_id','isbn','rating']], reader)

common_sim_opts = {'name':'pearson', 'min_support':3}

# User-KNN
user_knn = KNNWithMeans(k=40,
    sim_options={**common_sim_opts, 'user_based':True})
cv_user = cross_validate(user_knn, dataset,
    measures=['RMSE','MAE'], cv=5, verbose=False, n_jobs=-1)

# Item-KNN
item_knn = KNNWithMeans(k=40,
    sim_options={**common_sim_opts, 'user_based':False})
cv_item = cross_validate(item_knn, dataset,
    measures=['RMSE','MAE'], cv=5, verbose=False, n_jobs=-1)

print("=== Baseline Comparison (K=40, Pearson, 5-Fold CV) ===")
print(f"User-KNN  →  RMSE: {cv_user['test_rmse'].mean():.4f}  MAE: {cv_user['test_mae'].mean():.4f}")
print(f"Item-KNN  →  RMSE: {cv_item['test_rmse'].mean():.4f}  MAE: {cv_item['test_mae'].mean():.4f}")
print(f"\nUser-KNN fit time: {cv_user['fit_time'].mean():.1f}s")
print(f"Item-KNN fit time: {cv_item['fit_time'].mean():.1f}s")

Step 3 — GridSearch: Tune K and Metric

from surprise.model_selection import GridSearchCV

param_grid = {
    'k': [20, 40, 60, 80],
    'sim_options': [
        {'name':'pearson', 'user_based':False, 'min_support':3},
        {'name':'cosine',  'user_based':False, 'min_support':3},
        {'name':'msd',     'user_based':False, 'min_support':3},
    ]
}

gs = GridSearchCV(KNNWithMeans, param_grid,
                  measures=['rmse'], cv=3, n_jobs=-1)
gs.fit(dataset)

best_k      = gs.best_params['rmse']['k']
best_metric = gs.best_params['rmse']['sim_options']['name']
best_rmse   = gs.best_score['rmse']

print(f"Best K:      {best_k}")
print(f"Best metric: {best_metric}")
print(f"Best RMSE:   {best_rmse:.4f}")

results_gs = pd.DataFrame(gs.cv_results)
print(results_gs[['param_k', 'param_sim_options', 'mean_test_rmse']]
      .sort_values('mean_test_rmse')
      .head(6)
      .to_string(index=False))

Step 4 — "Readers Also Enjoyed" Shelf

# ── Train best configuration on full dataset ─────────────────
best_algo = KNNWithMeans(
    k=40,
    sim_options={'name':'msd', 'user_based':False, 'min_support':3}
)
full_trainset = dataset.build_full_trainset()
best_algo.fit(full_trainset)

def readers_also_enjoyed(isbn: str, algo, trainset,
                           top_n: int = 10) -> pd.DataFrame:
    """
    Item-KNN neighbour lookup — the 'Readers Also Enjoyed' shelf.
    Uses the pre-computed item similarity matrix from Surprise.
    """
    try:
        inner_id = trainset.to_inner_iid(isbn)
    except ValueError:
        raise ValueError(f"ISBN '{isbn}' not found in training data.")

    neighbours = algo.get_neighbors(inner_id, k=top_n)

    rows = []
    for nb_inner in neighbours:
        nb_isbn = trainset.to_raw_iid(nb_inner)
        sim     = float(algo.sim[inner_id, nb_inner])
        rows.append({
            'ISBN':       nb_isbn,
            'Similarity': round(sim, 4),
            'Strength':   'Strong' if sim > 0.7
                           else ('Medium' if sim > 0.4 else 'Weak')
        })
    return (pd.DataFrame(rows)
              .sort_values('Similarity', ascending=False)
              .reset_index(drop=True))


# Most-rated book in dataset
top_isbn = df['isbn'].value_counts().index[0]
shelf    = readers_also_enjoyed(top_isbn, best_algo, full_trainset)
print(f"=== Readers Also Enjoyed — ISBN: {top_isbn} ===")
print(shelf.to_string(index=False))

Step 5 — Personalised Top-10 for a Specific User

def personalised_top_n(user_id, algo, trainset,
                        df: pd.DataFrame,
                        top_n: int = 10) -> pd.DataFrame:
    """
    Personalised Item-KNN recommendations for one user.
    For every book the user rated, fetch its K nearest item neighbours.
    Accumulate weighted scores; normalise; rank; return top-N unread.

    Score(candidate) = Σ[sim(rated, candidate) × user_rating] / Σ sim(rated, candidate)
    """
    user_rows  = df[df['user_id'] == user_id]
    rated_set  = set(user_rows['isbn'])
    u_ratings  = dict(zip(user_rows['isbn'], user_rows['rating']))

    scores, sim_totals = {}, {}

    for isbn, rating in u_ratings.items():
        try:
            inner = trainset.to_inner_iid(isbn)
        except ValueError:
            continue   # book not in trainset — skip

        for nb_inner in algo.get_neighbors(inner, k=30):
            nb_isbn = trainset.to_raw_iid(nb_inner)
            if nb_isbn in rated_set:
                continue   # skip books user already rated
            sim = float(algo.sim[inner, nb_inner])
            if sim <= 0:
                continue   # ignore anti-signals
            scores[nb_isbn]     = scores.get(nb_isbn, 0.0)     + sim * rating
            sim_totals[nb_isbn] = sim_totals.get(nb_isbn, 0.0) + sim

    # Normalise
    predicted = {
        isbn: scores[isbn] / sim_totals[isbn]
        for isbn in scores if sim_totals[isbn] > 0
    }
    top = sorted(predicted.items(), key=lambda x: -x[1])[:top_n]

    return pd.DataFrame([
        {'Rank': r + 1, 'ISBN': isbn,
         'Predicted Rating': round(score, 2)}
        for r, (isbn, score) in enumerate(top)
    ])


# Pick a power reader with ≥20 ratings
power_readers = df['user_id'].value_counts()
sample_user   = power_readers[power_readers >= 20].index[5]

n_rated = len(df[df['user_id'] == sample_user])
print(f"User: {sample_user}  |  Books rated: {n_rated}")
print(f"Sample of their ratings:")
print(df[df['user_id'] == sample_user]
        [['isbn', 'rating']].head(5).to_string(index=False))

top10 = personalised_top_n(sample_user, best_algo, full_trainset, df)
print(f"\n=== Top-10 Personalised Recommendations ===")
print(top10.to_string(index=False))

KNN vs Other Recommender Approaches — Full Comparison

KNN is not the last word in recommendation systems — but it is the best starting point. Knowing exactly where it stands relative to other methods lets you make deliberate architectural choices rather than defaulting to complexity.

Golden Rules for KNN Recommenders

Summary — KNN Recommender at a Glance