IMDB Dataset을 활용한 나이브베이즈 예제

https://www.kaggle.com/lakshmi25npathi/imdb-dataset-of-50k-movie-reviews

IMDB Dataset of 50K Movie Reviews

---

텍스트 데이터 전처리(설명)

• 주어진 데이터를 처리하기 쉽도록 처리한다.
• 초기형태 : [오늘 나는 밥을 먹었다. 어제 나는 햄버거를 먹었다.]
• 특수문자(stopword) 제거 : [오늘 나는 밥을 먹었다 어제 나는 햄버거를 먹었다]
• Tokenize : [오늘, 나는, 밥을, 먹었다, 어제, 나는, 햄버거를, 먹었다]
• 이를 Dictionary에 빈도수 정보까지 포함해서 저장 -> Bag of Words(BOW) 생성
• {"오늘": 1, "나는": 2, "먹었다" : 2, "햄버거를" : 1, "밥을" : 1, "어제", 2}

import re

special_chars_remover = re.compile("[^\w'|_]")

def create_BOW(sentence):
    sentence = sentence.lower()
    sentence = remove_special_characters(sentence)
    word_list = sentence.split()
    bow = {}
    for word in word_list:
        if len(word) >= 1:
            if word in bow:
                bow[word] += 1
            else:
                bow[word] = 1

    return bow

def remove_special_characters(sentence):
    return special_chars_remover.sub(' ', sentence)

---

감정분류 알고리즘(설명)

• 긍정적 문서들과 부정적 문서들로 부터 미리 데이터 전처리 후 각 단어의 빈도수를 측정해놓는다.
• 그렇게 이미 나누어진 BOW에서 각각 확률을 얻어 이를 기반으로 현재 주어진 단어의 감정을 분류한다.
• 주어진 데이터가 '긍정'일 확률 = ('긍정' BOW에 주어진 데이터가 각각 존재할 확률) * ('긍정' BOW일 확률) / (데이터가 존재할 확률)
• $P(긍정|데이터) = P(데이터|긍정) * P(긍정) / P(데이터)$
• 주어진 데이터가 '부정'일 확률 = ('부정' BOW에 주어진 데이터가 각각 존재할 확률) * ('부정' BOW일 확률) / (데이터가 존재할 확률)
• $P(부정|데이터) = P(데이터|부정) * P(부정) / P(데이터)$
• 우리가 구할 것은 P(긍정|데이터)와 P(부정|데이터)의 '비율'이므로 P(데이터)는 제외하고 계산한다.
• 만약 각 BOW에 존재하지 않는 단어가 등장할 경우 임의의 확률을 리턴한다.

---

베이즈이론을 이용한 데이터 확률 분석

P(긍정|test_data) = P(test_data|긍정) * P(긍정) / P(test_data)

P(부정|test_data) = P(test_data|부정) * P(부정) / P(test_data)

import re # 정규표현식
import csv
from math import log, exp
from collections import Counter

# 우리가 오늘 최종으로 구할 것은 P(긍정|test_data) , P(부정|test_data)
def start():
    train_datas = open_csv()
    test_data = input()
    prob = naive_bayes(train_datas, test_data, 0.5, 0.5)

    print(f"{test_data}가 긍정적일 확률은 {prob[1]}이고 부정적일 확률은 {prob[0]}입니다")

• open_csv함수로 데이터를 불러오고 그를 기반으로 naive_bayes 수행
• main함수 역할

def open_csv():
    f = open('IMDB Dataset.csv','r',encoding='utf-8')
    csvreader = csv.reader(f)

    pos_doc = []
    neg_doc = []

    # 헤더 첫 줄 넘기기
    next(csvreader)

    for line in csvreader:
        if line[1] == 'positive':
            pos_doc.append(line[0])
        else:
            neg_doc.append(line[0])

    train_datas = [[],[]]
    train_datas[0] = neg_doc
    train_datas[1] = pos_doc

    # 리스트는 토큰화하기 어렵기 때문에, 전부 조인을 통해 하나의 문자열로 만들어줘야 함
    return [''.join(train_datas[0]), ''.join(train_datas[1])]

• csv파일로부터 데이터를 받고 긍정과 부정 분류
• train_data로 리턴

