留一法评估 HR 与 NDCG

  Posted by Mr.Zhang on 02 Aug, 2019

   PYTHON   

所谓留一法(leave-one-out cross validation)是指每次只留下一个样本作测试集,其它作为训练集,在推荐系统的模型评估中,计算命中率(hate ratio,HR)和归一化折损累积增益(Normalized Discounted Cumulative Gain,NDCG)所用时间很久,普遍采用留一法。

在MovieLens数据集中,根据时间戳保留最后一个项目评估,可以使用下列方法得到训练和测试数据集:

df.sort_values(by=['user', 'timestamp'], axis=0,
               ascending=True, inplace=True)
users = df.user.unique()
tr_raw, te_raw = None, None
for u in users:
    a = np.array(df[df.user == u].iloc[:-1,:3])
    b = np.array(df[df.user == u].iloc[-1:,:3])
    if tr_raw is None:
        tr_raw = a
    else:
        tr_raw = np.vstack((self.tr_raw, a))
    if te_raw is None:
        te_raw = b
    else:
        te_raw = np.vstack((self.te_raw, b))

所谓HR@10表示评估列表中,目标项目是否在列表中前10位中;NDCG@10则评估的是目标项目在前10位的位置,越靠前,值越大,通常计算公式为$\frac{1}{\log_2(i+2)}$;其中i为位置索引(从0开始索引)。为了节省计算开销和,同时评估多个值,可以采用下列类:

class EvalHRNDCG:
    """评估HR/NDCG类
        注意:使用前,指定被评估模型类变量EvalHRNDCG.Model = model!!!
        在使用多线程处理时,如果使用hybridize()函数后,会导致参数复制
        从而引起输入变量和参数变量不在同一缓存的问题,因此采用类变量
        调用eval_all_rating返回len(test) * 2*len(topK)的
        list[[hr@topk1, ndcg@tok1, hr@topk2,...]...[...]...]
            score = np.array(results)
            score.mean(axis=0)
            计算每列均值
    """
    Model = None

    def __init__(self, test, num_thread, ctx, topK=[5, 10]):
        self.test = test
        self.num_thread = num_thread
        self.ctx = ctx
        self.topK = topK
        self.maxK = max(topK)

    def eval_all_rating(self):
        logging.info('{}开始评估测试集中...'
                     .format(EvalHRNDCG.Model.model_name))
        if(self.num_thread > 1):
            # 使用多进程计算
            pool = multiprocessing.Pool(processes=self.num_thread)
            res = pool.map(self.eval_one_rating, tuple(self.test.keys()))
            pool.close()
            pool.join()
            logging.info('{}进程评估测试集结束.'.format(self.num_thread))
            return res

        results = []
        for u in self.test.keys():
            result = self.eval_one_rating(u)
            results.append(result)
        logging.info('单进程评估测试集结束.')
        return results

    def eval_one_rating(self, u):
        gt_item = self.test.get(u)[0]
        items = self.test.get(u)[1:]
        items.append(gt_item)

        users = np.full(len(items), u, dtype=np.int32)
        preds = EvalHRNDCG.Model(nd.array(users).as_in_context(self.ctx),
                           nd.array(items).as_in_context(self.ctx))
        item_score = dict(zip(items, preds))
        preds_list = heapq.nlargest(self.maxK, item_score, 
                                    key=item_score.get)
        result = []

        for topk in self.topK:
            rank_list = preds_list[:topk]
            hr, ndcg = EvalHRNDCG.get_hr_ndcg(rank_list, gt_item)
            result.append(hr)
            result.append(ndcg)
        return result

    @staticmethod
    def get_hr_ndcg(rank_list, gt_item):
        if gt_item in rank_list:
            idx = rank_list.index(gt_item)
            return 1, np.reciprocal(np.log2(idx + 2))
        return 0, 0.