Recommendation Engine

# What we need to query: Q1: Get personalized recommendations for user Q2: Get similar items (people who liked X also liked Y) Q3: Get user's interaction history (ratings, views, purchases) Q4: Find similar users (collaborative filtering) Q5: Get trending/popular items Q6: Update user preferences in real-time

from cassandra.cluster import Cluster from collections import defaultdict import math cluster = Cluster(['127.0.0.1']) session = cluster.connect('recommendations') def compute_item_similarities(): # Get all user-item interactions query = "SELECT user_id, item_id, rating FROM user_interactions WHERE rating IS NOT NULL" interactions = session.execute(query) # Build user-item matrix user_items = defaultdict(lambda: {}) item_users = defaultdict(set) for row in interactions: user_items[row.user_id][row.item_id] = row.rating item_users[row.item_id].add(row.user_id) # Compute item-item similarities (cosine similarity) items = list(item_users.keys()) for i, item1 in enumerate(items): for item2 in items[i+1:]: # Find common users common_users = item_users[item1] & item_users[item2] if len(common_users) >= 5: # Minimum threshold # Calculate cosine similarity dot_product = 0 norm1 = 0 norm2 = 0 for user in common_users: r1 = user_items[user][item1] r2 = user_items[user][item2] dot_product += r1 * r2 norm1 += r1 ** 2 norm2 += r2 ** 2 similarity = dot_product / (math.sqrt(norm1) * math.sqrt(norm2)) # Store similarity (both directions) insert = """ INSERT INTO item_similarities (item_id, similar_item_id, similarity_score, co_occurrence_count) VALUES (?, ?, ?, ?) """ session.execute(insert, (item1, item2, similarity, len(common_users))) session.execute(insert, (item2, item1, similarity, len(common_users))) print("Item similarities computed!") # Run as batch job (daily/weekly) compute_item_similarities()

def generate_recommendations_for_user(user_id): # Get user's highly rated items query = """ SELECT item_id, rating FROM user_interactions WHERE user_id = ? AND rating >= 4.0 LIMIT 50 """ liked_items = session.execute(query, (user_id,)) # Find similar items recommendations = {} for interaction in liked_items: # Get top 10 similar items similar_query = """ SELECT similar_item_id, similarity_score FROM item_similarities WHERE item_id = ? LIMIT 10 """ similar_items = session.execute(similar_query, (interaction.item_id,)) for item in similar_items: if item.similar_item_id not in recommendations: # Score = similarity * user's rating score = item.similarity_score * interaction.rating recommendations[item.similar_item_id] = score else: recommendations[item.similar_item_id] += item.similarity_score * interaction.rating # Sort by score and get top 20 top_recommendations = sorted( recommendations.items(), key=lambda x: x[1], reverse=True )[:20] # Store recommendations insert = """ INSERT INTO user_recommendations (user_id, recommendation_type, item_id, score, generated_at) VALUES (?, ?, ?, ?, ?) """ from datetime import datetime now = datetime.now() for item_id, score in top_recommendations: session.execute(insert, (user_id, 'personalized', item_id, score, now)) print(f"Generated {len(top_recommendations)} recommendations for {user_id}")

def build_user_profile(user_id): # Get user's liked items query = """ SELECT item_id, rating FROM user_interactions WHERE user_id = ? AND rating >= 4.0 """ interactions = session.execute(query, (user_id,)) # Extract item features (genres, tags, etc.) genre_scores = defaultdict(lambda: 0) tag_scores = defaultdict(lambda: 0) for interaction in interactions: # Get item metadata item = session.execute( "SELECT genres, tags FROM items WHERE item_id = ?", (interaction.item_id,) ).one() # Weight by rating weight = interaction.rating / 5.0 for genre in item.genres: genre_scores[genre] += weight for tag in item.tags: tag_scores[tag] += weight return { 'genres': dict(genre_scores), 'tags': dict(tag_scores) } def recommend_by_content(user_id): profile = build_user_profile(user_id) # Find items matching user's preferences # (This would typically use a feature vector comparison) # For simplicity: find items with matching genres top_genres = sorted(profile['genres'].items(), key=lambda x: x[1], reverse=True)[:3] recommendations = [] for genre, score in top_genres: # Find popular items in this genre (would need secondary index) # This is simplified - production would use ML models pass return recommendations

