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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277

import numpy as np
from collections import defaultdict
from typing import Dict, List, Tuple, Set, Optional
from dataclasses import dataclass
import heapq
 
@dataclass
class ItemNeighbors:
    """Precomputed neighbors for an item."""
    item_id: int
    neighbors: List[Tuple[int, float]]  # [(neighbor_id, similarity), ...]
 
 
class ItemBasedCF:
    """
    Production-quality Item-based Collaborative Filtering.
    
    Key features:
    - Adjusted cosine similarity (handles user rating bias)
    - Offline precomputation of item similarities
    - Efficient online prediction and recommendation
    - Configurable neighborhood size and minimum overlap
    """
    
    def __init__(
        self,
        k_neighbors: int = 50,
        min_common_users: int = 3,
        shrinkage: float = 100.0
    ):
        """
        Args:
            k_neighbors: Number of similar items to store per item
            min_common_users: Minimum users who rated both items
            shrinkage: Regularization to shrink similarities with few co-raters
        """
        self.k = k_neighbors
        self.min_common_users = min_common_users
        self.shrinkage = shrinkage
        
        # Data structures
        self.user_ratings: Dict[int, Dict[int, float]] = defaultdict(dict)
        self.item_ratings: Dict[int, Dict[int, float]] = defaultdict(dict)
        self.user_means: Dict[int, float] = {}
        self.item_neighbors: Dict[int, ItemNeighbors] = {}
        
    def fit(self, ratings: List[Tuple[int, int, float]]) -> 'ItemBasedCF':
        """
        Fit model: build indices and compute item similarities.
        
        Args:
            ratings: List of (user_id, item_id, rating) tuples
        """
        print("Building rating indices...")
        self._build_indices(ratings)
        
        print("Computing user means...")
        self._compute_user_means()
        
        print("Computing item-item similarities...")
        self._compute_all_similarities()
        
        return self
    
    def _build_indices(self, ratings: List[Tuple[int, int, float]]):
        """Build user->items and item->users indices."""
        for user_id, item_id, rating in ratings:
            self.user_ratings[user_id][item_id] = rating
            self.item_ratings[item_id][user_id] = rating
    
    def _compute_user_means(self):
        """Compute mean rating for each user."""
        for user_id, ratings in self.user_ratings.items():
            self.user_means[user_id] = np.mean(list(ratings.values()))
    
    def _adjusted_cosine(self, item_i: int, item_j: int) -> Optional[float]:
        """
        Compute adjusted cosine similarity between two items.
        
        Returns None if insufficient overlap.
        """
        users_i = set(self.item_ratings[item_i].keys())
        users_j = set(self.item_ratings[item_j].keys())
        common_users = users_i & users_j
        
        if len(common_users) < self.min_common_users:
            return None
        
        # Compute adjusted cosine components
        numerator = 0.0
        sum_sq_i = 0.0
        sum_sq_j = 0.0
        
        for user in common_users:
            r_ui = self.item_ratings[item_i][user]
            r_uj = self.item_ratings[item_j][user]
            mean_u = self.user_means[user]
            
            adj_i = r_ui - mean_u
            adj_j = r_uj - mean_u
            
            numerator += adj_i * adj_j
            sum_sq_i += adj_i ** 2
            sum_sq_j += adj_j ** 2
        
        denominator = np.sqrt(sum_sq_i) * np.sqrt(sum_sq_j)
        
        if denominator == 0:
            return 0.0
        
        similarity = numerator / denominator
        
        # Apply shrinkage regularization
        n_common = len(common_users)
        similarity *= n_common / (n_common + self.shrinkage)
        
        return similarity
    
    def _compute_all_similarities(self):
        """Compute and store top-k neighbors for each item."""
        items = list(self.item_ratings.keys())
        n_items = len(items)
        
        for idx, item_i in enumerate(items):
            if (idx + 1) % 1000 == 0:
                print(f"  Processed {idx + 1}/{n_items} items...")
            
            # Compute similarities to all other items
            similarities = []
            for item_j in items:
                if item_i == item_j:
                    continue
                sim = self._adjusted_cosine(item_i, item_j)
                if sim is not None and sim > 0:  # Only keep positive similarities
                    similarities.append((item_j, sim))
            
            # Keep top-k by similarity
            top_k = heapq.nlargest(self.k, similarities, key=lambda x: x[1])
            
            self.item_neighbors[item_i] = ItemNeighbors(
                item_id=item_i,
                neighbors=top_k
            )
    
    def predict(self, user_id: int, item_id: int) -> float:
        """
        Predict rating for a user-item pair.
        
        Uses deviation-based prediction:
        r̂_ui = r̄_u + Σ sim(i,j) * (r_uj - r̄_u) / Σ |sim(i,j)|
        """
        # Unknown user
        if user_id not in self.user_ratings:
            return 3.0
        
        # User already rated this item
        if item_id in self.user_ratings[user_id]:
            return self.user_ratings[user_id][item_id]
        
        # Unknown item (not in precomputed similarities)
        if item_id not in self.item_neighbors:
            return self.user_means.get(user_id, 3.0)
        
        user_mean = self.user_means[user_id]
        user_items = self.user_ratings[user_id]
        
        # Find intersection of item's neighbors and user's rated items
        neighbors = self.item_neighbors[item_id].neighbors
        
        numerator = 0.0
        denominator = 0.0
        
        for neighbor_id, sim in neighbors:
            if neighbor_id in user_items:
                user_rating = user_items[neighbor_id]
                deviation = user_rating - user_mean
                numerator += sim * deviation
                denominator += abs(sim)
        
        if denominator == 0:
            return user_mean
        
        prediction = user_mean + numerator / denominator
        return float(np.clip(prediction, 1.0, 5.0))
    
    def recommend(
        self,
        user_id: int,
        n: int = 10,
        exclude_rated: bool = True
    ) -> List[Tuple[int, float]]:
        """
        Generate top-N recommendations for a user.
        
        Returns list of (item_id, predicted_rating) sorted by rating.
        """
        if user_id not in self.user_ratings:
            return []
        
        user_items = set(self.user_ratings[user_id].keys())
        
        # Candidate items: all items with precomputed similarities
        # that the user hasn't rated
        candidates = set(self.item_neighbors.keys())
        if exclude_rated:
            candidates -= user_items
        
        # Predict ratings for all candidates
        predictions = []
        for item_id in candidates:
            pred = self.predict(user_id, item_id)
            predictions.append((item_id, pred))
        
        # Sort by predicted rating
        predictions.sort(key=lambda x: x[1], reverse=True)
        
        return predictions[:n]
    
    def similar_items(self, item_id: int, n: int = 10) -> List[Tuple[int, float]]:
        """Get most similar items to a given item."""
        if item_id not in self.item_neighbors:
            return []
        return self.item_neighbors[item_id].neighbors[:n]
 
 
