Coverage Metrics - API Reference

Auto-generated documentation for coverage metric classes.

warprec.evaluation.metrics.coverage.item_coverage.ItemCoverage

Bases: TopKMetric

The ItemCoverage@k metric counts the number of unique items that were recommended across all users.

Attributes:

Name	Type	Description
item_counts	Tensor	The tensor of item counts.

Parameters:

Name	Type	Description	Default
k	int	The cutoff.	required
num_items	int	Number of items in the training set.	required
dist_sync_on_step	bool	Torchmetrics parameter.	False
**kwargs	Any	The keyword argument dictionary.	{}

Source code in warprec/evaluation/metrics/coverage/item_coverage.py

@metric_registry.register("ItemCoverage")
class ItemCoverage(TopKMetric):
    """The ItemCoverage@k metric counts the number of unique items
       that were recommended across all users.

    Attributes:
        item_counts (Tensor): The tensor of item counts.

    Args:
        k (int): The cutoff.
        num_items (int): Number of items in the training set.
        dist_sync_on_step (bool): Torchmetrics parameter.
        **kwargs (Any): The keyword argument dictionary.
    """

    _REQUIRED_COMPONENTS: Set[MetricBlock] = {MetricBlock.TOP_K_INDICES}

    item_counts: Tensor

    def __init__(
        self,
        k: int,
        num_items: int,
        dist_sync_on_step: bool = False,
        **kwargs: Any,
    ):
        super().__init__(k, dist_sync_on_step)
        self.add_state(
            "item_counts", default=torch.zeros(num_items), dist_reduce_fx="sum"
        )

    def update(self, preds: Tensor, **kwargs: Any):
        # Retrieve top_k_indices from kwargs
        top_k_indices = kwargs.get(f"top_{self.k}_indices")

        # Remap top_k_indices to global
        item_indices = kwargs.get("item_indices")
        top_k_indices = self.remap_indices(top_k_indices, item_indices)

        # Flatten the indices to count occurrences across the entire batch
        flat_indices = top_k_indices.flatten()

        # Update counts
        batch_counts = torch.bincount(flat_indices, minlength=len(self.item_counts))
        self.item_counts += batch_counts.to(self.item_counts)

    def compute(self):
        item_coverage = (self.item_counts > 0).sum().item()
        return {self.name: item_coverage}

warprec.evaluation.metrics.coverage.user_coverage.UserCoverage

Bases: TopKMetric

The UserCoverage@k metric counts the number of users that received at least one recommendation.

Attributes:

Name	Type	Description
users	Tensor	The number of users evaluated.

Parameters:

Name	Type	Description	Default
k	int	The cutoff.	required
dist_sync_on_step	bool	Torchmetrics parameter.	False
**kwargs	Any	The keyword argument dictionary.	{}

Source code in warprec/evaluation/metrics/coverage/user_coverage.py

@metric_registry.register("UserCoverage")
class UserCoverage(TopKMetric):
    """The UserCoverage@k metric counts the number of users
       that received at least one recommendation.

    Attributes:
        users (Tensor): The number of users evaluated.

    Args:
        k (int): The cutoff.
        dist_sync_on_step (bool): Torchmetrics parameter.
        **kwargs (Any): The keyword argument dictionary.
    """

    _REQUIRED_COMPONENTS: Set[MetricBlock] = {
        MetricBlock.BINARY_RELEVANCE,
        MetricBlock.VALID_USERS,
    }

    users: Tensor

    def __init__(self, k: int, dist_sync_on_step: bool = False, **kwargs: Any):
        super().__init__(k, dist_sync_on_step)
        self.add_state("users", default=torch.tensor(0.0), dist_reduce_fx="sum")

    def update(self, preds: Tensor, **kwargs: Any):
        users = kwargs.get("valid_users")

        # Count only users with at least one interaction
        self.users += users.sum()

    def compute(self):
        return {self.name: int(self.users.item())}

warprec.evaluation.metrics.coverage.numretrieved.NumRetrieved

Bases: UserAverageTopKMetric

The NumRetrieved@k counts the number of items retrieved in the top-k list.

This metric simply counts how many items are present in the recommended list up to the specified cutoff k. It does not consider the relevance of the items.

Source code in warprec/evaluation/metrics/coverage/numretrieved.py

@metric_registry.register("NumRetrieved")
class NumRetrieved(UserAverageTopKMetric):
    """The NumRetrieved@k counts the number of items retrieved in the top-k list.

    This metric simply counts how many items are present in the recommended list up to
    the specified cutoff k. It does not consider the relevance of the items.
    """

    _REQUIRED_COMPONENTS: Set[MetricBlock] = {
        MetricBlock.BINARY_RELEVANCE,
        MetricBlock.VALID_USERS,
        MetricBlock.TOP_K_VALUES,
    }

    def compute_scores(
        self, preds: Tensor, target: Tensor, top_k_rel: Tensor, **kwargs: Any
    ) -> Tensor:
        # Retrieve top_k_values from kwargs
        top_k_values = kwargs.get(f"top_{self.k}_values")

        # Count items that are not -inf (valid recommendations)
        return (~torch.isinf(top_k_values)).sum(dim=1).float()

warprec.evaluation.metrics.coverage.user_coverage_at_n.UserCoverageAtN

Bases: TopKMetric

The UserCoverageAtN counts the number of user that retrieved correctly at least N recommendations.

This metric measures the system's ability to provide a sufficiently long recommendation list for users.

Attributes:

Name	Type	Description
users	Tensor	Number of user with at least 1 relevant item.

Parameters:

Name	Type	Description	Default
k	int	The cutoff.	required
dist_sync_on_step	bool	Torchmetrics parameter.	False
**kwargs	Any	The keyword argument dictionary.	{}

Source code in warprec/evaluation/metrics/coverage/user_coverage_at_n.py

@metric_registry.register("UserCoverageAtN")
class UserCoverageAtN(TopKMetric):
    """The UserCoverageAtN counts the number of user that retrieved
        correctly at least N recommendations.

    This metric measures the system's ability to provide a sufficiently long
    recommendation list for users.

    Attributes:
        users (Tensor): Number of user with at least 1 relevant item.

    Args:
        k (int): The cutoff.
        dist_sync_on_step (bool): Torchmetrics parameter.
        **kwargs (Any): The keyword argument dictionary.
    """

    users: Tensor

    def __init__(self, k: int, dist_sync_on_step: bool = False, **kwargs: Any):
        super().__init__(k, dist_sync_on_step)

        self.add_state("users", default=torch.tensor(0.0), dist_reduce_fx="sum")

    def update(self, preds: Tensor, **kwargs: Any):
        # Count how many items have a score > 0 for each user
        valid_items_per_user = (preds > 0).sum(dim=1)

        # Check if the count is >= k
        satisfied_users = valid_items_per_user.ge(self.k).sum()

        self.users += satisfied_users

    def compute(self):
        return {self.name: int(self.users.item())}