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

import numpy as np
from typing import Dict, Optional, Tuple
 
def pearson_correlation(
    ratings_u: Dict[int, float],
    ratings_v: Dict[int, float],
    min_overlap: int = 3
) -> Optional[float]:
    """
    Compute Pearson correlation between two users.
    
    Args:
        ratings_u: {item_id: rating} for user u
        ratings_v: {item_id: rating} for user v
        min_overlap: Minimum co-rated items required
    
    Returns:
        Pearson correlation in [-1, 1], or None if insufficient overlap
    """
    # Find co-rated items
    common = set(ratings_u.keys()) & set(ratings_v.keys())
    
    if len(common) < min_overlap:
        return None
    
    # Extract ratings for common items
    r_u = np.array([ratings_u[i] for i in common])
    r_v = np.array([ratings_v[i] for i in common])
    
    # Mean-center
    mean_u = np.mean(r_u)
    mean_v = np.mean(r_v)
    
    centered_u = r_u - mean_u
    centered_v = r_v - mean_v
    
    # Compute correlation
    numerator = np.dot(centered_u, centered_v)
    denom = np.sqrt(np.sum(centered_u**2)) * np.sqrt(np.sum(centered_v**2))
    
    if denom == 0:
        return 0.0
    
    return float(numerator / denom)
 
 
def cosine_similarity(
    ratings_u: Dict[int, float],
    ratings_v: Dict[int, float]
) -> float:
    """
    Compute cosine similarity between two users.
    
    Uses sparse representation - only considers rated items.
    """
    common = set(ratings_u.keys()) & set(ratings_v.keys())
    
    if not common:
        return 0.0
    
    # Numerator: dot product of common ratings
    dot = sum(ratings_u[i] * ratings_v[i] for i in common)
    
    # Denominators: norms of full rating vectors
    norm_u = np.sqrt(sum(r**2 for r in ratings_u.values()))
    norm_v = np.sqrt(sum(r**2 for r in ratings_v.values()))
    
    if norm_u == 0 or norm_v == 0:
        return 0.0
    
    return float(dot / (norm_u * norm_v))
 
 
def adjusted_cosine_similarity(
    ratings_u: Dict[int, float],
    ratings_v: Dict[int, float],
    user_mean_u: float,
    user_mean_v: float
) -> float:
    """
    Compute adjusted cosine similarity (for item-based CF).
    
    Subtracts user means to handle rating bias.
    """
    common = set(ratings_u.keys()) & set(ratings_v.keys())
    
    if not common:
        return 0.0
    
    # Centered ratings
    centered_u = {i: ratings_u[i] - user_mean_u for i in ratings_u}
    centered_v = {i: ratings_v[i] - user_mean_v for i in ratings_v}
    
    # Numerator: dot product of common centered ratings
    dot = sum(centered_u[i] * centered_v[i] for i in common)
    
    # Denominators: norms of full centered vectors
    norm_u = np.sqrt(sum(r**2 for r in centered_u.values()))
    norm_v = np.sqrt(sum(r**2 for r in centered_v.values()))
    
    if norm_u == 0 or norm_v == 0:
        return 0.0
    
    return float(dot / (norm_u * norm_v))
 
 
def jaccard_similarity(
    items_u: set,
    items_v: set
) -> float:
    """
    Compute Jaccard similarity between item sets.
    
    Used for binary/implicit feedback where only 'liked' items are known.
    """
    if not items_u or not items_v:
        return 0.0
    
    intersection = len(items_u & items_v)
    union = len(items_u | items_v)
    
    return intersection / union
 
 
def constrained_pearson(
    ratings_u: Dict[int, float],
    ratings_v: Dict[int, float],
    neutral_rating: float = 3.0,
    min_overlap: int = 3
) -> Optional[float]:
    """
    Constrained Pearson correlation.
    
    Centers around a fixed neutral value (e.g., 3 on a 1-5 scale)
    rather than user means. Useful when rating scale has inherent meaning.
    """
    common = set(ratings_u.keys()) & set(ratings_v.keys())
    
    if len(common) < min_overlap:
        return None
    
    numerator = sum(
        (ratings_u[i] - neutral_rating) * (ratings_v[i] - neutral_rating)
        for i in common
    )
    
    sum_sq_u = sum((ratings_u[i] - neutral_rating)**2 for i in common)
    sum_sq_v = sum((ratings_v[i] - neutral_rating)**2 for i in common)
    
    denom = np.sqrt(sum_sq_u) * np.sqrt(sum_sq_v)
    
    if denom == 0:
        return 0.0
    
    return float(numerator / denom)
 
 
