Skip to content

elementwise_top_k

logger = logging.getLogger(__name__) module-attribute

ElementwiseMetricK

Bases: MetricTopK

Base class for all elementwise metrics that can be calculated for each user-item pair in the Top-K recommendations.

:attr:results contains an entry for each user-item pair.

Examples are: HitK

This code is adapted from RecPack :cite:recpack

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

Source code in src/recnexteval/metrics/core/elementwise_top_k.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class ElementwiseMetricK(MetricTopK):
    """Base class for all elementwise metrics that can be calculated for
    each user-item pair in the Top-K recommendations.

    :attr:`results` contains an entry for each user-item pair.

    Examples are: HitK

    This code is adapted from RecPack :cite:`recpack`

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

    # TODO to fix this function
    @property
    def micro_result(self) -> dict[str, np.ndarray]:
        if not self._is_computed:
            raise ValueError("Metric has not been calculated yet.")
        elif self._scores is None:
            warn(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 hits per user
        sum_ones = np.zeros(len(unique_users))
        np.add.at(sum_ones, inv, scores)

        # Count recommendations 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:
        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

        scores = self._scores.toarray().reshape(-1)
        unique_users, inv = np.unique(self._user_id_sequence_array, return_inverse=True)
        # get all users that was recommended at least a relevant item
        sum_ones = np.zeros(len(unique_users))
        np.add.at(sum_ones, inv, scores)
        # Convert to binary: 1 if at least 1 hit, 0 otherwise
        binary_hits = (sum_ones > 0).astype(int)
        # Fraction of users with at least 1 hit
        return binary_hits.mean().item()

micro_result property

macro_result property

name property

Name of the metric.

params property

Parameters of the metric.

identifier property

Identifier of the object.

Identifier is made by combining the class name with the parameters passed at construction time.

Constructed by recreating the initialisation call. Example: Algorithm(param_1=value)

:return: Identifier of the object

IS_BASE = True class-attribute instance-attribute

is_time_aware property

Whether the metric is time-aware.

timestamp_limit property

The timestamp limit for the metric.

num_items property

Dimension of the item-space in both y_true and y_pred

num_users property

Dimension of the user-space in both y_true and y_pred after elimination of users without interactions in y_true.

K = K instance-attribute

col_names property

The names of the columns in the results DataFrame.

get_params()

Get the parameters of the metric.

Source code in src/recnexteval/metrics/core/base.py 
53
54
55
56
57
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}

calculate(y_true, y_pred)

Calculates this metric for all nonzero users in y_true, given true labels and predicted scores.

Source code in src/recnexteval/metrics/core/base.py 
116
117
118
119
120
121
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)

prepare_matrix(y_true, y_pred)

Source code in src/recnexteval/metrics/core/top_k.py 
57
58
59
60
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