annif.eval.false_negatives() - Code Metrics - Inspection of "Implement hyperparameter optimization of ensemble..." - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#414)

by Osma

created 2020-07-19 22:26 UTC

annif.eval.false_negatives() A

↳ Parent: annif.eval

Complexity

Conditions

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
nop	2

"""Evaluation metrics for Annif"""

import statistics
import warnings
import numpy as np
from sklearn.metrics import precision_score, recall_score, f1_score
from sklearn.metrics import label_ranking_average_precision_score
from annif.exception import NotSupportedException


def filter_pred_top_k(preds, limit):
    """filter a 2D prediction vector, retaining only the top K suggestions
    for each individual prediction; the rest will be set to zeros"""

    masks = []
    for pred in preds:
        mask = np.zeros_like(pred, dtype=np.bool)
        top_k = np.argsort(pred)[::-1][:limit]
        mask[top_k] = True
        masks.append(mask)
    return preds * np.array(masks)


def true_positives(y_true, y_pred):
    """calculate the number of true positives using bitwise operations,
    emulating the way sklearn evaluation metric functions work"""
    return (y_true & y_pred).sum()


def false_positives(y_true, y_pred):
    """calculate the number of false positives using bitwise operations,
    emulating the way sklearn evaluation metric functions work"""
    return (~y_true & y_pred).sum()


def false_negatives(y_true, y_pred):
    """calculate the number of false negatives using bitwise operations,
    emulating the way sklearn evaluation metric functions work"""
    return (y_true & ~y_pred).sum()


def precision_at_k_score(y_true, y_pred, limit):
    """calculate the precision at K, i.e. the number of relevant items
    among the top K predicted ones"""
    scores = []
    for true, pred in zip(y_true, y_pred):
        order = pred.argsort()[::-1]
        orderlimit = min(limit, np.count_nonzero(pred))
        order = order[:orderlimit]
        gain = true[order]
        if orderlimit > 0:
            scores.append(gain.sum() / orderlimit)
        else:
            scores.append(0.0)
    return statistics.mean(scores)


def dcg_score(y_true, y_pred, limit=None):
    """return the discounted cumulative gain (DCG) score for the selected
    labels vs. relevant labels"""
    order = y_pred.argsort()[::-1]
    n_pred = np.count_nonzero(y_pred)
    if limit is not None:
        n_pred = min(limit, n_pred)
    order = order[:n_pred]
    gain = y_true[order]
    discount = np.log2(np.arange(order.size) + 2)

    return (gain / discount).sum()


def ndcg_score(y_true, y_pred, limit=None):
    """return the normalized discounted cumulative gain (nDCG) score for the
    selected labels vs. relevant labels"""
    scores = []
    for true, pred in zip(y_true, y_pred):
        idcg = dcg_score(true, true, limit)
        dcg = dcg_score(true, pred, limit)
        if idcg > 0:
            scores.append(dcg / idcg)
        else:
            scores.append(1.0)  # perfect score for no relevant hits case
    return statistics.mean(scores)


