text: # AUTOGENERATED! DO NOT EDIT! File to edit: nbs/evaluations/behavioral.ipynb (unless otherwise specified).
  
  __all__ = ['BehavioralEvaluation', 'SimilarItemEvaluation', 'ComplementaryItemEvaluation', 'SessionItemEvaluation']
  
  # Cell
  from datetime import datetime
  from abc import ABC
  from functools import wraps
  from pathlib import Path
  import time
  import random
  import itertools
  import numpy as np
  import collections
  from collections import Counter, defaultdict
  import math
  import os
  import matplotlib.pyplot as plt
  from scipy.spatial.distance import cosine
  import json
  import networkx as nx
  from networkx.algorithms.shortest_paths.generic import shortest_path
  from statistics import mean
  from functools import partial
  
  # Internal Cell
  def rec_test(test_type: str):
      """
      Rec test decorator
      """
      def decorator(f):
          @wraps(f)
          def w(*args, **kwargs):
              return f(*args, **kwargs)
  
          # add attributes to f
          w.is_test = True
          w.test_type = test_type
          try:
              w.test_desc = f.__doc__.lstrip().rstrip()
          except:
              w.test_desc = ""
          try:
              # python 3
              w.name = w.__name__
          except:
              # python 2
              w.name = w.__func__.func_name
          return w
  
      return decorator
  
  # Cell
  class BehavioralEvaluation(ABC):
      def __init__(self, x_train, y_train, x_test, y_test, y_preds, model=None, save_path = '.'):
          self.name = self.__class__.__name__
          self._rec_tests = self.get_tests()
          self.data = {
              'x_train': x_train,
              'y_train': y_train,
              'x_test': x_test,
              'y_test': y_test,
              'y_preds': y_preds,
          }
          self.save_path = save_path
          self.model = model
          self._test_results = []
          self._test_data = {}
          self._dense_repr = {}
  
          assert len(self.data['y_test']) == len(self.data['y_preds'])
  
      def _statistics(self, data=None):
          train_size = len(self.data['x_train'])
          test_size = len(self.data['x_test'])
          # num non-zero preds
          num_preds = len([p for p in self.data['y_preds'] if p])
          return {
              'training_set__size': train_size,
              'test_set_size': test_size,
              'num_non_null_predictions': num_preds
          }
  
      def _sample_hits_at_k(self, k=3, size=3):
          hits = []
          for idx, (_p, _y) in enumerate(zip(self.data['y_preds'], self.data['y_test'])):
              if _y[0] in _p[:k]:
                  hit_info = {
                      'y_test': [_y[0]],
                      'y_preds': _p[:k],
                  }
                  if self.data['x_test']:
                      hit_info['x_test'] = [self.data['x_test'][idx][0]]
                  hits.append(hit_info)
  
          if len(hits) < size or size == -1:
              return hits
          return random.sample(hits, k=size)
  
      def _sample_misses_at_k(self, k=3, size=3, item_only=False):
          x_train = self.data['x_train']
          x_test = self.data['x_test']
          y_test = self.data['y_test']
          y_preds = self.data['y_preds']
          if item_only:
              _x_train = self.item_only(x_train)
              _x_test = self.item_only(x_test)
              _y_test = self.item_only(y_test)
              _y_preds = self.item_only(y_preds)
          misses = []
          for idx, (_p, _y) in enumerate(zip(_y_preds, _y_test)):
              if _y[0] not in _p[:k]:
                  miss_info =  {
                      'y_test': [_y[0]],
                      'y_preds': _p[:k],
                      'y_test_ext': [y_test[idx][0]],
                      'y_preds_ext': y_preds[idx][:k],
                  }
                  if x_test:
                      miss_info['x_test'] = [_x_test[idx][0]]
                      miss_info['x_test_ext'] = [x_test[idx][0]]
                  misses.append(miss_info)
  
          if len(misses) < size or size == -1:
              return misses
          return random.sample(misses, k=size)
  
      def _hit_rate_at_k(self, k=3, item_only=False, nep=False, y_preds=None, y_test=None):
          hits = 0
          y_test = y_test or self.data['y_test'].copy()
          y_preds = y_preds or self.data['y_preds'].copy()
          if item_only:
              y_test = self.item_only(y_test)
              y_preds = self.item_only(y_preds)
          if nep:
              y_test = [[k] for k in y_test]
          for _p, _y in zip(y_preds, y_test):
              if len(set(_p[:k]).intersection(set(_y))) > 0:
                  hits += 1
          return hits / len(y_test)
  
      def _mrr_at_k(self, k=3, item_only=False, nep=False):
          """
          Computes MRR
          :param self.data['y_preds']: predictions, as lists of lists
          :param self.data['y_test']: target data, as lists of lists (eventually [[sku1], [sku2],...]
          :param k: top-k
          """
          rr = []
          y_test = self.data['y_test'].copy()
          y_preds = self.data['y_preds'].copy()
          if item_only:
              y_test = self.item_only(y_test)
              y_preds = self.item_only(y_preds)
          if nep:
              y_test = [[k] for k in y_test]
          for _p, _y in zip(y_preds, y_test):
              for rank, p in enumerate(_p[:k], start=1):
                  if p in _y:
                      rr.append(1 / rank)
                      break
              else:
                  rr.append(0)
          assert len(rr) == len(y_preds)
          return np.mean(rr)
  
      def _coverage_at_k(self, product_data, k=3):
          pred_skus = set(itertools.chain.from_iterable(self.data['y_preds'][:k]))
          all_skus = set(product_data.keys())
          nb_overlap_skus = len(pred_skus.intersection(all_skus))
  
          return nb_overlap_skus / len(all_skus)
  
      def _popularity_bias_at_k(self, k=3, item_only=False):
          x_train = self.data['x_train'].copy()
          y_preds = self.data['y_preds'].copy()
          if item_only:
              x_train = self.item_only(x_train)
              y_preds = self.item_only(y_preds)
          # estimate popularity from training data
          pop_map = collections.defaultdict(lambda : 0)
          num_interactions = 0
          for session in x_train:
              for event in session:
                  pop_map[event] += 1
                  num_interactions += 1
          # normalize popularity
          pop_map = {k:v/num_interactions for k,v in pop_map.items()}
          all_popularity = []
          for p in y_preds:
              average_pop = sum(pop_map.get(_, 0.0) for _ in p[:k]) / len(p) if len(p) > 0 else 0
              all_popularity.append(average_pop)
          return sum(all_popularity) / len(y_preds)
  
      def _precision_at_k(self, k=3):
          precision_ls = [len(set(_y).intersection(set(_p[:k]))) / len(_p) if _p else 1 for _p, _y in zip(self.data['y_preds'], self.data['y_test'])]
          return np.average(precision_ls)
  
      def _recall_at_k(self, k=3):
          recall_ls = [len(set(_y).intersection(set(_p[:k]))) / len(_y) if _y else 1 for _p, _y in zip(self.data['y_preds'], self.data['y_test'])]
          return np.average(recall_ls)
  
      def _hits_distribution_by_rating(self, debug=False):
          """
          Calculates the distribution of hit-rate across movie ratings in testing data
          """
          hits = defaultdict(int)
          total = defaultdict(int)
  
          for target, pred in zip(self.data['y_test'], self.data['y_preds']):
              target_movie_id, target_rating = target[0]["sid"], target[0]["rating"]
              for movie in pred:
                  pred_movie_id, _ = movie["sid"], movie["rating"]
                  if target_movie_id == pred_movie_id:
                      hits[target_rating] += 1
                      break
              total[target_rating] += 1
          hits_distribution_by_rating = {}
          for target_rating in sorted(hits.keys()):
              hit_rate = hits[target_rating] / total[target_rating]
              hits_distribution_by_rating[target_rating] = hit_rate
  
          if debug:
              x_tick_names = list(hits_distribution_by_rating.keys())
              x_tick_idx = list(range(len(x_tick_names)))
              plt.figure(dpi=150)
              plt.bar(
                  x_tick_idx,
                  [hit_rate for hit_rate in hits_distribution_by_rating.values()],
                  align='center'
              )
              plt.xticks(
                  list(range(len(hits_distribution_by_rating))), x_tick_names, fontsize=10
              )
              plt.savefig(os.path.join(self.report_path,
                                      'plots',
                                      'hit_distribution_rating.png'))
              plt.clf()
  
          return hits_distribution_by_rating
  
      @staticmethod
      def roundup(x: int):
          div = 10.0 ** (len(str(x)))
          return int(math.ceil(x / div)) * div
  
      def _hits_distribution(self, k=3, item_only=False, debug=False):
          x_train = self.data['x_train']
          x_test = self.data['x_test']
          y_test = self.data['y_test']
          y_preds = self.data['y_preds']
          if item_only:
              x_train = self.item_only(x_train)
              x_test = self.item_only(x_test)
              y_test = self.item_only(y_test)
              y_preds = self.item_only(y_preds)
          # get product interaction frequency
          prod_interaction_cnt = Counter([_ for x in x_train for _ in x])
          hit_per_interaction_cnt = defaultdict(list)
          for _x, _y_test, _y_preds in zip(x_test, y_test, y_preds):
              _x_cnt = prod_interaction_cnt[_x[0]]
              # TODO: allow for generic metric
              hit_per_interaction_cnt[_x_cnt].append(self._hit_rate_at_k(y_preds=[_y_preds], y_test=[_y_test], k=k))
          # get max product frequency
          max_cnt = prod_interaction_cnt.most_common(1)[0][1]
          # round up to nearest place
          max_cnt = int(self.roundup(max_cnt))
          # generate log-bins
          indices = np.logspace(1, np.log10(max_cnt), num=int(np.log10(max_cnt))).astype(np.int64)
          indices = np.concatenate(([0], indices))
          counts_per_bin = [[_ for i in range(low, high) for _ in hit_per_interaction_cnt[i]]
                          for low, high in zip(indices[:-1], indices[1:])]
  
          histogram = [np.mean(counts) if counts else 0 for counts in counts_per_bin]
          count = [len(counts) for counts in counts_per_bin]
  
          if debug:
              # debug / visualization
              plt.bar(indices[1:], histogram, width=-np.diff(indices) / 1.05, align='edge')
              plt.xscale('log')
              plt.title('HIT Distribution Across Product Frequency')
              # plt.show()
              plt.savefig(os.path.join(self.report_path, 'plots', 'hit_distribution.png'))
              plt.clf()
  
          return {
              'histogram': {int(k): v for k, v in zip(indices[1:], histogram)},
              'counts': {int(k): v for k, v in zip(indices[1:], count)}
          }
  
      def _hits_distribution_by_slice(self, slices, k=3, sample_size=3, debug=False):
          hit_rate_per_slice = defaultdict(dict)
          for slice_name, slice_idx in slices.items():
              # get predictions for slice
              slice_y_preds = [self.data['y_preds'][_] for _ in slice_idx]
              # get labels for slice
              slice_y_test = [self.data['y_test'][_] for _ in slice_idx]
              # TODO: We may want to allow for generic metric to be used here
              slice_hr = self._hit_rate_at_k(y_preds=slice_y_preds, y_test=slice_y_test, k=k)
              # store results
              hit_rate_per_slice[slice_name]['hit_rate'] = slice_hr
              # hit_rate_per_slice[slice_name]['hits'] = self._sample_hits_at_k(slicey_preds, slice_self.data['y_test'], k=k, size=sample_size)
              # hit_rate_per_slice[slice_name]['misses'] = self._sample_misses_at_k(slicey_preds, slice_self.data['y_test'], k=k, size=sample_size)
  
          # debug / visualization
          if debug:
              x_tick_names = list(hit_rate_per_slice.keys())
              x_tick_idx = list(range(len(x_tick_names)))
              plt.figure(dpi=150)
              plt.bar(x_tick_idx, [_['hit_rate'] for _ in hit_rate_per_slice.values()], align='center')
              plt.xticks(list(range(len(hit_rate_per_slice))), x_tick_names, rotation=45, fontsize=5)
              plt.savefig(os.path.join(self.report_path,
                                      'plots',
                                      'hit_distribution_slice.png'))
              plt.clf()
  
          # cast to normal dict
          return dict(hit_rate_per_slice)
  
  
      # def error_by_cosine_distance(self, model, k=3, bins=25, debug=False):
      #     if not(hasattr(model.__class__, 'get_vector') and callable(getattr(model.__class__, 'get_vector'))):
      #         error_msg = "Error : Model {} does not support retrieval of vector embeddings".format(model.__class__)
      #         print(error_msg)
      #         return error_msg
      #     misses = self._sample_misses_at_k(self.data['y_preds'], self.data['y_test'], k=k, size=-1)
      #     cos_distances = []
      #     for m in misses:
      #         if m['y_preds']:
      #             vector_test = model.get_vector(m['y_test'][0])
      #             vector_pred = model.get_vector(m['y_preds'][0])
      #             if vector_pred and vector_test:
      #                 cos_dist = cosine(vector_pred, vector_test)
      #                 cos_distances.append(cos_dist)
  
      #     histogram = np.histogram(cos_distances, bins=bins, density=False)
      #     # cast to list
      #     histogram = (histogram[0].tolist(), histogram[1].tolist())
      #     # debug / viz
      #     if debug:
      #         plt.hist(cos_distances, bins=bins)
      #         plt.title('dist over cosine distance prod space')
      #         plt.savefig(os.path.join(self.report_path,
      #                                 'plots',
      #                                 'distance_to_predictions.png'))
      #         plt.clf()
      #         # plt.show()
  
      #     return {'mean': np.mean(cos_distances), 'histogram': histogram}
  
      def _distance_to_query(self, k=3, bins=25, item_only=False, debug=False):
          if not(hasattr(self.model.__class__, 'get_vector') and callable(getattr(self.model.__class__, 'get_vector'))):
              error_msg = "Error : Model {} does not support retrieval of vector embeddings".format(self.model.__class__)
              print(error_msg)
              return error_msg
          misses = self._sample_misses_at_k(k=k, size=-1, item_only=item_only)
          x_to_y_cos = []
          x_to_p_cos = []
          for m in misses:
              if m['y_preds']:
                  vector_x = self.model.get_vector(m['x_test_ext'][0])
                  vector_y = self.model.get_vector(m['y_test_ext'][0])
                  vectors_p = [self.model.get_vector(_) for _ in m['y_preds_ext']]
                  c_dists =[]
                  if not vector_x or not vector_y:
                      continue
                  x_to_y_cos.append(cosine(vector_x, vector_y))
                  for v_p in vectors_p:
                      if not v_p:
                          continue
                      cos_dist = cosine(v_p, vector_x)
                      if cos_dist:
                          c_dists.append(cos_dist)
                  if c_dists:
                      x_to_p_cos.append(mean(c_dists))
  
          h_xy = np.histogram(x_to_y_cos, bins=bins, density=False)
          h_xp = np.histogram(x_to_p_cos, bins=bins, density=False)
  
          h_xy = (h_xy[0].tolist(), h_xy[1].tolist())
          h_xp = (h_xp[0].tolist(), h_xp[1].tolist())
  
          # debug / viz
          if debug:
              plt.hist(x_to_y_cos, bins=bins, alpha=0.5)
              plt.hist(x_to_p_cos, bins=bins, alpha=0.5)
              plt.title('distribution of distance to input')
              plt.legend(['Distance from Input to Label',
                          'Distance from Input to Label'],
                      loc='upper right')
              # plt.show()
              plt.savefig(os.path.join(self.report_path,
                                      'plots',
                                      'distance_to_query.png'))
              plt.clf()
  
          return {
              'histogram_x_to_y': h_xy,
              'histogram_x_to_p': h_xp,
              'raw_distances_x_to_y': x_to_y_cos,
              'raw_distances_x_to_p': x_to_p_cos,
          }
  
      # def shortest_path_length():
      #     pass
  
      def _graph_distance_test(self, product_data, k=3):
          path_lengths = []
          misses = self._sample_misses_at_k(self.data['y_preds'], self.data['y_test'], k=k, size=-1)
          for _y, _y_p in zip([_['y_test'] for _ in misses],
                              [_['y_preds'] for _ in misses]):
              if not _y_p:
                  continue
              _y_sku = _y[0]
              _y_p_sku = _y_p[0]
  
              if _y_sku not in product_data or _y_p_sku not in product_data:
                  continue
              if not product_data[_y_sku]['CATEGORIES'] or not product_data[_y_p_sku]['CATEGORIES']:
                  continue
              # extract graph information
              get_nodes = lambda nodes, ancestors="": [] if not nodes else ['_'.join([ancestors, nodes[0]])] + \
                                                              get_nodes(nodes[1:], '_'.join([ancestors, nodes[0]]))
              catA = json.loads(product_data[_y_sku]['CATEGORIES'])
              catB = json.loads(product_data[_y_p_sku]['CATEGORIES'])
              catA_nodes = [get_nodes(c) for c in catA]
              catB_nodes = [get_nodes(c) for c in catB]
              all_nodes = list(set([n for c in catA_nodes + catB_nodes for n in c]))
              all_edges = [(n1, n2) for c in catA_nodes + catB_nodes for n1, n2 in zip(c[:-1], c[1:])]
              all_edges = list(set(all_edges))
  
              # build graph
              G = nx.Graph()
              G.add_nodes_from(all_nodes)
              G.add_edges_from(all_edges)
  
              # get leaves
              cat1_leaves = [c[-1] for c in catA_nodes]
              cat2_leaves = [c[-1] for c in catB_nodes]
  
              all_paths = [shortest_path(G, c1_l, c2_l) for c1_l in cat1_leaves for c2_l in cat2_leaves]
              s_path = min(all_paths, key=len)
              s_path_len = len(s_path) - 1
              path_lengths.append(s_path_len)
  
          histogram = np.histogram(path_lengths, bins=np.arange(0, max(path_lengths)))
          histogram = (histogram[0].tolist(), histogram[1].tolist())
          return {'mean': mean(path_lengths), 'hist': histogram}
  
      def _generic_cosine_distance(self,
                                  embeddings: dict,
                                  type_fn,
                                  k=10,
                                  bins=25,
                                  debug=False):
  
          misses = self._sample_misses_at_k(self.data['y_preds'], self.data['y_test'], k=k, size=-1)
          cos_distances = []
          for m in misses:
              if m['y_test'] and m['y_preds'] and type_fn(m['y_test'][0]) and type_fn(m['y_preds'][0]):
                  vector_test = embeddings.get(type_fn(m['y_test'][0]), None)
                  vector_pred = embeddings.get(type_fn(m['y_preds'][0]), None)
                  if vector_pred and vector_test:
                      cos_dist = cosine(vector_pred, vector_test)
                      cos_distances.append(cos_dist)
  
          # TODO: Maybe sample some examples from the bins
          histogram = np.histogram(cos_distances, bins=bins, density=False)
          # cast to list
          histogram = (histogram[0].tolist(), histogram[1].tolist())
          # debug / viz
          if debug:
              plt.hist(cos_distances, bins=bins)
              plt.title('cosine distance misses')
              plt.savefig(os.path.join(self.report_path,
                                      'plots',
                                      'cosine_distance_over_type.png'))
              plt.clf()
  
          return {'mean': np.mean(cos_distances), 'histogram': histogram}
  
      # def _session_perturbation_test(self,
      #                               perturbation_fn,
      #                               id_fn,
      #                               k):
      #     # generate perturbations
      #     perturbed_pairs = [(perturbation_fn(_x), _y_p) for _x, _y_p in zip(self.data['x_test'], self.data['y_preds']) if perturbation_fn(_x)]
      #     # extract perturbed x and original y
      #     x_test_p, y_preds_o = zip(*perturbed_pairs)
      #     # make new predictions over perturbed inputs
      #     y_preds_n = model.predict(x_test_p)
      #     # extract atomic unit for comparison
      #     y_preds_o, y_preds_n = id_fn(y_preds_o), id_fn(y_preds_n)
      #     # check for overlapping predictions
      #     overlap_ratios = []
      #     for _y_p_n, _y_p_o in zip(y_preds_o, y_preds_n):
      #         if _y_p_n and _y_p_o:
      #             # compute prediction intersection
      #             intersection = set(_y_p_o[:k]).intersection(_y_p_n[:k])
      #             overlap_ratio = len(intersection) / len(_y_p_n[:k])
      #             overlap_ratios.append(overlap_ratio)
      #         elif _y_p_n or _y_p_o:
      #             overlap_ratios.append(0)
      #     return np.mean(overlap_ratios)
  
      def _price_homogeneity(self, product_data, price_sel_fn, bins=25, debug=True):
          """
          :param self.data['y_test']: test data
          :param self.data['y_preds']: predicted data
          :param product_data: catalog data
          :param bins: bins on which we split the data for the product comparison
          :param debug: shows plots
          :param key: key from the product data to find the product price
          @return:
          """
          abs_log_price_diff = []
          for idx, (_y, y_preds) in enumerate(zip(self.data['y_test'], self.data['y_preds'])):
              # need >=1 predictions
              if not y_preds:
                  continue
              # item must be in product data
              if _y[0] not in product_data or y_preds[0] not in product_data:
                  continue
              _y_info = product_data[_y[0]]
              y_preds_info = product_data[y_preds[0]]
              if price_sel_fn(_y_info) and price_sel_fn(y_preds_info):
                  y_item_price = price_sel_fn(_y_info)
                  pred_item_price = price_sel_fn(y_preds_info)
                  price_diff = np.abs(np.log10(pred_item_price/y_item_price))
                  abs_log_price_diff.append(price_diff)
  
          histogram = np.histogram(abs_log_price_diff, bins=bins, density=False)
          histogram = (histogram[0].tolist(), histogram[1].tolist())
          if debug:
              plt.hist(abs_log_price_diff, bins=25)
              plt.savefig(os.path.join(self.report_path,
                                      'plots',
                                      'price_homogeneity.png'))
              plt.clf()
  
          return {
              'mean': np.mean(abs_log_price_diff),
              'histogram': histogram
          }
  
      def train_dense_repr(self, type_name: str, type_fn):
          """
          Train a dense representation over a type of meta-data & store into object
          """
          # type_fn: given a SKU returns some type i.e. brand
          x_train_transformed = [[type_fn(e) for e in session if type_fn(e)] for session in self.data['x_train']]
          wv = train_embeddings(x_train_transformed)
          # store a dict
          self._dense_repr[type_name] = {word: list(wv.get_vector(word)) for word in wv.key_to_index}
  
      def get_tests(self):
          """
          Helper to extract methods decorated with rec_test
          """
  
          nodes = {}
          for _ in self.__dir__():
              if not hasattr(self,_):
                  continue
              func = getattr(self, _)
              if hasattr(func, 'is_test'):
                  nodes[func.name] = func
  
          return nodes
  
      def __call__(self, verbose=True, *args, **kwargs):
          run_epoch_time_ms = round(time.time() * 1000)
          self.report_path = os.path.join(
              self.save_path,
              self.name,
              str(run_epoch_time_ms)
          )
          # create datastore
          Path(os.path.join(self.report_path, 'artifacts')).mkdir(parents=True, exist_ok=True)
          Path(os.path.join(self.report_path, 'results')).mkdir(parents=True, exist_ok=True)
          Path(os.path.join(self.report_path, 'plots')).mkdir(parents=True, exist_ok=True)
  
          # iterate through tests
          for test_func_name, test in self._rec_tests.items():
              test_result = test(*args, **kwargs)
              # we could store the results in the test function itself
              # test.__func__.test_result = test_result
              self._test_results.append({
                  'test_name': test.test_type,
                  'description': test.test_desc,
                  'test_result': test_result}
              )
              if verbose:
                  print("============= TEST RESULTS ===============")
                  print("Test Type        : {}".format(test.test_type))
                  print("Test Description : {}".format(test.test_desc))
                  print("Test Result      : {}\n".format(test_result))
          # at the end, we dump it locally
          if verbose:
              print("Generating reports at {}".format(datetime.utcnow()))
          # now, dump results
          self.dump_results_to_json(self._test_results, run_epoch_time_ms)
          # now, store artifacts
          self.store_artifacts()
  
      def store_artifacts(self):
          target_path = os.path.join(self.report_path, 'artifacts')
          # store predictions
          with open(os.path.join(target_path, 'model_predictions.json'), 'w') as f:
              json.dump({
                  'x_test': self.data['x_test'],
                  'y_test': self.data['y_test'],
                  'y_preds': self.data['y_preds']
              }, f)
  
      def dump_results_to_json(self, test_results: list, epoch_time_ms: int):
          target_path = os.path.join(self.report_path, 'results')
          # make sure the folder is there, with all intermediate parents
          Path(target_path).mkdir(parents=True, exist_ok=True)
          report = {
              'metadata': {
                  'run_time': epoch_time_ms,
                  'reclist': self.name,
                  'tests': list(self._rec_tests.keys())
              },
              'data': test_results
          }
          with open(os.path.join(target_path, 'report.json'), 'w') as f:
              json.dump(report, f)
  
      @property
      def test_results(self):
          return self._test_results
  
      @property
      def test_data(self):
          return self._test_data
  
      @property
      def rec_tests(self):
          return self._rec_tests
  
  # Cell
  class SimilarItemEvaluation(BehavioralEvaluation):
  
      @rec_test(test_type="stats")
      def basic_stats(self):
          """
          Basic statistics on training, test and prediction data
          """
          return self._statistics()
  
      @rec_test(test_type='HR@10')
      def hit_rate_at_k(self):
          """
          Compute the rate at which the top-k predictions contain the item to be predicted
          """
          return self._hit_rate_at_k(k=10, item_only=True)
  
      @rec_test(test_type='hits_distribution')
      def hits_distribution(self):
          """
          Compute the distribution of hit-rate across item frequency in training data
          """
          return self._hits_distribution(
              k=10,
              item_only=True,
              debug=True
          )
  
      @rec_test(test_type="hits_distribution_by_rating")
      def hits_distribution_by_rating(self):
          """
          Compute the distribution of hit-rate across item ratings in testing data
          """
          return self._hits_distribution_by_rating(
              debug=True
          )
  
      def item_only(self, items):
          return [[x["sid"] for x in y] for y in items]
  
  # Cell
  class ComplementaryItemEvaluation(BehavioralEvaluation):
  
      @rec_test(test_type='stats')
      def basic_stats(self):
          """
          Basic statistics on training, test and prediction data
          """
          return self._statistics()
  
      # @rec_test(test_type='price_homogeneity')
      # def price_test(self):
      #     """
      #     Measures the absolute log ratio of ground truth and prediction price
      #     """
      #     return price_homogeneity_test(y_test=self.sku_only(self._y_test),
      #                                   y_preds=self.sku_only(self._y_preds),
      #                                   product_data=self.product_data,
      #                                   price_sel_fn=lambda x: float(x['price_bucket'])
      #                                                          if x['price_bucket']
      #                                                          else None
      #                                   )
  
      # @rec_test(test_type='Coverage@10')
      # def coverage_at_k(self):
      #     """
      #     Coverage is the proportion of all possible products which the RS
      #     recommends based on a set of sessions
      #     """
      #     from reclist.metrics.standard_metrics import coverage_at_k
      #     return coverage_at_k(self.sku_only(self._y_preds),
      #                          self.product_data,
      #                          k=10)
  
      @rec_test(test_type='HR@10')
      def hit_rate_at_k(self):
          """
          Compute the rate at which the top-k predictions contain the item to be predicted
          """
          return self._hit_rate_at_k(k=10, item_only=True)
  
      @rec_test(test_type='hits_distribution')
      def hits_distribution(self):
          """
          Compute the distribution of hit-rate across item frequency in training data
          """
          return self._hits_distribution(
              k=10,
              item_only=True,
              debug=True
          )
  
      @rec_test(test_type='distance_to_query')
      def dist_to_query(self):
          """
          Compute the distribution of distance from query to label and query to prediction
          """
          return self._distance_to_query(k=10, bins=25, item_only=True, debug=True)
  
      def item_only(self, items):
          return [[x["sid"] for x in y] for y in items]
  
  # Cell
  class SessionItemEvaluation(BehavioralEvaluation):
  
      @rec_test(test_type='stats')
      def basic_stats(self):
          """
          Basic statistics on training, test and prediction data
          """
          return self._statistics()
  
      @rec_test(test_type='HR@10')
      def hit_rate_at_k(self):
          """
          Compute the rate at which the top-k predictions contain the item to be predicted
          """
          return self._hit_rate_at_k(k=10, item_only=True)
  
      # @rec_test(test_type='perturbation_test')
      # def perturbation_at_k(self):
      #     """
      #     Compute average consistency in model predictions when inputs are perturbed
      #     """
      #     # Step 1: Generate a map from artist uri to track uri
      #     substitute_mapping = defaultdict(list)
      #     for track_uri, row in self.product_data.items():
      #         substitute_mapping[row['artist_uri']].append(track_uri)
  
      #     # Step 2: define a custom perturbation function
      #     def perturb(session, sub_map):
      #         last_item = session[-1]
      #         last_item_artist = self.product_data[last_item['track_uri']]['artist_uri']
      #         substitutes = set(sub_map.get(last_item_artist,[])) - {last_item['track_uri']}
      #         if substitutes:
      #             similar_item = random.sample(substitutes, k=1)
      #             new_session = session[:-1] + [{"track_uri": similar_item[0]}]
      #             return new_session
      #         return []
  
      #     # Step 3: call test
      #     return session_perturbation_test(self.rec_model,
      #                                      partial(perturb, sub_map=substitute_mapping),
      #                                      self.uri_only,
      #                                      k=10)
  
      # @rec_test(test_type='shuffle_session')
      # def perturbation_shuffle_at_k(self):
      #     """
      #     Compute average consistency in model predictions when inputs are re-ordered
      #     """
      #     # Step 1: define a custom perturbation function
      #     def perturb(session):
      #         return random.sample(session, len(session))
  
      #     # Step 2: call test
      #     return self._session_perturbation_test(perturb, k=10)
  
      # @rec_test(test_type='hits_distribution_by_slice')
      # def hits_distribution_by_slice(self):
      #     """
      #     Compute the distribution of hit-rate across various slices of data
      #     """
      #     len_map = collections.defaultdict(list)
      #     for idx, playlist in enumerate(self._x_test):
      #         len_map[len(playlist)].append(idx)
      #     slices = collections.defaultdict(list)
      #     bins = [(x * 5, (x + 1) * 5) for x in range(max(len_map) // 5 + 1)]
      #     for bin_min, bin_max in bins:
      #         for i in range(bin_min + 1, bin_max + 1, 1):
      #             slices[f'({bin_min}, {bin_max}]'].extend(len_map[i])
      #             del len_map[i]
      #     assert len(len_map) == 0
  
      #     return hits_distribution_by_slice(slices,
      #                                       self.uri_only(self._y_test),
      #                                       self.uri_only(self._y_preds),
      #                                       debug=True)
  
      # @rec_test(test_type='Coverage@10')
      # def coverage_at_k(self):
      #     """
      #     Coverage is the proportion of all possible products which the RS
      #     recommends based on a set of sessions
      #     """
      #     from reclist.metrics.standard_metrics import coverage_at_k
      #     return coverage_at_k(self.uri_only(self._y_preds),
      #                          self.product_data,
      #                          # this contains all the track URIs from train and test sets
      #                          k=10)
  
      @rec_test(test_type='Popularity@10')
      def popularity_bias_at_k(self):
          """
          Compute average frequency of occurrence across recommended items in training data
          """
          return self._popularity_bias_at_k(k=10, item_only=True)
  
      @rec_test(test_type='MRR@10')
      def mrr_at_k(self):
          """
          MRR calculates the mean reciprocal of the rank at which the first
          relevant item was retrieved
          """
          return self._mrr_at_k(item_only=True)
  
      def item_only(self, items):
          return [[x["sid"] for x in y] for y in items]