推荐系统实践 0x07 基于邻域的算法（2）

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

基于邻域的算法（2）

上一篇我们讲了基于用户的协同过滤算法，基本流程就是寻找与目标用户兴趣相似的用户，按照他们对物品喜好的对目标用户进行推荐，其中哪些相似用户的评分要带上目标用户与相似用户的相似度作为权重来计算。但是，基于用户的协同过滤算法存在一些弊端，如计算用户兴趣相似度矩阵将越来越困难，其运算时间复杂度和空间复杂度的增长和用户数的增长近似于平方关系，另外也很难对推荐结果进行解释。那么，这一篇我们继续来了解一下基于物品的协同过滤算法。

基于物品的协同过滤算法（ItemCF）

基于物品的协同过滤算法是大多数网站常用的推荐算法的基础。ItemCF不会利用物品的内容属性计算物品之间的相似度，而是分析用户的行为记录计算物品之间的相似度。那么，ItemCF主要分为两个步骤：

1. 计算物品之间的相似度。
2. 根据物品的相似度和用户的历史行为给用户生成推荐列表。

物品相似度

我们可以用如下公式定义物品的相似度：

\(N(i)\)是指喜欢物品\(i\)的用户数量，分子部分表示既喜欢物品\(i\)又喜欢物品\(j\)的用户有多少，整个相似度公式表示的是喜欢物品\(i\)的用户中，同时喜欢物品\(j\)的用户比例是多少。可以使用归一化之后的结果作为物品相似度。但是如果物品\(j\)很热门人人都喜欢，那么整个相似度就会变成1，这对于推荐冷门物品的推荐系统来说并不是好事情，所以我们对物品相似度公式进行改进。

这个公式惩罚了热门物品\(j\)的权重，一定程度缓和了这个问题。同样的，我们在计算物品相似度的时候可以先建立一个用户-物品的倒排表，伪代码如下：

def ItemSimilarity(train):
# calculate co-rated users between items
C = dict()
N = dict()
for u, items in train.items():
for i in users:
N[i] += 1
for j in users:
if i == j:
continue
C[i][j] += 1
# finial similarity matrix W W = dict()
for i, related_items in C.items():
for j, cij in related_items.items():
W[u][v] = cij / math.sqrt(N[i] * N[j])
return W


用户对物品的兴趣

ItemCF虽然没有利用内容属性计算相似度，但是最后得到的结果仍然是内容上某种相似的，如同主演，同分类等等的电影。在得到物品相似度之后，我们用如下公式计算用户对物品的兴趣：

这里\(N(u)\)是用户喜欢的物品的集合，\(S(j,K)\)是和物品\(j\)最相似的\(K\)个物品的集合，\(w_{ji}\)是物品\(j\)和\(i\)的相似度，\(r_{ui}\)是用户\(u\)对物品\(i\)的兴趣。

def Recommendation(train, user_id, W, K):
rank = dict()
ru = train[user_id]
for i, pi in ru.items():
for j, wj in sorted(W[i].items(), key=itemgetter(1),
reverse=True)[0:K]:
if j in ru:
continue
rank[j] += pi * wj
return rank


另外加就是用户活跃度对物品相似度产生的影响。一个不活跃的用户含有大量的感兴趣的物品，那么会产生稠密的物品相似度大矩阵，所以活跃用户对物品相似度的贡献应该小于不活跃的用户。那么公式修正为：

跟基于用户的协同过滤的修正公式很像啊。

def ItemSimilarity(train):
#calculate co-rated users between items C = dict()
N = dict()
for u, items in train.items():
for i in users:
N[i] += 1
for j in users:
if i == j:
continue
C[i][j] += 1 / math.log(1 + len(items) * 1.0)
#calculate finial similarity matrix W W = dict()
for i,related_items in C.items():
for j, cij in related_items.items()
W[u][v] = cij / math.sqrt(N[i] * N[j])
return W


还是得感谢@Magic-Bubble分享在github上代码，清晰易懂，省去我重复造轮子的时间。那么给出在MovieLens数据集上的实验代码：