class EvaluationBatch:
    """A class for evaluating batches of results using all available metrics.
    The evaluate() method is called once per document in the batch.
    Final results can be queried using the results() method."""

    def __init__(self, subject_index):
        self._subject_index = subject_index
        self._samples = []

    def evaluate(self, hits, gold_subjects):
        self._samples.append((hits, gold_subjects))

    def _evaluate_samples(self, y_true, y_pred, metrics='all'):
        y_pred_binary = y_pred > 0.0

        # define the available metrics as lazy lambda functions
        # so we can execute only the ones actually requested
        all_metrics = {
            'Precision (doc avg)': lambda: precision_score(
                y_true, y_pred_binary, average='samples'),
            'Recall (doc avg)': lambda: recall_score(
                y_true, y_pred_binary, average='samples'),
            'F1 score (doc avg)': lambda: f1_score(
                y_true, y_pred_binary, average='samples'),
            'Precision (subj avg)': lambda: precision_score(
                y_true, y_pred_binary, average='macro'),
            'Recall (subj avg)': lambda: recall_score(
                y_true, y_pred_binary, average='macro'),
            'F1 score (subj avg)': lambda: f1_score(
                y_true, y_pred_binary, average='macro'),
            'Precision (weighted subj avg)': lambda: precision_score(
                y_true, y_pred_binary, average='weighted'),
            'Recall (weighted subj avg)': lambda: recall_score(
                y_true, y_pred_binary, average='weighted'),
            'F1 score (weighted subj avg)': lambda: f1_score(
                y_true, y_pred_binary, average='weighted'),
            'Precision (microavg)': lambda: precision_score(
                y_true, y_pred_binary, average='micro'),
            'Recall (microavg)': lambda: recall_score(
                y_true, y_pred_binary, average='micro'),
            'F1 score (microavg)': lambda: f1_score(
                y_true, y_pred_binary, average='micro'),
            'F1@5': lambda: f1_score(
                y_true, filter_pred_top_k(y_pred, 5) > 0.0, average='samples'),
            'NDCG': lambda: ndcg_score(y_true, y_pred),
            'NDCG@5': lambda: ndcg_score(y_true, y_pred, limit=5),
            'NDCG@10': lambda: ndcg_score(y_true, y_pred, limit=10),
            'Precision@1': lambda: precision_at_k_score(
                y_true, y_pred, limit=1),
            'Precision@3': lambda: precision_at_k_score(
                y_true, y_pred, limit=3),
            'Precision@5': lambda: precision_at_k_score(
                y_true, y_pred, limit=5),
            'LRAP': lambda: label_ranking_average_precision_score(
                y_true, y_pred),
            'True positives': lambda: true_positives(
                y_true, y_pred_binary),
            'False positives': lambda: false_positives(
                y_true, y_pred_binary),
            'False negatives': lambda: false_negatives(
                y_true, y_pred_binary),
        }

        if metrics == 'all':
            metrics = all_metrics.keys()

        with warnings.catch_warnings():
            warnings.simplefilter('ignore')

            return {metric: all_metrics[metric]() for metric in metrics}

    def _result_per_subject_header(self, results_file):
        print('\t'.join(['URI',
                         'Label',
                         'Support',
                         'True_positives',
                         'False_positives',
                         'False_negatives',
                         'Precision',
                         'Recall',
                         'F1_score']),
              file=results_file)

    def _result_per_subject_body(self, zipped_results, results_file):
        for row in zipped_results:
            print('\t'.join((str(e) for e in row)), file=results_file)


    def output_result_per_subject(self, y_true, y_pred, results_file):
        """Write results per subject (non-aggregated)
        to outputfile results_file"""

        y_pred = y_pred.T > 0.0
        y_true = y_true.T > 0.0

        true_pos = (y_true & y_pred)
        false_pos = (~y_true & y_pred)
        false_neg = (y_true & ~y_pred)

        r = len(y_true)

        zipped = zip(self._subject_index._uris,               # URI
                     self._subject_index._labels,             # Label
                     np.sum((true_pos + false_neg), axis=1),  # Support
                     np.sum(true_pos, axis=1),                # True_positives
                     np.sum(false_pos, axis=1),               # False_positives
                     np.sum(false_neg, axis=1),               # False_negatives
                     [precision_score(y_true[i], y_pred[i], zero_division=0)
                      for i in range(r)],                     # Precision
                     [recall_score(y_true[i], y_pred[i], zero_division=0)
                      for i in range(r)],                     # Recall
                     [f1_score(y_true[i], y_pred[i], zero_division=0)
                      for i in range(r)])                     # F1
        self._result_per_subject_header(results_file)
        self._result_per_subject_body(zipped, results_file)

    def results(self, metrics='all', results_file=None, warnings=False):
        """evaluate a set of selected subjects against a gold standard using
        different metrics. The set of metrics can be either 'all' or 'simple'.
        If results_file (file object) given, write results per subject to it"""

        if not self._samples:
            raise NotSupportedException("cannot evaluate empty corpus")

        y_true = np.array([gold_subjects.as_vector(self._subject_index,
                                                   warnings=warnings)
                           for hits, gold_subjects in self._samples])
        y_pred = np.array([hits.as_vector(self._subject_index)
                           for hits, gold_subjects in self._samples],
                          dtype=np.float32)

        results = self._evaluate_samples(
            y_true, y_pred, metrics)
        results['Documents evaluated'] = y_true.shape[0]

        if results_file:
            self.output_result_per_subject(y_true, y_pred, results_file)
        return results