# Example comparison
if __name__ == "__main__":
    # Two users with different rating scales but similar preferences
    user_a = {1: 5, 2: 4, 3: 3, 4: 5, 5: 4}  # Generous rater (mean ~4.2)
    user_b = {1: 3, 2: 2, 3: 1, 4: 3, 5: 2}  # Harsh rater (mean ~2.2)
    
    print("Comparing similarity measures:")
    print(f"  Pearson:          {pearson_correlation(user_a, user_b):.4f}")
    print(f"  Cosine:           {cosine_similarity(user_a, user_b):.4f}")
    print(f"  Adjusted Cosine:  {adjusted_cosine_similarity(user_a, user_b, 4.2, 2.2):.4f}")
    
    # Despite different absolute ratings, users have identical relative preferences
    # Pearson and adjusted cosine should show this (high similarity)
    # Raw cosine will be lower because of absolute value differences

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

from collections import defaultdict
from typing import Dict, Set, List, Tuple
import numpy as np
from scipy.sparse import csr_matrix
 
class EfficientSimilarityComputer:
    """
    Efficient similarity computation using inverted indices.
    
    Key optimizations:
    - Inverted index for fast co-rated item/user lookup
    - Early termination for insufficient overlap
    - Sparse matrix operations
    """
    
    def __init__(self, min_overlap: int = 3):
        self.min_overlap = min_overlap
        
        # Inverted indices
        self.item_to_users: Dict[int, Set[int]] = defaultdict(set)
        self.user_to_items: Dict[int, Set[int]] = defaultdict(set)
        
        # Rating data
        self.ratings: Dict[Tuple[int, int], float] = {}
        self.user_means: Dict[int, float] = {}
    
    def build_indices(self, ratings: List[Tuple[int, int, float]]):
        """Build inverted indices from rating data."""
        user_sums = defaultdict(float)
        user_counts = defaultdict(int)
        
        for user_id, item_id, rating in ratings:
            self.item_to_users[item_id].add(user_id)
            self.user_to_items[user_id].add(item_id)
            self.ratings[(user_id, item_id)] = rating
            
            user_sums[user_id] += rating
            user_counts[user_id] += 1
        
        self.user_means = {
            u: user_sums[u] / user_counts[u]
            for u in user_counts
        }
    
    def find_candidate_pairs(
        self, 
        target_user: int,
        max_candidates: int = 10000
    ) -> List[int]:
        """
        Find users who share items with target user.
        
        Uses inverted index for efficient lookup.
        """
        candidate_overlap: Dict[int, int] = defaultdict(int)
        
        # For each item the target user rated
        for item_id in self.user_to_items[target_user]:
            # Find other users who rated this item
            for other_user in self.item_to_users[item_id]:
                if other_user != target_user:
                    candidate_overlap[other_user] += 1
        
        # Filter by minimum overlap
        candidates = [
            user_id for user_id, overlap in candidate_overlap.items()
            if overlap >= self.min_overlap
        ]
        
        # Sort by overlap (highest first) and truncate
        candidates.sort(key=lambda u: candidate_overlap[u], reverse=True)
        return candidates[:max_candidates]
    
    def pearson_pair(self, user_u: int, user_v: int) -> float:
        """Compute Pearson correlation for a user pair."""
        # Get co-rated items (already filtered by inverted index)
        common = self.user_to_items[user_u] & self.user_to_items[user_v]
        
        if len(common) < self.min_overlap:
            return 0.0
        
        mean_u = self.user_means[user_u]
        mean_v = self.user_means[user_v]
        
        numerator = 0.0
        sum_sq_u = 0.0
        sum_sq_v = 0.0
        
        for item_id in common:
            r_u = self.ratings[(user_u, item_id)] - mean_u
            r_v = self.ratings[(user_v, item_id)] - mean_v
            
            numerator += r_u * r_v
            sum_sq_u += r_u ** 2
            sum_sq_v += r_v ** 2
        
        denom = np.sqrt(sum_sq_u) * np.sqrt(sum_sq_v)
        
        return 0.0 if denom == 0 else float(numerator / denom)
    
    def compute_neighbors(
        self, 
        target_user: int, 
        k: int = 50
    ) -> List[Tuple[int, float]]:
        """
        Compute k-nearest neighbors for a user.
        
        Returns: List of (user_id, similarity) tuples.
        """
        candidates = self.find_candidate_pairs(target_user)
        
        similarities = [
            (user_id, self.pearson_pair(target_user, user_id))
            for user_id in candidates
        ]
        
        # Sort by absolute similarity (descending)
        similarities.sort(key=lambda x: abs(x[1]), reverse=True)
        
        return similarities[:k]
 
 
# Complexity analysis
print("Without inverted index:")
print("  Finding co-raters: O(m) scan per user pair")
print("  All pairs: O(m² · n) total")
print()
print("With inverted index:")
print("  Finding candidates: O(|I_u| · avg_item_popularity)")
print("  All neighbors for one user: O(|I_u| · p · n')")
print("    where p = avg popularity, n' = overlap size")
print("  Speedup: Often 100-1000x for sparse data")