SGD를 사용한 Matrix Factorization 알고리즘

• MF(Matrix Factorization)
  • 행렬분해로 추천시스템에서 사용자, 아이템의 관계를 가장 잘 설명하는 P, Q행렬로 분해하는 것을 의미한다.
• MF 알고리즘
  • 1. 잠재요인의 개수 K를 설정
  • 2. 주어진 K에 따라서 P(MxK)와 Q(NxK)행렬을 만들고 초기화한다. 
  • 3. 주어진 P(User Latent Matrix), Q(Item Latent Matrix)을 사용해서 예측 평점을 구한다.  $$\hat{R}=PQ^{^{T}}$$
  • 4. R(User-Item Matrix)에 있는 실제 평점과 예측 평점의 오차를 구하고 이 오차를 줄이기 위해서 P, Q 값을 업데이트한다.
  • 5. 오차가 일정 Threshold 이하가 되거나 미리 정해진 Iteration에 도달할 때 까지 3번으로 돌아가서 반복한다.
• SGD(Stochastic Gradient Descent)
  • 3번에서 오차를 줄이기 위해서 SGD방식을 적용

• 사용자 i의 아이템 j에 대한 예측값 - (1)

$$\hat{r_{ij}} = p_{i}^{T}q_{j} = \sum_{k=1}^{K}p_{ik}q_{kj}$$

• 예측 오차(실제값 - 예측값)

$$e_{ij} = (r_{ij}-\hat{r_{ij}})$$

• 정확도 지표 RMSE를 향상시키기 위해서 오차의 제곱을 최소화해야 하기 때문에 오차의 제곱식

$$e_{ij}^{2} = (r_{ij}-\sum_{k=1}^{K}p_{ik}q_{kj})^{2}$$

• 오차의 제곱식 p와 q에 대해서 편미분

$$\\\ \frac{\partial }{\partial p_{ik}}e_{ij}^{2} = -2(r_{ij}-\hat{r_{ij}})(q_{kj}) = -2e_{ij}q_{kj}\\
\\\ \frac{\partial }{\partial q_{kj}}e_{ij}^{2} = -2(r_{ij}-\hat{r_{ij}})(p_{ik}) = -2e_{ij}p_{ik}$$

• 새로운 p, q 값 update(a는 학습률)

$$\\\ {p}'_{ik} = p_{ik}-{\alpha}\frac{\partial}{\partial p_{ik}}e_{ij}^{2} = p_{ik}+2{\alpha}e_{ij}q_{kj}\\
\\\ {q}'_{ik} = q_{ik}-{\alpha}\frac{\partial}{\partial q_{kj}}e_{ij}^{2} = q_{kj}+2{\alpha}e_{ij}p_{ik}$$

위의 식이 반복되면서 오차가 점차 줄어들면서 원행렬과 유사한 값이 담겨있는 P, Q행렬을 구할 수 있다.

• 과적합 방지를 위한 Regularization term 추가(Beta는 정규화 정도)

$$e_{ij}^{2} = (r_{ij} - \sum_{k=1}^{K}p_{ik}q_{kj})^{2} + \frac{\beta}{2}\sum_{k=1}^{K}(\parallel P \parallel^{2} + \parallel Q \parallel ^{2})$$

• 새로운 p, q 값 update - (2)

$$\\\ {p}'_{ik} = p_{ik}-{\alpha}\frac{\partial}{\partial p_{ik}}e_{ij}^{2} = p_{ik}+{\alpha}(e_{ij}q_{kj}-{\beta}p_{ik})\\ 
\\\ {q}'_{ik} = q_{ik}-{\alpha}\frac{\partial}{\partial q_{kj}}e_{ij}^{2} = q_{kj}+ {\alpha}(e_{ij}p_{ik}-{\beta}q_{kj})$$

• 사용자와 아이템의 경향성을 고려한 (1)식 수정

$$\hat{r_{ij}} = b + bu_{i} + bd_{j} + \sum_{k=1}^{K}p_{ik}q_{kj}$$

bu는 전체평균을 제거한 후 사용자 i의 평가경향, bd는 전체평균을 제거한 후 아이템 j의 평가경향이라고 생각하면 된다.

• 사용자 평가경향, 아이템 평가 경향 값 update - (4)

$$\\\ {bu}'_{i} = bu_{i}+ {\alpha}(e_{ij} - {\beta}bu_{i})\\ 
\\\ {bd}'_{j} = bd_{j}+ {\alpha}(e_{ij} - {\beta}bd_{j})$$

• 종합하면 식(2)와 식(4)를 사용해서 P, Q행렬과 bu, bd를 반복적으로 값을 update 진행한다.

---

코드

class MF():
    def __init__(self, ratings, K, alpha, beta, iteration, verbose=True):
        self.R = np.array(ratings)                        # 원행렬
        self.num_users, self.num_items = np.shape(self.R) # 유저, 아이템 수
        self.K = K                                        # 잠재요인 수
        self.alpha = alpha                                # 학습률
        self.beta = beta                                  # 정규화 계수
        self.iteration = iteration                        # 반복횟수
        self.verbose = verbose                            # 중간과정 표시
        
    
    def RMSE(self):
        xs, ys = self.R.nonzero()
        self.predictions = []
        self.errors = []
        for x, y in zip(xs, ys):
            prediction = self.get_prediction(x, y)
            self.predictions.append(prediction)
            self.errors.append(self.R[x, y] - prediction)
        
        # 계산을 위해 numpy array로 변경
        self.predictions = np.array(self.predictions)
        self.errors = np.array(self.errors)
        return np.sqrt(np.mean(self.errors**2))
    
    
    def train(self): 
        # 초기값 설정
        self.P = np.random.normal(scale=1./self.K, size=(self.num_users, self.K))
        self.Q = np.random.normal(scale=1./self.K, size=(self.num_items, self.K))

        # bias 초기값 설정
        self.b_u = np.zeros(self.num_users)
        self.b_d = np.zeros(self.num_items)
        self.b = np.mean(self.R[self.R.nonzero()])

        # 값이 존재하는 data 불러오기
        rows, columns = self.R.nonzero()
        self.samples = [(i, j, self.R[i,j]) for i, j in zip(rows, columns)]

        # SGD 진행
        training_process = []
        for i in range(self.iteration):
            np.random.shuffle(self.samples)
            self.sgd()
            rmse = self.RMSE()
            training_process.append((i+1, rmse))
            if self.verbose:
                if (i+1) % 10 == 0:
                    print("Iteration: %d : Train RMSE = %.4f " % (i+1, rmse))
        return training_process
    
    def get_prediction(self, i, j):
        prediction = self.b + self.b_u[i] + self.b_d[j] + self.P[i,:].dot(self.Q[j,:].T)
        return prediction
    
    def sgd(self):
        for i, j, r in self.samples:
            prediction = self.get_prediction(i, j)
            e = (r - prediction)

            self.b_u[i] += self.alpha * (e - self.beta * self.b_u[i])
            self.b_d[j] += self.alpha * (e - self.beta * self.b_d[j])

            self.P[i, :] += self.alpha * (e * self.Q[j, :] - self.beta * self.P[i,:])
            self.Q[j, :] += self.alpha * (e * self.P[i, :] - self.beta * self.Q[j,:])

 

---

출처 : Python을 이용한 개인화 추천시스템