<jupyter_start><jupyter_text>Metrics> Metrics.<jupyter_code>#hide
from nbdev.showdoc import *
from fastcore.nb_imports import *
from fastcore.test import *
#export
import numpy as np
import torch
from sklearn.metrics import precision_score, recall_score, ndcg_score
#export
def calculate_precision_recall(X, y_true, y_pred, N, threshold):
    """Calculate the precision and recall scores.

    Args:
        X
        y_true
        y_pred
        N
        threshold
    
    Returns:
        precision_score (float)
        recall_score (float)
    """
    precision = 0
    recall = 0
    count = 0
    
    rec_true = np.array([1 if rating >= threshold else 0 for rating in y_true])
    rec_pred = np.zeros(y_pred.size)
    
    for user_id in np.unique(X[:,0]):
        indices = np.where(X[:,0] == user_id)[0]
        
        rec_true = np.array([1 if y_true[i] >= threshold else 0 for i in indices])

        if (np.count_nonzero(rec_true) > 0): # ignore test users without relevant ratings
        
            user_pred = np.array([y_pred[i] for i in indices])
            rec_pred = np.zeros(indices.size)

            for pos in np.argsort(user_pred)[-N:]:
                if user_pred[pos] >= threshold:
                    rec_pred[pos] = 1
            
            precision += precision_score(rec_true, rec_pred, zero_division=0)
            recall += recall_score(rec_true, rec_pred)
            count += 1
        
    return precision/count, recall/count
#export
def calculate_ndcg(X, y_true, y_pred, N):
    """Calculate the NDCG score.

    Args:
        X
        y_true
        y_pred
        N
    
    Returns:
        ndcg_score (float)
    """
    ndcg = 0
    count = 0
    
    for user_id in np.unique(X[:,0]):
        indices = np.where(X[:,0] == user_id)[0]
        
        user_true = np.array([y_true[i] for i in indices])
        user_pred = np.array([y_pred[i] for i in indices])  
        
        user_true = np.expand_dims(user_true, axis=0)
        user_pred = np.expand_dims(user_pred, axis=0)
                
        if user_true.size > 1:
            ndcg += ndcg_score(user_true, user_pred, k=N, ignore_ties=False)
            count += 1
    
    return ndcg / count
#export
def recall(scores, labels, k):
    scores = scores.cpu()
    labels = labels.cpu()
    rank = (-scores).argsort(dim=1)
    cut = rank[:, :k]
    hit = labels.gather(1, cut)
    return (hit.sum(1).float() / torch.min(torch.Tensor([k]).to(hit.device), labels.sum(1).float())).mean().cpu().item()
#export
def ndcg(scores, labels, k):
    scores = scores.cpu()
    labels = labels.cpu()
    rank = (-scores).argsort(dim=1)
    cut = rank[:, :k]
    hits = labels.gather(1, cut)
    position = torch.arange(2, 2+k)
    weights = 1 / torch.log2(position.float())
    dcg = (hits.float() * weights).sum(1)
    idcg = torch.Tensor([weights[:min(int(n), k)].sum() for n in labels.sum(1)])
    ndcg = dcg / idcg
    return ndcg.mean()
#export
def recalls_and_ndcgs_for_ks(scores, labels, ks):
    metrics = {}

    scores = scores
    labels = labels
    answer_count = labels.sum(1)

    labels_float = labels.float()
    rank = (-scores).argsort(dim=1)
    cut = rank
    for k in sorted(ks, reverse=True):
       cut = cut[:, :k]
       hits = labels_float.gather(1, cut)
       metrics['Recall@%d' % k] = \
           (hits.sum(1) / torch.min(torch.Tensor([k]).to(labels.device), labels.sum(1).float())).mean().cpu().item()

       position = torch.arange(2, 2+k)
       weights = 1 / torch.log2(position.float())
       dcg = (hits * weights.to(hits.device)).sum(1)
       idcg = torch.Tensor([weights[:min(int(n), k)].sum() for n in answer_count]).to(dcg.device)
       ndcg = (dcg / idcg).mean()
       metrics['NDCG@%d' % k] = ndcg.cpu().item()

    return metrics
