""" Comprehensive Movie Recommendation System ======================================== This system implements multiple recommendation approaches: 1. User-based Collaborative Filtering 2. Item-based Collaborative Filtering 3. Matrix Factorization (SVD) 4. Neural Embeddings (Deep Learning) Author: DataSynthis ML Team Date: 2024 """ import pandas as pd import numpy as np import pickle import os from typing import List, Tuple, Dict, Optional from sklearn.metrics.pairwise import cosine_similarity from sklearn.decomposition import TruncatedSVD from sklearn.neural_network import MLPRegressor from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split from sklearn.metrics import mean_squared_error, mean_absolute_error import matplotlib.pyplot as plt import seaborn as sns import warnings warnings.filterwarnings('ignore') class ComprehensiveRecommendationSystem: """ Comprehensive recommendation system implementing multiple approaches """ def __init__(self, data_path: str = None): """ Initialize the recommendation system Args: data_path: Path to the MovieLens dataset directory """ self.data_path = data_path self.ratings = None self.movies = None self.user_item_matrix = None self.user_similarity_matrix = None self.item_similarity_matrix = None self.svd_model = None self.neural_model = None self.scaler = None self.user_means = None self.item_means = None def load_data(self, ratings_file: str, movies_file: str): """ Load and preprocess the MovieLens dataset Args: ratings_file: Path to ratings file movies_file: Path to movies file """ print("Loading MovieLens dataset...") # Load ratings data - handle both CSV and tab-separated formats if ratings_file.endswith('.csv'): self.ratings = pd.read_csv(ratings_file) # Rename columns to standard format if 'userId' in self.ratings.columns: self.ratings = self.ratings.rename(columns={'userId': 'user_id', 'movieId': 'movie_id'}) else: # Handle tab-separated format (ml-100k) self.ratings = pd.read_csv( ratings_file, sep="\t", names=["user_id", "movie_id", "rating", "timestamp"] ) # Load movies data - handle both CSV and pipe-separated formats if movies_file.endswith('.csv'): self.movies = pd.read_csv(movies_file) # Rename columns to standard format if 'movieId' in self.movies.columns: self.movies = self.movies.rename(columns={'movieId': 'movie_id'}) # Select relevant columns for movies if 'title' in self.movies.columns: self.movies = self.movies[['movie_id', 'title']] else: # Fallback to first two columns if title column not found self.movies = self.movies.iloc[:, [0, 1]] self.movies.columns = ["movie_id", "title"] else: # Handle pipe-separated format (ml-100k) self.movies = pd.read_csv( movies_file, sep="|", encoding="latin-1", header=None ) # Select relevant columns for movies self.movies = self.movies[[0, 1]] self.movies.columns = ["movie_id", "title"] print(f"Ratings shape: {self.ratings.shape}") print(f"Movies shape: {self.movies.shape}") print(f"Unique users: {self.ratings['user_id'].nunique()}") print(f"Unique movies: {self.movies['movie_id'].nunique()}") print(f"Rating range: {self.ratings['rating'].min()} - {self.ratings['rating'].max()}") # Create user-item matrix self._create_user_item_matrix() def _create_user_item_matrix(self): """Create user-item rating matrix""" print("Creating user-item matrix...") self.user_item_matrix = self.ratings.pivot_table( index="user_id", columns="movie_id", values="rating" ).fillna(0) print(f"User-Item Matrix shape: {self.user_item_matrix.shape}") print(f"Sparsity: {(1 - (self.user_item_matrix != 0).sum().sum() / self.user_item_matrix.size) * 100:.2f}%") def train_user_based_cf(self): """Train user-based collaborative filtering""" print("Training user-based collaborative filtering...") # Compute user similarity matrix self.user_similarity_matrix = cosine_similarity(self.user_item_matrix) # Calculate user means for bias correction self.user_means = self.user_item_matrix.mean(axis=1) print("User-based CF trained successfully!") def train_item_based_cf(self): """Train item-based collaborative filtering""" print("Training item-based collaborative filtering...") # Compute item similarity matrix self.item_similarity_matrix = cosine_similarity(self.user_item_matrix.T) # Calculate item means for bias correction self.item_means = self.user_item_matrix.mean(axis=0) print("Item-based CF trained successfully!") def train_svd(self, n_components: int = 50): """Train SVD matrix factorization""" print(f"Training SVD with {n_components} components...") # Create a copy of the matrix for SVD matrix_for_svd = self.user_item_matrix.copy() # Replace 0s with user means for proper SVD user_means = matrix_for_svd.mean(axis=1) for user_id in matrix_for_svd.index: user_ratings = matrix_for_svd.loc[user_id] zero_mask = user_ratings == 0 matrix_for_svd.loc[user_id, zero_mask] = user_means[user_id] # Apply SVD self.svd_model = TruncatedSVD(n_components=n_components, random_state=42) self.svd_model.fit(matrix_for_svd) print("SVD trained successfully!") def train_neural_embeddings(self, hidden_layers: Tuple = (100, 50), max_iter: int = 100): """Train neural network for embeddings""" print("Training neural embeddings...") # Prepare data for neural network X = [] y = [] for user_id in self.user_item_matrix.index: for movie_id in self.user_item_matrix.columns: rating = self.user_item_matrix.loc[user_id, movie_id] if rating > 0: # Only use actual ratings X.append([user_id, movie_id]) y.append(rating) X = np.array(X) y = np.array(y) # Scale features self.scaler = StandardScaler() X_scaled = self.scaler.fit_transform(X) # Train neural network self.neural_model = MLPRegressor( hidden_layer_sizes=hidden_layers, max_iter=max_iter, random_state=42 ) self.neural_model.fit(X_scaled, y) print("Neural embeddings trained successfully!") def recommend_movies_user_based(self, user_id: int, N: int = 10, top_k: int = 10) -> pd.DataFrame: """ Generate recommendations using user-based collaborative filtering Args: user_id: ID of the user to recommend for N: Number of movies to recommend top_k: Number of similar users to consider Returns: DataFrame with recommended movies """ if user_id not in self.user_item_matrix.index: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Get user's ratings user_ratings = self.user_item_matrix.loc[user_id] rated_movies = set(user_ratings[user_ratings > 0].index) # Get all movies all_movies = set(self.user_item_matrix.columns) unrated_movies = all_movies - rated_movies if not unrated_movies: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Get similar users user_idx = list(self.user_item_matrix.index).index(user_id) similarities = self.user_similarity_matrix[user_idx] similar_users_idx = np.argsort(similarities)[::-1][1:top_k+1] # Predict ratings for unrated movies predictions = [] for movie_id in unrated_movies: predicted_rating = self._predict_rating_user_based(user_id, movie_id, similar_users_idx) if predicted_rating > 0: predictions.append({ 'movie_id': movie_id, 'predicted_rating': predicted_rating }) if not predictions: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Sort by predicted rating and get top-N predictions_df = pd.DataFrame(predictions) predictions_df = predictions_df.sort_values('predicted_rating', ascending=False).head(N) # Merge with movie information recommendations = predictions_df.merge( self.movies[['movie_id', 'title']], on='movie_id', how='left' ) return recommendations[['movie_id', 'title', 'predicted_rating']] def recommend_movies_item_based(self, user_id: int, N: int = 10, top_k: int = 10) -> pd.DataFrame: """ Generate recommendations using item-based collaborative filtering Args: user_id: ID of the user to recommend for N: Number of movies to recommend top_k: Number of similar items to consider Returns: DataFrame with recommended movies """ if user_id not in self.user_item_matrix.index: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Get user's ratings user_ratings = self.user_item_matrix.loc[user_id] rated_movies = set(user_ratings[user_ratings > 0].index) # Get all movies all_movies = set(self.user_item_matrix.columns) unrated_movies = all_movies - rated_movies if not unrated_movies: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Predict ratings for unrated movies predictions = [] for movie_id in unrated_movies: predicted_rating = self._predict_rating_item_based(user_id, movie_id, top_k) if predicted_rating > 0: predictions.append({ 'movie_id': movie_id, 'predicted_rating': predicted_rating }) if not predictions: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Sort by predicted rating and get top-N predictions_df = pd.DataFrame(predictions) predictions_df = predictions_df.sort_values('predicted_rating', ascending=False).head(N) # Merge with movie information recommendations = predictions_df.merge( self.movies[['movie_id', 'title']], on='movie_id', how='left' ) return recommendations[['movie_id', 'title', 'predicted_rating']] def recommend_movies_svd(self, user_id: int, N: int = 10) -> pd.DataFrame: """ Generate recommendations using SVD matrix factorization Args: user_id: ID of the user to recommend for N: Number of movies to recommend Returns: DataFrame with recommended movies """ if user_id not in self.user_item_matrix.index: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Get user's ratings user_ratings = self.user_item_matrix.loc[user_id] rated_movies = set(user_ratings[user_ratings > 0].index) # Get all movies all_movies = set(self.user_item_matrix.columns) unrated_movies = all_movies - rated_movies if not unrated_movies: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Predict ratings for unrated movies predictions = [] for movie_id in unrated_movies: predicted_rating = self._predict_rating_svd(user_id, movie_id) if predicted_rating > 0: predictions.append({ 'movie_id': movie_id, 'predicted_rating': predicted_rating }) if not predictions: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Sort by predicted rating and get top-N predictions_df = pd.DataFrame(predictions) predictions_df = predictions_df.sort_values('predicted_rating', ascending=False).head(N) # Merge with movie information recommendations = predictions_df.merge( self.movies[['movie_id', 'title']], on='movie_id', how='left' ) return recommendations[['movie_id', 'title', 'predicted_rating']] def recommend_movies_neural(self, user_id: int, N: int = 10) -> pd.DataFrame: """ Generate recommendations using neural embeddings Args: user_id: ID of the user to recommend for N: Number of movies to recommend Returns: DataFrame with recommended movies """ if user_id not in self.user_item_matrix.index: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Get user's ratings user_ratings = self.user_item_matrix.loc[user_id] rated_movies = set(user_ratings[user_ratings > 0].index) # Get all movies all_movies = set(self.user_item_matrix.columns) unrated_movies = all_movies - rated_movies if not unrated_movies: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Predict ratings for unrated movies predictions = [] for movie_id in unrated_movies: predicted_rating = self._predict_rating_neural(user_id, movie_id) if predicted_rating > 0: predictions.append({ 'movie_id': movie_id, 'predicted_rating': predicted_rating }) if not predictions: return pd.DataFrame(columns=['movie_id', 'title', 'predicted_rating']) # Sort by predicted rating and get top-N predictions_df = pd.DataFrame(predictions) predictions_df = predictions_df.sort_values('predicted_rating', ascending=False).head(N) # Merge with movie information recommendations = predictions_df.merge( self.movies[['movie_id', 'title']], on='movie_id', how='left' ) return recommendations[['movie_id', 'title', 'predicted_rating']] def recommend_movies(self, user_id: int, N: int = 10, method: str = "ensemble") -> pd.DataFrame: """ Main recommendation function with the required signature Args: user_id: ID of the user to recommend for N: Number of movies to recommend method: Recommendation method ('user_based', 'item_based', 'svd', 'neural', 'ensemble') Returns: DataFrame with recommended movies """ if method == "user_based": return self.recommend_movies_user_based(user_id, N) elif method == "item_based": return self.recommend_movies_item_based(user_id, N) elif method == "svd": return self.recommend_movies_svd(user_id, N) elif method == "neural": return self.recommend_movies_neural(user_id, N) elif method == "ensemble": return self._ensemble_recommendations(user_id, N) else: raise ValueError("Method must be 'user_based', 'item_based', 'svd', 'neural', or 'ensemble'") def _ensemble_recommendations(self, user_id: int, N: int) -> pd.DataFrame: """Combine recommendations from all methods""" # Get recommendations from all methods user_recs = self.recommend_movies_user_based(user_id, N*2) item_recs = self.recommend_movies_item_based(user_id, N*2) svd_recs = self.recommend_movies_svd(user_id, N*2) neural_recs = self.recommend_movies_neural(user_id, N*2) # Combine and average scores all_recs = [] movie_scores = {} for recs in [user_recs, item_recs, svd_recs, neural_recs]: for _, rec in recs.iterrows(): movie_id = rec['movie_id'] score = rec['predicted_rating'] if movie_id not in movie_scores: movie_scores[movie_id] = [] movie_scores[movie_id].append(score) # Average scores for each movie for movie_id, scores in movie_scores.items(): avg_score = np.mean(scores) all_recs.append({ 'movie_id': movie_id, 'predicted_rating': avg_score }) # Sort by average score and get top-N all_recs_df = pd.DataFrame(all_recs) all_recs_df = all_recs_df.sort_values('predicted_rating', ascending=False).head(N) # Merge with movie information recommendations = all_recs_df.merge( self.movies[['movie_id', 'title']], on='movie_id', how='left' ) return recommendations[['movie_id', 'title', 'predicted_rating']] def _predict_rating_user_based(self, user_id: int, movie_id: int, similar_users_idx: np.ndarray) -> float: """Predict rating using user-based collaborative filtering""" user_idx = list(self.user_item_matrix.index).index(user_id) movie_idx = list(self.user_item_matrix.columns).index(movie_id) numerator = 0.0 denominator = 0.0 for similar_user_idx in similar_users_idx: similar_user_id = self.user_item_matrix.index[similar_user_idx] rating = self.user_item_matrix.iloc[similar_user_idx, movie_idx] if rating > 0: similarity = self.user_similarity_matrix[user_idx, similar_user_idx] user_mean = self.user_means[similar_user_id] numerator += similarity * (rating - user_mean) denominator += abs(similarity) if denominator == 0: return 0.0 user_mean = self.user_means[user_id] predicted_rating = user_mean + (numerator / denominator) return max(0.0, min(5.0, predicted_rating)) def _predict_rating_item_based(self, user_id: int, movie_id: int, top_k: int) -> float: """Predict rating using item-based collaborative filtering""" user_ratings = self.user_item_matrix.loc[user_id] movie_idx = list(self.user_item_matrix.columns).index(movie_id) # Get similar items similarities = self.item_similarity_matrix[movie_idx] similar_items_idx = np.argsort(similarities)[::-1][1:top_k+1] numerator = 0.0 denominator = 0.0 for similar_item_idx in similar_items_idx: similar_movie_id = self.user_item_matrix.columns[similar_item_idx] rating = user_ratings[similar_movie_id] if rating > 0: similarity = similarities[similar_item_idx] item_mean = self.item_means[similar_movie_id] numerator += similarity * (rating - item_mean) denominator += abs(similarity) if denominator == 0: return 0.0 item_mean = self.item_means[movie_id] predicted_rating = item_mean + (numerator / denominator) return max(0.0, min(5.0, predicted_rating)) def _predict_rating_svd(self, user_id: int, movie_id: int) -> float: """Predict rating using SVD""" user_idx = list(self.user_item_matrix.index).index(user_id) movie_idx = list(self.user_item_matrix.columns).index(movie_id) # Transform user and movie to latent space user_latent = self.svd_model.transform(self.user_item_matrix.iloc[user_idx:user_idx+1])[0] movie_latent = self.svd_model.components_[:, movie_idx] # Predict rating predicted_rating = np.dot(user_latent, movie_latent) return max(0.0, min(5.0, predicted_rating)) def _predict_rating_neural(self, user_id: int, movie_id: int) -> float: """Predict rating using neural network""" # Prepare input X = np.array([[user_id, movie_id]]) X_scaled = self.scaler.transform(X) # Predict predicted_rating = self.neural_model.predict(X_scaled)[0] return max(0.0, min(5.0, predicted_rating)) def evaluate_model(self, test_users: List[int], method: str = "ensemble", N: int = 10) -> Dict: """ Evaluate model performance using precision@K, recall@K, and NDCG@K Args: test_users: List of user IDs to evaluate method: Recommendation method to evaluate N: Number of recommendations to generate Returns: Dictionary with evaluation metrics """ print(f"Evaluating {method} method on {len(test_users)} users...") precision_scores = [] recall_scores = [] ndcg_scores = [] for user_id in test_users: try: # Get recommendations recommendations = self.recommend_movies(user_id, N=N, method=method) recommended_movies = set(recommendations['movie_id'].values) # Get actual liked movies (rating >= 4) user_ratings = self.ratings[ (self.ratings['user_id'] == user_id) & (self.ratings['rating'] >= 4) ] liked_movies = set(user_ratings['movie_id'].values) if len(recommended_movies) == 0: precision_scores.append(0.0) recall_scores.append(0.0) ndcg_scores.append(0.0) continue # Calculate precision@K precision = len(recommended_movies & liked_movies) / len(recommended_movies) precision_scores.append(precision) # Calculate recall@K if len(liked_movies) > 0: recall = len(recommended_movies & liked_movies) / len(liked_movies) else: recall = 0.0 recall_scores.append(recall) # Calculate NDCG@K ndcg = self._calculate_ndcg(list(recommendations['movie_id']), liked_movies, N) ndcg_scores.append(ndcg) except Exception as e: print(f"Error evaluating user {user_id}: {str(e)}") precision_scores.append(0.0) recall_scores.append(0.0) ndcg_scores.append(0.0) return { 'precision': np.mean(precision_scores), 'recall': np.mean(recall_scores), 'ndcg': np.mean(ndcg_scores), 'std_precision': np.std(precision_scores), 'std_recall': np.std(recall_scores), 'std_ndcg': np.std(ndcg_scores), 'num_users': len(test_users) } def _calculate_ndcg(self, recommended: List[int], relevant: set, k: int = None) -> float: """Calculate Normalized Discounted Cumulative Gain""" if k is None: k = len(recommended) recommended = recommended[:k] if len(relevant) == 0: return 0.0 # Calculate DCG dcg = 0.0 for i, movie_id in enumerate(recommended): if movie_id in relevant: dcg += 1.0 / np.log2(i + 2) # Calculate IDCG idcg = 0.0 for i in range(min(len(relevant), len(recommended))): idcg += 1.0 / np.log2(i + 2) if idcg == 0: return 0.0 return dcg / idcg def train_all_models(self): """Train all recommendation models""" print("Training all recommendation models...") # Train user-based CF self.train_user_based_cf() # Train item-based CF self.train_item_based_cf() # Train SVD self.train_svd(n_components=50) # Train neural embeddings self.train_neural_embeddings(hidden_layers=(100, 50), max_iter=50) print("All models trained successfully!") def save_model(self, filepath: str): """Save the trained model""" model_data = { 'user_item_matrix': self.user_item_matrix, 'user_similarity_matrix': self.user_similarity_matrix, 'item_similarity_matrix': self.item_similarity_matrix, 'svd_model': self.svd_model, 'neural_model': self.neural_model, 'scaler': self.scaler, 'user_means': self.user_means, 'item_means': self.item_means, 'movies': self.movies, 'ratings': self.ratings } with open(filepath, 'wb') as f: pickle.dump(model_data, f) print(f"Model saved to {filepath}") def load_model(self, filepath: str): """Load a trained model""" with open(filepath, 'rb') as f: model_data = pickle.load(f) self.user_item_matrix = model_data['user_item_matrix'] self.user_similarity_matrix = model_data['user_similarity_matrix'] self.item_similarity_matrix = model_data['item_similarity_matrix'] self.svd_model = model_data['svd_model'] self.neural_model = model_data['neural_model'] self.scaler = model_data['scaler'] self.user_means = model_data['user_means'] self.item_means = model_data['item_means'] self.movies = model_data['movies'] self.ratings = model_data['ratings'] print(f"Model loaded from {filepath}") def main(): """Main function to demonstrate the comprehensive recommendation system""" print("Comprehensive Movie Recommendation System") print("=" * 50) # Initialize the system rec_system = ComprehensiveRecommendationSystem() # Load data rec_system.load_data("u.data", "u.item") # Train all models rec_system.train_all_models() # Test recommendations for a specific user test_user = 1 print(f"\nRecommendations for User {test_user}:") print("-" * 40) # Test different methods methods = ["user_based", "item_based", "svd", "neural", "ensemble"] for method in methods: print(f"\n{method.upper()} Method:") recommendations = rec_system.recommend_movies(test_user, N=5, method=method) print(recommendations) # Evaluate models print("\nEvaluating models...") print("-" * 40) # Select a sample of users for evaluation all_users = rec_system.ratings['user_id'].unique() test_users = np.random.choice(all_users, min(20, len(all_users)), replace=False) for method in methods: results = rec_system.evaluate_model(test_users, method=method, N=10) print(f"{method:12} | Precision: {results['precision']:.4f} | Recall: {results['recall']:.4f} | NDCG: {results['ndcg']:.4f}") # Save model rec_system.save_model("comprehensive_recommendation_model.pkl") return rec_system if __name__ == "__main__": rec_system = main()