ch12 영화 추천시스템¶

! cat ./data/ml-100k/u.user | head -n 2

! cat ./data/ml-100k/u.data | head -n 2

! cat ./data/ml-100k/u.item | head -n 2

# 작업에 사용할 파일목록들을 연결합니다
fin_user   = "./data/ml-100k/u.user" # 사용자 개인정보
fin_movie  = "./data/ml-100k/u.item" # 영화정보
fin_rating = "./data/ml-100k/u.data" # 영화 별점정보

# 데이터셋 불러오기 (delim:구분자)
import codecs
def read_data(fin, delim):
    info_li = []
    for line in codecs.open(fin, "r", encoding="latin-1"):
        line_items = line.strip().split(delim)
        key        = int(line_items[0])
        if (len(info_li)+1) != key:
            print('errors at data_id'); exit(0)
        info_li.append(line_items[1:])
    print('{} 데이터의 행수: {:5,}'.format(fin, len(info_li)))
    return(info_li)

user_info_li  = read_data(fin_user, '|')
movie_info_li = read_data(fin_movie,'|')

# 영화 별점정보로 유틸리티 행렬 만들기
import numpy as np
def read_rating_data(fin):
    Q = np.zeros((len(user_info_li), len(movie_info_li)), dtype=np.float64)
    for line in open(fin):
        user, movie, rating, date = line.strip().split("\t")
        user_index  = int(user)  - 1
        movie_index = int(movie) - 1
        Q[user_index,movie_index] = float(rating)
    return(Q)

R = read_rating_data(fin_rating)
print("별점정보 영화 Matrix :", R.shape)

# 개별 사용자의 별점평균 List
from scipy import stats

user_mean_li = []
for i in range(0,R.shape[0]):
    user_rating = [x for x in R[i] if x>0.0]
    user_mean_li.append(stats.describe(user_rating).mean)

# 사용자 별점평균의 기초통계량 분석
def print_statistics(array, title=""):
    sta = stats.describe(array)
    print(title, "\n%14s %15s"%('statistic','value'), "\n", 30*'-', "\n%14s %15.5f"%('size',sta[0]),\
          "\n%14s %15.5f"%('min',sta[1][0]), "\n%14s %15.5f"%('max', sta[1][1]),\
          "\n%14s %15.5f"%('mean',sta[2]), "\n%14s %15.5f"%('std', np.sqrt(sta[3])),\
          "\n%14s %15.5f"%('skew',sta[4]), "\n%14s %15.5f"%('kutosis', sta[5])) 

print_statistics(user_mean_li, "<사용자별 별점평균 기초통계분석>")

import warnings
warnings.simplefilter(action='ignore', category=RuntimeWarning)

# 영화의 평균 별점 기초 통계량 구하기
movie_mean_li = []
for i in range(0, R.shape[1]):
    R_T = R.T
    movie_rating = [x  for x in R_T[i] if x>0.0]
    movie_mean_li.append(stats.describe(movie_rating).mean)

warnings.simplefilter('default')
print_statistics(movie_mean_li, "<영화별 별점평균 기초통계분석>")
# stats.describe(movie_mean_li)

# 영화 플롯 데이터 호출 (예제 12-5, 12-6을 대체)
movie_plot_li =[]
warnings.simplefilter(action='ignore', category=ImportWarning)
warnings.simplefilter(action='ignore', category=ResourceWarning)
for line in open('./data/ml-100k-plot.txt'):
    try:
        ml_id, plot = line.strip().split('|',1)
        movie_plot_li.append(plot)            
    except ValueError:
        movie_plot_li.append('')

print("info, plot 갯수일치: {}\nmovie_plot_li[0]:{}\nmovie_plot_li[0]:{}".format(
    len(movie_plot_li) == len(movie_info_li),
    movie_info_li[0][:3], movie_plot_li[0][:80]))

warnings.simplefilter(action='ignore', category=PendingDeprecationWarning)