import pandas as pd

sample_data = pd.DataFrame.from_records(
    [{'deepmf': 2.7268011569976807,
  'item': 496,
  'ncf': 2.854853630065918,
  'user': 68,
  'vdeepmf': 2.4722371101379395,
  'vncf': 2.620150566101074,
  'y_test': 3.0},
 {'deepmf': 3.3491923809051514,
  'item': 473,
  'ncf': 3.0023105144500732,
  'user': 633,
  'vdeepmf': 2.847424030303955,
  'vncf': 2.6900570392608643,
  'y_test': 3.5},
 {'deepmf': 3.7268624305725098,
  'item': 329,
  'ncf': 3.605560779571533,
  'user': 1405,
  'vdeepmf': 3.810497283935547,
  'vncf': 3.466035842895508,
  'y_test': 4.0},
 {'deepmf': 3.4670088291168213,
  'item': 328,
  'ncf': 3.389759063720703,
  'user': 1240,
  'vdeepmf': 3.6399013996124268,
  'vncf': 3.205043315887451,
  'y_test': 0.5},
 {'deepmf': 3.140076160430908,
  'item': 54,
  'ncf': 3.1944096088409424,
  'user': 841,
  'vdeepmf': 2.887760877609253,
  'vncf': 2.848487138748169,
  'y_test': 3.0}]
)
sample_data
# test Precision Recall
num_recommendations = 2
method = 'ncf'
like_threshold = 3

ids = sample_data[['user', 'item']].to_numpy()
y_true = sample_data['y_test'].to_numpy()
y_pred = sample_data[method].to_numpy()
precision, recall = calculate_precision_recall(ids, y_true, y_pred, num_recommendations, like_threshold)

test_eq(precision, 0.75)
test_eq(recall, 0.75)
# sample data of 5 users/items
sample_data_2 = {
    'y_true':np.array([1,2,3,4,5]),
    'y_pred':np.array([1,3,3,2,4]),
    'ids':np.array([[1,2],[1,3],[2,4],[2,5],[3,2]]),
}
# test NDCG
num_recommendations = 2
ids = sample_data_2['ids']
y_true = sample_data_2['y_true']
y_pred = sample_data_2['y_pred']
ndcg = calculate_ndcg(ids, y_true, y_pred, num_recommendations)
test_close(ndcg, 0.9686, eps=1e-4)<jupyter_output><empty_output><jupyter_text>NDCG> Normalized Discounted Cumulative Gain.NDCG is a metric that evaluates how well the recommender performs in recommending ranked items to users. Therefore both hit of relevant items and correctness in ranking of these items matter to the NDCG evaluation. The total NDCG score is normalized by the total number of users.nDCG has three parts. First is ‘CG’ which stands for Cumulative Gains. It deals with the fact that most relevant items are more useful than somewhat relevant items that are more useful than irrelevant items. It sums the items based on its relevancy, hence, the term cumulative.But CG doesn’t account for the position of the items on the list. And hence, changing the item's position won’t change the CG. This is where the second part of nDCG comes in to play i.e. ‘D’. Discounted Cumulative Gain, DCG for short, penalized the items that appear lower in the list. A relevant item appearing at the end of the list is a result of a bad recommender system and hence that item should be discounted to indicate the bad performance of the model.nDCG normalized the DCG values of the different number of the items lists. To do so we sort the item list by relevancy and calculate the DCG for that list. This will be the perfect DCG score as items are sorted by their relevancy score.<jupyter_code>#export
def ndcg_at_k(y_true_list, y_reco_list, users=None, k=10, next_item=False,
              all_item=False):
    if next_item:
        ndcg_all = []
        y_true_list = y_true_list.tolist()
        y_reco_list = y_reco_list.tolist()
        for y_true, y_reco in zip(y_true_list, y_reco_list):
            if y_true in y_reco:
                index = y_reco.index(y_true)
                ndcg = 1. / np.log2(index + 2)
            else:
                ndcg = 0.
            ndcg_all.append(ndcg)
        return np.mean(ndcg_all)

    elif all_item:
        ndcg_all = []
        users = users.tolist()
        y_reco_list = y_reco_list.tolist()
        for i in range(len(y_reco_list)):
            y_true = y_true_list[users[i]]
            y_reco = y_reco_list[i]
            ndcg_all.append(ndcg_one(y_true, y_reco, k))
        return np.mean(ndcg_all)

    else:
        ndcg_all = list()
        for u in users:
            y_true = y_true_list[u]
            y_reco = y_reco_list[u]
            ndcg_all.append(ndcg_one(y_true, y_reco, k))
        return np.mean(ndcg_all)
