Collaborative Filtering

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

import numpy as np
from collections import defaultdict
from typing import Dict, List, Tuple, Optional
 
class UserBasedCF:
    """
    Production-quality User-based Collaborative Filtering implementation.
    
    Supports:
    - Pearson correlation and cosine similarity
    - Configurable neighborhood size
    - Significance weighting for sparse overlaps
    - Efficient similarity caching
    """
    
    def __init__(
        self, 
        k_neighbors: int = 50,
        similarity_metric: str = "pearson",
        min_common_items: int = 3,
        significance_weight: bool = True,
        significance_threshold: int = 50
    ):
        self.k = k_neighbors
        self.similarity_metric = similarity_metric
        self.min_common_items = min_common_items
        self.use_significance_weight = significance_weight
        self.significance_threshold = significance_threshold
        
        # Data structures populated on fit()
        self.ratings: Dict[int, Dict[int, float]] = {}  # user -> {item: rating}
        self.user_means: Dict[int, float] = {}
        self.item_users: Dict[int, set] = defaultdict(set)  # item -> set of users
        self.similarity_cache: Dict[Tuple[int, int], float] = {}
    
    def fit(self, ratings_data: List[Tuple[int, int, float]]) -> 'UserBasedCF':
        """
        Fit the model with rating data.
        
        Args:
            ratings_data: List of (user_id, item_id, rating) tuples
        """
        # Build user-item rating dict
        for user_id, item_id, rating in ratings_data:
            if user_id not in self.ratings:
                self.ratings[user_id] = {}
            self.ratings[user_id][item_id] = rating
            self.item_users[item_id].add(user_id)
        
        # Compute user means
        for user_id, user_ratings in self.ratings.items():
            self.user_means[user_id] = np.mean(list(user_ratings.values()))
        
        return self
    
    def _pearson_correlation(self, user_u: int, user_v: int) -> Optional[float]:
        """Compute Pearson correlation between two users."""
        # Find co-rated items
        items_u = set(self.ratings[user_u].keys())
        items_v = set(self.ratings[user_v].keys())
        common_items = items_u & items_v
        
        if len(common_items) < self.min_common_items:
            return None
        
        mean_u = self.user_means[user_u]
        mean_v = self.user_means[user_v]
        
        # Compute correlation components
        numerator = 0.0
        sum_sq_u = 0.0
        sum_sq_v = 0.0
        
        for item in common_items:
            dev_u = self.ratings[user_u][item] - mean_u
            dev_v = self.ratings[user_v][item] - mean_v
            numerator += dev_u * dev_v
            sum_sq_u += dev_u ** 2
            sum_sq_v += dev_v ** 2
        
        denominator = np.sqrt(sum_sq_u) * np.sqrt(sum_sq_v)
        
        if denominator == 0:
            return 0.0
        
        correlation = numerator / denominator
        
        # Apply significance weighting
        if self.use_significance_weight:
            n = len(common_items)
            if n < self.significance_threshold:
                correlation *= (n / self.significance_threshold)
        
        return correlation
    
    def _get_similarity(self, user_u: int, user_v: int) -> Optional[float]:
        """Get similarity with caching."""
        cache_key = (min(user_u, user_v), max(user_u, user_v))
        
        if cache_key not in self.similarity_cache:
            if self.similarity_metric == "pearson":
                sim = self._pearson_correlation(user_u, user_v)
            else:
                raise ValueError(f"Unknown similarity metric: {self.similarity_metric}")
            self.similarity_cache[cache_key] = sim
        
        return self.similarity_cache[cache_key]
    
    def predict(self, user_id: int, item_id: int) -> float:
        """
        Predict rating for a user-item pair.
        
        Returns the predicted rating or user's mean if prediction is not possible.
        """
        if user_id not in self.ratings:
            return 3.0  # Global default for unknown users
        
        # Get users who rated this item
        candidate_neighbors = self.item_users.get(item_id, set()) - {user_id}
        
        if not candidate_neighbors:
            return self.user_means[user_id]
        
        # Compute similarities to all candidates
        similarities = []
        for neighbor_id in candidate_neighbors:
            sim = self._get_similarity(user_id, neighbor_id)
            if sim is not None:
                similarities.append((neighbor_id, sim))
        
        if not similarities:
            return self.user_means[user_id]
        
        # Select top-k neighbors by absolute similarity
        similarities.sort(key=lambda x: abs(x[1]), reverse=True)
        top_k = similarities[:self.k]
        
        # Compute weighted prediction
        numerator = 0.0
        denominator = 0.0
        
        for neighbor_id, sim in top_k:
            neighbor_rating = self.ratings[neighbor_id][item_id]
            neighbor_mean = self.user_means[neighbor_id]
            deviation = neighbor_rating - neighbor_mean
            
            numerator += sim * deviation
            denominator += abs(sim)
        
        if denominator == 0:
            return self.user_means[user_id]
        
        prediction = self.user_means[user_id] + (numerator / denominator)
        
        # Clamp to valid rating range
        return max(1.0, min(5.0, prediction))
    
    def recommend(
        self, 
        user_id: int, 
        n_recommendations: int = 10,
        exclude_rated: bool = True
    ) -> List[Tuple[int, float]]:
        """
        Generate top-N recommendations for a user.
        
        Returns list of (item_id, predicted_rating) tuples, sorted by predicted rating.
        """
        if user_id not in self.ratings:
            return []
        
        rated_items = set(self.ratings[user_id].keys()) if exclude_rated else set()
        candidate_items = set(self.item_users.keys()) - rated_items
        
        # Predict ratings for all candidates
        predictions = []
        for item_id in candidate_items:
            pred = self.predict(user_id, item_id)
            predictions.append((item_id, pred))
        
        # Sort by predicted rating descending
        predictions.sort(key=lambda x: x[1], reverse=True)
        
        return predictions[:n_recommendations]
 
 
