src/evaluation.py - master - katherinesiv/study_material_recommender

src/evaluation.py
291 строка9 KB
Сиваева Екатерина
update src/evaluation.py
25 дек 2025, 11:25
25 дек 2025, 11:251b39e9f
"""
Evaluation metrics for recommendation models
"""

import logging
from typing import Any, Dict, List, Tuple

import numpy as np
import pandas as pd
from sklearn.metrics import mean_squared_error


logger = logging.getLogger(__name__)


class RecommendationEvaluator:
    """Evaluates recommendation models using various metrics"""

    def __init__(self, test_data: pd.DataFrame):
        """
        Initialize evaluator with test data.

        Args:
            test_data: DataFrame with test ratings (user_id, material_id, rating)
        """
        self.test_data = test_data
        self.test_matrix = None
        self._prepare_test_matrix()

    def _prepare_test_matrix(self) -> None:
        """Prepare test user-item matrix"""
        self.test_matrix = self.test_data.pivot_table(
            index="user_id",
            columns="material_id",
            values="rating",
            fill_value=0
        )

    def calculate_rmse(self, predictions: pd.DataFrame) -> float:
        """
        Calculate Root Mean Squared Error.

        Args:
            predictions: DataFrame with columns (user_id, material_id, predicted_rating)

        Returns:
            RMSE score
        """
        # Merge predictions with actual ratings
        merged = pd.merge(
            predictions,
            self.test_data,
            on=["user_id", "material_id"],
            suffixes=("_pred", "_actual")
        )

        if len(merged) == 0:
            logger.warning(
                "No overlapping predictions and test data for RMSE calculation"
            )
            return float("inf")

        rmse = np.sqrt(mean_squared_error(
            merged["rating_actual"],
            merged["predicted_rating"]
        ))

        return rmse

    def calculate_precision_at_k(
        self, recommendations: Dict[int, List[int]], k: int = 10
    ) -> float:
        """
        Calculate Precision@K for recommendations.

        Args:
            recommendations: Dictionary mapping user_id to list of recommended material_ids
            k: Number of top recommendations to consider

        Returns:
            Precision@K score
        """
        precisions = []

        for user_id, rec_items in recommendations.items():
            # Get actual items user rated highly (rating >= 4)
            actual_items = self.test_data[
                (self.test_data["user_id"] == user_id) &
                (self.test_data["rating"] >= 4)
            ]["material_id"].tolist()

            # Take top-k recommendations
            top_k_recs = rec_items[:k]

            # Calculate precision for this user
            if actual_items:
                hits = len(set(top_k_recs) & set(actual_items))
                precision = hits / min(k, len(actual_items))
                precisions.append(precision)
            else:
                # If user has no highly rated items in test set, skip
                continue

        if not precisions:
            return 0.0

        return np.mean(precisions)

    def calculate_recall_at_k(
        self, recommendations: Dict[int, List[int]], k: int = 10
    ) -> float:
        """
        Calculate Recall@K for recommendations.

        Args:
            recommendations: Dictionary mapping user_id to list of recommended material_ids
            k: Number of top recommendations to consider

        Returns:
            Recall@K score
        """
        recalls = []

        for user_id, rec_items in recommendations.items():
            # Get actual items user rated highly (rating >= 4)
            actual_items = self.test_data[
                (self.test_data["user_id"] == user_id) &
                (self.test_data["rating"] >= 4)
            ]["material_id"].tolist()

            # Take top-k recommendations
            top_k_recs = rec_items[:k]

            # Calculate recall for this user
            if actual_items:
                hits = len(set(top_k_recs) & set(actual_items))
                recall = hits / len(actual_items)
                recalls.append(recall)
            else:
                # If user has no highly rated items in test set, skip
                continue

        if not recalls:
            return 0.0

        return np.mean(recalls)

    def calculate_ndcg(
        self, recommendations: Dict[int, List[int]], k: int = 10
    ) -> float:
        """
        Calculate Normalized Discounted Cumulative Gain.

        Args:
            recommendations: Dictionary mapping user_id to list of recommended material_ids
            k: Number of top recommendations to consider

        Returns:
            NDCG score
        """
        ndcg_scores = []

        for user_id, rec_items in recommendations.items():
            # Get actual ratings for this user
            user_ratings = self.test_data[self.test_data["user_id"] == user_id]
            rating_dict = dict(zip(
                user_ratings["material_id"],
                user_ratings["rating"]
            ))

            # Take top-k recommendations
            top_k_recs = rec_items[:k]

            # Calculate DCG
            dcg = 0
            for i, item_id in enumerate(top_k_recs, 1):
                rel = rating_dict.get(item_id, 0)
                dcg += rel / np.log2(i + 1)

            # Calculate IDCG (ideal DCG)
            ideal_ratings = sorted(rating_dict.values(), reverse=True)[:k]
            idcg = sum(r / np.log2(i + 1) for i, r in enumerate(ideal_ratings, 1))

            # Calculate NDCG
            ndcg = dcg / idcg if idcg > 0 else 0
            ndcg_scores.append(ndcg)

        if not ndcg_scores:
            return 0.0

        return np.mean(ndcg_scores)

    def evaluate_model(self, model_predictions: Dict[str, Any]) -> Dict[str, float]:
        """
        Comprehensive evaluation of a model.

        Args:
            model_predictions: Dictionary containing:
                - 'predictions': DataFrame with predicted ratings
                - 'recommendations': Dict mapping user_id to recommended items

        Returns:
            Dictionary of evaluation metrics
        """
        metrics = {}

        # Calculate RMSE if predictions are available
        if "predictions" in model_predictions:
            rmse = self.calculate_rmse(model_predictions["predictions"])
            metrics["rmse"] = rmse

        # Calculate ranking metrics if recommendations are available
        if "recommendations" in model_predictions:
            for k in [5, 10, 20]:
                precision = self.calculate_precision_at_k(
                    model_predictions["recommendations"], k
                )
                recall = self.calculate_recall_at_k(
                    model_predictions["recommendations"], k
                )
                ndcg = self.calculate_ndcg(
                    model_predictions["recommendations"], k
                )

                metrics[f"precision@{k}"] = precision
                metrics[f"recall@{k}"] = recall
                metrics[f"ndcg@{k}"] = ndcg

        # Calculate coverage
        if "recommendations" in model_predictions:
            coverage = self._calculate_coverage(model_predictions["recommendations"])
            metrics["coverage"] = coverage

        logger.info(f"Evaluation metrics: {metrics}")
        return metrics

    def _calculate_coverage(self, recommendations: Dict[int, List[int]]) -> float:
        """Calculate catalog coverage of recommendations"""
        all_recommended_items = set()

        for user_recs in recommendations.values():
            all_recommended_items.update(user_recs)

        total_items = len(self.test_data["material_id"].unique())

        if total_items == 0:
            return 0.0

        return len(all_recommended_items) / total_items

    def cross_validate(
        self, model, train_test_splits: List[Tuple[pd.DataFrame, pd.DataFrame]]
    ) -> Dict[str, List[float]]:
        """
        Perform cross-validation.

        Args:
            model: Recommendation model with fit() and recommend() methods
            train_test_splits: List of (train_data, test_data) tuples

        Returns:
            Dictionary of metric scores for each fold
        """
        fold_metrics = {
            "rmse": [],
            "precision@10": [],
            "recall@10": [],
            "ndcg@10": []
        }

        for fold, (train_data, test_data) in enumerate(train_test_splits, 1):
            logger.info(f"Evaluating fold {fold}/{len(train_test_splits)}")

            # Create evaluator for this fold
            _ = RecommendationEvaluator(test_data)

            # For now, we'll skip the actual training and just show the structure
            # TODO: Implement actual cross-validation with your model

            fold_metrics["rmse"].append(0.0)
            fold_metrics["precision@10"].append(0.0)
            fold_metrics["recall@10"].append(0.0)
            fold_metrics["ndcg@10"].append(0.0)

        # Calculate mean and std for each metric
        summary = {}
        for metric, values in fold_metrics.items():
            summary[f"{metric}_mean"] = np.mean(values)
            summary[f"{metric}_std"] = np.std(values)

        return summary