#export
def ndcg_one(y_true, y_reco, k):
    rank_list = np.zeros(k)
    common_items, indices_in_true, indices_in_reco = np.intersect1d(
        y_true, y_reco, assume_unique=False, return_indices=True)

    if common_items.size > 0:
        rank_list[indices_in_reco] = 1
        ideal_list = np.sort(rank_list)[::-1]
        #  np.sum(rank_list / np.log2(2, k+2))
        dcg = np.sum(rank_list / np.log2(np.arange(2, k + 2)))
        idcg = np.sum(ideal_list / np.log2(np.arange(2, k + 2)))
        ndcg = dcg / idcg
    else:
        ndcg = 0.
    return ndcg
#export
def dcg_at_k(r, k, method=0):
    """
    Args:
        r: Relevance scores (list or numpy) in rank order
            (first element is the first item)
        k: Number of results to consider
        method: If 0 then weights are [1.0, 1.0, 0.6309, 0.5, 0.4307, ...]
                If 1 then weights are [1.0, 0.6309, 0.5, 0.4307, ...]
    Returns:
        Discounted cumulative gain
    """
    r = np.asfarray(r)[:k]
    if r.size:
        if method == 0:
            return r[0] + np.sum(r[1:] / np.log2(np.arange(2, r.size + 1)))
        elif method == 1:
            return np.sum(r / np.log2(np.arange(2, r.size + 2)))
        else:
            raise ValueError('method must be 0 or 1.')
    return 0.
result = dcg_at_k([3, 2, 3, 0, 1, 2], 6)
test_eq(np.round(result,4), 8.0972)
#export
def ndcg_at_k_v2(r, k, method=0):
    dcg_max = dcg_at_k(sorted(r, reverse=True), k, method)
    if not dcg_max:
        return 0.
    return dcg_at_k(r, k, method) / dcg_max
result = ndcg_at_k_v2([3, 2, 3, 0, 1, 2], 6, method=1)
test_eq(np.round(result,4), 0.9608)
# !pip install -q torchmetrics
#export
from abc import abstractmethod
from typing import Dict, Optional, Union

import torch
import torchmetrics as tm
#export
FLOAT_TYPES = [
    torch.float,
    torch.double,
    torch.half,
    torch.float32,
    torch.float64,
    torch.float16,
    torch.bfloat16,
]
#export
def _check_inputs(ks, scores, labels):
    if len(ks.shape) > 1:
        raise ValueError("ks should be a 1-dimensional tensor")

    if len(scores.shape) != 2:
        raise ValueError("scores must be a 2-dimensional tensor")

    if len(labels.shape) != 2:
        raise ValueError("labels must be a 2-dimensional tensor")

    if scores.shape != labels.shape:
        raise ValueError("scores and labels must be the same shape")

    if ks.device != scores.device:
        ks = ks.to(device=scores.device)

    if labels.device != scores.device:
        labels = labels.to(device=scores.device)

    if scores.dtype not in FLOAT_TYPES:
        scores = scores.to(dtype=torch.float)

    if labels.dtype not in FLOAT_TYPES:
        labels = labels.to(dtype=torch.float)

    return (
        ks,
        scores,
        labels,
    )


def _extract_topk(ks, scores, labels):
    max_k = int(max(ks))
    topk_scores, topk_indices = torch.topk(scores, max_k)
    topk_labels = torch.gather(labels, 1, topk_indices)
    return topk_scores, topk_indices, topk_labels