# Example usage
if __name__ == "__main__":
    # Sample rating data: (user_id, item_id, rating)
    ratings = [
        (1, 1, 5), (1, 2, 3), (1, 3, 4), (1, 5, 1),
        (2, 1, 4), (2, 3, 5), (2, 4, 4), (2, 5, 2),
        (3, 2, 1), (3, 3, 2), (3, 4, 4), (3, 5, 5),
        (4, 1, 5), (4, 2, 4), (4, 3, 4), (4, 4, 3),
        (5, 1, 1), (5, 2, 2), (5, 4, 5), (5, 5, 4),
    ]
    
    cf = UserBasedCF(k_neighbors=3, min_common_items=2)
    cf.fit(ratings)
    
    # Predict Alice's (user 1) rating for item 4
    prediction = cf.predict(user_id=1, item_id=4)
    print(f"Predicted rating for User 1, Item 4: {prediction:.2f}")
    
    # Get recommendations for Alice
    recommendations = cf.recommend(user_id=1, n_recommendations=3)
    print(f"Top recommendations for User 1: {recommendations}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

from sklearn.decomposition import TruncatedSVD
from sklearn.neighbors import NearestNeighbors
import numpy as np
from scipy.sparse import csr_matrix
 
class ScalableUserCF:
    """
    Scalable user-based CF using dimensionality reduction
    and approximate nearest neighbor search.
    """
    
    def __init__(
        self, 
        n_neighbors: int = 50,
        n_components: int = 100,
        metric: str = "cosine"
    ):
        self.k = n_neighbors
        self.n_components = n_components
        self.metric = metric
        
        self.svd = TruncatedSVD(n_components=n_components, random_state=42)
        self.nn_model = NearestNeighbors(
            n_neighbors=n_neighbors + 1,  # +1 because user is their own neighbor
            metric=metric,
            algorithm='ball_tree'  # Efficient for moderate dimensions
        )
        
        self.user_factors = None
        self.user_ids = None
        self.user_id_to_idx = {}
        self.ratings_matrix = None
        self.user_means = None
    
    def fit(self, ratings_matrix: csr_matrix, user_ids: np.ndarray):
        """
        Fit the model using sparse ratings matrix.
        
        Args:
            ratings_matrix: Sparse m x n matrix of user-item ratings
            user_ids: Array of user IDs corresponding to matrix rows
        """
        self.ratings_matrix = ratings_matrix
        self.user_ids = user_ids
        self.user_id_to_idx = {uid: idx for idx, uid in enumerate(user_ids)}
        
        # Compute user means (for non-zero entries only)
        row_sums = np.array(ratings_matrix.sum(axis=1)).flatten()
        row_counts = np.array((ratings_matrix != 0).sum(axis=1)).flatten()
        self.user_means = np.divide(
            row_sums, row_counts, 
            out=np.full_like(row_sums, 3.0),  # Default mean
            where=row_counts != 0
        )
        
        # Reduce dimensionality
        # Mean-center before SVD for better representation
        centered = ratings_matrix.copy()
        # Note: Full mean-centering of sparse matrix requires care
        # Here we use SVD directly on ratings for simplicity
        
        self.user_factors = self.svd.fit_transform(ratings_matrix)
        
        # Build approximate nearest neighbor index
        self.nn_model.fit(self.user_factors)
        
        return self
    
    def find_neighbors(self, user_idx: int) -> tuple:
        """Find k nearest neighbors for a user."""
        user_vector = self.user_factors[user_idx].reshape(1, -1)
        distances, indices = self.nn_model.kneighbors(user_vector)
        
        # Remove self from neighbors
        mask = indices[0] != user_idx
        neighbor_indices = indices[0][mask][:self.k]
        neighbor_distances = distances[0][mask][:self.k]
        
        # Convert distances to similarities
        # For cosine distance: similarity = 1 - distance
        similarities = 1 - neighbor_distances
        
        return neighbor_indices, similarities
    
    def predict(self, user_id: int, item_idx: int) -> float:
        """Predict rating for user-item pair."""
        if user_id not in self.user_id_to_idx:
            return 3.0  # Unknown user
        
        user_idx = self.user_id_to_idx[user_id]
        neighbor_indices, similarities = self.find_neighbors(user_idx)
        
        # Get neighbors who rated this item
        neighbor_ratings = self.ratings_matrix[neighbor_indices, item_idx].toarray().flatten()
        rated_mask = neighbor_ratings != 0
        
        if not np.any(rated_mask):
            return self.user_means[user_idx]
        
        # Filter to neighbors who rated the item
        valid_similarities = similarities[rated_mask]
        valid_ratings = neighbor_ratings[rated_mask]
        valid_indices = neighbor_indices[rated_mask]
        valid_means = self.user_means[valid_indices]
        
        # Compute prediction with mean-centering
        deviations = valid_ratings - valid_means
        numerator = np.dot(valid_similarities, deviations)
        denominator = np.sum(np.abs(valid_similarities))
        
        if denominator == 0:
            return self.user_means[user_idx]
        
        prediction = self.user_means[user_idx] + numerator / denominator
        return np.clip(prediction, 1.0, 5.0)
 
 
# Complexity comparison
print("Complexity Analysis:")
print("─" * 50)
print("Naive User-based CF:")
print("  - Similarity computation: O(m² · n) precompute")
print("  - Prediction: O(m) per query")
print("  - Space: O(m²) for similarity matrix")
print()
print("Scalable User-based CF (this implementation):")
print("  - SVD: O(m · n · k) one-time")
print("  - NN Index: O(m · k · log m) one-time")
print("  - Prediction: O(k · log m) per query")
print("  - Space: O(m · k) for user factors")