Recommendation Engine

Recommendation Engine

Build recommendation systems for personalized content and product suggestions.

Recommendation Approaches

Collaborative Filtering

import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity

class CollaborativeFilter:
    def __init__(self):
        self.user_similarity = None
        self.item_similarity = None

    def fit(self, user_item_matrix):
        # User-based similarity
        self.user_similarity = cosine_similarity(user_item_matrix)
        # Item-based similarity
        self.item_similarity = cosine_similarity(user_item_matrix.T)

    def recommend_for_user(self, user_id, n=10):
        scores = self.user_similarity[user_id].dot(self.user_item_matrix)
        # Exclude already interacted items
        already_interacted = self.user_item_matrix[user_id].nonzero()[0]
        scores[already_interacted] = -np.inf
        return np.argsort(scores)[-n:][::-1]

Matrix Factorization (SVD)

from sklearn.decomposition import TruncatedSVD

class MatrixFactorization:
    def __init__(self, n_factors=50):
        self.svd = TruncatedSVD(n_components=n_factors)

    def fit(self, user_item_matrix):
        self.user_factors = self.svd.fit_transform(user_item_matrix)
        self.item_factors = self.svd.components_.T

    def predict(self, user_id, item_id):
        return np.dot(self.user_factors[user_id], self.item_factors[item_id])

Hybrid Recommender

class HybridRecommender:
    def __init__(self, collab_weight=0.7, content_weight=0.3):
        self.collab = CollaborativeFilter()
        self.content = ContentBasedFilter()
        self.weights = (collab_weight, content_weight)

    def recommend(self, user_id, n=10):
        collab_scores = self.collab.score(user_id)
        content_scores = self.content.score(user_id)
        combined = self.weights[0] * collab_scores + self.weights[1] * content_scores
        return np.argsort(combined)[-n:][::-1]

Evaluation Metrics

• Precision@K, Recall@K
• NDCG (ranking quality)
• Coverage (catalog diversity)
• A/B test conversion rate

Cold Start Solutions

• New users: Popular items, onboarding preferences, demographic-based
• New items: Content-based bootstrapping, active learning
• Exploration strategies: ε-greedy, Thompson sampling bandits

Quick Start: Build a Recommender in 5 Steps

from scipy.sparse import csr_matrix
import numpy as np

# 1. Prepare user-item interaction matrix
# rows = users, cols = items, values = ratings/interactions
ratings_data = [(0, 5, 5), (0, 10, 4), (1, 5, 3), ...]  # (user, item, rating)
n_users, n_items = 1000, 5000

row_idx = [r[0] for r in ratings_data]
col_idx = [r[1] for r in ratings_data]
ratings = [r[2] for r in ratings_data]
user_item_matrix = csr_matrix((ratings, (row_idx, col_idx)), shape=(n_users, n_items))

# 2. Choose and train model
from recommendation_engine import ItemBasedCollaborativeFilter  # See references

model = ItemBasedCollaborativeFilter(similarity_metric='cosine', k_neighbors=20)
model.fit(user_item_matrix)

# 3. Generate recommendations
recommendations = model.recommend(user_id=42, n=10)
print(recommendations)  # [(item_id, score), ...]

# 4. Evaluate on test set
from evaluation_metrics import precision_at_k, recall_at_k

test_items = {42: {10, 25, 30}}  # True relevant items for user 42
rec_items = [item for item, score in recommendations]

precision = precision_at_k(rec_items, test_items[42], k=10)
recall = recall_at_k(rec_items, test_items[42], k=10)
print(f"Precision@10: {precision:.3f}, Recall@10: {recall:.3f}")

# 5. Handle cold start
from cold_start import PopularityRecommender

popularity_model = PopularityRecommender()
popularity_model.fit(interactions_with_timestamps)
new_user_recs = popularity_model.recommend(n=10)

Known Issues Prevention

1. Popularity Bias

Problem: Recommending only popular items, ignoring long tail. Reduces diversity and serendipity.

Solution: Balance popularity with personalization, apply re-ranking for diversity:

def diversify_recommendations(
    recommendations: List[Tuple[int, float]],
    item_features: np.ndarray,
    diversity_weight: float = 0.3
) -> List[Tuple[int, float]]:
    """Re-rank to increase diversity while maintaining relevance."""
    from sklearn.metrics.pairwise import cosine_distances

    selected = []
    candidates = recommendations.copy()

    while len(selected) < len(recommendations) and candidates:
        if not selected:
            # First item: highest score
            selected.append(candidates.pop(0))
            continue

        # Compute diversity scores
        selected_features = item_features[[item for item, _ in selected]]
        diversity_scores = []

        for item, relevance in candidates:
            item_feature = item_features[item].reshape(1, -1)
            # Average distance to already selected items
            avg_distance = cosine_distances(item_feature, selected_features).mean()
            # Combined score: relevance + diversity
            combined = (1 - diversity_weight) * relevance + diversity_weight * avg_distance
            diversity_scores.append((item, relevance, combined))

        # Select item with best combined score
        best = max(diversity_scores, key=lambda x: x[2])
        selected.append((best[0], best[1]))
        candidates = [(i, s) for i, s, _ in diversity_scores if i != best[0]]

    return selected

2. Data Sparsity (Matrix >99% Empty)

Problem: Collaborative filtering fails when most users have rated <1% of items.

Solution: Use matrix factorization (SVD, ALS) instead of memory-based CF:

# ❌ Bad: User-based CF on sparse data (fails to find similar users)
user_cf = UserBasedCollaborativeFilter()
user_cf.fit(sparse_matrix)  # Most users have <10 ratings

# ✅ Good: Matrix factorization handles sparsity
from sklearn.decomposition import TruncatedSVD

svd = TruncatedSVD(n_components=50)
user_factors = svd.fit_transform(sparse_matrix)
item_factors = svd.components_.T

# Predict rating: user_factors[u] @ item_factors[i]

3. Cold Start Without Fallback

Problem: Recommender crashes or returns empty results for new users/items.

Solution: Always implement fallback chain:

def recommend_with_fallback(user_id, n=10):
    """Graceful degradation through fallback chain."""
    try:
        # Try personalized recommendations
        if has_sufficient_history(user_id, min_interactions=5):
            return collaborative_filter.recommend(user_id, n)
    except Exception as e:
        logger.warning(f"CF failed for user {user_id}: {e}")

    # Fallback 1: Demographic-based
    if user_demographics_available(user_id):
        return demographic_recommender.recommend(user_id, n)

    # Fallback 2: Popularity
    return popularity_recommender.recommend(n)

4. Not Excluding Already-Interacted Items

Problem: Recommending items user already purchased/viewed wastes recommendation slots.

Solution: Always filter interacted items:

# ✅ Correct: Exclude interacted items
user_items = user_item_matrix[user_id].nonzero()[1]
scores[user_items] = -np.inf  # Ensure they don't appear in top-K
recommendations = np.argsort(scores)[-n:][::-1]

# ❌ Wrong: Forgetting to filter
recommendations = np.argsort(scores)[-n:][::-1]  # May include already purchased!

5. Ignoring Implicit Feedback Confidence

Problem: Treating all clicks/views equally. 1 view ≠ 100 views.

Solution: Weight by interaction strength (view count, watch time, etc.):

# For implicit feedback, use confidence weighting
confidence_matrix = 1 + alpha * np.log(1 + interaction_counts)

# In ALS: C_ui * (P_ui - X_ui)²
# Higher confidence for items with more interactions

6. Not Evaluating Ranking Quality (Using Only Accuracy)

Problem: High prediction accuracy (RMSE) doesn't mean good top-K recommendations.

Solution: Use ranking metrics (NDCG, MAP@K):

# ❌ Bad: Only RMSE
from sklearn.metrics import mean_squared_error
rmse = np.sqrt(mean_squared_error(y_true, y_pred))

# ✅ Good: Ranking metrics for top-K evaluation
from evaluation_metrics import ndcg_at_k, mean_average_precision_at_k

# NDCG rewards putting highly relevant items first
ndcg = ndcg_at_k(recommendations, relevance_scores, k=10)

# MAP@K considers precision at each relevant item position
map_score = mean_average_precision_at_k(all_recommendations, ground_truth, k=10)

7. Filter Bubble (Lack of Exploration)

Problem: Always recommending similar items limits discovery, reduces user engagement over time.

Solution: Implement explore-exploit strategy:

class ExploreExploitRecommender:
    def __init__(self, base_model, epsilon=0.1):
        self.base_model = base_model
        self.epsilon = epsilon  # 10% exploration

    def recommend(self, user_id, n=10):
        # Exploit: Use trained model for most recommendations
        n_exploit = int(n * (1 - self.epsilon))
        exploitative_recs = self.base_model.recommend(user_id, n=n_exploit)

        # Explore: Add random diverse items
        n_explore = n - n_exploit
        explored_items = sample_diverse_items(n_explore)

        return exploitative_recs + explored_items

When to Load References

Load reference files when you need detailed implementations:

• Collaborative Filtering: Load references/collaborative-filtering-deep-dive.md for complete user-based and item-based CF implementations with similarity metrics (cosine, Pearson, Jaccard), scalability optimizations (sparse matrices, approximate nearest neighbors), and handling edge cases (cold start, sparsity)

• Matrix Factorization: Load references/matrix-factorization-methods.md for SVD, ALS, and NMF implementations with hyperparameter tuning, implicit feedback handling, and advanced techniques (BPR, WARP)

• Evaluation Metrics: Load references/evaluation-metrics-implementation.md for Precision@K, Recall@K, NDCG, coverage, diversity metrics, cross-validation strategies, and statistical significance testing (paired t-test, bootstrap confidence intervals)

• Cold Start Solutions: Load references/cold-start-strategies.md for new user/item strategies (popularity-based, onboarding, demographic, content-based bootstrapping, active learning), explore-exploit approaches (ε-greedy, Thompson sampling), and hybrid fallback chains

Cold Start Strategies

Complete solutions for cold start problems in recommendation systems: new users, new items, and system bootstrapping.

The Cold Start Problem

Three types of cold start: 1. New User: No interaction history to base recommendations on 2. New Item: No users have interacted with it yet 3. New System: Sparse data overall, can't train reliable models

New User Cold Start

1. Popular Items Fallback

Simplest approach: Recommend most popular items.

from typing import List, Tuple
import numpy as np
from collections import Counter
import logging

logger = logging.getLogger(__name__)


class PopularityRecommender:
    """
    Fallback recommender based on item popularity.

    Use for new users with no interaction history.
    """

    def __init__(self, decay_factor: float = 0.95):
        """
        Args:
            decay_factor: Time decay factor for recency weighting
        """
        self.decay_factor = decay_factor
        self.item_popularity: dict = {}
        self.item_recency_score: dict = {}

    def fit(self, interactions: np.ndarray, timestamp_col: int = 2):
        """
        Compute popularity scores from interaction history.

        Args:
            interactions: Array with columns [user_id, item_id, timestamp, ...]
            timestamp_col: Column index for timestamps
        """
        # Count interactions per item
        item_counts = Counter(interactions[:, 1].astype(int))

        # Compute recency-weighted popularity
        max_timestamp = interactions[:, timestamp_col].max()

        for user_id, item_id, timestamp in interactions[:, :3]:
            item_id = int(item_id)
            recency_weight = self.decay_factor ** ((max_timestamp - timestamp) / 86400)  # Days

            if item_id not in self.item_recency_score:
                self.item_recency_score[item_id] = 0.0

            self.item_recency_score[item_id] += recency_weight

        # Normalize popularity scores
        max_score = max(self.item_recency_score.values()) if self.item_recency_score else 1.0
        self.item_popularity = {
            item: score / max_score
            for item, score in self.item_recency_score.items()
        }

        logger.info(f"Computed popularity for {len(self.item_popularity)} items")

    def recommend(self, n: int = 10, exclude_items: set = None) -> List[Tuple[int, float]]:
        """
        Return top-N popular items.

        Args:
            n: Number of recommendations
            exclude_items: Items to exclude (already purchased, etc.)

        Returns:
            List of (item_id, popularity_score) tuples
        """
        exclude_items = exclude_items or set()

        # Sort by popularity, exclude specified items
        sorted_items = sorted(
            [(item, score) for item, score in self.item_popularity.items()
             if item not in exclude_items],
            key=lambda x: x[1],
            reverse=True
        )

        return sorted_items[:n]


### 2. Onboarding Preference Elicitation

Gather initial preferences during user onboarding.

class OnboardingRecommender: """ Collect initial preferences to bootstrap recommendations.

Shows carefully selected items to maximize information gain. """

def __init__( self, item_features: np.ndarray, n_explore: int = 10, popularity_recommender: 'PopularityRecommender' = None ): """ Args: item_features: Feature matrix (n_items, n_features) n_explore: Number of items to show during onboarding popularity_recommender: Optional popularity recommender for fallback """ self.item_features = item_features self.n_explore = n_explore self.popularity_recommender = popularity_recommender

def select_onboarding_items(self) -> List[int]: """ Select diverse, representative items for onboarding.

Uses k-means clustering to find cluster centroids. """ from sklearn.cluster import KMeans

Cluster items into k groups

kmeans = KMeans(n_clusters=self.n_explore, random_state=42) kmeans.fit(self.item_features)

Find items closest to each cluster centroid

from sklearn.metrics.pairwise import euclidean_distances

onboarding_items = [] for centroid in kmeans.cluster_centers_: distances = euclidean_distances([centroid], self.item_features)[0] closest_item = np.argmin(distances) onboarding_items.append(int(closest_item))

logger.info(f"Selected {len(onboarding_items)} diverse items for onboarding") return onboarding_items

def bootstrap_from_preferences( self, liked_items: List[int], disliked_items: List[int], n: int = 10 ) -> List[int]: """ Generate initial recommendations from onboarding preferences.

Args: liked_items: Items user liked during onboarding disliked_items: Items user disliked n: Number of recommendations

Returns: List of recommended item IDs """ if not liked_items: logger.warning("No liked items provided")

Actually fall back to popularity recommendations

if self.popularity_recommender: logger.info("Falling back to popularity-based recommendations") popular_items = self.popularity_recommender.recommend(n=n)

Extract just the item IDs from (item_id, score) tuples

return [item_id for item_id, _ in popular_items] else:

No popularity recommender available - return random sample

logger.warning("No popularity recommender available, using random sample") item_ids = np.arange(len(self.item_features)) return np.random.choice(item_ids, size=min(n, len(item_ids)), replace=False).tolist()

Compute user preference vector (average of liked items)

liked_features = self.item_features[liked_items] user_profile = liked_features.mean(axis=0)

Compute similarity to all items

from sklearn.metrics.pairwise import cosine_similarity similarities = cosine_similarity([user_profile], self.item_features)[0]

Penalize disliked items

for item_id in disliked_items: similarities[item_id] = -1.0

Exclude items already rated

for item_id in liked_items + disliked_items: similarities[item_id] = -1.0

Get top-N

top_items = np.argsort(similarities)[-n:][::-1] return top_items.tolist()

### 3. Demographic-Based Recommendations

Use user demographics when available.

class DemographicRecommender: """ Recommend based on user demographics (age, gender, location, etc.).

Finds similar users demographically and recommends what they liked. """

def __init__(self): self.user_demographics: dict = {} # user_id -> feature vector self.user_interactions: dict = {} # user_id -> set of items

def fit(self, user_demo_matrix: np.ndarray, interactions: np.ndarray): """ Args: user_demo_matrix: Array (n_users, n_demo_features) interactions: Array with [user_id, item_id] pairs """

Store demographics

for user_id, features in enumerate(user_demo_matrix): self.user_demographics[user_id] = features

Store interactions

from collections import defaultdict self.user_interactions = defaultdict(set) for user_id, item_id in interactions: self.user_interactions[int(user_id)].add(int(item_id))

def recommend_for_new_user( self, user_features: np.ndarray, n: int = 10, k_neighbors: int = 50 ) -> List[int]: """ Recommend for new user based on similar users' preferences.

Args: user_features: Demographic features for new user n: Number of recommendations k_neighbors: Number of similar users to consider

Returns: List of recommended item IDs """ from sklearn.metrics.pairwise import cosine_similarity

Find k most similar users demographically

all_user_features = np.array([ self.user_demographics[uid] for uid in self.user_demographics ]) similarities = cosine_similarity([user_features], all_user_features)[0]

Get top-k similar users

top_k_users = np.argsort(similarities)[-k_neighbors:]

Aggregate their interactions

from collections import Counter item_votes = Counter() for user_idx in top_k_users: user_id = list(self.user_demographics.keys())[user_idx] for item_id in self.user_interactions.get(user_id, []): item_votes[item_id] += similarities[user_idx]

Return top-N items

top_items = [item for item, score in item_votes.most_common(n)] return top_items

## New Item Cold Start

### 1. Content-Based Bootstrapping

Use item metadata to find similar items.

class ContentBasedBootstrap: """ Recommend new items based on content similarity to items user liked. """

def __init__(self, item_features: np.ndarray): """ Args: item_features: Matrix (n_items, n_features) with item metadata """ self.item_features = item_features

def add_new_item(self, item_id: int, features: np.ndarray): """ Add a new item with its features.

Args: item_id: New item ID features: Feature vector for new item """

Expand feature matrix

if item_id >= len(self.item_features):

Pad with zeros

padding_size = item_id - len(self.item_features) + 1 padding = np.zeros((padding_size, self.item_features.shape[1])) self.item_features = np.vstack([self.item_features, padding])

self.item_features[item_id] = features

def recommend_new_item_to_users( self, new_item_id: int, user_profiles: dict, n_users: int = 100 ) -> List[int]: """ Find users most likely to like the new item.

Args: new_item_id: ID of new item user_profiles: Dict[user_id -> preference vector] n_users: Number of users to target

Returns: List of user IDs to show new item to """ from sklearn.metrics.pairwise import cosine_similarity