def _create_output_placeholder(scores, ks):
    return torch.zeros(
        scores.shape[0], len(ks), dtype=scores.dtype, device=scores.device
    )


def _mask_with_nans(metrics, labels):
    for row, values in enumerate(labels):
        if values.sum() == 0.0:
            metrics[row, :].fill_(float("NaN"))

    return metrics
#export
def one_hot_1d(
    labels: torch.Tensor,
    num_classes: int,
    device: Optional[torch.device] = None,
    dtype: Optional[torch.dtype] = torch.float32,
) -> torch.Tensor:
    r"""Coverts a 1d label tensor to one-hot representation
    Args:
        labels (torch.Tensor) : tensor with labels of shape :math:`(N, H, W)`,
                                where N is batch siz. Each value is an integer
                                representing correct classification.
        num_classes (int): number of classes in labels.
        device (Optional[torch.device]): the desired device of returned tensor.
         Default: if None, uses the current device for the default tensor type
         (see torch.set_default_tensor_type()). device will be the CPU for CPU
         tensor types and the current CUDA device for CUDA tensor types.
        dtype (Optional[torch.dtype]): the desired data type of returned
         tensor. Default: torch.float32
    Returns:
        torch.Tensor: the labels in one hot tensor.
    Examples::
        >>> labels = torch.LongTensor([0, 1, 2, 0])
        >>> one_hot_1d(labels, num_classes=3)
        tensor([[1., 0., 0.],
                [0., 1., 0.],
                [0., 0., 1.],
                [1., 0., 0.],
               ])
    """
    if not torch.is_tensor(labels):
        raise TypeError("Input labels type is not a torch.Tensor. Got {}".format(type(labels)))
    if not len(labels.shape) == 1:
        raise ValueError("Expected tensor should have 1 dim. Got: {}".format(labels.shape))
    if not labels.dtype == torch.int64:
        raise ValueError(
            "labels must be of the same dtype torch.int64. Got: {}".format(labels.dtype)
        )
    if num_classes < 1:
        raise ValueError(
            "The number of classes must be bigger than one." " Got: {}".format(num_classes)
        )
    if device is None:
        device = labels.device
    labels_size = labels.shape[0]
    one_hot = torch.zeros(labels_size, num_classes, device=device, dtype=dtype)
    return one_hot.scatter_(1, labels.unsqueeze(-1), 1.0)
#export
def _tranform_label_to_onehot(labels, vocab_size):
    return one_hot_1d(labels.reshape(-1), vocab_size, dtype=torch.float32).detach()
#export
class RankingMetric(tm.Metric):
    """
    Metric wrapper for computing ranking metrics@K for session-based task.
    Parameters
    ----------
    top_ks : list, default [2, 5])
        list of cutoffs
    labels_onehot : bool
        Enable transform the labels to one-hot representation
    """

    def __init__(self, top_ks=None, labels_onehot=False):
        super(RankingMetric, self).__init__()
        self.top_ks = top_ks or [2, 5]
        self.labels_onehot = labels_onehot
        # Store the mean of the batch metrics (for each cut-off at topk)
        self.add_state("metric_mean", default=[], dist_reduce_fx="cat")

    def update(self, preds: torch.Tensor, target: torch.Tensor, **kwargs):  # type: ignore
        # Computing the metrics at different cut-offs
        if self.labels_onehot:
            target = _tranform_label_to_onehot(target, preds.size(-1))
        metric = self._metric(torch.LongTensor(self.top_ks), preds.view(-1, preds.size(-1)), target)
        self.metric_mean.append(metric)  # type: ignore

    def compute(self):
        # Computing the mean of the batch metrics (for each cut-off at topk)
        return torch.cat(self.metric_mean, axis=0).mean(0)

    @abstractmethod
    def _metric(self, ks: torch.Tensor, preds: torch.Tensor, target: torch.Tensor) -> torch.Tensor:
        """
        Compute a ranking metric over a predictions and one-hot targets.
        This method should be overridden by subclasses.
        """
