# ============================================================================ # MOVIELENS RECOMMENDATION SYSTEM - PURE IMPLEMENTATION # ============================================================================ import numpy as np import pandas as pd from scipy.sparse.linalg import svds from sklearn.metrics.pairwise import cosine_similarity from sklearn.model_selection import train_test_split import pickle import os import warnings warnings.filterwarnings('ignore') # ============================================================================ # DATA LOADING & PREPROCESSING # ============================================================================ def load_movielens_data(ratings_path='ratings.csv', movies_path='movies.csv'): """Load MovieLens data""" ratings = pd.read_csv(ratings_path) movies = pd.read_csv(movies_path) print(f"Loaded {len(ratings)} ratings") print(f"Loaded {len(movies)} movies") print(f"Users: {ratings['userId'].nunique()}") print(f"Rating distribution:\n{ratings['rating'].value_counts().sort_index()}") print(f"Mean rating: {ratings['rating'].mean():.3f}") print(f"Median rating: {ratings['rating'].median():.3f}") return ratings, movies def create_user_item_matrix(ratings): """Create user-item rating matrix""" user_item_matrix = ratings.pivot_table( index='userId', columns='movieId', values='rating' ).fillna(0) sparsity = 100 * (1 - (user_item_matrix > 0).sum().sum() / (user_item_matrix.shape[0] * user_item_matrix.shape[1])) print(f"Matrix shape: {user_item_matrix.shape}") print(f"Sparsity: {sparsity:.2f}%") return user_item_matrix # ============================================================================ # USER-BASED COLLABORATIVE FILTERING # ============================================================================ class UserBasedCF: """User-based collaborative filtering using cosine similarity""" def __init__(self, user_item_matrix): self.matrix = user_item_matrix self.user_similarity = None def fit(self): """Compute user-user similarity matrix""" print("Computing user similarity matrix...") self.user_similarity = cosine_similarity(self.matrix) np.fill_diagonal(self.user_similarity, 0) print("User similarity matrix computed") def predict(self, user_id, k=50): """Predict ratings for a user based on similar users""" if user_id not in self.matrix.index: return pd.Series(dtype=float) user_idx = self.matrix.index.get_loc(user_id) user_similarities = self.user_similarity[user_idx] # Get top-k similar users top_k_indices = np.argsort(user_similarities)[::-1][:k] top_k_similarities = user_similarities[top_k_indices] # Filter out negative similarities positive_mask = top_k_similarities > 0 top_k_indices = top_k_indices[positive_mask] top_k_similarities = top_k_similarities[positive_mask] if len(top_k_indices) == 0: return pd.Series(0, index=self.matrix.columns, dtype=float) # Get ratings from similar users similar_users_ratings = self.matrix.iloc[top_k_indices] # Weighted sum of ratings weighted_ratings = similar_users_ratings.T.dot(top_k_similarities) sum_of_weights = np.sum(top_k_similarities) # Calculate predicted ratings predicted_ratings = weighted_ratings / (sum_of_weights + 1e-10) # Exclude already rated items user_ratings = self.matrix.loc[user_id] predicted_ratings[user_ratings > 0] = 0 return predicted_ratings # ============================================================================ # ITEM-BASED COLLABORATIVE FILTERING # ============================================================================ class ItemBasedCF: """Item-based collaborative filtering using cosine similarity""" def __init__(self, user_item_matrix): self.matrix = user_item_matrix self.item_similarity = None def fit(self): """Compute item-item similarity matrix""" print("Computing item similarity matrix...") self.item_similarity = cosine_similarity(self.matrix.T) np.fill_diagonal(self.item_similarity, 0) print("Item similarity matrix computed") def predict(self, user_id, k=50): """Predict ratings for a user based on similar items""" if user_id not in self.matrix.index: return pd.Series(dtype=float) user_ratings = self.matrix.loc[user_id] rated_items = user_ratings[user_ratings > 0] if len(rated_items) == 0: return pd.Series(0, index=self.matrix.columns, dtype=float) predicted_ratings = pd.Series(0.0, index=self.matrix.columns) for item_id, rating in rated_items.items(): item_idx = self.matrix.columns.get_loc(item_id) item_similarities = self.item_similarity[item_idx] # Get top-k similar items top_k_indices = np.argsort(item_similarities)[::-1][:k] for similar_idx in top_k_indices: similar_item_id = self.matrix.columns[similar_idx] similarity = item_similarities[similar_idx] if similarity > 0 and user_ratings[similar_item_id] == 0: predicted_ratings[similar_item_id] += similarity * rating # Exclude already rated items predicted_ratings[user_ratings > 0] = 0 return predicted_ratings # ============================================================================ # SINGULAR VALUE DECOMPOSITION (SVD) # ============================================================================ class SVDRecommender: """Matrix factorization using SVD""" def __init__(self, user_item_matrix, n_factors=50): self.matrix = user_item_matrix self.n_factors = n_factors self.predictions = None def fit(self): """Perform SVD decomposition""" print(f"Performing SVD with {self.n_factors} factors...") # Mean center the matrix matrix_mean = np.mean(self.matrix.values[np.where(self.matrix.values != 0)]) matrix_centered = self.matrix.values.copy() matrix_centered[matrix_centered != 0] -= matrix_mean # Perform SVD U, sigma, Vt = svds(matrix_centered, k=self.n_factors) sigma = np.diag(sigma) # Reconstruct the matrix predicted_ratings = np.dot(np.dot(U, sigma), Vt) + matrix_mean self.predictions = pd.DataFrame( predicted_ratings, index=self.matrix.index, columns=self.matrix.columns ) print("SVD decomposition complete") def predict(self, user_id): """Get predicted ratings for a user""" if user_id not in self.predictions.index: return pd.Series(dtype=float) user_predictions = self.predictions.loc[user_id].copy() user_ratings = self.matrix.loc[user_id] # Exclude already rated items user_predictions[user_ratings > 0] = 0 return user_predictions # ============================================================================ # EVALUATION METRICS # ============================================================================ def precision_at_k(recommended, relevant, k): """Precision@K: fraction of recommended items that are relevant""" recommended_k = set(recommended[:k]) relevant_set = set(relevant) if k == 0: return 0.0 return len(recommended_k & relevant_set) / k def recall_at_k(recommended, relevant, k): """Recall@K: fraction of relevant items that are recommended""" recommended_k = set(recommended[:k]) relevant_set = set(relevant) if len(relevant_set) == 0: return 0.0 return len(recommended_k & relevant_set) / len(relevant_set) def ndcg_at_k(recommended, relevant, k): """NDCG@K: Normalized Discounted Cumulative Gain""" dcg = 0.0 for i, item in enumerate(recommended[:k]): if item in relevant: dcg += 1.0 / np.log2(i + 2) idcg = sum([1.0 / np.log2(i + 2) for i in range(min(len(relevant), k))]) if idcg == 0: return 0.0 return dcg / idcg def evaluate_model(model, test_data, user_item_matrix, k=10, threshold=4.0): """Evaluate recommendation model""" precisions = [] recalls = [] ndcgs = [] test_users = test_data['userId'].unique() print(f"Evaluating on {len(test_users)} test users...") evaluated_count = 0 for user_id in test_users: if user_id not in user_item_matrix.index: continue # Get relevant items for this user (rated >= threshold) user_test_data = test_data[test_data['userId'] == user_id] relevant_items = user_test_data[user_test_data['rating'] >= threshold]['movieId'].tolist() if len(relevant_items) == 0: continue # Get predictions predictions = model.predict(user_id) if len(predictions) == 0 or predictions.sum() == 0: continue # Get top-k recommendations top_k_items = predictions.nlargest(k).index.tolist() # Calculate metrics precisions.append(precision_at_k(top_k_items, relevant_items, k)) recalls.append(recall_at_k(top_k_items, relevant_items, k)) ndcgs.append(ndcg_at_k(top_k_items, relevant_items, k)) evaluated_count += 1 if evaluated_count >= 100: # Limit for computational efficiency break print(f"Evaluated {evaluated_count} users") if len(precisions) == 0: return { 'Precision@K': 0.0, 'Recall@K': 0.0, 'NDCG@K': 0.0 } return { 'Precision@K': np.mean(precisions), 'Recall@K': np.mean(recalls), 'NDCG@K': np.mean(ndcgs) } # ============================================================================ # RECOMMENDATION FUNCTION # ============================================================================ def recommend_movies(user_id, N, model, movies_df): """ Recommend top N movies for a user Parameters: - user_id: User ID - N: Number of recommendations - model: Trained recommendation model - movies_df: DataFrame with movie information Returns: - DataFrame with recommended movies """ predictions = model.predict(user_id) if len(predictions) == 0: return pd.DataFrame(columns=['movieId', 'title', 'predicted_rating']) # Get top N predictions top_n = predictions.nlargest(N) recommendations = pd.DataFrame({ 'movieId': top_n.index, 'predicted_rating': top_n.values }) # Merge with movie titles recommendations = recommendations.merge( movies_df[['movieId', 'title']], on='movieId', how='left' ) return recommendations[['movieId', 'title', 'predicted_rating']] # ============================================================================ # MAIN EXECUTION # ============================================================================ def main(): print("="*70) print("MOVIELENS RECOMMENDATION SYSTEM") print("="*70) # Load data print("\n[1/6] Loading data...") ratings, movies = load_movielens_data() # Split data print("\n[2/6] Splitting data (80% train, 20% test)...") train_data, test_data = train_test_split(ratings, test_size=0.2, random_state=42) print(f"Training set: {len(train_data)} ratings") print(f"Test set: {len(test_data)} ratings") # Create user-item matrix print("\n[3/6] Creating user-item matrix...") user_item_matrix = create_user_item_matrix(train_data) # Train User-Based CF print("\n[4/6] Training User-Based Collaborative Filtering...") user_cf = UserBasedCF(user_item_matrix) user_cf.fit() print("Evaluating User-Based CF...") metrics_user_cf = evaluate_model(user_cf, test_data, user_item_matrix) print(f"User-Based CF Results:") for metric, value in metrics_user_cf.items(): print(f" {metric}: {value:.4f}") # Train Item-Based CF print("\n[5/6] Training Item-Based Collaborative Filtering...") item_cf = ItemBasedCF(user_item_matrix) item_cf.fit() print("Evaluating Item-Based CF...") metrics_item_cf = evaluate_model(item_cf, test_data, user_item_matrix) print(f"Item-Based CF Results:") for metric, value in metrics_item_cf.items(): print(f" {metric}: {value:.4f}") # Train SVD print("\n[6/6] Training SVD (Matrix Factorization)...") svd = SVDRecommender(user_item_matrix, n_factors=50) svd.fit() print("Evaluating SVD...") metrics_svd = evaluate_model(svd, test_data, user_item_matrix) print(f"SVD Results:") for metric, value in metrics_svd.items(): print(f" {metric}: {value:.4f}") # Model comparison print("\n" + "="*70) print("MODEL COMPARISON") print("="*70) comparison_df = pd.DataFrame({ 'User-Based CF': metrics_user_cf, 'Item-Based CF': metrics_item_cf, 'SVD': metrics_svd }) print(comparison_df.to_string()) # Determine best model best_model_name = comparison_df.loc['NDCG@K'].idxmax() print(f"\n*** Best Model (by NDCG@K): {best_model_name} ***") if best_model_name == 'User-Based CF': best_model = user_cf elif best_model_name == 'Item-Based CF': best_model = item_cf else: best_model = svd # Example recommendations print("\n" + "="*70) print("EXAMPLE RECOMMENDATIONS") print("="*70) sample_user_id = user_item_matrix.index[0] print(f"\nTop 10 recommendations for User {sample_user_id} using {best_model_name}:") recommendations = recommend_movies(sample_user_id, 10, best_model, movies) print(recommendations.to_string(index=False)) # Save models for deployment print("\n" + "="*70) print("SAVING MODELS FOR DEPLOYMENT") print("="*70) save_models_for_deployment( user_cf, item_cf, svd, user_item_matrix, movies, metrics_user_cf, metrics_item_cf, metrics_svd ) return best_model, user_item_matrix, movies def save_models_for_deployment(user_cf, item_cf, svd, user_item_matrix, movies, metrics_user_cf, metrics_item_cf, metrics_svd): """Save all models and data for Hugging Face deployment""" output_dir = 'deployment_files' os.makedirs(output_dir, exist_ok=True) print(f"Saving models to {output_dir}/...") with open(f'{output_dir}/user_cf_model.pkl', 'wb') as f: pickle.dump(user_cf, f) print(" ✓ User-Based CF model saved") with open(f'{output_dir}/item_cf_model.pkl', 'wb') as f: pickle.dump(item_cf, f) print(" ✓ Item-Based CF model saved") with open(f'{output_dir}/svd_model.pkl', 'wb') as f: pickle.dump(svd, f) print(" ✓ SVD model saved") with open(f'{output_dir}/user_item_matrix.pkl', 'wb') as f: pickle.dump(user_item_matrix, f) print(" ✓ User-item matrix saved") metrics = { 'User-Based CF': metrics_user_cf, 'Item-Based CF': metrics_item_cf, 'SVD': metrics_svd } with open(f'{output_dir}/metrics.pkl', 'wb') as f: pickle.dump(metrics, f) print(" ✓ Metrics saved") movies.to_csv(f'{output_dir}/movies.csv', index=False) print(" ✓ Movies data saved") print("\nAll files ready for Hugging Face deployment!") if __name__ == "__main__": best_model, user_item_matrix, movies = main() import gradio as gr import pickle import pandas as pd import numpy as np import os # Determine file location BASE_DIR = 'deployment_files' if os.path.exists('deployment_files') else '.' # Load models and data print("Loading models...") with open(f'{BASE_DIR}/user_cf_model.pkl', 'rb') as f: user_cf = pickle.load(f) with open(f'{BASE_DIR}/item_cf_model.pkl', 'rb') as f: item_cf = pickle.load(f) with open(f'{BASE_DIR}/svd_model.pkl', 'rb') as f: svd = pickle.load(f) with open(f'{BASE_DIR}/user_item_matrix.pkl', 'rb') as f: user_item_matrix = pickle.load(f) movies = pd.read_csv(f'{BASE_DIR}/movies.csv') with open(f'{BASE_DIR}/metrics.pkl', 'rb') as f: metrics = pickle.load(f) MODELS = { 'User-Based CF': user_cf, 'Item-Based CF': item_cf, 'SVD': svd } print("Models loaded successfully!") def recommend_movies(user_id, N, model_name='SVD'): """Generate movie recommendations""" try: user_id = int(user_id) N = int(N) if user_id not in user_item_matrix.index: return pd.DataFrame({'Error': ['User ID not found in system']}), "" model = MODELS[model_name] predictions = model.predict(user_id) if len(predictions) == 0 or predictions.sum() == 0: return pd.DataFrame({'Error': ['No predictions available for this user']}), "" # Get top N recommendations top_n = predictions.nlargest(N) recommendations = pd.DataFrame({ 'movieId': top_n.index, 'predicted_rating': top_n.values }) # Add movie titles recommendations = recommendations.merge( movies[['movieId', 'title']], on='movieId', how='left' ) result = recommendations[['movieId', 'title', 'predicted_rating']] # Format metrics metrics_text = f""" ### {model_name} Performance Metrics - **Precision@10**: {metrics[model_name]['Precision@K']:.4f} - **Recall@10**: {metrics[model_name]['Recall@K']:.4f} - **NDCG@10**: {metrics[model_name]['NDCG@K']:.4f} *Metrics evaluated on test set with relevance threshold = 4.0* """ return result, metrics_text except Exception as e: return pd.DataFrame({'Error': [f'Error: {str(e)}']}), "" def show_model_comparison(): """Display model comparison report""" # Determine best model ndcg_scores = {name: m['NDCG@K'] for name, m in metrics.items()} best_model = max(ndcg_scores, key=ndcg_scores.get) report = f""" # Model Comparison Report ## Performance Metrics | Model | Precision@10 | Recall@10 | NDCG@10 | |-------|--------------|-----------|---------| | User-Based CF | {metrics['User-Based CF']['Precision@K']:.4f} | {metrics['User-Based CF']['Recall@K']:.4f} | {metrics['User-Based CF']['NDCG@K']:.4f} | | Item-Based CF | {metrics['Item-Based CF']['Precision@K']:.4f} | {metrics['Item-Based CF']['Recall@K']:.4f} | {metrics['Item-Based CF']['NDCG@K']:.4f} | | SVD | {metrics['SVD']['Precision@K']:.4f} | {metrics['SVD']['Recall@K']:.4f} | {metrics['SVD']['NDCG@K']:.4f} | ## Best Model: {best_model} ### Why {best_model} Performs Best **Matrix Factorization (SVD) Advantages:** - Captures latent factors in user-movie interactions - Handles sparse data through dimensionality reduction - Generalizes better than similarity-based methods - Computationally efficient for prediction **Collaborative Filtering Trade-offs:** - **User-Based**: Intuitive but computationally expensive, struggles with sparsity - **Item-Based**: More stable than user-based, but limited to similar items - **SVD**: Best balance of accuracy and efficiency ### Implementation Details - **SVD**: 50 latent factors via Singular Value Decomposition - **CF**: Cosine similarity with k=50 neighbors - **Evaluation**: 80/20 train-test split, threshold=4.0 for relevance - **Metrics**: Precision, Recall, and NDCG at K=10 ### Conclusion SVD achieves the best performance by learning compressed representations of user preferences and movie characteristics, making it the recommended approach for production deployment. """ return report def get_dataset_info(): """Display dataset statistics""" min_user = int(user_item_matrix.index.min()) max_user = int(user_item_matrix.index.max()) num_users = len(user_item_matrix.index) num_movies = len(movies) info = f""" ### Dataset Information - **Total Users**: {num_users:,} - **Total Movies**: {num_movies:,} - **User ID Range**: {min_user} to {max_user} - **Rating Scale**: 0.5 to 5.0 stars - **Source**: MovieLens Dataset """ return info # Build Gradio Interface with gr.Blocks(title="MovieLens Recommendation System", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🎬 MovieLens Recommendation System ## DataSynthis_ML_JobTask Compare three recommendation algorithms: User-Based CF, Item-Based CF, and SVD Matrix Factorization """) with gr.Tab("🎯 Get Recommendations"): gr.Markdown(get_dataset_info()) with gr.Row(): with gr.Column(): user_id_input = gr.Number( label="User ID", value=1, precision=0, info="Enter a valid user ID from the dataset" ) n_input = gr.Number( label="Number of Recommendations (N)", value=10, precision=0, info="How many movies to recommend (1-20)" ) model_select = gr.Dropdown( choices=['User-Based CF', 'Item-Based CF', 'SVD'], value='SVD', label="Recommendation Algorithm", info="Select which model to use" ) recommend_btn = gr.Button("🎬 Get Recommendations", variant="primary", size="lg") recommendations_output = gr.Dataframe( label="📋 Recommended Movies", wrap=True ) metrics_output = gr.Markdown(label="📊 Model Performance") recommend_btn.click( fn=recommend_movies, inputs=[user_id_input, n_input, model_select], outputs=[recommendations_output, metrics_output] ) with gr.Tab("📊 Model Comparison"): gr.Markdown(show_model_comparison()) with gr.Tab("ℹ️ Documentation"): gr.Markdown(""" ## Implementation Overview ### Algorithms **1. User-Based Collaborative Filtering** - Finds users with similar rating patterns - Recommends items liked by similar users - Uses cosine similarity with k=50 neighbors **2. Item-Based Collaborative Filtering** - Finds items similar to those the user has rated - Recommends items similar to user's preferences - Uses cosine similarity with k=50 neighbors **3. Singular Value Decomposition (SVD)** - Matrix factorization with 50 latent factors - Learns low-dimensional representations of users and items - Predicts ratings via reconstructed matrix ### Evaluation Metrics - **Precision@K**: Fraction of recommended items that are relevant - **Recall@K**: Fraction of relevant items that are recommended - **NDCG@K**: Normalized Discounted Cumulative Gain (considers ranking order) ### Technical Stack - Python 3.10+ - NumPy, Pandas for data processing - SciPy for SVD computation - Scikit-learn for similarity metrics - Gradio for web interface ### Dataset - Source: MovieLens - Split: 80% training, 20% testing - Relevance Threshold: 4.0 stars --- **Project**: DataSynthis ML Job Task **Task**: Movie Recommendation System """) demo.launch()