# 导入包
import random
import math
import time
from tqdm import tqdm# 定义装饰器，监控运行时间
def timmer(func):
def wrapper(*args, **kwargs):
start_time = time.time()
res = func(*args, **kwargs)
stop_time = time.time()
print('Func %s, run time: %s' %
(func.__name__, stop_time - start_time))
return res return wrapperclass Dataset():
def __init__(self, fp):
# fp: data file path
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)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# 1. 基于物品余弦相似度的推荐
def ItemCF(train, K, N):
'''
:params: train, 训练数据集
:params: K, 超参数，设置取TopK相似物品数目
:params: N, 超参数，设置取TopN推荐物品数目
:return: GetRecommendation, 推荐接口函数
'''
# 计算物品相似度矩阵
sim = {}
num = {}
for user in train:
items = train[user]
for i in range(len(items)):
u = items[i]
if u not in num:
num[u] = 0
num[u] += 1
if u not in sim:
sim[u] = {}
for j in range(len(items)):
if j == i: continue
v = items[j]
if v not in sim[u]:
sim[u][v] = 0
sim[u][v] += 1
for u in sim:
for v in sim[u]:
sim[u][v] /= math.sqrt(num[u] * num[v]) # 按照相似度排序
sorted_item_sim = {k: list(sorted(v.items(), \
key=lambda x: x[1], reverse=True)) \
for k, v in sim.items()} # 获取接口函数
def GetRecommendation(user):
items = {}
seen_items = set(train[user])
for item in train[user]:
for u, _ in sorted_item_sim[item][:K]:
if u not in seen_items:
if u not in items:
items[u] = 0
items[u] += sim[item][u]
recs = list(sorted(items.items(), key=lambda x: x[1],
reverse=True))[:N]
return recs return GetRecommendation# 2. 基于改进的物品余弦相似度的推荐
def ItemIUF(train, K, N):
'''
:params: train, 训练数据集
:params: K, 超参数，设置取TopK相似物品数目
:params: N, 超参数，设置取TopN推荐物品数目
:return: GetRecommendation, 推荐接口函数
'''
# 计算物品相似度矩阵
sim = {}
num = {}
for user in train:
items = train[user]
for i in range(len(items)):
u = items[i]
if u not in num:
num[u] = 0
num[u] += 1
if u not in sim:
sim[u] = {}
for j in range(len(items)):
if j == i: continue
v = items[j]
if v not in sim[u]:
sim[u][v] = 0
# 相比ItemCF，主要是改进了这里
sim[u][v] += 1 / math.log(1 + len(items))
for u in sim:
for v in sim[u]:
sim[u][v] /= math.sqrt(num[u] * num[v]) # 按照相似度排序
sorted_item_sim = {k: list(sorted(v.items(), \
key=lambda x: x[1], reverse=True)) \
for k, v in sim.items()} # 获取接口函数
def GetRecommendation(user):
items = {}
seen_items = set(train[user])
for item in train[user]:
for u, _ in sorted_item_sim[item][:K]:
# 要去掉用户见过的
if u not in seen_items:
if u not in items:
items[u] = 0
items[u] += sim[item][u]
recs = list(sorted(items.items(), key=lambda x: x[1],
reverse=True))[:N]
return recs return GetRecommendation# 3. 基于归一化的物品余弦相似度的推荐
def ItemCF_Norm(train, K, N):
'''
:params: train, 训练数据集
:params: K, 超参数，设置取TopK相似物品数目
:params: N, 超参数，设置取TopN推荐物品数目
:return: GetRecommendation, 推荐接口函数
'''
# 计算物品相似度矩阵
sim = {}
num = {}
for user in train:
items = train[user]
for i in range(len(items)):
u = items[i]
if u not in num:
num[u] = 0
num[u] += 1
if u not in sim:
sim[u] = {}
for j in range(len(items)):
if j == i: continue
v = items[j]
if v not in sim[u]:
sim[u][v] = 0
sim[u][v] += 1
for u in sim:
for v in sim[u]:
sim[u][v] /= math.sqrt(num[u] * num[v]) # 对相似度矩阵进行按行归一化
for u in sim:
s = 0
for v in sim[u]:
s += sim[u][v]
if s > 0:
for v in sim[u]:
sim[u][v] /= s # 按照相似度排序
sorted_item_sim = {k: list(sorted(v.items(), \
key=lambda x: x[1], reverse=True)) \
for k, v in sim.items()} # 获取接口函数
def GetRecommendation(user):
items = {}
seen_items = set(train[user])
for item in train[user]:
for u, _ in sorted_item_sim[item][:K]:
if u not in seen_items:
if u not in items:
items[u] = 0
items[u] += sim[item][u]
recs = list(sorted(items.items(), key=lambda x: x[1],
reverse=True))[:N]
return recs return GetRecommendationclass Experiment():
def __init__(self, M, K, N, fp='../dataset/ml-1m/ratings.dat',
rt='ItemCF'):
'''
:params: M, 进行多少次实验
:params: K, TopK相似物品的个数
:params: N, TopN推荐物品的个数
:params: fp, 数据文件路径
:params: rt, 推荐算法类型
'''
self.M = M
self.K = K
self.N = N
self.fp = fp
self.rt = rt
self.alg = {
'ItemCF': ItemCF,
'ItemIUF': ItemIUF,
'ItemCF-Norm': ItemCF_Norm
} # 定义单次实验
@timmer
def worker(self, train, test):
'''
:params: train, 训练数据集
:params: test, 测试数据集
:return: 各指标的值
'''
getRecommendation = self.alg[self.rt](train, self.K, 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={}, K={}, N={}): {}'.format(\
self.M, self.K, self.N, metrics))# 1. ItemCF实验
M, N = 8, 10
for K in [5, 10, 20, 40, 80, 160]:
cf_exp = Experiment(M, K, N, rt='ItemCF')
cf_exp.run()# 2. ItemIUF实验
M, N = 8, 10
K = 10 # 与书中保持一致
iuf_exp = Experiment(M, K, N, rt='ItemIUF')
iuf_exp.run()# 3. ItemCF-Norm实验
M, N = 8, 10
K = 10 # 与书中保持一致
norm_exp = Experiment(M, K, N, rt='ItemCF-Norm')
norm_exp.run()


参考

《推荐系统实践》(项亮等著) —— 代码实现