1			"""Evaluation metrics for Annif"""
2
3			import statistics
4			import warnings
5			import numpy as np
6			from sklearn.metrics import precision_score, recall_score, f1_score
7			from sklearn.metrics import label_ranking_average_precision_score
8			from annif.exception import NotSupportedException
9
10
11			def filter_pred_top_k(preds, limit):
12			"""filter a 2D prediction vector, retaining only the top K suggestions
13			for each individual prediction; the rest will be set to zeros"""
14
15			masks = []
16			for pred in preds:
17			mask = np.zeros_like(pred, dtype=np.bool)
18			top_k = np.argsort(pred)[::-1][:limit]
19			mask[top_k] = True
20			masks.append(mask)
21			return preds * np.array(masks)
22
23
24			def true_positives(y_true, y_pred):
25			"""calculate the number of true positives using bitwise operations,
26			emulating the way sklearn evaluation metric functions work"""
27			return (y_true & y_pred).sum()
28
29
30			def false_positives(y_true, y_pred):
31			"""calculate the number of false positives using bitwise operations,
32			emulating the way sklearn evaluation metric functions work"""
33			return (~y_true & y_pred).sum()
34
35
36			def false_negatives(y_true, y_pred):
37			"""calculate the number of false negatives using bitwise operations,
38			emulating the way sklearn evaluation metric functions work"""
39			return (y_true & ~y_pred).sum()
40
41
42			def precision_at_k_score(y_true, y_pred, limit):
43			"""calculate the precision at K, i.e. the number of relevant items
44			among the top K predicted ones"""
45			scores = []
46			for true, pred in zip(y_true, y_pred):
47			order = pred.argsort()[::-1]
48			orderlimit = min(limit, np.count_nonzero(pred))
49			order = order[:orderlimit]
50			gain = true[order]
51			if orderlimit > 0:
52			scores.append(gain.sum() / orderlimit)
53			else:
54			scores.append(0.0)
55			return statistics.mean(scores)
56
57
58			def dcg_score(y_true, y_pred, limit=None):
59			"""return the discounted cumulative gain (DCG) score for the selected
60			labels vs. relevant labels"""
61			order = y_pred.argsort()[::-1]
62			n_pred = np.count_nonzero(y_pred)
63			if limit is not None:
64			n_pred = min(limit, n_pred)
65			order = order[:n_pred]
66			gain = y_true[order]
67			discount = np.log2(np.arange(order.size) + 2)
68
69			return (gain / discount).sum()
70
71
72			def ndcg_score(y_true, y_pred, limit=None):
73			"""return the normalized discounted cumulative gain (nDCG) score for the
74			selected labels vs. relevant labels"""
75			scores = []
76			for true, pred in zip(y_true, y_pred):
77			idcg = dcg_score(true, true, limit)
78			dcg = dcg_score(true, pred, limit)
79			if idcg > 0:
80			scores.append(dcg / idcg)
81			else:
82			scores.append(1.0) # perfect score for no relevant hits case
83			return statistics.mean(scores)
84
85
86			class EvaluationBatch:
87			"""A class for evaluating batches of results using all available metrics.
88			The evaluate() method is called once per document in the batch.
89			Final results can be queried using the results() method."""
90
91			def __init__(self, subject_index):
92			self._subject_index = subject_index
93			self._samples = []
94
95			def evaluate(self, hits, gold_subjects):
96			self._samples.append((hits, gold_subjects))
97
98			def _evaluate_samples(self, y_true, y_pred, metrics='all'):
99			y_pred_binary = y_pred > 0.0
100
101			# define the available metrics as lazy lambda functions
102			# so we can execute only the ones actually requested
103			all_metrics = {
104			'Precision (doc avg)': lambda: precision_score(
105			y_true, y_pred_binary, average='samples'),
106			'Recall (doc avg)': lambda: recall_score(
107			y_true, y_pred_binary, average='samples'),
108			'F1 score (doc avg)': lambda: f1_score(
109			y_true, y_pred_binary, average='samples'),
110			'Precision (subj avg)': lambda: precision_score(
111			y_true, y_pred_binary, average='macro'),
112			'Recall (subj avg)': lambda: recall_score(
113			y_true, y_pred_binary, average='macro'),
114			'F1 score (subj avg)': lambda: f1_score(
115			y_true, y_pred_binary, average='macro'),
116			'Precision (weighted subj avg)': lambda: precision_score(
117			y_true, y_pred_binary, average='weighted'),
118			'Recall (weighted subj avg)': lambda: recall_score(
119			y_true, y_pred_binary, average='weighted'),
120			'F1 score (weighted subj avg)': lambda: f1_score(
121			y_true, y_pred_binary, average='weighted'),
122			'Precision (microavg)': lambda: precision_score(
123			y_true, y_pred_binary, average='micro'),
124			'Recall (microavg)': lambda: recall_score(
125			y_true, y_pred_binary, average='micro'),
126			'F1 score (microavg)': lambda: f1_score(
127			y_true, y_pred_binary, average='micro'),
128			'F1@5': lambda: f1_score(
129			y_true, filter_pred_top_k(y_pred, 5) > 0.0, average='samples'),
130			'NDCG': lambda: ndcg_score(y_true, y_pred),
131			'NDCG@5': lambda: ndcg_score(y_true, y_pred, limit=5),
132			'NDCG@10': lambda: ndcg_score(y_true, y_pred, limit=10),
133			'Precision@1': lambda: precision_at_k_score(
134			y_true, y_pred, limit=1),
135			'Precision@3': lambda: precision_at_k_score(
136			y_true, y_pred, limit=3),
137			'Precision@5': lambda: precision_at_k_score(
138			y_true, y_pred, limit=5),
139			'LRAP': lambda: label_ranking_average_precision_score(
140			y_true, y_pred),
141			'True positives': lambda: true_positives(
142			y_true, y_pred_binary),
143			'False positives': lambda: false_positives(
144			y_true, y_pred_binary),
145			'False negatives': lambda: false_negatives(
146			y_true, y_pred_binary),
147			}
148
149			if metrics == 'all':
150			metrics = all_metrics.keys()
151
152			with warnings.catch_warnings():
153			warnings.simplefilter('ignore')
154
155			return {metric: all_metrics[metric]() for metric in metrics}
156
157			def _result_per_subject_header(self, results_file):
158			print('\t'.join(['URI',
159			'Label',
160			'Support',
161			'True_positives',
162			'False_positives',
163			'False_negatives',
164			'Precision',
165			'Recall',
166			'F1_score']),
167			file=results_file)
168
169			def _result_per_subject_body(self, zipped_results, results_file):
170			for row in zipped_results:
171			print('\t'.join((str(e) for e in row)), file=results_file)
			0 ignored issues – show introduced 2020-03-20 15:53 UTC by Report Bug Copy Issue Report The variable `e` does not seem to be defined in case the `for` loop on line `170` is not entered. Are you sure this can never be the case? Loading history...
