인프런 커뮤니티 질문&답변

안녕하세요.

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

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

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

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

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()