推薦系統實踐 0x07 基於鄰域的演算法（2）
阿新 • • 發佈：2020-11-26
# 基於鄰域的演算法（2）
上一篇我們講了基於使用者的協同過濾演算法，基本流程就是尋找與目標使用者興趣相似的使用者，按照他們對物品喜好的對目標使用者進行推薦，其中哪些相似使用者的評分要帶上目標使用者與相似使用者的相似度作為權重來計算。但是，基於使用者的協同過濾演算法存在一些弊端，如計算使用者興趣相似度矩陣將越來越困難，其運算時間複雜度和空間複雜度的增長和使用者數的增長近似於平方關係，另外也很難對推薦結果進行解釋。那麼，這一篇我們繼續來了解一下基於物品的協同過濾演算法。
## 基於物品的協同過濾演算法（ItemCF）
基於物品的協同過濾演算法是大多數網站常用的推薦演算法的基礎。ItemCF不會利用物品的內容屬性計算物品之間的相似度，而是分析使用者的行為記錄計算物品之間的相似度。那麼，ItemCF主要分為兩個步驟：
1. 計算物品之間的相似度。
2. 根據物品的相似度和使用者的歷史行為給使用者生成推薦列表。
### 物品相似度
我們可以用如下公式定義物品的相似度：
$$
w_{ij}=\frac{|N(i)\cap N(j)|}{|N(i)|}
$$
$N(i)$是指喜歡物品$i$的使用者數量，分子部分表示既喜歡物品$i$又喜歡物品$j$的使用者有多少，整個相似度公式表示的是喜歡物品$i$的使用者中，同時喜歡物品$j$的使用者比例是多少。可以使用**歸一化**之後的結果作為物品相似度。但是如果物品$j$很熱門人人都喜歡，那麼整個相似度就會變成1，這對於推薦冷門物品的推薦系統來說並不是好事情，所以我們對物品相似度公式進行改進。
$$
w_{ij}=\frac{|N(i)\cap N(j)|}{\sqrt{|N(i)||N(j)|}}
$$
這個公式懲罰了熱門物品$j$的權重，一定程度緩和了這個問題。同樣的，我們在計算物品相似度的時候可以先建立一個**使用者-物品的倒排表**，虛擬碼如下：
```python
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雖然沒有利用內容屬性計算相似度，但是最後得到的結果仍然是內容上某種相似的，如同主演，同分類等等的電影。在得到物品相似度之後，我們用如下公式計算使用者對物品的興趣：
$$
p_{uj}=\sum_{i\in N(u)\cap S(j,K)}w_{ji}r_{ui}
$$
這裡$N(u)$是使用者喜歡的物品的集合，$S(j,K)$是和物品$j$最相似的$K$個物品的集合，$w_{ji}$是物品$j$和$i$的相似度，$r_{ui}$是使用者$u$對物品$i$的興趣。
```python
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
```
另外加就是使用者活躍度對物品相似度產生的影響。一個不活躍的使用者含有大量的感興趣的物品，那麼會產生稠密的物品相似度大矩陣，所以活躍使用者對物品相似度的貢獻應該小於不活躍的使用者。那麼公式修正為：
$$
w_{ij}=\frac{\sum_{u\in N(i)\cap N(j)}\frac{1}{\log 1+|N(u)|}}{\sqrt{|N(i)||N(j)|}}
$$
跟基於使用者的協同過濾的修正公式很像啊。
```python
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資料集上的實驗程式碼：
```python
# 匯入包
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 wrapper
class 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 GetRecommendation
class 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()
```
## 參考
[《推薦系統實踐》(項亮等著) —— 程式碼實現](https://github.com/Magic-Bubble/RecommendSystemPr