# TF-IDF로 변환한 키워드의 리스트 (X의 0번 열 키워드는 feature_names[0])
from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer    = TfidfVectorizer(min_df=3, stop_words='english')
X             = vectorizer.fit_transform(movie_plot_li[:100])
feature_names = vectorizer.get_feature_names()
warnings.simplefilter('default')
print(feature_names[:10])

warnings.simplefilter(action='ignore', category=ImportWarning)
warnings.simplefilter(action='ignore', category=ResourceWarning)
warnings.simplefilter(action='ignore', category=DeprecationWarning)
warnings.simplefilter(action='ignore', category=PendingDeprecationWarning)

# 사용자 정의 Tokenizer 정의
from nltk.stem import WordNetLemmatizer
from nltk.tokenize import RegexpTokenizer

class LemmaTokenizer(object):
    def __init__(self):
        self.wnl       = WordNetLemmatizer()
        # self.tokenizer = RegexpTokenizer('(?u)\w\w+')
        self.tokenizer = RegexpTokenizer('(?u)[A-z]+')
    
    # 클래스 객체를 호출시 마다 실행(Tf-idf Vector 키워드 호출)
    def __call__(self, doc): 
        return([self.wnl.lemmatize(t) for t in self.tokenizer.tokenize(doc)])

# 사이킷런에 위에서 정의한 토크나이저를 입력으로 넣습니다.
vectorizer2   = TfidfVectorizer(min_df=3, tokenizer=LemmaTokenizer(), stop_words='english')
X             = vectorizer2.fit_transform(movie_plot_li[:100])
feature_names = vectorizer2.get_feature_names()

# 비슷한 영화 추천하기
from sklearn.metrics.pairwise import cosine_similarity
movie_sim = cosine_similarity(X)
print(movie_sim.shape)

def similar_recommend_by_movie_id(movielens_id):
    # enumerate 함수로 [(리스트 인덱스 0, 유사도 0), (리스트 인덱스 1, 유사도 1)...]의
    # 리스트를 만듭니다. 그 후 각 튜플의 두 번째 항목, 즉 유사도를 이용하여 내림차순 정렬합니다.
    # 이렇게 만든 리스트의 가장 앞 튜플의 첫 번째 항목이 영화 ID가 됩니다.
    movie_index    = movielens_id - 1
    similar_movies = sorted(list(enumerate(movie_sim[movie_index])),key=lambda x:x[1], reverse=True)
    recommended    = 1
    print("----- {} : 관람객 추천영화 ------".format(movie_info_li[movie_index][0]))
    for movie_info in similar_movies[1:8]:
        movie_title = movie_info_li[movie_info[0]] # 자신은 추천서 제외
        print('추천영화 %d순위 : %s'%(recommended, movie_title[0]))
        recommended += 1

similar_recommend_by_movie_id(1)

similar_recommend_by_movie_id(22)

%%time
import numpy as np
from sklearn.metrics import mean_squared_error

# R(사용자 별점정보) 교대 최소제곱법 구현
def compute_ALS(R, n_iter:int, lambda_:float, k):
    '''임의의 사용자 행렬 X, 영화 행렬 Y를 생성한 뒤
    교대 최소제곱법을 이용하여 유틸리티 행렬 R을 근사합니다.
    :param R(ndarray)     : 유틸리티 행렬
    :param n_iter(fint)   : X와 Y 반복횟수
    :param lambda_(float) : 정규화 파라미터
    '''
    m, n         = R.shape
    X, Y, errors = np.random.rand(m, k), np.random.rand(k, n), [] # 갱신마다 에러저장
    for i in range(0, n_iter): # np.eye : n x n 크기 단위행렬
        X = np.linalg.solve(np.dot(Y, Y.T) + lambda_ * np.eye(k), np.dot(Y, R.T)).T
        Y = np.linalg.solve(np.dot(X.T, X) + lambda_ * np.eye(k), np.dot(X.T, R))        
        errors.append(mean_squared_error(R, np.dot(X, Y)))
        if i % 10 == 0: print('{:3} 번 iteration 완료'.format(i))
    R_hat = np.dot(X, Y)
    print('MSE rate(작을수록 일치도 높음): %.5f'%(mean_squared_error(R, np.dot(X, Y))))
    return(R_hat, errors) # 생성한 배열과 오류값 출력합니다

