[추천시스템] (잠재요인)모델 기반 협업 필터링(4) - NMF

NMF(Non-negative Matrix Factorization)

NMF는 음수 미포함 행렬 분해(비음행렬 인수분해)라고 하여 말 그대로 음수를 포함하지 않는 행렬 $X$를 음수를 포함하지 않는 행렬$W$와  $H$의 곱으로 분해하는 알고리즘 입니다.

 

$$X = WH$$

 

NMF와 관련하여 포스팅에 참고한 링크는 다음과 같습니다.

참고링크1 : https://angeloyeo.github.io/2020/10/15/NMF.html
참고링크2 : https://github.com/ahmadvh/Non-Negative-Matrix-factorization---Implemented-in-python

 

음수를 포함하지 않은 행렬(비음행렬) 암시적 피드백 데이터 세트에 일반적으로 사용됩니다.

암시적 피드백 데이터 세트의 예는 다음과 같습니다.

1. 사용자가 특정 아이템에 대하여 구매 하였을 때 아이템 구매 횟수
2. 웹에서 사용자가 특정 아이템에 대해서 클릭한 클릭 횟수
3. 페이스북의 "좋아요" 버튼을 선택한 횟수

- 다른 평점의 종류에 대한 포스팅은 아래에 있습니다.

[추천시스템] 평점의 종류

위의 예시처럼 사용자가 부여한 평점의 표현은 음수가 아닌 임의의 숫자로 표현할 수 있고 이러한 숫자는 아이템에 대해서 호감도를 표현하지만 비호감도를 나타내는 것은 아닙니다.

또한 암시적 피드백 데이터는 보통 신뢰를 나타내고 명시적 피드백의 경우에는 선호를 나타냅니다.

 

NMF의 목적은 $W$, $H$가 최대한 $X$와 가깝게 만드는 것입니다.

최대한 $X$와 가깝다 라는 의미는 $W$, $H$와 $X$의 차이가 0에 가깝다라는 의미이고 차이를 최소화해야합니다.

즉 우리의 목적함수는 프로베니우스 노름(Frobenius norm)$||X-WH||^{2}_{F}$로 정의할 수 있습니다.

$$min(||X-WH||^{2}_{F})$$

 

프로베니우스 노름이 0에 가까워야 하면 거리가 0에 가깝다는 의미로 해석할 수 있습니다.

 

$W$, $H$는 경사하강법을 이용하여 $W$, $H$를 업데이트 할수 있습니다.

위의 참고링크1을 보면 $W$, $H$가 업데이트 하는 과정이 상세하게 잘 나와 있습니다.

 

결론만 가지고 오면

 

$$H \Leftarrow H \circ \frac{W^{T}X}{W^{T}WH}$$

 

$$W \Leftarrow W \circ \frac{XH^{T}}{WHH^{T}}$$  

 

여기서 $\circ$ 기호는 원소별 곱(Hadamard product)를 의미합니다.

 

초기화 단계에서는 (0, 1)의 임의의 값으로 초기화 되며 $\epsilon \approx 10^{-9}$와 같은 작은 값으로 분모가 0이 되지 않게 하여 안정성을 높입니다. 
그리고 $U$, $V$가 업데이트 될때 모든 항목이 "동시에" 업데이트가 됩니다.  

 

식을 다시 수정하면 아래와 같습니다.

$$H \Leftarrow H \circ \frac{W^{T}X}{W^{T}WH + \epsilon}$$
$$W \Leftarrow W \circ \frac{XH^{T}}{WHH^{T}+ \epsilon}$$  

 

간단한 NMF 코드

import numpy as np

R = np.array([[2, 2, 1, 2, 0, 0],
       [2, 3, 3, 3, 0, 0],
       [1, 1, 1, 1, 0, 0],
       [2, 2, 2, 3, 1, 1],
       [0, 0, 0, 1, 1, 1],
       [0, 0, 0, 2, 1, 2]])

예제로 사용할 행렬 $R$을 정의하였습니다.

- 0의 값은 원래 사용자가 평가하지 않은 데이터 입니다.

 

class NMF_CF:
  def __init__(self, R, k, max_iters):
    self.R =R
    self.num_users, self.num_items = R.shape
    self.k = k
    self.max_iters =max_iters
  def fit(self):
    W = np.random.uniform(0, 1, size=(self.num_users, self.k))
    H = np.random.uniform(0, 1, size=(self.k, self.num_items))
    
    norms = []
    e = 1e-10
    for n in range(self.max_iters):
        # Update H
        W_TA = W.T @ self.R 
        W_TWH = W.T @ W @ H
        for i in range(self.k):
            for j in range(self.num_items):
                H[i, j] = H[i, j] * W_TA[i, j] / (W_TWH[i, j] + e)
        # Update W
        AH_T = self.R @ H.T
        WHH_T =  W @ H @ H.T

        for i in range(self.num_users):
            for j in range(self.k):
                W[i, j] = W[i, j] * AH_T[i, j] / (WHH_T[i, j] + e)

        norm = np.linalg.norm(self.R - W@H, 'fro')
        norms.append(norm)
    return norms

생성자(__init__)

self.R : 원본 행렬

self.num_users, self._num_items : $R$이 ($m \times n$)이라고 가정하면 self.num_users = $m$, self.num_items = $n$

self.k : 잠재요인의 수(차원 수)

self.max_iters : 최대 반복 횟수

 

fit

W = np.random.uniform(0, 1, size=(self.num_users, self.k))
H = np.random.uniform(0, 1, size=(self.k, self.num_items))

norms = []
e = 1e-10

먼저 0~1사이의 값으로 $W$, $H$ 행렬을 초기화 합니다.

- 이때 $W$는 ($m \times k$), $H$는 ($k \times n$)입니다.

norms는 실제값과 예측값의 차이를 담을 리스트입니다.

e = $\epsilon$에 해당합니다.

 

for n in range(self.max_iters):
        # Update H
        W_TA = W.T @ self.R 
        W_TWH = W.T @ W @ H
        for i in range(self.k):
            for j in range(self.num_items):
                H[i, j] = H[i, j] * W_TA[i, j] / (W_TWH[i, j] + e)
        # Update W
        AH_T = self.R @ H.T
        WHH_T =  W @ H @ H.T

        for i in range(self.num_users):
            for j in range(self.k):
                W[i, j] = W[i, j] * AH_T[i, j] / (WHH_T[i, j] + e)

        norm = np.linalg.norm(self.R - W@H, 'fro')
        norms.append(norm)

식으로는 아래와 같습니다.

 

$$H \Leftarrow H \circ \frac{W^{T}X}{W^{T}WH + \epsilon}$$
$$W \Leftarrow W \circ \frac{XH^{T}}{WHH^{T}+ \epsilon}$$  

 

mat_iter만큼 반복문을 돌면서 $W$, $H$의 행렬을 업데이트 합니다.

그리고 나온 예측값 $WH$과 실제값(self.R)의 거리(프로베니우스 노름)을 구하여 norms 리스트에 저장합니다.