-
Notifications
You must be signed in to change notification settings - Fork 125
/
evaluate_foldout.py
82 lines (64 loc) · 2.62 KB
/
evaluate_foldout.py
1
2
3
4
5
6
7
8
9
10
11
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
76
77
78
79
80
81
82
"""
@author: Zhongchuan Sun
"""
import itertools
import numpy as np
import sys
import heapq
from concurrent.futures import ThreadPoolExecutor
def argmax_top_k(a, top_k=50):
ele_idx = heapq.nlargest(top_k, zip(a, itertools.count()))
return np.array([idx for ele, idx in ele_idx], dtype=np.intc)
def precision(rank, ground_truth):
hits = [1 if item in ground_truth else 0 for item in rank]
result = np.cumsum(hits, dtype=np.float)/np.arange(1, len(rank)+1)
return result
def recall(rank, ground_truth):
hits = [1 if item in ground_truth else 0 for item in rank]
result = np.cumsum(hits, dtype=np.float) / len(ground_truth)
return result
def map(rank, ground_truth):
pre = precision(rank, ground_truth)
pre = [pre[idx] if item in ground_truth else 0 for idx, item in enumerate(rank)]
sum_pre = np.cumsum(pre, dtype=np.float32)
gt_len = len(ground_truth)
# len_rank = np.array([min(i, gt_len) for i in range(1, len(rank)+1)])
result = sum_pre/gt_len
return result
def ndcg(rank, ground_truth):
len_rank = len(rank)
len_gt = len(ground_truth)
idcg_len = min(len_gt, len_rank)
# calculate idcg
idcg = np.cumsum(1.0 / np.log2(np.arange(2, len_rank + 2)))
idcg[idcg_len:] = idcg[idcg_len-1]
# idcg = np.cumsum(1.0/np.log2(np.arange(2, len_rank+2)))
dcg = np.cumsum([1.0/np.log2(idx+2) if item in ground_truth else 0.0 for idx, item in enumerate(rank)])
result = dcg/idcg
return result
def mrr(rank, ground_truth):
last_idx = sys.maxsize
for idx, item in enumerate(rank):
if item in ground_truth:
last_idx = idx
break
result = np.zeros(len(rank), dtype=np.float32)
result[last_idx:] = 1.0/(last_idx+1)
return result
def eval_score_matrix_foldout(score_matrix, test_items, top_k=50, thread_num=None):
def _eval_one_user(idx):
scores = score_matrix[idx] # all scores of the test user
test_item = test_items[idx] # all test items of the test user
ranking = argmax_top_k(scores, top_k) # Top-K items
result = []
result.extend(precision(ranking, test_item))
result.extend(recall(ranking, test_item))
result.extend(map(ranking, test_item))
result.extend(ndcg(ranking, test_item))
result.extend(mrr(ranking, test_item))
result = np.array(result, dtype=np.float32).flatten()
return result
with ThreadPoolExecutor(max_workers=thread_num) as executor:
batch_result = executor.map(_eval_one_user, range(len(test_items)))
result = list(batch_result) # generator to list
return np.array(result) # list to ndarray