# Example usage and evaluation
if __name__ == "__main__":
    # Sample data: (user_id, item_id, rating)
    ratings = [
        # User 1: Likes sci-fi (items 1, 2, 3)
        (1, 1, 5), (1, 2, 4), (1, 3, 5), (1, 6, 2),
        # User 2: Similar to user 1
        (2, 1, 4), (2, 2, 5), (2, 3, 4), (2, 4, 3), (2, 6, 1),
        # User 3: Likes romance (items 4, 5, 6)  
        (3, 4, 5), (3, 5, 5), (3, 6, 4), (3, 1, 2),
        # User 4: Similar to user 3
        (4, 4, 4), (4, 5, 5), (4, 6, 5), (4, 2, 2),
        # User 5: Mixed tastes
        (5, 1, 4), (5, 3, 3), (5, 5, 4), (5, 6, 3),
        # User 6: Provides more ratings for items
        (6, 1, 5), (6, 2, 5), (6, 3, 4), (6, 4, 2), (6, 5, 2),
        # User 7: More data
        (7, 2, 4), (7, 3, 5), (7, 4, 3), (7, 5, 4), (7, 6, 3),
    ]
    
    # Create and fit model
    cf = ItemBasedCF(k_neighbors=5, min_common_users=2)
    cf.fit(ratings)
    
    print("
" + "="*50)
    print("Item-based CF Results")
    print("="*50)
    
    # Show similar items
    print("
Items similar to Item 1 (sci-fi):")
    for item_id, sim in cf.similar_items(1):
        print(f"  Item {item_id}: similarity = {sim:.3f}")
    
    print("
Items similar to Item 4 (romance):")
    for item_id, sim in cf.similar_items(4):
        print(f"  Item {item_id}: similarity = {sim:.3f}")
    
    # Predictions
    print("
Predictions for User 1:")
    for item_id in [4, 5]:
        pred = cf.predict(1, item_id)
        print(f"  Item {item_id}: predicted rating = {pred:.2f}")
    
    # Recommendations
    print("
Top recommendations for User 1:")
    for item_id, score in cf.recommend(1, n=5):
        print(f"  Item {item_id}: predicted = {score:.2f}")

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

import numpy as np
from scipy.sparse import csr_matrix
from sklearn.decomposition import TruncatedSVD
from sklearn.metrics.pairwise import cosine_similarity
from typing import Dict, List, Tuple
import heapq
 
class EfficientItemSimilarity:
    """
    Efficient item similarity computation using dimensionality reduction.
    
    Strategy:
    1. Reduce m-dimensional rating vectors to k-dimensional factors
    2. Compute similarities in reduced space
    3. Use sparse operations throughout
    """
    
    def __init__(
        self,
        n_factors: int = 100,
        k_neighbors: int = 50
    ):
        self.n_factors = n_factors
        self.k = k_neighbors
        self.item_factors: np.ndarray = None
        self.item_ids: List[int] = None
        self.item_id_to_idx: Dict[int, int] = {}
        self.neighbors: Dict[int, List[Tuple[int, float]]] = {}
    
    def fit(self, ratings_matrix: csr_matrix, item_ids: List[int]):
        """
        Compute similarities via dimensionality reduction.
        
        Args:
            ratings_matrix: Sparse n_items × n_users matrix
            item_ids: List of item IDs corresponding to rows
        """
        self.item_ids = item_ids
        self.item_id_to_idx = {iid: idx for idx, iid in enumerate(item_ids)}
        
        print(f"Reducing to {self.n_factors} dimensions...")
        
        # Mean-center rows (items) for adjusted cosine
        # Note: Full mean-centering of sparse matrix is complex
        # SVD on raw matrix approximates this
        svd = TruncatedSVD(n_components=self.n_factors, random_state=42)
        self.item_factors = svd.fit_transform(ratings_matrix)
        
        print(f"Explained variance: {svd.explained_variance_ratio_.sum():.2%}")
        print("Computing neighbors...")
        
        # Compute all pairwise similarities in reduced space
        # This is O(n² · k) but k << m
        similarities = cosine_similarity(self.item_factors)
        
        # Extract top-k neighbors for each item
        for idx, item_id in enumerate(item_ids):
            # Get similarities for this item
            item_sims = similarities[idx]
            
            # Find top-k (excluding self)
            top_indices = np.argpartition(item_sims, -self.k-1)[-self.k-1:]
            top_indices = top_indices[top_indices != idx][:self.k]
            
            neighbors = [
                (item_ids[j], float(item_sims[j]))
                for j in top_indices
            ]
            neighbors.sort(key=lambda x: x[1], reverse=True)
            
            self.neighbors[item_id] = neighbors
        
        return self
    
    def get_neighbors(self, item_id: int) -> List[Tuple[int, float]]:
        """Get precomputed neighbors for an item."""
        return self.neighbors.get(item_id, [])
 
 
# Complexity analysis
print("Complexity Comparison")
print("=" * 60)
print()
print("Naive approach:")
print("  • Similarity computation: O(n² · m)")
print("  • For n=1M items, m=100M users: ~10^20 operations")
print()
print("Dimensionality reduction approach:")
print("  • SVD: O(n · m · k) where k = 100")
print("  • Similarity: O(n² · k)")
print("  • For n=1M items, k=100: ~10^14 operations")
print("  • Speedup: ~1,000,000x")
print()
print("Additional optimizations:")
print("  • Block-wise similarity computation (memory efficient)")
print("  • GPU acceleration (cosine similarity is parallelizable)")
print("  • Approximate nearest neighbors (FAISS, Annoy)")