new_item_features = self.item_features[new_item_id].reshape(1, -1)

Compute similarity between new item and each user's profile

user_scores = {} for user_id, profile in user_profiles.items(): similarity = cosine_similarity(new_item_features, [profile])[0][0] user_scores[user_id] = similarity

Return top-N users

top_users = sorted(user_scores.items(), key=lambda x: x[1], reverse=True) return [user_id for user_id, score in top_users[:n_users]]

### 2. Active Learning for New Items

Strategically show new item to diverse users to gather feedback quickly.

class ActiveLearningBootstrap: """ Intelligently select which users to show new item to maximize learning. """

def __init__(self): self.user_diversity_scores: dict = {}

def select_exploration_users( self, user_features: np.ndarray, n_users: int = 50 ) -> List[int]: """ Select diverse set of users for initial exposure.

Uses diversity sampling to cover different user segments.

Args: user_features: Matrix (n_users, n_features) n_users: Number of users to select

Returns: List of user IDs """ from sklearn.cluster import KMeans

Cluster users

n_clusters = min(n_users, len(user_features) // 10) kmeans = KMeans(n_clusters=n_clusters, random_state=42) kmeans.fit(user_features)

Select users from each cluster

selected_users = [] for cluster_id in range(n_clusters): cluster_users = np.where(kmeans.labels_ == cluster_id)[0]

Sample proportionally from each cluster

n_from_cluster = max(1, n_users // n_clusters) sampled = np.random.choice( cluster_users, size=min(n_from_cluster, len(cluster_users)), replace=False ) selected_users.extend(sampled.tolist())

return selected_users[:n_users]

## Hybrid Cold Start Approach

Combine multiple strategies for robust cold start handling.

class HybridColdStartRecommender: """ Combines popularity, content, and demographic approaches.

Automatically selects best strategy based on available data. """

def __init__( self, popularity_recommender: PopularityRecommender, content_recommender: ContentBasedBootstrap, demographic_recommender: DemographicRecommender ): self.popularity = popularity_recommender self.content = content_recommender self.demographic = demographic_recommender

def recommend_for_cold_user( self, user_id: int, user_features: np.ndarray = None, onboarding_likes: List[int] = None, n: int = 10 ) -> List[Tuple[int, float]]: """ Recommend for cold user using best available strategy.

Strategy selection priority: 1. If onboarding preferences available -> use content-based 2. If demographics available -> use demographic-based 3. Else -> use popularity-based

Args: user_id: User ID user_features: Demographic features (optional) onboarding_likes: Items liked during onboarding (optional) n: Number of recommendations

Returns: List of (item_id, score) tuples """

Strategy 1: Onboarding preferences (best signal)

if onboarding_likes: logger.info(f"Using content-based recommendations for user {user_id}") items = self.content.bootstrap_from_preferences( liked_items=onboarding_likes, disliked_items=[], n=n ) return [(item, 1.0) for item in items]

Strategy 2: Demographics (good signal)

if user_features is not None: logger.info(f"Using demographic-based recommendations for user {user_id}") items = self.demographic.recommend_for_new_user( user_features=user_features, n=n ) return [(item, 0.8) for item in items]

Strategy 3: Popularity (weak signal)

logger.info(f"Using popularity-based recommendations for user {user_id}") return self.popularity.recommend(n=n)

def recommend_new_item( self, item_id: int, item_features: np.ndarray, n_target_users: int = 100 ) -> List[int]: """ Bootstrap recommendations for new item.

Returns: List of user IDs to show item to """

Add item to content recommender

self.content.add_new_item(item_id, item_features)

Find target users (content-based similarity)

This requires building user profiles first

logger.info(f"Targeting {n_target_users} users for new item {item_id}") return [] # Would implement user profile building

## Explore-Exploit Strategies

Balance showing proven recommendations with exploring new items.

class EpsilonGreedyRecommender: """ ε-greedy strategy: exploit with probability (1-ε), explore with ε. """

def __init__( self, base_recommender, epsilon: float = 0.1, catalog_items: List[int] = None ): """ Args: base_recommender: Main recommendation model epsilon: Exploration probability (0 to 1) catalog_items: Full list of item IDs """ self.base_recommender = base_recommender self.epsilon = epsilon self.catalog_items = catalog_items or []

def recommend(self, user_id: int, n: int = 10) -> List[Tuple[int, float]]: """ Generate recommendations with exploration.

Args: user_id: User ID n: Number of recommendations

Returns: List of (item_id, score) tuples """

Decide: explore or exploit

if np.random.random() < self.epsilon:

Explore: random items

logger.debug(f"Exploring for user {user_id}") random_items = np.random.choice(self.catalog_items, size=n, replace=False) return [(int(item), 0.5) for item in random_items] else:

Exploit: use base recommender

logger.debug(f"Exploiting for user {user_id}") return self.base_recommender.recommend(user_id, n=n)

## Multi-Armed Bandit for Cold Start

Use bandit algorithms to learn quickly which items work best.

class ThompsonSamplingRecommender: """ Thompson Sampling for item recommendation.

Models each item as a Bernoulli bandit, learns click-through rates. """

def __init__(self, n_items: int): """ Args: n_items: Number of items in catalog """

Beta distribution parameters for each item

self.alpha = np.ones(n_items) # Successes + 1 self.beta = np.ones(n_items) # Failures + 1

def recommend(self, n: int = 10) -> List[int]: """ Sample from each item's posterior and return top-N.

Returns: List of item IDs """

Sample from Beta distribution for each item

sampled_ctrs = np.random.beta(self.alpha, self.beta)

Return top-N by sampled CTR

top_items = np.argsort(sampled_ctrs)[-n:][::-1] return top_items.tolist()

def update(self, item_id: int, clicked: bool): """ Update beliefs after user interaction.

Args: item_id: Item that was shown clicked: Whether user clicked/liked it """ if clicked: self.alpha[item_id] += 1 else: self.beta[item_id] += 1

## When to Use Each Strategy

**Popularity-Based**:
- ✅ Absolute cold start (new system, very sparse data)
- ✅ Quick baseline
- ❌ Low personalization

**Onboarding Elicitation**:
- ✅ Can ask users questions
- ✅ Need fast personalization
- ❌ User friction (some drop-off)

**Demographic-Based**:
- ✅ Demographics available
- ✅ Regulated industries (explainability)
- ❌ Privacy concerns

**Content-Based**:
- ✅ Rich item metadata
- ✅ New items frequently added
- ❌ Limited by feature quality

**Active Learning**:
- ✅ Can afford exploration cost
- ✅ Need quick learning
- ❌ Complex to implement

**Bandit Algorithms**:
- ✅ Online learning scenario
- ✅ Fast feedback loop
- ❌ Requires real-time updates

Collaborative Filtering Deep Dive

Complete implementations of user-based and item-based collaborative filtering with similarity metrics and scalability optimizations.

Overview

Collaborative filtering recommends items based on patterns in user-item interactions. Two main approaches:

• User-based: Find similar users, recommend what they liked
• Item-based: Find similar items, recommend items similar to what user liked

Complete User-Based Implementation

from typing import List, Tuple, Dict, Optional
import numpy as np
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity, pairwise_distances
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class UserBasedCollaborativeFilter:
    """
    User-based collaborative filtering recommender.

    Finds similar users and recommends items they liked.
    Works well when #users << #items and user preferences are stable.
    """

    def __init__(
        self,
        similarity_metric: str = 'cosine',
        k_neighbors: int = 50,
        min_similarity: float = 0.0
    ):
        """
        Initialize user-based CF.

        Args:
            similarity_metric: 'cosine', 'pearson', or 'jaccard'
            k_neighbors: Number of similar users to consider
            min_similarity: Minimum similarity threshold
        """
        self.similarity_metric = similarity_metric
        self.k_neighbors = k_neighbors
        self.min_similarity = min_similarity
        self.user_similarity: Optional[np.ndarray] = None
        self.user_item_matrix: Optional[csr_matrix] = None

    def fit(self, user_item_matrix: csr_matrix) -> 'UserBasedCollaborativeFilter':
        """
        Compute user-user similarity matrix.

        Args:
            user_item_matrix: Sparse matrix (n_users, n_items) with ratings/interactions

        Returns:
            self for chaining
        """
        logger.info(f"Fitting user-based CF with {user_item_matrix.shape[0]} users, "
                   f"{user_item_matrix.shape[1]} items")

        self.user_item_matrix = user_item_matrix

        # Compute similarity based on metric
        if self.similarity_metric == 'cosine':
            self.user_similarity = cosine_similarity(user_item_matrix)
        elif self.similarity_metric == 'pearson':
            # Pearson = cosine on mean-centered data
            mean_centered = user_item_matrix.copy()
            row_means = np.array(mean_centered.mean(axis=1)).flatten()
            mean_centered.data -= np.repeat(row_means, np.diff(mean_centered.indptr))
            self.user_similarity = cosine_similarity(mean_centered)
        elif self.similarity_metric == 'jaccard':
            # Jaccard distance
            self.user_similarity = 1 - pairwise_distances(
                user_item_matrix.astype(bool), metric='jaccard'
            )
        else:
            raise ValueError(f"Unknown similarity metric: {self.similarity_metric}")

        # Zero out diagonal (user similarity with self)
        np.fill_diagonal(self.user_similarity, 0)

        logger.info(f"Computed similarity matrix shape: {self.user_similarity.shape}")
        return self

    def recommend(
        self,
        user_id: int,
        n: int = 10,
        exclude_interacted: bool = True
    ) -> List[Tuple[int, float]]:
        """
        Generate top-N recommendations for a user.

        Args:
            user_id: User ID to generate recommendations for
            n: Number of recommendations to return
            exclude_interacted: Whether to exclude items user already interacted with

        Returns:
            List of (item_id, score) tuples, sorted by score descending
        """
        if self.user_similarity is None or self.user_item_matrix is None:
            raise ValueError("Model not fitted. Call fit() first.")

        if user_id >= self.user_similarity.shape[0]:
            raise ValueError(f"User ID {user_id} out of range")

        # Find k most similar users
        user_sims = self.user_similarity[user_id]

        # Apply minimum similarity threshold
        user_sims[user_sims < self.min_similarity] = 0

        # Get top-k similar users
        top_k_users = np.argsort(user_sims)[-self.k_neighbors:]
        top_k_sims = user_sims[top_k_users]

        # Weighted sum of similar users' ratings
        similar_users_matrix = self.user_item_matrix[top_k_users]
        scores = np.array(top_k_sims.dot(similar_users_matrix.toarray())).flatten()

        # Exclude already interacted items
        if exclude_interacted:
            interacted_items = self.user_item_matrix[user_id].nonzero()[1]
            scores[interacted_items] = -np.inf

        # Get top-N items
        top_n_items = np.argsort(scores)[-n:][::-1]
        top_n_scores = scores[top_n_items]

        # Normalize scores to [0, 1]
        if top_n_scores.max() > 0:
            top_n_scores = top_n_scores / top_n_scores.max()

        return list(zip(top_n_items.tolist(), top_n_scores.tolist()))


class ItemBasedCollaborativeFilter:
    """
    Item-based collaborative filtering recommender.

    Finds similar items and recommends items similar to what user liked.
    Works well when #items << #users and item features are stable.
    More scalable than user-based for large user bases.
    """

    def __init__(
        self,
        similarity_metric: str = 'cosine',
        k_neighbors: int = 20,
        min_similarity: float = 0.1
    ):
        """
        Initialize item-based CF.

        Args:
            similarity_metric: 'cosine', 'pearson', or 'adjusted_cosine'
            k_neighbors: Number of similar items to consider
            min_similarity: Minimum similarity threshold
        """
        self.similarity_metric = similarity_metric
        self.k_neighbors = k_neighbors
        self.min_similarity = min_similarity
        self.item_similarity: Optional[np.ndarray] = None
        self.user_item_matrix: Optional[csr_matrix] = None

    def fit(self, user_item_matrix: csr_matrix) -> 'ItemBasedCollaborativeFilter':
        """
        Compute item-item similarity matrix.

        Args:
            user_item_matrix: Sparse matrix (n_users, n_items) with ratings/interactions

        Returns:
            self for chaining
        """
        logger.info(f"Fitting item-based CF with {user_item_matrix.shape[0]} users, "
                   f"{user_item_matrix.shape[1]} items")

        self.user_item_matrix = user_item_matrix

        # Transpose to get item-user matrix
        item_user_matrix = user_item_matrix.T

        if self.similarity_metric == 'cosine':
            self.item_similarity = cosine_similarity(item_user_matrix)
        elif self.similarity_metric == 'pearson':
            mean_centered = item_user_matrix.copy()
            row_means = np.array(mean_centered.mean(axis=1)).flatten()
            mean_centered.data -= np.repeat(row_means, np.diff(mean_centered.indptr))
            self.item_similarity = cosine_similarity(mean_centered)
        elif self.similarity_metric == 'adjusted_cosine':
            # Adjusted cosine: normalize by user mean (better for ratings)
            user_means = np.array(user_item_matrix.mean(axis=1)).flatten()
            adjusted = user_item_matrix.copy()
            for i in range(adjusted.shape[0]):
                adjusted.data[adjusted.indptr[i]:adjusted.indptr[i+1]] -= user_means[i]
            self.item_similarity = cosine_similarity(adjusted.T)
        else:
            raise ValueError(f"Unknown similarity metric: {self.similarity_metric}")

        # Zero out diagonal
        np.fill_diagonal(self.item_similarity, 0)

        logger.info(f"Computed item similarity matrix shape: {self.item_similarity.shape}")
        return self

    def recommend(
        self,
        user_id: int,
        n: int = 10,
        exclude_interacted: bool = True
    ) -> List[Tuple[int, float]]:
        """
        Generate top-N recommendations for a user based on item similarity.

        Args:
            user_id: User ID to generate recommendations for
            n: Number of recommendations to return
            exclude_interacted: Whether to exclude items user already interacted with

        Returns:
            List of (item_id, score) tuples, sorted by score descending
        """
        if self.item_similarity is None or self.user_item_matrix is None:
            raise ValueError("Model not fitted. Call fit() first.")

        if user_id >= self.user_item_matrix.shape[0]:
            raise ValueError(f"User ID {user_id} out of range")

        # Get user's interacted items and their ratings
        user_items = self.user_item_matrix[user_id].toarray().flatten()
        interacted_items = user_items.nonzero()[0]

        if len(interacted_items) == 0:
            logger.warning(f"User {user_id} has no interactions, returning empty recommendations")
            return []

        # For each candidate item, compute weighted similarity to user's items
        scores = np.zeros(self.item_similarity.shape[0])

        for item_id in interacted_items:
            # Get similarities of this item to all other items
            item_sims = self.item_similarity[item_id]

            # Apply threshold
            item_sims[item_sims < self.min_similarity] = 0

            # Weight by user's rating/interaction strength
            item_rating = user_items[item_id]
            scores += item_sims * item_rating

        # Exclude already interacted items
        if exclude_interacted:
            scores[interacted_items] = -np.inf

        # Get top-N items
        top_n_items = np.argsort(scores)[-n:][::-1]
        top_n_scores = scores[top_n_items]

        # Normalize scores
        if top_n_scores.max() > 0:
            top_n_scores = top_n_scores / top_n_scores.max()

        return list(zip(top_n_items.tolist(), top_n_scores.tolist()))

Similarity Metrics Comparison

Cosine Similarity

Best for: Binary interactions (clicks, views), sparse data

# Formula: cos(A, B) = (A · B) / (||A|| * ||B||)
# Range: [-1, 1], typically [0, 1] for non-negative data
similarity = cosine_similarity(user_item_matrix)

Pros: Fast, works well with sparse data, scale-invariant Cons: Ignores rating magnitude differences

Pearson Correlation

Best for: Rating data where user bias matters

# Centers data by subtracting means before computing cosine
# Accounts for users who rate everything high/low
mean_centered = user_item_matrix - user_means[:, np.newaxis]
similarity = cosine_similarity(mean_centered)

Pros: Accounts for user bias, better for ratings Cons: Computationally expensive, requires dense data

Jaccard Similarity

Best for: Binary data, set overlap

# Formula: J(A, B) = |A ∩ B| / |A ∪ B|
# Treats interactions as sets
from sklearn.metrics import jaccard_score

Pros: Simple, interpretable, good for binary Cons: Ignores rating values, sensitive to popularity

Scalability Optimizations

1. Sparse Matrix Operations

from scipy.sparse import csr_matrix

# Always use sparse matrices for large datasets
user_item_matrix = csr_matrix(ratings_array)

# Sparse matrix multiplication is O(nnz) not O(n²)
scores = user_similarity_sparse.dot(user_item_matrix)

2. Approximate Nearest Neighbors

For very large datasets (millions of users/items):

from sklearn.neighbors import NearestNeighbors

class ApproximateCF:
    def __init__(self, k_neighbors=50):
        self.nn_model = NearestNeighbors(
            n_neighbors=k_neighbors,
            algorithm='auto',  # Uses ball_tree or kd_tree
            metric='cosine'
        )

    def fit(self, user_item_matrix):
        self.nn_model.fit(user_item_matrix)
        self.user_item_matrix = user_item_matrix

    def recommend(self, user_id, n=10):
        # Find k-nearest neighbors efficiently
        distances, indices = self.nn_model.kneighbors(
            self.user_item_matrix[user_id],
            return_distance=True
        )

        # Convert distances to similarities
        similarities = 1 - distances.flatten()
        neighbor_ids = indices.flatten()

        # Weighted recommendation
        scores = similarities.dot(
            self.user_item_matrix[neighbor_ids].toarray()
        )
        return np.argsort(scores)[-n:][::-1]

3. Pre-compute Item Similarities

# For item-based CF, pre-compute and store only top-k similar items
def compute_sparse_similarity(item_matrix, k=20):
    """
    Compute top-k similar items for each item, store as sparse matrix.
    Reduces memory from O(n²) to O(nk).
    """
    n_items = item_matrix.shape[0]
    similarities = []

    for i in range(n_items):
        sims = cosine_similarity(item_matrix[i:i+1], item_matrix).flatten()
        top_k = np.argpartition(sims, -k)[-k:]
        similarities.append((i, top_k, sims[top_k]))

    return similarities  # Store as {item_id: [(similar_id, score), ...]}

Handling Edge Cases

Cold Start Users (No Interactions)

def recommend_for_cold_user(self, n=10):
    """Recommend popular items for users with no interaction history."""
    item_popularity = np.array(self.user_item_matrix.sum(axis=0)).flatten()
    top_items = np.argsort(item_popularity)[-n:][::-1]
    return [(item_id, float(item_popularity[item_id])) for item_id in top_items]

Cold Start Items (No Interactions)

# Fall back to content-based filtering (covered in hybrid section)
# Or use item metadata to find similar items

Data Sparsity

# Use matrix factorization (SVD, ALS) for extremely sparse data
# See matrix-factorization-methods.md

Production Considerations

Memory Usage

# User-based similarity matrix: O(n_users²)
# For 1M users: ~4TB for float32!

# Solutions:
# 1. Use item-based (usually fewer items)
# 2. Use approximate methods (LSH, ANNOY)
# 3. Store only top-k similarities per user
# 4. Use model-based methods (matrix factorization)

Computation Time

import time

def timed_fit(model, data):
    start = time.time()
    model.fit(data)
    elapsed = time.time() - start
    logger.info(f"Fit completed in {elapsed:.2f}s")
    return model

# Typical times:
# - 10K users, 1K items: ~5 seconds
# - 100K users, 10K items: ~5 minutes
# - 1M users, 100K items: Use approximate methods

Incremental Updates

class IncrementalCF:
    """Update recommendations without full recomputation."""

    def update_user(self, user_id, new_interactions):
        """Update single user's recommendations."""
        # Recompute only affected similarity scores
        user_vector = self.user_item_matrix[user_id]
        user_vector[new_interactions] = 1

        # Update similarity with all other users
        self.user_similarity[user_id] = cosine_similarity(
            user_vector.reshape(1, -1),
            self.user_item_matrix
        ).flatten()

        self.user_similarity[:, user_id] = self.user_similarity[user_id]
        np.fill_diagonal(self.user_similarity, 0)

When to Use User-Based vs Item-Based

User-Based:

• ✅ Small number of users (<100K)
• ✅ Many items per user
• ✅ User preferences change slowly
• ✅ Need serendipity (discover new genres)

Item-Based:

• ✅ Small number of items (<100K)
• ✅ Many users per item
• ✅ Item features stable
• ✅ Need stability (consistent recommendations)
• ✅ Real-time recommendations (pre-computed similarities)

Neither (use matrix factorization):

• ❌ Millions of users AND items
• ❌ Extremely sparse data (<0.1% density)
• ❌ Need latent features

Evaluation Metrics Implementation

Complete implementations of Precision@K, Recall@K, NDCG, coverage metrics, and statistical significance testing for recommendation systems.

Overview

Recommendation systems require specialized metrics beyond traditional ML metrics because:

• We care about top-K results, not all predictions
• Ranking quality matters (order of recommendations)
• Diversity and coverage are important
• Online A/B tests measure real business impact

Precision@K and Recall@K

from typing import List, Set, Dict
import numpy as np
from collections import defaultdict
import logging

logger = logging.getLogger(__name__)


def precision_at_k(recommended: List[int], relevant: Set[int], k: int) -> float:
    """
    Precision@K: Fraction of recommended items in top-K that are relevant.

    Args:
        recommended: List of recommended item IDs (ordered by score)
        relevant: Set of truly relevant item IDs
        k: Number of top recommendations to consider

    Returns:
        Precision@K score in [0, 1]

    Example:
        >>> recommended = [1, 3, 5, 7, 9]
        >>> relevant = {1, 5, 10}
        >>> precision_at_k(recommended, relevant, k=5)
        0.4  # 2 out of 5 recommendations are relevant
    """
    if k <= 0:
        return 0.0

    top_k = recommended[:k]
    relevant_in_top_k = len([item for item in top_k if item in relevant])

    return relevant_in_top_k / k


def recall_at_k(recommended: List[int], relevant: Set[int], k: int) -> float:
    """
    Recall@K: Fraction of relevant items that appear in top-K.

    Args:
        recommended: List of recommended item IDs
        relevant: Set of truly relevant item IDs
        k: Number of top recommendations

    Returns:
        Recall@K score in [0, 1]

    Example:
        >>> recommended = [1, 3, 5, 7, 9]
        >>> relevant = {1, 5, 10}
        >>> recall_at_k(recommended, relevant, k=5)
        0.667  # 2 out of 3 relevant items retrieved
    """
    if len(relevant) == 0:
        return 0.0

    top_k = recommended[:k]
    relevant_in_top_k = len([item for item in top_k if item in relevant])

    return relevant_in_top_k / len(relevant)


def f1_at_k(recommended: List[int], relevant: Set[int], k: int) -> float:
    """
    F1@K: Harmonic mean of Precision@K and Recall@K.

    Returns:
        F1 score in [0, 1]
    """
    prec = precision_at_k(recommended, relevant, k)
    rec = recall_at_k(recommended, relevant, k)

    if prec + rec == 0:
        return 0.0

    return 2 * (prec * rec) / (prec + rec)


def average_precision_at_k(recommended: List[int], relevant: Set[int], k: int) -> float:
    """
    Average Precision@K (AP@K): Rewards putting relevant items earlier.

    Computes precision at each relevant item position, then averages.

    Args:
        recommended: Ordered list of recommendations
        relevant: Set of relevant items
        k: Maximum number of recommendations to consider

    Returns:
        AP@K score in [0, 1]

    Example:
        >>> recommended = [1, 2, 3, 4, 5]
        >>> relevant = {1, 3, 5}
        >>> average_precision_at_k(recommended, relevant, k=5)
        0.756  # (1/1 + 2/3 + 3/5) / 3
    """
    if len(relevant) == 0:
        return 0.0

    top_k = recommended[:k]
    score = 0.0
    num_hits = 0.0

    for i, item in enumerate(top_k):
        if item in relevant:
            num_hits += 1.0
            score += num_hits / (i + 1.0)

    return score / min(len(relevant), k)


def mean_average_precision_at_k(
    recommendations: Dict[int, List[int]],
    ground_truth: Dict[int, Set[int]],
    k: int
) -> float:
    """
    Mean Average Precision@K (MAP@K): Average AP@K across all users.

    Args:
        recommendations: Dict mapping user_id -> list of recommended items
        ground_truth: Dict mapping user_id -> set of relevant items
        k: Number of recommendations

    Returns:
        MAP@K score in [0, 1]
    """
    ap_scores = []

    for user_id in recommendations:
        if user_id in ground_truth:
            ap = average_precision_at_k(
                recommendations[user_id],
                ground_truth[user_id],
                k
            )
            ap_scores.append(ap)

    return np.mean(ap_scores) if ap_scores else 0.0

Normalized Discounted Cumulative Gain (NDCG)

NDCG measures ranking quality with position-based discounting.

def dcg_at_k(recommended: List[int], relevance_scores: Dict[int, float], k: int) -> float:
    """
    Discounted Cumulative Gain@K.

    DCG = sum(rel_i / log2(i + 1)) for i in [1, k]

    Args:
        recommended: Ordered list of recommended items
        relevance_scores: Dict mapping item_id -> relevance score
        k: Number of top items

    Returns:
        DCG score
    """
    dcg = 0.0
    for i, item_id in enumerate(recommended[:k]):
        rel = relevance_scores.get(item_id, 0.0)
        # Position discount: log2(i + 2) because i starts at 0
        dcg += rel / np.log2(i + 2)

    return dcg


def ndcg_at_k(
    recommended: List[int],
    relevance_scores: Dict[int, float],
    k: int
) -> float:
    """
    Normalized Discounted Cumulative Gain@K.

    NDCG = DCG / IDCG

    Where IDCG is DCG of the ideal ranking (sorted by relevance).

    Args:
        recommended: Ordered list of recommended items
        relevance_scores: Dict mapping item_id -> relevance (higher = more relevant)
        k: Number of top items

    Returns:
        NDCG score in [0, 1]

    Example:
        >>> recommended = [3, 1, 5, 2, 4]
        >>> relevance = {1: 3, 2: 2, 3: 3, 4: 1, 5: 2}
        >>> ndcg_at_k(recommended, relevance, k=5)
        0.952  # Close to ideal ranking
    """
    # Compute DCG for actual recommendations
    dcg = dcg_at_k(recommended, relevance_scores, k)

    # Compute IDCG (ideal DCG) by sorting items by relevance
    ideal_ranking = sorted(
        relevance_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )
    ideal_items = [item_id for item_id, _ in ideal_ranking]
    idcg = dcg_at_k(ideal_items, relevance_scores, k)

    if idcg == 0:
        return 0.0

    return dcg / idcg


def mean_ndcg_at_k(
    recommendations: Dict[int, List[int]],
    relevance_scores: Dict[int, Dict[int, float]],
    k: int
) -> float:
    """
    Mean NDCG@K across all users.

    Args:
        recommendations: Dict[user_id -> list of recommended items]
        relevance_scores: Dict[user_id -> Dict[item_id -> relevance]]
        k: Number of recommendations

    Returns:
        Mean NDCG@K
    """
    ndcg_scores = []

    for user_id, recs in recommendations.items():
        if user_id in relevance_scores:
            ndcg = ndcg_at_k(recs, relevance_scores[user_id], k)
            ndcg_scores.append(ndcg)

    return np.mean(ndcg_scores) if ndcg_scores else 0.0

Coverage and Diversity Metrics

def catalog_coverage(
    recommendations: Dict[int, List[int]],
    catalog_size: int,
    k: int
) -> float:
    """
    Catalog Coverage: Fraction of catalog items that appear in recommendations.

    Measures diversity across all users.

    Args:
        recommendations: Dict[user_id -> list of recommendations]
        catalog_size: Total number of items in catalog
        k: Number of recommendations per user

    Returns:
        Coverage in [0, 1]

    Example:
        >>> recs = {
        ...     0: [1, 2, 3],
        ...     1: [2, 3, 4],
        ...     2: [3, 4, 5]
        ... }
        >>> catalog_coverage(recs, catalog_size=100, k=3)
        0.05  # 5 unique items out of 100
    """
    recommended_items = set()

    for user_recs in recommendations.values():
        recommended_items.update(user_recs[:k])

    return len(recommended_items) / catalog_size


def gini_coefficient(recommendations: Dict[int, List[int]], k: int) -> float:
    """
    Gini Coefficient: Measure of recommendation concentration.

    0 = perfectly equal distribution (all items recommended equally)
    1 = perfect inequality (only one item recommended to everyone)

    Lower is better (more diverse).

    Args:
        recommendations: Dict[user_id -> list of recommendations]
        k: Number of recommendations per user

    Returns:
        Gini coefficient in [0, 1]
    """
    # Count how many times each item was recommended
    item_counts = defaultdict(int)
    for user_recs in recommendations.values():
        for item in user_recs[:k]:
            item_counts[item] += 1

    # Sort counts
    counts = sorted(item_counts.values())
    n = len(counts)

    if n == 0:
        return 0.0

    # Compute Gini
    # G = (2 * sum(i * x_i)) / (n * sum(x_i)) - (n + 1) / n
    cumsum = np.cumsum(counts)
    gini = (2 * np.sum((np.arange(1, n + 1) * counts))) / (n * cumsum[-1]) - (n + 1) / n

    return gini


def intra_list_diversity(recommended: List[int], item_features: np.ndarray) -> float:
    """
    Intra-List Diversity: Average dissimilarity between recommended items.

    Measures how diverse a single recommendation list is.

    Args:
        recommended: List of recommended item IDs
        item_features: Array (n_items, n_features) of item feature vectors

    Returns:
        Average pairwise distance in [0, 1]
    """
    from sklearn.metrics.pairwise import cosine_distances

    if len(recommended) < 2:
        return 0.0

    # Get feature vectors for recommended items
    rec_features = item_features[recommended]

    # Compute pairwise distances
    distances = cosine_distances(rec_features)

    # Average distance (excluding diagonal)
    n = len(recommended)
    total_distance = np.sum(distances) - np.trace(distances)  # Exclude diagonal
    avg_distance = total_distance / (n * (n - 1))

    return avg_distance

Cross-Validation for Recommenders

from sklearn.model_selection import KFold


def time_based_split(
    interactions: np.ndarray,
    timestamp_col: int,
    test_ratio: float = 0.2
):
    """
    Split based on time (realistic for recommendations).

    Train on past interactions, test on future.

    Args:
        interactions: Array with columns [user_id, item_id, timestamp, ...]
        timestamp_col: Column index containing timestamps
        test_ratio: Fraction of data for test set

    Returns:
        train_interactions, test_interactions
    """
    # Sort by timestamp
    sorted_interactions = interactions[interactions[:, timestamp_col].argsort()]

    # Split at time threshold
    split_idx = int(len(sorted_interactions) * (1 - test_ratio))
    train = sorted_interactions[:split_idx]
    test = sorted_interactions[split_idx:]

    logger.info(f"Time-based split: {len(train)} train, {len(test)} test")
    return train, test


def leave_one_out_cv(user_item_matrix: np.ndarray):
    """
    Leave-one-out cross-validation for each user.

    For each user, hold out one interaction for testing.

    Args:
        user_item_matrix: Matrix (n_users, n_items)

    Yields:
        (train_matrix, test_matrix) pairs
    """
    n_users, n_items = user_item_matrix.shape

    for user_id in range(n_users):
        # Get user's interactions
        user_interactions = user_item_matrix[user_id].nonzero()[0]

        if len(user_interactions) < 2:
            continue  # Need at least 2 interactions

        # Randomly select one to hold out
        test_item = np.random.choice(user_interactions)

        # Create train and test matrices
        train_matrix = user_item_matrix.copy()
        train_matrix[user_id, test_item] = 0

        test_matrix = np.zeros_like(user_item_matrix)
        test_matrix[user_id, test_item] = user_item_matrix[user_id, test_item]

        yield train_matrix, test_matrix

Statistical Significance Testing

from scipy import stats


def paired_t_test(
    metrics_a: List[float],
    metrics_b: List[float],
    alpha: float = 0.05
) -> Dict:
    """
    Paired t-test to compare two recommendation models.

    Tests if difference in metrics is statistically significant.

    Args:
        metrics_a: Metric values (e.g., NDCG@10) for model A, one per user
        metrics_b: Metric values for model B, same users
        alpha: Significance level

    Returns:
        Dict with test results

    Example:
        >>> ndcg_baseline = [0.7, 0.8, 0.75, 0.82, 0.79]
        >>> ndcg_new_model = [0.72, 0.83, 0.76, 0.85, 0.81]
        >>> result = paired_t_test(ndcg_baseline, ndcg_new_model)
        >>> print(result['significant'])  # True if new model is better
    """
    if len(metrics_a) != len(metrics_b):
        raise ValueError("Metric lists must have same length")

    # Compute differences
    differences = np.array(metrics_b) - np.array(metrics_a)

    # Perform paired t-test
    t_statistic, p_value = stats.ttest_rel(metrics_a, metrics_b)

    # Effect size (Cohen's d)
    mean_diff = np.mean(differences)
    std_diff = np.std(differences, ddof=1)
    cohens_d = mean_diff / std_diff if std_diff > 0 else 0.0

    return {
        'mean_improvement': mean_diff,
        'p_value': p_value,
        'significant': p_value < alpha,
        't_statistic': t_statistic,
        'cohens_d': cohens_d,
        'interpretation': 'significant' if p_value < alpha else 'not significant'
    }


def bootstrap_confidence_interval(
    metric_values: List[float],
    n_bootstrap: int = 1000,
    confidence: float = 0.95
) -> tuple:
    """
    Compute confidence interval for a metric using bootstrap.

    Args:
        metric_values: List of metric values
        n_bootstrap: Number of bootstrap samples
        confidence: Confidence level (e.g., 0.95 for 95% CI)

    Returns:
        (lower_bound, upper_bound)
    """
    bootstrap_means = []

    for _ in range(n_bootstrap):
        # Resample with replacement
        sample = np.random.choice(metric_values, size=len(metric_values), replace=True)
        bootstrap_means.append(np.mean(sample))

    # Compute percentiles
    alpha = 1 - confidence
    lower = np.percentile(bootstrap_means, alpha / 2 * 100)
    upper = np.percentile(bootstrap_means, (1 - alpha / 2) * 100)

    return lower, upper

Complete Evaluation Pipeline

class RecommenderEvaluator:
    """
    Comprehensive evaluation of recommendation models.
    """

    def __init__(self, k_values: List[int] = [5, 10, 20]):
        """
        Args:
            k_values: List of K values to evaluate at
        """
        self.k_values = k_values

    def evaluate(
        self,
        model,
        test_data: Dict[int, Set[int]],
        catalog_size: int
    ) -> Dict:
        """
        Evaluate model on test data.

        Args:
            model: Trained recommender model with recommend() method
            test_data: Dict[user_id -> set of relevant items]
            catalog_size: Total number of items

        Returns:
            Dict of metrics
        """
        results = defaultdict(dict)

        # Generate recommendations for all users
        recommendations = {}
        for user_id in test_data:
            try:
                recs = model.recommend(user_id, n=max(self.k_values))
                recommendations[user_id] = [item for item, score in recs]
            except Exception as e:
                logger.warning(f"Failed to generate recs for user {user_id}: {e}")
                continue

        # Compute metrics at each K
        for k in self.k_values:
            # Precision, Recall, F1
            precisions = []
            recalls = []

            for user_id, relevant_items in test_data.items():
                if user_id in recommendations:
                    prec = precision_at_k(recommendations[user_id], relevant_items, k)
                    rec = recall_at_k(recommendations[user_id], relevant_items, k)
                    precisions.append(prec)
                    recalls.append(rec)

            results[f'precision@{k}'] = np.mean(precisions) if precisions else 0.0
            results[f'recall@{k}'] = np.mean(recalls) if recalls else 0.0
            results[f'f1@{k}'] = (
                2 * results[f'precision@{k}'] * results[f'recall@{k}']
                / (results[f'precision@{k}'] + results[f'recall@{k}'])
                if (results[f'precision@{k}'] + results[f'recall@{k}']) > 0 else 0.0
            )

            # Coverage
            results[f'coverage@{k}'] = catalog_coverage(recommendations, catalog_size, k)

            # Diversity (Gini)
            results[f'gini@{k}'] = gini_coefficient(recommendations, k)

        logger.info("Evaluation complete:")
        for metric, value in results.items():
            logger.info(f"  {metric}: {value:.4f}")

        return dict(results)

Example Usage

# Evaluate model
evaluator = RecommenderEvaluator(k_values=[5, 10, 20])

# Test data: dict of user_id -> set of relevant items
test_data = {
    0: {10, 25, 42},
    1: {5, 17, 33},
    # ... more users
}

# Run evaluation
metrics = evaluator.evaluate(
    model=my_recommender,
    test_data=test_data,
    catalog_size=1000
)

# Compare two models statistically
baseline_ndcg = [...]  # NDCG@10 for each test user with baseline
new_model_ndcg = [...]  # NDCG@10 for each test user with new model

test_result = paired_t_test(baseline_ndcg, new_model_ndcg)
if test_result['significant']:
    print(f"New model is significantly better (p={test_result['p_value']:.4f})")

Matrix Factorization Methods

Complete implementations of SVD, ALS, and NMF for recommendation systems with hyperparameter tuning and implicit feedback handling.

Overview

Matrix factorization decomposes the user-item interaction matrix into lower-dimensional user and item latent factor matrices:

R (n_users × n_items) ≈ U (n_users × k) × V^T (k × n_items)

Advantages:

• Handles sparsity better than CF
• Discovers latent features
• More scalable than memory-based CF
• Works well with implicit feedback

Singular Value Decomposition (SVD)

Basic SVD Implementation

from typing import Optional, Tuple
import numpy as np
from scipy.sparse import csr_matrix
from scipy.sparse.linalg import svds
import logging

logger = logging.getLogger(__name__)


class SVDRecommender:
    """
    SVD-based recommender using truncated singular value decomposition.

    Decomposes user-item matrix into:
    R ≈ U × Σ × V^T

    Works best with explicit ratings data.
    """

    def __init__(
        self,
        n_factors: int = 50,
        normalize: bool = True,
        random_state: Optional[int] = 42
    ):
        """
        Initialize SVD recommender.

        Args:
            n_factors: Number of latent factors (rank of approximation)
            normalize: Whether to normalize by singular values
            random_state: Random seed for reproducibility
        """
        self.n_factors = n_factors
        self.normalize = normalize
        self.random_state = random_state

        self.user_factors: Optional[np.ndarray] = None
        self.item_factors: Optional[np.ndarray] = None
        self.singular_values: Optional[np.ndarray] = None
        self.global_mean: float = 0.0

    def fit(self, user_item_matrix: csr_matrix) -> 'SVDRecommender':
        """
        Fit SVD model to user-item matrix.

        Args:
            user_item_matrix: Sparse matrix (n_users, n_items)

        Returns:
            self for chaining
        """
        logger.info(f"Fitting SVD with {self.n_factors} factors")

        # Compute global mean (for mean-centering)
        self.global_mean = user_item_matrix.data.mean()

        # Mean-center the ratings
        centered_matrix = user_item_matrix.copy()
        centered_matrix.data -= self.global_mean

        # Compute truncated SVD
        # Note: svds returns in ascending order of singular values
        U, sigma, Vt = svds(
            centered_matrix,
            k=self.n_factors,
            random_state=self.random_state
        )

        # Reverse to get descending order
        U = U[:, ::-1]
        sigma = sigma[::-1]
        Vt = Vt[::-1, :]

        self.singular_values = sigma

        if self.normalize:
            # Distribute singular values into factors
            self.user_factors = U * np.sqrt(sigma)
            self.item_factors = (Vt.T * np.sqrt(sigma)).T
        else:
            # Keep singular values separate for numerical stability
            self.user_factors = U
            self.item_factors = Vt

        logger.info(f"SVD fit complete. Explained variance: "
                   f"{self._explained_variance():.2%}")

        return self

    def predict(self, user_id: int, item_id: int) -> float:
        """
        Predict rating for user-item pair.

        Args:
            user_id: User index
            item_id: Item index

        Returns:
            Predicted rating
        """
        if self.user_factors is None or self.item_factors is None:
            raise ValueError("Model not fitted")

        if self.normalize:
            prediction = np.dot(
                self.user_factors[user_id],
                self.item_factors[:, item_id]
            )
        else:
            prediction = np.dot(
                self.user_factors[user_id] * self.singular_values,
                self.item_factors[:, item_id]
            )

        # Add back global mean
        return prediction + self.global_mean

    def recommend(
        self,
        user_id: int,
        n: int = 10,
        exclude_interacted: Optional[np.ndarray] = None
    ) -> list:
        """
        Generate top-N recommendations for a user.

        Args:
            user_id: User index
            n: Number of recommendations
            exclude_interacted: Array of item indices to exclude

        Returns:
            List of (item_id, predicted_rating) tuples
        """
        if self.user_factors is None or self.item_factors is None:
            raise ValueError("Model not fitted")

        # Compute predictions for all items
        if self.normalize:
            scores = self.user_factors[user_id] @ self.item_factors
        else:
            scores = (self.user_factors[user_id] * self.singular_values) @ self.item_factors

        scores += self.global_mean

        # Exclude already interacted items
        if exclude_interacted is not None:
            scores[exclude_interacted] = -np.inf

        # Get top-N items
        top_items = np.argsort(scores)[-n:][::-1]
        return [(int(item), float(scores[item])) for item in top_items]

    def _explained_variance(self) -> float:
        """Calculate proportion of variance explained by the model."""
        if self.singular_values is None:
            return 0.0
        total_variance = np.sum(self.singular_values ** 2)
        explained = np.sum(self.singular_values[:self.n_factors] ** 2)
        return explained / total_variance if total_variance > 0 else 0.0

Alternating Least Squares (ALS)

ALS is better for implicit feedback and very large datasets (scales to billions of interactions).

from scipy.sparse import csr_matrix, diags
import numpy as np


class ALSRecommender:
    """
    Alternating Least Squares for implicit feedback.

    Optimizes: min ||C ⊙ (R - UV^T)||² + λ(||U||² + ||V||²)

    Where:
    - R: user-item confidence matrix
    - C: confidence weights
    - ⊙: element-wise product
    - λ: regularization parameter
    """

    def __init__(
        self,
        n_factors: int = 50,
        regularization: float = 0.01,
        iterations: int = 15,
        alpha: float = 40.0,
        random_state: Optional[int] = 42
    ):
        """
        Initialize ALS model.

        Args:
            n_factors: Number of latent factors
            regularization: L2 regularization strength
            iterations: Number of alternating iterations
            alpha: Confidence scaling factor for implicit feedback
            random_state: Random seed
        """
        self.n_factors = n_factors
        self.regularization = regularization
        self.iterations = iterations
        self.alpha = alpha
        self.random_state = random_state

        self.user_factors: Optional[np.ndarray] = None
        self.item_factors: Optional[np.ndarray] = None

    def fit(self, user_item_matrix: csr_matrix) -> 'ALSRecommender':
        """
        Fit ALS model using alternating least squares.

        Args:
            user_item_matrix: Sparse binary or count matrix (n_users, n_items)

        Returns:
            self for chaining
        """
        np.random.seed(self.random_state)

        n_users, n_items = user_item_matrix.shape
        logger.info(f"Fitting ALS: {n_users} users, {n_items} items, "
                   f"{self.iterations} iterations")

        # Initialize factors randomly
        self.user_factors = np.random.normal(
            scale=0.01, size=(n_users, self.n_factors)
        ).astype(np.float32)
        self.item_factors = np.random.normal(
            scale=0.01, size=(n_items, self.n_factors)
        ).astype(np.float32)

        # Confidence matrix: C = 1 + alpha * R
        confidence_matrix = user_item_matrix.copy()
        confidence_matrix.data = 1 + self.alpha * confidence_matrix.data

        # Preference matrix: P (binary)
        preference_matrix = user_item_matrix.copy()
        preference_matrix.data = np.ones_like(preference_matrix.data)

        # Alternating least squares
        for iteration in range(self.iterations):
            logger.info(f"ALS iteration {iteration + 1}/{self.iterations}")

            # Fix item factors, solve for user factors
            self.user_factors = self._solve_factors(
                self.item_factors,
                confidence_matrix.T,
                preference_matrix.T
            )

            # Fix user factors, solve for item factors
            self.item_factors = self._solve_factors(
                self.user_factors,
                confidence_matrix,
                preference_matrix
            )

            # Compute loss (optional, for monitoring)
            if (iteration + 1) % 5 == 0:
                loss = self._compute_loss(
                    user_item_matrix, confidence_matrix, preference_matrix
                )
                logger.info(f"  Loss: {loss:.4f}")

        return self

    def _solve_factors(
        self,
        fixed_factors: np.ndarray,
        confidence: csr_matrix,
        preference: csr_matrix
    ) -> np.ndarray:
        """
        Solve for one set of factors with the other held fixed.

        Args:
            fixed_factors: The fixed factor matrix
            confidence: Confidence matrix C
            preference: Preference matrix P

        Returns:
            Updated factor matrix
        """
        n_entities = confidence.shape[0]
        n_factors = fixed_factors.shape[1]
        new_factors = np.zeros((n_entities, n_factors), dtype=np.float32)

        # Precompute Y^T Y (fixed_factors^T @ fixed_factors)
        YtY = fixed_factors.T @ fixed_factors

        # Regularization term
        lambda_I = self.regularization * np.eye(n_factors, dtype=np.float32)

        # Solve for each entity (user or item)
        for entity_id in range(n_entities):
            # Get confidence and preference for this entity
            c_u = confidence[entity_id].toarray().flatten()  # Shape: (n_factors,)
            p_u = preference[entity_id].toarray().flatten()

            # Construct diagonal confidence matrix
            C_u = diags(c_u, format='csr')

            # Solve: (Y^T C_u Y + λI) x_u = Y^T C_u p_u
            # Where Y is fixed_factors, x_u is the factor we're solving for
            A = YtY + fixed_factors.T @ (C_u - diags(np.ones(len(c_u)))).toarray() @ fixed_factors + lambda_I
            b = fixed_factors.T @ (c_u * p_u)

            # Solve linear system
            new_factors[entity_id] = np.linalg.solve(A, b)

        return new_factors

    def _compute_loss(
        self,
        original_matrix: csr_matrix,
        confidence: csr_matrix,
        preference: csr_matrix
    ) -> float:
        """Compute weighted squared loss."""
        predictions = self.user_factors @ self.item_factors.T

        # Weighted loss: sum(C_ui * (P_ui - X_ui)²)
        loss = 0.0
        for u in range(original_matrix.shape[0]):
            items = original_matrix[u].indices
            c_u = confidence[u].data
            p_u = preference[u].data
            pred_u = predictions[u, items]

            loss += np.sum(c_u * (p_u - pred_u) ** 2)

        # Add regularization
        reg_loss = self.regularization * (
            np.sum(self.user_factors ** 2) + np.sum(self.item_factors ** 2)
        )

        return loss + reg_loss

    def recommend(self, user_id: int, n: int = 10, filter_interacted: bool = True) -> list:
        """Generate recommendations using dot product of factors."""
        if self.user_factors is None or self.item_factors is None:
            raise ValueError("Model not fitted")

        scores = self.user_factors[user_id] @ self.item_factors.T

        if filter_interacted:
            # Set interacted items to -inf
            # (assuming original matrix is available)
            pass

        top_items = np.argsort(scores)[-n:][::-1]
        return [(int(item), float(scores[item])) for item in top_items]

Non-Negative Matrix Factorization (NMF)

NMF enforces non-negativity constraints, leading to more interpretable factors.

from sklearn.decomposition import NMF


class NMFRecommender:
    """
    NMF for recommendation with non-negative constraints.

    Useful when factors should represent additive components
    (e.g., genre preferences, topic interests).
    """

    def __init__(
        self,
        n_factors: int = 50,
        init: str = 'nndsvd',
        max_iter: int = 200,
        random_state: Optional[int] = 42
    ):
        """
        Args:
            n_factors: Number of components
            init: Initialization method ('random', 'nndsvd', 'nndsvda')
            max_iter: Maximum iterations
            random_state: Random seed
        """
        self.nmf = NMF(
            n_components=n_factors,
            init=init,
            max_iter=max_iter,
            random_state=random_state
        )
        self.user_factors = None
        self.item_factors = None

    def fit(self, user_item_matrix):
        """Fit NMF model."""
        # Convert sparse to dense (NMF doesn't handle sparse well)
        dense_matrix = user_item_matrix.toarray() if hasattr(user_item_matrix, 'toarray') else user_item_matrix

        self.user_factors = self.nmf.fit_transform(dense_matrix)
        self.item_factors = self.nmf.components_

        logger.info(f"NMF reconstruction error: {self.nmf.reconstruction_err_:.4f}")
        return self

    def recommend(self, user_id: int, n: int = 10) -> list:
        """Generate recommendations."""
        scores = self.user_factors[user_id] @ self.item_factors
        top_items = np.argsort(scores)[-n:][::-1]
        return [(int(item), float(scores[item])) for item in top_items]

Hyperparameter Tuning

from sklearn.model_selection import GridSearchCV
from sklearn.metrics import mean_squared_error


def tune_svd_hyperparameters(train_matrix, val_matrix):
    """
    Tune number of factors and normalization.

    Returns:
        Best hyperparameters
    """
    param_grid = {
        'n_factors': [20, 50, 100, 200],
        'normalize': [True, False]
    }

    best_rmse = float('inf')
    best_params = None

    for n_factors in param_grid['n_factors']:
        for normalize in param_grid['normalize']:
            model = SVDRecommender(n_factors=n_factors, normalize=normalize)
            model.fit(train_matrix)

            # Evaluate on validation set
            predictions = []
            actuals = []
            for u, i in zip(*val_matrix.nonzero()):
                predictions.append(model.predict(u, i))
                actuals.append(val_matrix[u, i])

            rmse = np.sqrt(mean_squared_error(actuals, predictions))

            if rmse < best_rmse:
                best_rmse = rmse
                best_params = {'n_factors': n_factors, 'normalize': normalize}

    logger.info(f"Best params: {best_params}, RMSE: {best_rmse:.4f}")
    return best_params


def tune_als_hyperparameters(train_matrix):
    """Tune ALS regularization and number of factors."""
    param_grid = {
        'n_factors': [20, 50, 100],
        'regularization': [0.001, 0.01, 0.1],
        'alpha': [1, 10, 40]
    }

    # Grid search (simplified)
    best_model = None
    best_loss = float('inf')

    for n_factors in param_grid['n_factors']:
        for reg in param_grid['regularization']:
            for alpha in param_grid['alpha']:
                model = ALSRecommender(
                    n_factors=n_factors,
                    regularization=reg,
                    alpha=alpha,
                    iterations=10  # Fewer iterations for tuning
                )
                model.fit(train_matrix)

                # Compute confidence matrix: C = 1 + alpha * R
                confidence_matrix = train_matrix.copy()
                confidence_matrix.data = 1 + model.alpha * confidence_matrix.data

                # Compute preference matrix: P (binary indicator)
                preference_matrix = train_matrix.copy()
                preference_matrix.data = np.ones_like(preference_matrix.data)

                # Now call with proper matrices
                loss = model._compute_loss(train_matrix, confidence_matrix, preference_matrix)
                if loss < best_loss:
                    best_loss = loss
                    best_model = model

    return best_model

Handling Implicit Feedback

Implicit feedback (clicks, views) requires different treatment than explicit ratings:

def convert_implicit_to_confidence(interaction_matrix, alpha=40):
    """
    Convert implicit feedback to confidence matrix.

    C_ui = 1 + alpha * r_ui

    Where r_ui is the interaction count (views, clicks, etc.)
    """
    confidence = interaction_matrix.copy()
    confidence.data = 1 + alpha * np.log(1 + confidence.data)
    return confidence

Advanced Techniques

BPR (Bayesian Personalized Ranking)

For implicit feedback, optimizes pairwise ranking:

# User u should prefer item i over item j
# max P(i >_u j) = sigmoid(x_ui - x_uj)

WARP (Weighted Approximate-Rank Pairwise)

Optimizes top-k ranking directly (available in LightFM library).

When to Use Each Method

SVD:

• ✅ Explicit ratings (1-5 stars)
• ✅ Need fast inference
• ✅ Moderate sparsity (<99%)
• ❌ Implicit feedback (use ALS)

ALS:

• ✅ Implicit feedback (clicks, views)
• ✅ Very large scale (billions of interactions)
• ✅ High sparsity (>99%)
• ✅ Need interpretable latent factors
• ❌ Explicit ratings with wide range

NMF:

• ✅ Need interpretable non-negative factors
• ✅ Topic modeling style decomposition
• ✅ Smaller datasets (dense computation)
• ❌ Very large or very sparse data