# P(test_data|긍정), P(test_data|부정)
def calculate_doc_prob(train_data, test_data, nowwords_weight): # train_data에서 test_data가 나올 확률 추정
    log_prob = 0

    # stopwords 제거, 소문자화
    sw_train_data = re.compile('[^\w]').sub(' ', train_data.lower())
    # 토큰화
    sw_train_token = sw_train_data.split()
    # 빈도수정리
    train_vector = dict(Counter(sw_train_token))

    # stopwords 제거, 소문자화
    sw_test_data = re.compile('[^\w]').sub(' ', test_data.lower())
    # 토큰화
    sw_test_token = sw_test_data.split()
    # 빈도수정리
    test_vector = dict(Counter(sw_test_token))

    # 전체 단어 갯수
    total_wc = len(sw_train_token)

    # P(test|긍정)
    # test 문장에 있는 단어 별로 긍정인 문장에서 등장한 확률을 곱해주는 것
    for word in test_vector:
        # test 문장에 있는 단어 하나를 뽑아서
        if word in train_vector:
            log_prob += log(train_vector[word]/total_wc) # 로그처리하는 이유 : 값이 깨지는 것 방지            
            # train 문장에 있는지 확인하고
            # 있으면 현재확률 * (train 문장에서 해당 단어의 빈도수 / train 문장 전체 단어의 빈도수)
        else:
            # train 문장에 없으면, 해당 단어의 빈도수도 없다. -> 단어가 나올 확률을 추정할 수 없음
            # 해당 단어의 빈도수를 임의로 nowords_weight 변수로 지정
            log_prob += log(nowwords_weight/total_wc)


    return log_prob

• train_data와 test_data에서 stopword를 제거하고, split 및 dictionary화를 진행한다.
• test_data의 단어를 돌면서 해당 단어가 train_data에 존재할 확률을 구한다.
• 해당 확률을 곱해 리턴한다. (언더플로우를 방지하기 위해 log화를 적용한다.)

def naive_bayes(train_data, test_data, pos_prob, neg_prob):
    # P (긍정|test) = P(test|긍정)) * P(긍정) / P(test)
    test_pos_prob = calculate_doc_prob(train_data[1], test_data, 0.1) + log(pos_prob)

    # P (부정|test) = P(test|부정)) * P(부정) / P(test)
    test_neg_prob = calculate_doc_prob(train_data[0], test_data, 0.1) + log(neg_prob)

    # 상대적인 확률을 구하는 것이기 때문에 각각의 확률을 정확히 도출하는 것은 의미가 없다.
    # log(P(test))를 빼지 않는 이유
    # 둘다 0.00000000000000 에 수렴한다.
    # 따라 이를 보정해야 한다.
    maxprob = max(test_neg_prob, test_pos_prob)
    test_pos_prob -= maxprob
    test_neg_prob -= maxprob
    test_pos_prob = exp(test_pos_prob)
    test_neg_prob = exp(test_neg_prob)

    # 두 확률 값의 상대적 비율
    normalized_prob = [test_neg_prob/(test_pos_prob + test_neg_prob), test_pos_prob/(test_pos_prob + test_neg_prob)]
    return normalized_prob

• 받은 데이터를 기반으로 긍정적인 확률과 부정적인 확률을 구한다.
• 상대적 확률을 구하는 것이므로 log(P(test))는 굳이 찾아서 빼줄 필요없다.
• 마찬가지 이유로 상대적확률만 구하면 되므로 둘 중 큰 값으로 나눠준다. (로그형태니까 빼준다.)
• 그리고 최종적 활용을 위해 다시 exp해주고 리턴한다.

def open_csv_validate():
    f = open('IMDB Dataset.csv', 'r', encoding='utf-8')
    csvreader = csv.reader(f)

    pos_doc = []
    neg_doc = []

    # csv파일의 첫줄이 헤더라서, 그거를 스킵
    next(csvreader)
    for line in csvreader:
        if line[1] == 'positive':
            pos_doc.append(line[0])
        else:
            neg_doc.append(line[0])

    train_datas = [[], []]
    # 25000개씩 각각있어서, 4:1로 나눈다!
    train_datas[0] = neg_doc[:20000]
    train_datas[1] = pos_doc[:20000]
    test_datas = [neg_doc[20000:], pos_doc[20000:]]

    # 리스트 형태는 토큰화하기가 어렵기 때문에, 전부 조인을 해서 하나의 문자열로 만들어준다
    return [' '.join(train_datas[0]), ' '.join(train_datas[1])], [' '.join(test_datas[0]), ' '.join(test_datas[1])]

• csv파일을 열어 학습 데이터와 테스트 데이터로 나눠주는 함수이다.
• 이를 활용해 현 데이터의 신용성을 한번에 평가할 수 있다.

def start_validate():
    train_datas, test_datas = open_csv_validate()
    # 긍정인 문장을 넣고, 실제로 긍정이라고 나오는지 확인
    prob = naive_bayes(train_datas, test_datas[1], 0.5, 0.5)
    print(f'긍정적인 데이터가 부정적일 확률 : {prob[0]}, 긍정적일 확률 : {prob[1]}')
    # 부정인 문장을 넣고, 실제로 부정이라고 나오는지 확인
    prob = naive_bayes(train_datas, test_datas[0], 0.5, 0.5)
    print(f'부정적인 데이터가 부정적일 확률 : {prob[0]}, 긍정적일 확률 : {prob[1]}')

• 전체데이터에 대한 신용성 평가는 위 함수로 수행한다.