Recommendation Systems

Recommendation Systems

💡 Recommendation systems predict user preferences and suggest items. They power Netflix, Spotify, Amazon, and countless other platforms. This lesson covers collaborative filtering, content-based methods, matrix factorization, and hybrid approaches.

1. Types of Recommendation Systems

                    Recommendation Systems
                            │
            ┌───────────────┼───────────────┐
            │               │               │
    Collaborative     Content-Based      Hybrid
       Filtering          Filtering       Methods
            │               │               │
    ┌───────┴───────┐      │          ┌────┴────┐
    │               │      │          │         │
  User-Based    Item-Based   │    Weighted  Feature
    CF             CF        │      CF     Combination
    │               │      │          │
  Matrix Factorization      │    Neural CF
  (SVD, ALS, NMF)          │    (Deep Learning)

2. Collaborative Filtering

User-Based CF

Item-Based CF

Idea: Items rated similarly by users are similar.

Similarity Metrics

Pearson Correlation:

Jaccard Similarity (for implicit feedback):

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class UserBasedCF:
    def __init__(self, ratings_matrix):
        """
        ratings_matrix: (n_users, n_items) sparse or dense matrix
        """
        self.ratings = np.array(ratings_matrix)
        self.user_sim = cosine_similarity(self.ratings)
        np.fill_diagonal(self.user_sim, 0)  # exclude self-similarity

    def predict(self, user_id, item_id, k=10):
        # Find k most similar users who rated this item
        rated_mask = self.ratings[:, item_id] > 0
        similarities = self.user_sim[user_id] * rated_mask

        # Top-k similar users
        top_k = np.argsort(similarities)[-k:]
        top_sims = similarities[top_k]
        top_ratings = self.ratings[top_k, item_id]

        # Weighted average
        if np.sum(np.abs(top_sims)) == 0:
            return np.mean(self.ratings[user_id][self.ratings[user_id] > 0])

        pred = np.dot(top_sims, top_ratings) / np.sum(np.abs(top_sims))
        return np.clip(pred, 1, 5)  # clamp to rating range

    def recommend(self, user_id, n=5, k=10):
        unrated = np.where(self.ratings[user_id] == 0)[0]
        predictions = [(item, self.predict(user_id, item, k)) for item in unrated]
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:n]

# Example
ratings = np.array([
    [5, 3, 0, 1, 4],
    [4, 0, 0, 1, 2],
    [1, 1, 0, 5, 4],
    [0, 0, 5, 4, 0],
    [0, 4, 4, 0, 3],
])

cf = UserBasedCF(ratings)
print("Prediction (user 0, item 2):", cf.predict(0, 2))
print("Recommendations for user 0:", cf.recommend(0, n=3))

3. Matrix Factorization

SVD-Based Approach

User-Item Matrix R (5 users × 5 items)
┌────┬────┬────┬────┬────┐
│ 5  │ 3  │  ? │ 1  │ 4  │    Decompose into:
├────┼────┼────┼────┼────┤
│ 4  │  ? │  ? │ 1  │ 2  │    P (5×2)          Q (5×2)
├────┼────┼────┼────┼────┤    ┌────┐           ┌────┐
│ 1  │ 1  │  ? │ 5  │ 4  │    │p₁₁│           │q₁₁│
├────┼────┼────┼────┼────┤    │p₂₁│    ×      │q₂₁│
│  ? │  ? │ 5  │ 4  │  ? │    │p₃₁│    ×      │q₃₁│
├────┼────┼────┼────┼────┤    │p₄₁│    ×      │q₄₁│
│  ? │ 4  │ 4  │  ? │ 3  │    │p₅₁│    ×      │q₅₁│
└────┴────┴────┴────┴────┘    └────┘           └────┘

? = predicted: r̂ᵤᵢ = pᵤ · qᵢ

Training with SGD

Gradient updates:

class MatrixFactorization:
    def __init__(self, n_users, n_items, n_factors=20, lr=0.01, reg=0.02):
        self.P = np.random.normal(0, 0.1, (n_users, n_factors))
        self.Q = np.random.normal(0, 0.1, (n_items, n_factors))
        self.lr = lr
        self.reg = reg

    def fit(self, ratings, n_epochs=100):
        """
        ratings: list of (user, item, rating) tuples
        """
        for epoch in range(n_epochs):
            np.random.shuffle(ratings)
            total_error = 0

            for u, i, r in ratings:
                pred = np.dot(self.P[u], self.Q[i])
                error = r - pred
                total_error += error ** 2

                # Update factors
                p_old = self.P[u].copy()
                self.P[u] += self.lr * (error * self.Q[i] - self.reg * self.P[u])
                self.Q[i] += self.lr * (error * p_old - self.reg * self.Q[i])

            rmse = np.sqrt(total_error / len(ratings))
            if (epoch + 1) % 20 == 0:
                print(f"Epoch {epoch+1:3d} | RMSE: {rmse:.4f}")

    def predict(self, user, item):
        return np.clip(np.dot(self.P[user], self.Q[item]), 1, 5)

    def recommend(self, user, n=5):
        scores = np.dot(self.P[user], self.Q.T)
        top_items = np.argsort(scores)[::-1][:n]
        return [(item, scores[item]) for item in top_items]

# Example
ratings_data = [
    (0, 0, 5), (0, 1, 3), (0, 3, 1), (0, 4, 4),
    (1, 0, 4), (1, 3, 1), (1, 4, 2),
    (2, 0, 1), (2, 1, 1), (2, 3, 5), (2, 4, 4),
    (3, 2, 5), (3, 3, 4),
    (4, 1, 4), (4, 2, 4), (4, 4, 3),
]

mf = MatrixFactorization(n_users=5, n_items=5, n_factors=10, lr=0.01, reg=0.02)
mf.fit(ratings_data, n_epochs=100)
print("\nPredictions:")
for u in range(5):
    for i in range(5):
        if ratings[u][i] == 0:
            print(f"  User {u}, Item {i}: {mf.predict(u, i):.2f}")

ALS (Alternating Least Squares)

Fixes one factor matrix and solves for the other:

class ALS:
    def __init__(self, n_users, n_items, n_factors=20, reg=0.1):
        self.P = np.random.normal(0, 0.1, (n_users, n_factors))
        self.Q = np.random.normal(0, 0.1, (n_items, n_factors))
        self.reg = reg

    def fit(self, ratings_matrix, n_iterations=15):
        """
        ratings_matrix: (n_users, n_items) with 0 for missing
        """
        for iteration in range(n_iterations):
            # Fix Q, solve for P
            for u in range(len(ratings_matrix)):
                idx = np.where(ratings_matrix[u] > 0)[0]
                if len(idx) == 0:
                    continue
                Q_u = self.Q[idx]
                r_u = ratings_matrix[u, idx]
                self.P[u] = np.linalg.solve(
                    Q_u.T @ Q_u + self.reg * np.eye(Q_u.shape[1]),
                    Q_u.T @ r_u
                )

            # Fix P, solve for Q
            for i in range(ratings_matrix.shape[1]):
                idx = np.where(ratings_matrix[:, i] > 0)[0]
                if len(idx) == 0:
                    continue
                P_i = self.P[idx]
                r_i = ratings_matrix[idx, i]
                self.Q[i] = np.linalg.solve(
                    P_i.T @ P_i + self.reg * np.eye(P_i.shape[1]),
                    P_i.T @ r_i
                )

    def predict(self):
        return self.P @ self.Q.T

als = ALS(n_users=5, n_items=5, n_factors=10, reg=0.1)
als.fit(ratings)
predicted = als.predict()
print("ALS predicted ratings:\n", np.round(predicted, 2))

4. Content-Based Filtering

Item Feature Representation

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import linear_kernel

