code: # 导入包
  import random
  import math
  import numpy as np
  import time
  from tqdm import tqdm
  from tqdm import trange
  
  # 1 通用函数定义
  ## 定义装饰器，监控运行时间
  def timmer(func):
      def wrapper(*args, **kwargs):
          start_time = time.time()
          res = func(*args, **kwargs)
          stop_time = time.time()
          print('Func {},run time:{}'.format(func.__name__,stop_time - start_time))
          return res
      return wrapper
  
  ## 数据处理相关
  ### load data
  ### split data
  class Dataset():
      def __init__(self,fp):
          self.data = self.loadData(fp)
  
      @timmer
      def loadData(self,fp):
          data = []
          for l in open(fp):
              data.append(tuple(map(int, l.strip().split('::')[:2])))
          return data
  
      @timmer
      def splitData(self, M, k, seed=1):
          '''
          :params: data, 加载的所有(user, item)数据条目
          :params: M, 划分的数目，最后需要取M折的平均
          :params: k, 本次是第几次划分，k~[0, M)
          :params: seed, random的种子数，对于不同的k应设置成一样的
          :return: train, test
          '''
          train , test = [], []
          random.seed(seed)
          for user, item in self.data:
              # 这里与书中的不一致，本人认为取M-1较为合理，因randint是左右都覆盖的
              if random.randint(0, M-1) == k:  
                  test.append((user, item))
              else:
                  train.append((user, item))
  
          # 处理成字典的形式，user->set(items)
          def convert_dict(data):
              data_dict = {}
              for user, item in data:
                  if user not in data_dict:
                      data_dict[user] = set()
                  data_dict[user].add(item)
              data_dict = {k: list(data_dict[k]) for k in data_dict}
              return data_dict
  
          return convert_dict(train), convert_dict(test)
  
  ## 评价指标
  ### Precision
  ### Recall
  ### Coverage
  ### Popularity(Novelty)
  
  class Metric():
      def __init__(self, train, test, GetRecommendation):
          '''
          :params: train, 训练数据
          :params: test, 测试数据
          :params: GetRecommendation, 为某个用户获取推荐物品的接口函数
          '''
          self.train = train
          self.test = test
          self.GetRecommendation = GetRecommendation
          self.recs = self.getRec()
          
      # 为test中的每个用户进行推荐
      def getRec(self):
          recs = {}
          for user in self.test:
              rank = self.GetRecommendation(user)
              recs[user] = rank
          return recs
          
      # 定义精确率指标计算方式
      def precision(self):
          all, hit = 0, 0
          for user in self.test:
              test_items = set(self.test[user])
              rank = self.recs[user]
              for item, score in rank:
                  if item in test_items:
                      hit += 1
              all += len(rank)
          return round(hit / all * 100, 2)
      
      # 定义召回率指标计算方式
      def recall(self):
          all, hit = 0, 0
          for user in self.test:
              test_items = set(self.test[user])
              rank = self.recs[user]
              for item, score in rank:
                  if item in test_items:
                      hit += 1
              all += len(test_items)
          return round(hit / all * 100, 2)
      
      # 定义覆盖率指标计算方式
      def coverage(self):
          all_item, recom_item = set(), set()
          for user in self.test:
              for item in self.train[user]:
                  all_item.add(item)
              rank = self.recs[user]
              for item, score in rank:
                  recom_item.add(item)
          return round(len(recom_item) / len(all_item) * 100, 2)
      
      # 定义新颖度指标计算方式
      def popularity(self):
          # 计算物品的流行度
          item_pop = {}
          for user in self.train:
              for item in self.train[user]:
                  if item not in item_pop:
                      item_pop[item] = 0
                  item_pop[item] += 1
  
          num, pop = 0, 0
          for user in self.test:
              rank = self.recs[user]
              for item, score in rank:
                  # 取对数，防止因长尾问题带来的被流行物品所主导
                  pop += math.log(1 + item_pop[item])
                  num += 1
          return round(pop / num, 6)
      
      def eval(self):
          metric = {'Precision': self.precision(),
                      'Recall': self.recall(),
                      'Coverage': self.coverage(),
                      'Popularity': self.popularity()}
          print('Metric:', metric)
          return metric
  
  # 2 LFM算法实现
  def LFM(train,ratio,K,lr,step,lmbda,N):
      '''
      :params: train, 训练数据
      :params: ratio, 负采样的正负比例
      :params: K, 隐语义个数
      :params: lr, 初始学习率
      :params: step, 迭代次数
      :params: lmbda, 正则化系数
      :params: N, 推荐TopN物品的个数
      :return: GetRecommendation, 获取推荐结果的接口
      '''
      all_items = {}
      for user in train:
          for item in train[user]:
              if item not in all_items:
                  all_items[item] = 0
              all_items[item] += 1
  
      all_items = list(all_items.items())
      items = [x[0] for x in all_items]
      pops = [x[1] for x in all_items]
  
      # 负采样函数（按照流行度就行采样）
      def nSample(data,ratio):
          new_data = {}
          # 正样本
          for user in data:
              if user not in new_data:
                  new_data[user] = {}
              for item in data[user]:
                  new_data[user][item] = 1
          # 负样本
          for user in new_data:
              seen = set(new_data[user])
              pos_num = len(seen)
              item = np.random.choice(items, int(pos_num * ratio * 3), pops)
              item = [x for x in item if x not in seen][:int(pos_num * ratio)]
              new_data[user].update({x: 0 for x in item})
          
          return new_data
  
      # 训练
      P, Q = {}, {}
      for user in train:
          P[user] = np.random.random(K)
      for item in items:
          Q[item] = np.random.random(K)
              
      for s in trange(step):
          data = nSample(train, ratio)
          for user in data:
              for item in data[user]:
                  eui = data[user][item] - (P[user] * Q[item]).sum()
                  P[user] += lr * (Q[item] * eui - lmbda * P[user])
                  Q[item] += lr * (P[user] * eui - lmbda * Q[item])
          lr *= 0.9 # 调整学习率
          
      # 获取接口函数
      def GetRecommendation(user):
          seen_items = set(train[user])
          recs = {}
          for item in items:
              if item not in seen_items:
                  recs[item] = (P[user] * Q[item]).sum()
          recs = list(sorted(recs.items(), key=lambda x: x[1], reverse=True))[:N]
          return recs
      
      return GetRecommendation
  
  # 3 LFM实验
  ## M=8, N=10, ratio=[1, 2, 3, 5, 10, 20]
  
  class Experiment():
      def __init__(self, M, N, ratio=1,
                      K=100, lr=0.02, step=100, lmbda=0.01, fp='../dataset/ml-1m/ratings.dat'):
          '''
          :params: M, 进行多少次实验
          :params: N, TopN推荐物品的个数
          :params: ratio, 正负样本比例
          :params: K, 隐语义个数
          :params: lr, 学习率
          :params: step, 训练步数
          :params: lmbda, 正则化系数
          :params: fp, 数据文件路径
          '''
          self.M = M
          self.K = K
          self.N = N
          self.ratio = ratio
          self.lr = lr
          self.step = step
          self.lmbda = lmbda
          self.fp = fp
          self.alg = LFM
      
      # 定义单次实验
      @timmer
      def worker(self, train, test):
          '''
          :params: train, 训练数据集
          :params: test, 测试数据集
          :return: 各指标的值
          '''
          getRecommendation = self.alg(train, self.ratio, self.K, 
                                          self.lr, self.step, self.lmbda, self.N)
          metric = Metric(train, test, getRecommendation)
          return metric.eval()
      
      # 多次实验取平均
      @timmer
      def run(self):
          metrics = {'Precision': 0, 'Recall': 0, 
                      'Coverage': 0, 'Popularity': 0}
          dataset = Dataset(self.fp)
          for ii in range(self.M):
              train, test = dataset.splitData(self.M, ii)
              print('Experiment {}:'.format(ii))
              metric = self.worker(train, test)
              metrics = {k: metrics[k]+metric[k] for k in metrics}
          metrics = {k: metrics[k] / self.M for k in metrics}
          print('Average Result (M={}, N={}, ratio={}): {}'.format(\
                                  self.M, self.N, self.ratio, metrics))
  
  # LFM实验(运行时间较长，这里没贴实验结果)
  M, N = 8, 10
  for r in [1, 2, 3, 5, 10, 20]:
      exp = Experiment(M, N, ratio=r)
      exp.run()
apis: numpy.random.random