from datetime import datetime def track_interaction(user_id, item_id, interaction_type, rating=None, duration=None): # Record interaction insert = """ INSERT INTO user_interactions (user_id, item_id, interaction_type, timestamp, rating, duration) VALUES (?, ?, ?, ?, ?, ?) """ session.execute(insert, ( user_id, item_id, interaction_type, datetime.now(), rating, duration )) # Update item popularity if interaction_type == 'view': session.execute(""" UPDATE items SET total_ratings = total_ratings + 1 WHERE item_id = ? """, (item_id,)) # Trigger real-time recommendation update (optional) if rating and rating >= 4.0: update_recommendations_realtime(user_id, item_id, rating) def update_recommendations_realtime(user_id, liked_item_id, rating): # Get similar items immediately query = """ SELECT similar_item_id, similarity_score FROM item_similarities WHERE item_id = ? LIMIT 5 """ similar_items = session.execute(query, (liked_item_id,)) # Add to user's recommendations insert = """ INSERT INTO user_recommendations (user_id, recommendation_type, item_id, score, reason, generated_at) VALUES (?, ?, ?, ?, ?, ?) """ for item in similar_items: score = item.similarity_score * rating reason = f"Because you rated {liked_item_id} highly" session.execute(insert, ( user_id, 'realtime', item.similar_item_id, score, reason, datetime.now() )) print(f"Updated real-time recommendations for {user_id}")

from flask import Flask, request, jsonify app = Flask(__name__) @app.route('/recommendations/<user_id>') def get_recommendations(user_id): # Get personalized recommendations query = """ SELECT item_id, score, reason FROM user_recommendations WHERE user_id = ? AND recommendation_type = 'personalized' LIMIT 20 """ results = session.execute(query, (user_id,)) recommendations = [] for row in results: # Get item details item = session.execute( "SELECT title, category FROM items WHERE item_id = ?", (row.item_id,) ).one() recommendations.append({ 'item_id': row.item_id, 'title': item.title, 'category': item.category, 'score': float(row.score), 'reason': row.reason }) return jsonify(recommendations) @app.route('/track', methods=['POST']) def track(): data = request.json track_interaction( data['user_id'], data['item_id'], data['type'], data.get('rating'), data.get('duration') ) return jsonify({'status': 'success'})

def generate_hybrid_recommendations(user_id): recommendations = {} # Signal 1: Collaborative filtering (40% weight) collab_recs = get_collaborative_recommendations(user_id) for item_id, score in collab_recs: recommendations[item_id] = score * 0.4 # Signal 2: Content-based (30% weight) content_recs = get_content_based_recommendations(user_id) for item_id, score in content_recs: if item_id in recommendations: recommendations[item_id] += score * 0.3 else: recommendations[item_id] = score * 0.3 # Signal 3: Trending items (20% weight) trending = get_trending_items() for item_id, score in trending: if item_id in recommendations: recommendations[item_id] += score * 0.2 else: recommendations[item_id] = score * 0.2 # Signal 4: Popularity boost (10% weight) popular = get_popular_items() for item_id, score in popular: if item_id in recommendations: recommendations[item_id] += score * 0.1 else: recommendations[item_id] = score * 0.1 # Sort and return top N top_recs = sorted(recommendations.items(), key=lambda x: x[1], reverse=True)[:50] return top_recs def get_trending_items(): query = """ SELECT item_id, trend_score FROM trending_items WHERE category = 'all' AND time_window = 'daily' LIMIT 20 """ results = session.execute(query) return [(r.item_id, r.trend_score) for r in results]