core

class Metric(BaseModel, ParamMixin):
    """Base class for all metrics.

    A Metric object is stateful, i.e. after `calculate`
    the results can be retrieved in one of two ways:
      - Detailed results are stored in :attr:`results`,
      - Aggregated result value can be retrieved using :attr:`value`
    """

    _scores: None | csr_matrix
    _user_id_map: np.ndarray
    _y_true: csr_matrix
    _y_pred: csr_matrix
    _user_id_sequence_array: np.ndarray
    """Sequence of user IDs in the evaluation data."""
    _num_users: int
    _true_positive: int
    """Number of true positives computed. Used for caching to obtain macro results."""
    _false_negative: int
    """Number of false negatives computed. Used for caching to obtain macro results."""
    _false_positive: int
    """Number of false positives computed. Used for caching to obtain macro results."""

    def __init__(
        self,
        user_id_sequence_array: np.ndarray,
        user_item_shape: tuple[int, int],
        timestamp_limit: None | int = None,
    ) -> None:
        self._user_id_sequence_array = user_id_sequence_array
        self._num_users, self._num_items = user_item_shape
        self._timestamp_limit: None | int = timestamp_limit

    @property
    def _is_computed(self) -> bool:
        """Whether the metric has been computed."""
        return hasattr(self, "_scores")

    def get_params(self) -> dict[str, int | None]:
        """Get the parameters of the metric."""
        if not self.is_time_aware:
            return {}
        return {"timestamp_limit": self._timestamp_limit}

    @property
    def micro_result(self) -> dict[str, np.ndarray]:
        """Micro results for the metric.

        :return: Detailed results for the metric.
        :rtype: dict[str, np.ndarray]
        """
        return {"score": np.array(self.macro_result)}

    @property
    def macro_result(self) -> None | float:
        """The global metric value."""
        raise NotImplementedError()

    @property
    def is_time_aware(self) -> bool:
        """Whether the metric is time-aware."""
        return self._timestamp_limit is not None

    @property
    def timestamp_limit(self) -> int:
        """The timestamp limit for the metric."""
        if not self.is_time_aware or self._timestamp_limit is None:
            raise ValueError("This metric is not time-aware.")
        return self._timestamp_limit

    @property
    def num_items(self) -> int:
        """Dimension of the item-space in both `y_true` and `y_pred`"""
        return self._num_items

    @property
    def num_users(self) -> int:
        """Dimension of the user-space in both `y_true` and `y_pred`
        after elimination of users without interactions in `y_true`.
        """
        return self._num_users

    def _prepare_matrix(
        self, y_true: csr_matrix, y_pred: csr_matrix
    ) -> tuple[csr_matrix, csr_matrix]:
        """Prepare the matrices for the metric calculation.

        This method is used to prepare the matrices for the metric calculation.
        It is used to eliminate empty users and to set the shape of the matrices.
        """
        if not y_true.shape == y_pred.shape:
            raise AssertionError(
                f"Shape mismatch between y_true: {y_true.shape} and y_pred: {y_pred.shape}"
            )
        self._set_shape(y_true=y_true)
        return y_true, y_pred

    @abstractmethod
    def _calculate(self, y_true: csr_matrix, y_pred: csr_matrix) -> None:
        raise NotImplementedError()

    def calculate(self, y_true: csr_matrix, y_pred: csr_matrix) -> None:
        """Calculates this metric for all nonzero users in `y_true`,
        given true labels and predicted scores.
        """
        y_true, y_pred = self._prepare_matrix(y_true, y_pred)
        self._calculate(y_true, y_pred)

    def _set_shape(self, y_true: csr_matrix) -> None:
        """Set the number of users and items based on the shape of y_true.
        """
        self._num_users, self._num_items = y_true.shape  # type: ignore

class ListwiseMetricK(MetricTopK):
    """Base class for all listwise metrics that can be calculated for every Top-K recommendation list,
    i.e. one value for each user.
    Examples are: PrecisionK, RecallK, DCGK, NDCGK.

    :param K: Size of the recommendation list consisting of the Top-K item predictions.
    :type K: int
    """

    @property
    def micro_result(self) -> dict[str, np.ndarray]:
        """User level results for the metric.

        Contains an entry for every user.

        :return: The results DataFrame with columns: user_id, score
        :rtype: pd.DataFrame
        """
        if not self._is_computed:
            raise ValueError("Metric has not been calculated yet.")
        elif self._scores is None:
            logger.warning(UserWarning("No scores were computed. Returning empty dict."))
            return dict(zip(self.col_names, (np.array([]), np.array([]))))

        scores = self._scores.toarray().reshape(-1)

        unique_users, inv = np.unique(self._user_id_sequence_array, return_inverse=True)

        # sum of scores per user
        sum_ones = np.zeros(len(unique_users))
        np.add.at(sum_ones, inv, scores)

        # count per user
        count_all = np.zeros(len(unique_users))
        np.add.at(count_all, inv, 1)

        # aggregated score per user
        agg_score = sum_ones / count_all

        return dict(zip(self.col_names, (unique_users, agg_score)))

    @property
    def macro_result(self) -> None | float:
        """Global metric value obtained by taking the average over all users.

        :raises ValueError: If the metric has not been calculated yet.
        :return: The global metric value.
        :rtype: float, optional
        """
        if not self._is_computed:
            raise ValueError("Metric has not been calculated yet.")
        elif self._scores is None:
            logger.warning(UserWarning("No scores were computed. Returning Null value."))
            return None
        elif self._scores.size == 0:
            logger.warning(
                UserWarning(
                    f"All predictions were off or the ground truth matrix was empty during compute of {self.identifier}."
                )
            )
            return 0
        return self._scores.mean().item()

class MetricTopK(Metric):
    """Base class for all metrics computed based on the Top-K recommendations for every user.

    A MetricTopK object is stateful, i.e. after `calculate`
    the results can be retrieved in one of two ways:
      - Detailed results are stored in :attr:`results`,
      - Aggregated result value can be retrieved using :attr:`value`

    :param K: Size of the recommendation list consisting of the Top-K item predictions.
    :type K: int
    """

    def __init__(
        self,
        user_id_sequence_array: np.ndarray,
        user_item_shape: tuple[int, int],
        timestamp_limit: None | int = None,
        K: int = 10,
    ) -> None:
        super().__init__(
            user_id_sequence_array=user_id_sequence_array,
            user_item_shape=user_item_shape,
            timestamp_limit=timestamp_limit,
        )
        if K is None:
            warn(f"K not specified, using default value {K}.")
        self.K = K

    @property
    def name(self) -> str:
        """Name of the metric."""
        return f"{super().name}_{self.K}"

    @property
    def params(self) -> dict[str, int | None]:
        """Parameters of the metric."""
        return super().params | {"K": self.K}

    @property
    def col_names(self) -> list[str]:
        """The names of the columns in the results DataFrame."""
        return ["user_id", "score"]

    def prepare_matrix(self, y_true: csr_matrix, y_pred: csr_matrix) -> tuple[csr_matrix, csr_matrix]:
        y_true, y_pred = super()._prepare_matrix(y_true, y_pred)
        y_pred = get_top_K_ranks(y_pred, self.K)
        return y_true, y_pred