172
173			def output_result_per_subject(self, y_true, y_pred, results_file):
174			"""Write results per subject (non-aggregated)
175			to outputfile results_file"""
176
177			y_pred = y_pred.T > 0.0
178			y_true = y_true.T > 0.0
179
180			true_pos = (y_true & y_pred)
181			false_pos = (~y_true & y_pred)
182			false_neg = (y_true & ~y_pred)
183
184			r = len(y_true)
185
186			zipped = zip(self._subject_index._uris, # URI
187			self._subject_index._labels, # Label
188			np.sum((true_pos + false_neg), axis=1), # Support
189			np.sum(true_pos, axis=1), # True_positives
190			np.sum(false_pos, axis=1), # False_positives
191			np.sum(false_neg, axis=1), # False_negatives
192			[precision_score(y_true[i], y_pred[i], zero_division=0)
193			for i in range(r)], # Precision
194			[recall_score(y_true[i], y_pred[i], zero_division=0)
195			for i in range(r)], # Recall
196			[f1_score(y_true[i], y_pred[i], zero_division=0)
197			for i in range(r)]) # F1
198			self._result_per_subject_header(results_file)
199			self._result_per_subject_body(zipped, results_file)
200
201			def results(self, metrics='all', results_file=None, warnings=False):
202			"""evaluate a set of selected subjects against a gold standard using
203			different metrics. The set of metrics can be either 'all' or 'simple'.
204			If results_file (file object) given, write results per subject to it"""
205
206			if not self._samples:
207			raise NotSupportedException("cannot evaluate empty corpus")
208
209			y_true = np.array([gold_subjects.as_vector(self._subject_index,
210			warnings=warnings)
211			for hits, gold_subjects in self._samples])
212			y_pred = np.array([hits.as_vector(self._subject_index)
213			for hits, gold_subjects in self._samples],
214			dtype=np.float32)
215
216			results = self._evaluate_samples(
217			y_true, y_pred, metrics)
218			results['Documents evaluated'] = y_true.shape[0]
219
220			if results_file:
221			self.output_result_per_subject(y_true, y_pred, results_file)
222			return results
223

NatLibFi / Annif

Pull Request — master (#414)

annif.eval.false_negatives() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like