4장 SGD를 사용한 MF 기본 알고리즘 오류

안녕하세요.

일단 전달주신 부분으로만으로 확인했을 때는 크게 오류가 없어보이는데,

강의에 대한 원본 코드를 드려보겠습니다.

혹시 전달 드린 코드 보시고,

코드 오류가 없는지 확인 부탁드려도 괜찮을까요 ? :)

import os
import numpy as np
import pandas as pd

base_src = 'drive/MyDrive/RecoSys/Data'
u_data_src = os.path.join(base_src,'u.data')
r_cols = ['user_id','movie_id','rating','timestamp']
ratings = pd.read_csv(u_data_src,
        sep = '\t',
        names = r_cols,
        encoding='latin-1')
# timestamp 제거 
ratings = ratings[['user_id','movie_id','rating']].astype(int)


# MF class
# MF라는 이름의 클래스는 MF를 위한 데이터와 메소드를 가진 클래스이다.
class MF():
  # 클래스가 생성될 때 실행되는 초기화 함수이다.
  def __init__(self,ratings,hyper_params):
    # DataFrame 형식으로 전달된 평점(ratings)을 numpy array 형태로 바꿔서 클래스 변수인 self.R에 저장
    self.R = np.array(ratings)
    # 사용자 수(num_users)와 아이템 수(num_iterms)를 받아온다.
    self.num_users,self.num_items = np.shape(self.R)
    # 아래는 MF weight 조절을 위한 하이퍼파라미터이다.
    # K : 잠재요인(latent factor)의 수
    self.K = hyper_params['K']
    # alpha : 학습률
    self.alpha = hyper_params['alpha']
    # beta : 정규화 계수
    self.beta = hyper_params['beta']
    # iterations : SGD의 계산을 할 때의 반복 횟수
    self.iterations = hyper_params['iterations']
    # verbose : SGD의 학습 과정을 중간중간에 출력할 것인지에 대한 여부
    self.verbose = hyper_params['verbose']

  # 현재의 P,Q를 가지고 Root Mean Squared Error(RMSE) 계산하는 함수이다.
  def rmse(self):
    # self.R에서 평점이 있는(0이 아닌) 요소의 인덱스를 가져온다.
    xs, ys = self.R.nonzero()
    # prediction과 error를 담을 리스트 변수 초기화
    self.predictions = []
    self.errors = []
    # 평점이 있는 요소(사용자 x, 아이템 y) 각각에 대해서 아래의 코드를 실행한다.
    for x,y in zip(xs,ys):
      # 사용자 x, 아이템 y에 대해서 평점 예측치를 get_prediction() 함수를 사용해서 계산한다.
      prediction = self.get_prediction(x,y)
      # 예측값을 예측값 리스트에 추가한다.
      self.predictions.append(prediction)
      # 실제값(R)과 예측값의 차이(errors) 계산해서 오차값 리스트에 추가한다.
      self.errors.append(self.R[x,y] - prediction)
    # 예측값 리스트와 오차값 리스트를 numpy array형태로 변환한다.
    self.predictions = np.array(self.predictions)
    self.errors = np.array(self.errors)
    # error를 활용해서 RMSE 도출
    return np.sqrt(np.mean(self.errors**2))

  # 정해진 반복 횟수만큼 P(사용자 요인), Q(아이템 요인), bu(사용자 평가경향),bd(아이템 평가경향)을 업데이트하는 함수
  def train(self):
    # Initializing user-feature and movie-feature matrix
    # P행렬 정규분포(평균=0, 표준편차=1/K)로 난수 초기화
    self.P = np.random.normal(scale=1./self.K,
                              size=(self.num_users,self.K))
    # Q행렬 정규분포(평균=0, 표준편차=1/K)로 난수 초기화
    self.Q = np.random.normal(scale=1./self.K,
                              size=(self.num_items,self.K))
   
    # Initializing the bias terms
    # 사용자 평가경향 0으로 초기화
    self.b_u = np.zeros(self.num_users)
    # 아이템 평가경향 0으로 초기화
    self.b_d = np.zeros(self.num_items)
    # 평점의 전체 평균으로 b로 할당
    self.b = np.mean(self.R[self.R.nonzero()])

    # List of training samples
    # 평점행렬 R 중에서 평점이 있는 요소의 인덱스를 가져온다.
    rows, columns = self.R.nonzero()
    # SGD를 적용할 대상, 즉 평점이 있는 요소의 인덱스와 평점을 리스트로 만들어서 samples에 저장한다.
    self.samples = [(i,j,self.R[i,j]) for i,j in zip(rows,columns)]

    # Stochastic Gradient Descent for given number of iterations
    # training_process 리스트에 SGD가 한 번 실행될 때마다 RMSE가 얼마나 개선되는지 기록
    training_process = []
    for i in range(self.iterations):
      # 기계학습 알고리즘과 비슷하게 SGD를 어디서 시작하느냐에 따라서 수렴의 속도가 달라질 수 있기 때문에
      # 매 반복마다 다양한 시작점에서 출발하기 위해 samples를 임의로 섞는다.
      np.random.shuffle(self.samples)
      # SGD를 실행하는 함수 호출
      self.sgd()
      # SGD로 P,Q,bu,bd가 업데이트되었으므로 이에 따른 새로운 RMSE를 계산
      rmse = self.rmse()
      # 결과 저장
      training_process.append((i+1,rmse))
      # verbose가 True일 때 10회 반복마다 중간 결과 출력
      if self.verbose:
        if (i+1) % 10 == 0:
          print("Iteration : %d ; Train RMSE = %.4f"%(i+1,rmse))
    # 최종 결과 반환
    return training_process
 
 # Rating prediction for user i and item j
 # 평점 예측값을 구하는 함수이다.
  def get_prediction(self,i,j):
    # 사용자 i, 아이템 j에 대한 평점 예측치를 앞에서 배웠던 식을 이용해서 구한다.
    prediction = self.b + self.b_u[i] + self.b_d[j] + self.P[i,:].dot(self.Q[j,:].T)
    return prediction
 
  # Stochastic gradient descent to get optimized P and Q matrix
  # 실제로 최적의 P, Q, b_u, b_d를 구하기 위한 SGD 실행 함수
  def sgd(self):
    for i,j,r in self.samples:
      # 사용자 i, 아이템 j에 대한 평점 예측치 계산
      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]))

      # P 행렬 계산 및 업데이트
      self.P[i,:] += self.alpha * ((e * self.Q[j,:]) - (self.beta * self.P[i,:]))
      # Q 행렬 계산 및 업데이트
      self.Q[j,:] += self.alpha * ((e * self.P[i,:]) - (self.beta * self.Q[j,:]))
 

# 전체 데이터 사용 MF
# DataFrame 형식으로 된 ratings 데이터를 full matrix로 변환
R_temp = ratings.pivot(index='user_id',
                       columns='movie_id',
                       values='rating').fillna(0)

# 모델 MF 계산에 필요한 하이퍼파라미터 정의                       
hyper_params = {
    "K":30,
    "alpha":0.001,
    "beta":0.02,
    "iterations":100,
    "verbose":True
}
# MF 클래스 생성
mf = MF(R_temp,hyper_params)
# MF 모델 학습
train_process = mf.train()