# 검사행렬, 반복횟수, 변화율, 행렬의 크기
R_hat, errors = compute_ALS(R, 20, 0.1, 100)

%%time
# 가중치 교대 최소제곱법 구현 (가중치 행렬을 mean_squared_error 함수의 인자로 사용) 
def compute_wALS(R,W, n_iter, lambda_, k):
    m, n = R.shape
    weighted_errors = []
    X, Y = np.random.rand(m, k), np.random.rand(k, n)
    
    for ii in range(n_iter): # 가중치 행렬을 넣어서 계산 (사용자와 영화 가중치 행렬로 X와 Y를 갱신)
        for u, Wu in enumerate(W):
            X[u,:] = np.linalg.solve(np.dot(Y, np.dot(np.diag(Wu), Y.T)) +\
                                     lambda_ * np.eye(k), np.dot(Y, np.dot(np.diag(Wu),R[u,:].T))).T
        for i, Wi in enumerate(W.T):
            Y[:, i] = np.linalg.solve(np.dot(X.T, np.dot(np.diag(Wi), X)) +\
                                      lambda_ * np.eye(k), np.dot(X.T, np.dot(np.diag(Wi), R[:, i])))
        
        weighted_errors.append(mean_squared_error(R, np.dot(X, Y),sample_weight=W))
        if ii % 3 == 0: print('{:3} 번 iteration 완료'.format(ii))
    
    R_hat = np.dot(X, Y)
    print('MSE rate(작을수록 일치도 높음): %.5f'%(mean_squared_error(R, np.dot(X, Y), sample_weight=W)))
    return(R_hat, errors)

# R 테이블의 가중치 행렬을 생성후 계산을 한다 (별점 존재시1, 별점 없으면0)
W                            = R > 0.0
W[W == True], W[W == False]  = 1, 0
W                            = W.astype(np.float64, copy=False)
R_hat, errors                = compute_wALS(R, W, 20, 0.1, 100)

%%time
# 경사하강법 구현
def compute_GD(R,n_iter, lambda_, learning_rate, k):
    m,n    = R.shape
    errors = []
    X, Y   = np.random.rand(m, k), np.random.rand(k, n)
    for ii in range(n_iter):   # 입력받은 반복 횟수만큼 갱신을 반복
        for u in range(m):
            for i in range(n):
                if R[u,i] > 0: # 각 사용자 및 상품의 행렬에 대해 하나씩 계산
                    e_ui   = R[u,i]-np.dot(X[u, :], Y[:,i])
                    X[u,:] = X[u,:] + learning_rate * (e_ui* Y[:,i] - lambda_ * X[u,:])
                    Y[:,i] = Y[:,i] + learning_rate * (e_ui * X[u,:] - lambda_ * Y[:,i])                      
        errors.append(mean_squared_error(R, np.dot(X, Y)))
        if ii % 10 == 0: print('{:3} 번 iteration 완료'.format(ii))

    R_hat = np.dot(X, Y)
    print('MSE rate(작을수록 일치도 높음): %.5f'%(mean_squared_error(R, R_hat)))
    return(R_hat, errors)

R_hat, errors= compute_GD(R, 20, 1, 0.001, 100)

R_hat, errors= compute_GD(R, 40, 1, 0.001, 100)

%%time
from sklearn.metrics import mean_squared_error
import numpy as np

