Coverage for hopwise/evaluator/base_metric.py: 86%
120 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-09-30 13:25 +0000
1# @Time : 2020/10/21
2# @Author : Kaiyuan Li
3# @email : tsotfsk@outlook.com
5# UPDATE
6# @Time : 2020/10/21, 2021/8/29
7# @Author : Kaiyuan Li, Zhichao Feng
8# @email : tsotfsk@outlook.com, fzcbupt@gmail.com
10"""hopwise.evaluator.abstract_metric
11#####################################
12"""
14import numpy as np
15import pandas as pd
16import torch
18from hopwise.utils import EvaluatorType
21class AbstractMetric:
22 """:class:`AbstractMetric` is the base object of all metrics. If you want to
23 implement a metric, you should inherit this class.
25 Args:
26 config (Config): the config of evaluator.
27 """
29 smaller = False
30 metric_need = []
32 def __init__(self, config):
33 self.decimal_place = config["metric_decimal_place"]
35 def __init_subclass__(cls, **kwargs):
36 """Automatically extend parent's metric_need if subclass defines metric_need."""
37 super().__init_subclass__(**kwargs)
39 if hasattr(cls, "metric_need") and cls.metric_need is not cls.__bases__[0].metric_need:
40 # Get parents' metric_need
41 parent_metric_need = []
42 for base in cls.__bases__:
43 if hasattr(base, "metric_need"):
44 parent_metric_need.extend(base.metric_need)
46 cls.metric_need = list(set(parent_metric_need + cls.metric_need))
48 def calculate_metric(self, dataobject):
49 """Get the dictionary of a metric.
51 Args:
52 dataobject(DataStruct): it contains all the information needed to calculate metrics.
54 Returns:
55 dict: such as ``{'metric@10': 3153, 'metric@20': 0.3824}``
56 """
57 raise NotImplementedError("Method [calculate_metric] should be implemented.")
60class TopkMetric(AbstractMetric):
61 """:class:`TopkMetric` is a base object of top-k metrics. If you want to
62 implement an top-k metric, you can inherit this class.
64 Args:
65 config (Config): The config of evaluator.
66 """
68 metric_type = EvaluatorType.RANKING
69 metric_need = ["rec.topk"]
71 def __init__(self, config):
72 super().__init__(config)
73 self.topk = config["topk"]
75 def used_info(self, dataobject):
76 """Get the bool matrix indicating whether the corresponding item is positive
77 and number of positive items for each user.
78 """
79 rec_mat = dataobject.get("rec.topk")
80 self.topk = dataobject.get("topk")
81 topk_idx, pos_len_list = torch.split(rec_mat, [max(self.topk), 1], dim=1)
82 return topk_idx.to(torch.bool).numpy(), pos_len_list.squeeze(-1).numpy()
84 def topk_result(self, metric, value):
85 """Match the metric value to the `k` and put them in `dictionary` form.
87 Args:
88 metric(str): the name of calculated metric.
89 value(numpy.ndarray): metrics for each user, including values from `metric@1` to `metric@max(self.topk)`.
91 Returns:
92 dict: metric values required in the configuration.
93 """
94 metric_dict = {}
95 avg_result = value.mean(axis=0)
96 for k in self.topk:
97 key = f"{metric}@{k}"
98 metric_dict[key] = round(avg_result[k - 1], self.decimal_place)
99 return metric_dict
101 def metric_info(self, pos_index, pos_len=None):
102 """Calculate the value of the metric.
104 Args:
105 pos_index(numpy.ndarray): a bool matrix, shape of ``n_users * max(topk)``. The item with the (j+1)-th \
106 highest score of i-th user is positive if ``pos_index[i][j] == True`` and negative otherwise.
107 pos_len(numpy.ndarray): a vector representing the number of positive items per user, shape of ``(n_users,)``.
109 Returns:
110 numpy.ndarray: metrics for each user, including values from `metric@1` to `metric@max(self.topk)`.
111 """ # noqa: E501
112 raise NotImplementedError("Method [metric_info] of top-k metric should be implemented.")
115class LossMetric(AbstractMetric):
116 """:class:`LossMetric` is a base object of loss based metrics and AUC. If you want to
117 implement an loss based metric, you can inherit this class.
119 Args:
120 config (Config): The config of evaluator.
121 """
123 metric_type = EvaluatorType.VALUE
124 metric_need = ["rec.score", "data.label"]
126 def __init__(self, config):
127 super().__init__(config)
129 def used_info(self, dataobject):
130 """Get scores that model predicted and the ground truth."""
131 preds = dataobject.get("rec.score")
132 trues = dataobject.get("data.label")
134 return preds.squeeze(-1).numpy(), trues.squeeze(-1).numpy()
136 def output_metric(self, metric, dataobject):
137 preds, trues = self.used_info(dataobject)
138 result = self.metric_info(preds, trues)
139 return {metric: round(result, self.decimal_place)}
141 def metric_info(self, preds, trues):
142 """Calculate the value of the metric.
144 Args:
145 preds (numpy.ndarray): the scores predicted by model, a one-dimensional vector.
146 trues (numpy.ndarray): the label of items, which has the same shape as ``preds``.
148 Returns:
149 float: The value of the metric.
150 """
151 raise NotImplementedError("Method [metric_info] of loss-based metric should be implemented.")
154class ConsumerTopKMetric(AbstractMetric):
155 """:class:`ConsumerTopKMetric` is a base object of consumer-based metrics. If you want to
156 implement a consumer-based metric, you can inherit this class.
157 The consumer-based metrics are based on a binary partition of users and on the demographic parity notion,
158 commonly measured as the absolute difference between the two groups in terms of a ranking metric.
160 Args:
161 config (Config): The config of evaluator.
162 """
164 metric_type = EvaluatorType.RANKING
165 metric_need = ["eval_data.user_feat", "rec.users"]
166 smaller = True
167 USER_GROUP_1 = 1
168 USER_GROUP_2 = 2
170 def __init__(self, config):
171 super().__init__(config)
172 self._ranking_metric = None
173 self.sensitive_attribute = config["sensitive_attribute"]
175 if self.sensitive_attribute is None:
176 raise ValueError("The sensitive attribute is not specified in the config. Consumer metrics require it.")
178 @property
179 def ranking_metric(self):
180 if self._ranking_metric is None:
181 raise NotImplementedError("Use a subclass of ConsumerTopKMetric to calculate a specific ranking metric")
182 return self._ranking_metric
184 @ranking_metric.setter
185 def ranking_metric(self, value):
186 if not isinstance(value, TopkMetric):
187 raise TypeError(f"The type of ranking_metric [{type(value)}] is not supported for ConsumerTopKMetric.")
189 self._ranking_metric = value
191 def get_group_mask(self, user_feat, interaction_users):
192 # 0 is the padding sensitive attribute
193 group1_mask = user_feat[self.sensitive_attribute][interaction_users] == self.USER_GROUP_1
194 group2_mask = user_feat[self.sensitive_attribute][interaction_users] == self.USER_GROUP_2
196 return group1_mask, group2_mask
198 def used_info(self, dataobject):
199 """Get the users features and the users in the interaction batch."""
200 user_feat = dataobject.get("eval_data.user_feat")
201 interaction_users = dataobject.get("rec.users")
203 return self.get_group_mask(user_feat, interaction_users)
205 def get_dp(self, result, group1_mask, group2_mask):
206 """Get the absolute difference between the two groups in terms of a ranking metric.
208 Args:
209 group1_result (torch.Tensor): the result of the first group.
210 group2_result (torch.Tensor): the result of the second group.
212 Returns:
213 torch.Tensor: the difference between the two groups in terms of a ranking metric.
214 """
215 group1_result = result[group1_mask, :].mean(axis=0, keepdims=True)
216 group2_result = result[group2_mask, :].mean(axis=0, keepdims=True)
218 return np.abs(group1_result - group2_result)
220 def ranking_metric_info(self, pos_index, pos_len):
221 raise NotImplementedError("Use a subclass of ConsumerTopKMetric to calculate a specific ranking metric")
223 def calculate_metric(self, dataobject):
224 group_mask1, group_mask2 = self.used_info(dataobject)
225 pos_index, pos_len = self.ranking_metric.used_info(dataobject)
226 ranking_result = self.ranking_metric_info(pos_index, pos_len)
227 result = self.get_dp(ranking_result, group_mask1, group_mask2)
228 metric_dict = self.ranking_metric.topk_result(self.__class__.__name__.lower(), result)
229 return metric_dict
232class PathQualityMetric(AbstractMetric):
233 """:class:`PathQualityMetric` is a base object of path-based metrics. If you want to
234 implement a path based metric, you can inherit this class.
236 Args:
237 config (Config): The config of evaluator.
238 """
240 metric_type = EvaluatorType.RANKING
241 metric_need = ["rec.paths"]
243 def __init__(self, config):
244 super().__init__(config)
245 self.topk = config["topk"]
247 def used_info(self, dataobject):
248 paths = dataobject.get("rec.paths")
249 return paths
251 def normalized_ema(self, values):
252 if max(values) == min(values):
253 values = list(range(len(values)))
255 values = pd.Series(values)
256 ema_vals = values.ewm(span=len(values)).mean()
257 normalized_ema_vals = (ema_vals - ema_vals.min()) / (ema_vals.max() - ema_vals.min())
258 return normalized_ema_vals.to_numpy()
260 def topk_result(self, metric, value):
261 """Match the metric value to the `k` and put them in `dictionary` form.
263 Path quality metrics aggregate one value per user or per path and have no rank dimension,
264 because `rec.paths` only holds the paths of the top-`max(topk)` recommended items.
265 The same average is therefore reported for every `k`.
267 Args:
268 metric(str): the name of calculated metric.
269 value(numpy.ndarray): metric value of each user or of each path, shape of ``(n_users,)`` or \
270 ``(n_paths,)``.
272 Returns:
273 dict: metric values required in the configuration.
274 """
275 metric_dict = {}
276 avg_result = value.mean(axis=0)
277 for k in self.topk:
278 key = f"{metric}@{k}"
279 metric_dict[key] = round(avg_result, self.decimal_place)
280 return metric_dict