class ContentBasedRecommender:
    def __init__(self):
        self.tfidf = TfidfVectorizer(stop_words='english')
        self.cosine_sim = None
        self.items = None

    def fit(self, items_df, content_column='description'):
        """items_df: DataFrame with 'id' and content_column"""
        self.items = items_df.reset_index(drop=True)

        # Build TF-IDF matrix
        tfidf_matrix = self.tfidf.fit_transform(self.items[content_column])

        # Compute cosine similarity
        self.cosine_sim = linear_kernel(tfidf_matrix, tfidf_matrix)

    def get_recommendations(self, item_id, n=10):
        idx = self.items[self.items['id'] == item_id].index[0]

        # Get similarity scores
        sim_scores = list(enumerate(self.cosine_sim[idx]))
        sim_scores = sorted(sim_scores, key=lambda x: x[1], reverse=True)

        # Exclude itself
        sim_scores = sim_scores[1:n+1]
        item_indices = [s[0] for s in sim_scores]

        return self.items.iloc[item_indices][['id', 'title', 'genres']]

# Example
movies = pd.DataFrame({
    'id': [1, 2, 3, 4, 5],
    'title': ['The Matrix', 'Inception', 'Toy Story', 'The Dark Knight', 'Finding Nemo'],
    'genres': ['Action Sci-Fi', 'Action Sci-Fi Thriller', 'Animation Comedy',
               'Action Crime Thriller', 'Animation Comedy Adventure'],
    'description': [
        'A computer hacker learns about the true nature of reality',
        'A thief who steals secrets through dream-sharing technology',
        'A cowboy doll is threatened by a new spaceman toy',
        'Batman faces the Joker, a criminal mastermind',
        'A clownfish searches for his missing son',
    ]
})

rec = ContentBasedRecommender()
rec.fit(movies, 'description')
print(rec.get_recommendations(1, n=3))

5. Neural Collaborative Filtering

Deep learning approach that learns non-linear user-item interactions:

import torch
import torch.nn as nn

class NeuralCF(nn.Module):
    def __init__(self, n_users, n_items, embed_dim=64, hidden_dims=[128, 64, 32]):
        super().__init__()
        self.user_embed = nn.Embedding(n_users, embed_dim)
        self.item_embed = nn.Embedding(n_items, embed_dim)

        layers = []
        input_dim = embed_dim * 2
        for h_dim in hidden_dims:
            layers.extend([
                nn.Linear(input_dim, h_dim),
                nn.ReLU(),
                nn.BatchNorm1d(h_dim),
                nn.Dropout(0.2),
            ])
            input_dim = h_dim
        layers.append(nn.Linear(input_dim, 1))
        self.mlp = nn.Sequential(*layers)

    def forward(self, user_ids, item_ids):
        user_emb = self.user_embed(user_ids)
        item_emb = self.item_embed(item_ids)
        x = torch.cat([user_emb, item_emb], dim=-1)
        return self.mlp(x).squeeze()

# Training
model = NeuralCF(n_users=1000, n_items=5000)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

# Example training loop
for epoch in range(20):
    model.train()
    for batch_users, batch_items, batch_ratings in train_loader:
        optimizer.zero_grad()
        predictions = model(batch_users, batch_items)
        loss = criterion(predictions, batch_ratings.float())
        loss.backward()
        optimizer.step()

6. Evaluation Metrics

Rating Prediction Metrics

RMSE (Root Mean Squared Error):

MAE (Mean Absolute Error):

Ranking Metrics

NDCG@K (Normalized Discounted Cumulative Gain):

MAP@K (Mean Average Precision):

import numpy as np

class RecommenderEvaluator:
    @staticmethod
    def rmse(predictions, actuals):
        return np.sqrt(np.mean((np.array(predictions) - np.array(actuals)) ** 2))

    @staticmethod
    def mae(predictions, actuals):
        return np.mean(np.abs(np.array(predictions) - np.array(actuals)))

    @staticmethod
    def precision_at_k(recommended, relevant, k):
        recommended = recommended[:k]
        relevant_set = set(relevant)
        hits = len([r for r in recommended if r in relevant_set])
        return hits / k

    @staticmethod
    def recall_at_k(recommended, relevant, k):
        recommended = recommended[:k]
        relevant_set = set(relevant)
        hits = len([r for r in recommended if r in relevant_set])
        return hits / len(relevant) if relevant else 0

    @staticmethod
    def ndcg_at_k(recommended, relevant_scores, k):
        """
        recommended: list of item ids in recommended order
        relevant_scores: dict {item_id: relevance_score}
        """
        dcg = 0.0
        for i, item in enumerate(recommended[:k]):
            rel = relevant_scores.get(item, 0)
            dcg += (2 ** rel - 1) / np.log2(i + 2)

        # Ideal DCG
        ideal = sorted(relevant_scores.values(), reverse=True)[:k]
        idcg = sum((2 ** r - 1) / np.log2(i + 2) for i, r in enumerate(ideal))

        return dcg / idcg if idcg > 0 else 0.0

    @staticmethod
    def mean_average_precision(user_recommendations, user_relevant, k=10):
        aps = []
        for rec, rel in zip(user_recommendations, user_relevant):
            rel_set = set(rel)
            hits = 0
            precision_sum = 0.0
            for i, item in enumerate(rec[:k]):
                if item in rel_set:
                    hits += 1
                    precision_sum += hits / (i + 1)
            ap = precision_sum / min(k, len(rel)) if rel else 0
            aps.append(ap)
        return np.mean(aps)