def compute_ALS(R, test, n_iter, lambda_, k):
    '''임의의 사용자 요인 행렬 X와 임의의 영화 요인 행렬 Y를 생성하고 교대 최소제곱법을 이용하여
    유틸리티 행렬 R을 근사합니다. 그후 test행렬을 이용하여 평가합니다.
    R(ndarray) : 유틸리티 행렬
    test: 평가행렬
    lambda_(float) : 정규화 파라미터
    n_iter(fint) : X와 Y의 갱신 횟수
    '''
    m,n  = R.shape
    X, Y, errors = np.random.rand(m, k), np.random.rand(k, n), []
    for i in range(0, n_iter): # 갱신 시마다 계산한 에러를 저장
        X = np.linalg.solve(np.dot(Y, Y.T) + lambda_ * np.eye(k),np.dot(Y, R.T)).T
        Y = np.linalg.solve(np.dot(X.T, X) + lambda_ * np.eye(k), np.dot(X.T, R))
        errors.append(get_test_mse(test,np.dot(X, Y)))
        if i % 10 == 0: print('{:3} 번 iteration 완료'.format(i))
    R_hat = np.dot(X, Y)
    print('MSE rate(작을수록 일치도 높음): %.5f'%(get_test_mse(test, R_hat)))
    return(R_hat, errors)


train,test          = train_test_split(R, 10)
# 생성한 근사 Matrix와 train 데이터 Set 일치도
R_hat, train_errors = compute_ALS(train, train, 20, 0.1,100)

# 생성한 근사 Matrix와 test 데이터 Set 일치도
R_hat, test_errors = compute_ALS(train, test,20, 0.1,100)

# 반복에 따른 학습-평가 에러변화 출력(평가에러: 점선, 학습에러: 실선)
# 1회 학습후 test 오류가 늘어나는 모습으로 OverFit을 의심가능
from matplotlib import pyplot as plt
x = range(0,20)
plt.xlim(0,20); plt.xlabel('iteration'); plt.xticks(x, range(0,20));
plt.ylim(0,15); plt.ylabel('MSE')
plt.plot(x,train_errors,      label='train_error')
plt.plot(x,test_errors, '--', label='test_error')
plt.legend()#handles=[train_plot, test_plot])
plt.show()

%%time
# 정규화 파라미터를 50으로 학습한 경우, 일반화 성능 개선여부
R_hat, train_errors = compute_ALS(train, train,20, 50,100)
R_hat, test_errors  = compute_ALS(train, test, 20, 50,100)

# 반복에 따른 학습-평가 에러변화 출력(평가에러: 점선, 학습에러: 실선)
# 훈연을 반복할수록 test 의 MSE도 감소합니다
x = range(0,20)
plt.xlim(0,20); plt.xlabel('iteration'); plt.xticks(x, range(0,20));
plt.ylim(0,15); plt.ylabel('MSE')
plt.plot(x,train_errors,      label='train_error')
plt.plot(x,test_errors, '--', label='test_error')
plt.legend() #handles=[train_plot, test_plot]); 
plt.show()

%%time
# 최솟값 0: 전체에 최소값을 뺀다 (X, Y를 0~5의 수로 초기화)
# 큰 값 5: 최대 예측값(np.max(R_hat))으로 나눈 값을 곱합니다.
R_hat -= np.min(R_hat)
R_hat *= float(5) / np.max(R_hat)

# 사용자의 ID 입력, 보지 않은 영화 중 추천합니다.
def recommend_by_user(user):
    user_index       = user - 1
    user_seen_movies = sorted(list(enumerate(R_hat[user_index])),key=lambda x:x[1], reverse=True)
    recommended      = 1
    print("----- {} 관람객 추천영화 ------".format(" ".join(user_info_li[0][:-1])))
    for movie_info in user_seen_movies:
        if W[user][movie_info[0]] == 0:
            movie_title = movie_info_li[int(movie_info[0]+1)]
            movie_score = movie_info[1]
            print("%d 순위 추천영화: 평점(%.3f) %s"%(recommended, movie_score, movie_title[0]))
            recommended += 1
        if recommended == 6: break

recommend_by_user(1)