#export
class PrecisionAt(RankingMetric):
    def __init__(self, top_ks=None, labels_onehot=False):
        super(PrecisionAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)

    def _metric(self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
        """Compute precision@K for each of the provided cutoffs
        Parameters
        ----------
        ks : torch.Tensor or list
            list of cutoffs
        scores : torch.Tensor
            predicted item scores
        labels : torch.Tensor
            true item labels
        Returns
        -------
        torch.Tensor:
            list of precisions at cutoffs
        """

        ks, scores, labels = _check_inputs(ks, scores, labels)
        _, _, topk_labels = _extract_topk(ks, scores, labels)
        precisions = _create_output_placeholder(scores, ks)

        for index, k in enumerate(ks):
            precisions[:, index] = torch.sum(topk_labels[:, : int(k)], dim=1) / float(k)

        return precisions
#export
class RecallAt(RankingMetric):
    def __init__(self, top_ks=None, labels_onehot=False):
        super(RecallAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)

    def _metric(self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
        """Compute recall@K for each of the provided cutoffs
        Parameters
        ----------
        ks : torch.Tensor or list
            list of cutoffs
        scores : torch.Tensor
            predicted item scores
        labels : torch.Tensor
            true item labels
        Returns
        -------
            torch.Tensor: list of recalls at cutoffs
        """

        ks, scores, labels = _check_inputs(ks, scores, labels)
        _, _, topk_labels = _extract_topk(ks, scores, labels)
        recalls = _create_output_placeholder(scores, ks)

        # Compute recalls at K
        num_relevant = torch.sum(labels, dim=-1)
        rel_indices = (num_relevant != 0).nonzero()
        rel_count = num_relevant[rel_indices].squeeze()

        if rel_indices.shape[0] > 0:
            for index, k in enumerate(ks):
                rel_labels = topk_labels[rel_indices, : int(k)].squeeze()

                recalls[rel_indices, index] = (
                    torch.div(torch.sum(rel_labels, dim=-1), rel_count)
                    .reshape(len(rel_indices), 1)
                    .to(dtype=torch.float32)
                )  # Ensuring type is double, because it can be float if --fp16

        return recalls
#export
class AvgPrecisionAt(RankingMetric):
    def __init__(self, top_ks=None, labels_onehot=False):
        super(AvgPrecisionAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)
        self.precision_at = PrecisionAt(top_ks)._metric

    def _metric(self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
        """Compute average precision at K for provided cutoffs
        Parameters
        ----------
        ks : torch.Tensor or list
            list of cutoffs
        scores : torch.Tensor
            2-dim tensor of predicted item scores
        labels : torch.Tensor
            2-dim tensor of true item labels
        Returns
        -------
        torch.Tensor:
            list of average precisions at cutoffs
        """
        ks, scores, labels = _check_inputs(ks, scores, labels)
        topk_scores, _, topk_labels = _extract_topk(ks, scores, labels)
        avg_precisions = _create_output_placeholder(scores, ks)

        # Compute average precisions at K
        num_relevant = torch.sum(labels, dim=1)
        max_k = ks.max().item()

        precisions = self.precision_at(1 + torch.arange(max_k), topk_scores, topk_labels)
        rel_precisions = precisions * topk_labels

        for index, k in enumerate(ks):
            total_prec = rel_precisions[:, : int(k)].sum(dim=1)
            avg_precisions[:, index] = total_prec / num_relevant.clamp(min=1, max=k).to(
                dtype=torch.float32, device=scores.device
            )  # Ensuring type is double, because it can be float if --fp16

        return avg_precisions
# class MRRAt(RankingMetric):
#     def __init__(self, top_ks=None, labels_onehot=False):
#         super(MRRAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)

#     def _metric(self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
#         """Compute mean reciprocal rank at K for provided cutoffs
#         Parameters
#         ----------
#         ks : torch.Tensor or list
#             list of cutoffs
#         scores : torch.Tensor
#             2-dim tensor of predicted item scores
#         labels : torch.Tensor
#             2-dim tensor of true item labels
#         Returns
#         -------
#         torch.Tensor:
#             list of mrr scores at cutoffs
#         """
#         ks, scores, labels = _check_inputs(ks, scores, labels)
#         topk_scores, _, topk_labels = _extract_topk(ks, scores, labels)
#         mrrs = _create_output_placeholder(scores, ks)

#         # Compute mrr scores at K
#         # for index, k in enumerate(ks):
#             # topk_labels = topk_labels[torch.argsort(topk_scores[:,:k], dim=-1, descending=True)]
#             # position = torch.nonzero(topk_labels)
#             # mrrs[:, index] = 1.0 / (position[0] + 1.0).to(
#             #     dtype=torch.float32, device=scores.device
#             # )

#         return topk_labels
# num_items = 10000 # The number of candidate items to rank
# batch_size = 10 # The number of ranked lists to process
# cutoffs = [10, 20, 50, 100] # Cutoffs for the number of positions to include when computing metrics

# # labels: A large tensor of binary relevance labels
# probabilities = torch.empty(batch_size, num_items).fill_(0.5)
# labels = torch.bernoulli(probabilities)

# # scores: A large tensor of simulated relevance scores
# scores = torch.empty(batch_size, num_items).uniform_(0, 1)

# metric_at_k = MRRAt(cutoffs)

# # metric has correct shape?
# p_at_ks = metric_at_k(scores, labels)
# # test_eq(len(p_at_ks.shape), 1)
# # test_eq(len(p_at_ks), len(cutoffs))

# # # metric is all zero when nothing is relevant
# # p_at_ks = metric_at_k(scores, torch.zeros(batch_size, num_items))
# # test_eq(int(torch.count_nonzero(p_at_ks)), 0)

# # # metric is all one when everything is relevant
# # p_at_ks = metric_at_k(scores, torch.ones(batch_size, num_items))
# # test_eq(int(torch.count_nonzero(p_at_ks)), len(cutoffs))
#export
class DCGAt(RankingMetric):
    def __init__(self, top_ks=None, labels_onehot=False):
        super(DCGAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)

    def _metric(
        self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor, log_base: int = 2
    ) -> torch.Tensor:
        """Compute discounted cumulative gain at K for provided cutoffs (ignoring ties)
        Parameters
        ----------
        ks : torch.Tensor or list
            list of cutoffs
        scores : torch.Tensor
            predicted item scores
        labels : torch.Tensor
            true item labels
        Returns
        -------
        torch.Tensor :
            list of discounted cumulative gains at cutoffs
        """
        ks, scores, labels = _check_inputs(ks, scores, labels)
        topk_scores, topk_indices, topk_labels = _extract_topk(ks, scores, labels)
        dcgs = _create_output_placeholder(scores, ks)

        # Compute discounts
        discount_positions = torch.arange(ks.max().item()).to(
            device=scores.device, dtype=torch.float32
        )

        discount_log_base = torch.log(
            torch.Tensor([log_base]).to(device=scores.device, dtype=torch.float32)
        ).item()

        discounts = 1 / (torch.log(discount_positions + 2) / discount_log_base)

        # Compute DCGs at K
        for index, k in enumerate(ks):
            dcgs[:, index] = torch.sum(
                (topk_labels[:, :k] * discounts[:k].repeat(topk_labels.shape[0], 1)), dim=1
            ).to(
                dtype=torch.float32, device=scores.device
            )  # Ensuring type is double, because it can be float if --fp16

        return dcgs
#export
class NDCGAt(RankingMetric):
    def __init__(self, top_ks=None, labels_onehot=False):
        super(NDCGAt, self).__init__(top_ks=top_ks, labels_onehot=labels_onehot)
        self.dcg_at = DCGAt(top_ks)._metric

    def _metric(
        self, ks: torch.Tensor, scores: torch.Tensor, labels: torch.Tensor, log_base: int = 2
    ) -> torch.Tensor:
        """Compute normalized discounted cumulative gain at K for provided cutoffs (ignoring ties)
        Parameters
        ----------
        ks : torch.Tensor or list
            list of cutoffs
        scores : torch.Tensor
            predicted item scores
        labels : torch.Tensor
            true item labels
        Returns
        -------
        torch.Tensor :
            list of discounted cumulative gains at cutoffs
        """
        ks, scores, labels = _check_inputs(ks, scores, labels)
        topk_scores, topk_indices, topk_labels = _extract_topk(ks, scores, labels)
        # ndcgs = _create_output_placeholder(scores, ks) #TODO track if this line is needed

        # Compute discounted cumulative gains
        gains = self.dcg_at(ks, topk_scores, topk_labels)
        normalizing_gains = self.dcg_at(ks, topk_labels, topk_labels)

        # Prevent divisions by zero
        relevant_pos = (normalizing_gains != 0).nonzero(as_tuple=True)
        irrelevant_pos = (normalizing_gains == 0).nonzero(as_tuple=True)

        gains[irrelevant_pos] = 0
        gains[relevant_pos] /= normalizing_gains[relevant_pos]

        return gains
num_items = 10000 # The number of candidate items to rank
batch_size = 10 # The number of ranked lists to process
cutoffs = [10, 20, 50, 100] # Cutoffs for the number of positions to include when computing metrics

# labels: A large tensor of binary relevance labels
probabilities = torch.empty(batch_size, num_items).fill_(0.5)
labels = torch.bernoulli(probabilities)

# scores: A large tensor of simulated relevance scores
scores = torch.empty(batch_size, num_items).uniform_(0, 1)

ranking_metrics = [PrecisionAt, RecallAt, AvgPrecisionAt, DCGAt, NDCGAt]

for metric in ranking_metrics:
    metric_at_k = metric(cutoffs)

    # metric has correct shape?
    p_at_ks = metric_at_k(scores, labels)
    test_eq(len(p_at_ks.shape), 1)
    test_eq(len(p_at_ks), len(cutoffs))

    # metric is all zero when nothing is relevant
    p_at_ks = metric_at_k(scores, torch.zeros(batch_size, num_items))
    test_eq(int(torch.count_nonzero(p_at_ks)), 0)

    # metric is all one when everything is relevant
    p_at_ks = metric_at_k(scores, torch.ones(batch_size, num_items))
    test_eq(int(torch.count_nonzero(p_at_ks)), len(cutoffs))<jupyter_output><empty_output><jupyter_text>> **References:-**- https://github.com/massquantity/DBRL/blob/master/dbrl/evaluate/metrics.py- [https://github.com/NVIDIA-Merlin/Transformers4Rec/blob/main/transformers4rec/torch/ranking_metric.py](https://github.com/NVIDIA-Merlin/Transformers4Rec/blob/main/transformers4rec/torch/ranking_metric.py)- [https://github.com/karlhigley/ranking-metrics-torch](https://github.com/karlhigley/ranking-metrics-torch)- [https://github.com/mquad/sars_tutorial/blob/master/util/metrics.py](https://github.com/mquad/sars_tutorial/blob/master/util/metrics.py)<jupyter_code>#hide
%reload_ext watermark
%watermark -a "Sparsh A." -m -iv -u -t -d -p recohut<jupyter_output>Author: Sparsh A.

Last updated: 2022-01-06 09:02:26

recohut: 0.0.9

Compiler    : GCC 7.5.0
OS          : Linux
Release     : 5.4.144+
Machine     : x86_64
Processor   : x86_64
CPU cores   : 2
Architecture: 64bit

torchmetrics: 0.6.2
numpy       : 1.19.5
torch       : 1.10.0+cu111
PIL         : 7.1.2
matplotlib  : 3.2.2
IPython     : 5.5.0
{'fraction_non_alphanumeric': 0.11642535452413742, 'fraction_numerical': 0.028845795957866526, 'mean_word_length': 2.6381855111712933, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 37, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '556927', 'license': 'Apache-2.0', 'n_tokens_mistral': 9442, 'n_tokens_neox': 8863, 'n_words': 2294, 'path': 'tutorials/evaluation/evaluation.metrics.ipynb', 'repo_name': 'sparsh-ai/recohut', 'size': '26867'}