# Example evaluation
evaluator = RecommenderEvaluator()

predictions = [4.2, 3.8, 5.0, 2.1, 4.5]
actuals = [4, 4, 5, 2, 5]

print(f"RMSE: {evaluator.rmse(predictions, actuals):.4f}")
print(f"MAE:  {evaluator.mae(predictions, actuals):.4f}")

recommended = [1, 5, 3, 7, 2, 8, 10, 4, 6, 9]
relevant = [1, 3, 5, 7]
print(f"Precision@5: {evaluator.precision_at_k(recommended, relevant, 5):.4f}")
print(f"Recall@5:    {evaluator.recall_at_k(recommended, relevant, 5):.4f}")

relevant_scores = {1: 3, 3: 2, 5: 3, 7: 1}
print(f"NDCG@5:      {evaluator.ndcg_at_k(recommended, relevant_scores, 5):.4f}")

7. Hybrid Methods

Weighted Hybrid

Feature-Combination Hybrid

Combine user and item features with collaborative signals:

class HybridRecommender(nn.Module):
    def __init__(self, n_users, n_items, user_features_dim, item_features_dim, embed_dim=64):
        super().__init__()
        self.user_embed = nn.Embedding(n_users, embed_dim)
        self.item_embed = nn.Embedding(n_items, embed_dim)
        self.user_features_fc = nn.Linear(user_features_dim, embed_dim)
        self.item_features_fc = nn.Linear(item_features_dim, embed_dim)

        self.classifier = nn.Sequential(
            nn.Linear(embed_dim * 4, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 1),
        )

    def forward(self, user_ids, item_ids, user_features, item_features):
        user_emb = self.user_embed(user_ids)
        item_emb = self.item_embed(item_ids)
        user_feat = self.user_features_fc(user_features)
        item_feat = self.item_features_fc(item_features)

        x = torch.cat([user_emb, item_emb, user_feat, item_feat], dim=-1)
        return self.classifier(x).squeeze()

8. Cold Start Problem

New User (No History)

• Use content-based recommendations (demographics, popular items)
• Ask for explicit preferences (onboarding questionnaire)
• Use bandit algorithms to explore quickly

New Item (No Interactions)

• Use content features to find similar existing items
• Popularity-based initial exposure
• Knowledge-based recommendations (item attributes)

9. Key Takeaways

10. Practice Exercises

Exercise 1: Build a Collaborative Filter

# TODO: Implement item-based CF on MovieLens dataset
# Compare cosine vs Pearson similarity
# Evaluate with RMSE on held-out test set
# Target: RMSE < 0.9

Exercise 2: Matrix Factorization from Scratch

# TODO: Implement SGD-based matrix factorization
# Tune: learning rate, regularization, latent factors
# Plot training RMSE vs epochs
# Compare with ALS implementation

Exercise 3: Content-Based + Hybrid

# TODO: Build a content-based recommender for books
# Use: title, author, genre, description as features
# Combine with CF using weighted hybrid
# Find optimal alpha

Exercise 4: Deep Learning Recommender

# TODO: Implement Neural Collaborative Filtering
# Use: embedding layers + MLP
# Compare with matrix factorization baseline
# Experiment with architecture (layers, dropout)