Spaces:
Sleeping
Sleeping
| """ | |
| 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() | |