Implementing a Naive Bayes Twitter Sentiment Classifier

# python
from collections import Counter, defaultdict
from functools import partial
from pathlib import Path

import os
import pickle

# pypi
from dotenv import load_dotenv
from tabulate import tabulate

import numpy
import pandas

# my stuff
from neurotic.nlp.twitter.counter import WordCounter

TABLE = partial(tabulate, tablefmt="orgtbl", headers="keys", showindex=False)

env_path = Path("posts/nlp/.env")
assert env_path.is_file()
load_dotenv(env_path)

train_raw = pandas.read_feather(
    Path(os.environ["TWITTER_TRAINING_RAW"]).expanduser())

test_raw = pandas.read_feather(
    Path(os.environ["TWITTER_TEST_RAW"]).expanduser()
)

print(f"Training: {len(train_raw):,}")
print(f"Testing: {len(test_raw):,}")

counter = WordCounter(train_raw.tweet, train_raw.label)

with open(os.environ["TWITTER_SENTIMENT"], "rb") as reader:
    Sentiment = pickle.load(reader)
print(Sentiment)

def train_naive_bayes(counts: Counter,
                      train_x: pandas.Series,
                      train_y: pandas.Series) -> tuple:
    """
    Args:
       counts: Counter from (word, label) to how often the word appears
       train_x: a list of tweets
       train_y: a list of labels correponding to the tweets (0,1)

    Returns:
       logprior: the log odds ratio
       loglikelihood: log likelihood dictionary for the Naive bayes equation
    """
    loglikelihood = defaultdict(lambda: 0)
    logprior = 0

    vocabulary = set([pair[0] for pair in counts])
    V = len(vocabulary)

    # number of positive and negative words in the training set
    N_pos = sum((counts[(token, sentiment)] for token, sentiment in counts
                 if sentiment == Sentiment.positive))
    N_neg = sum((counts[(token, sentiment)] for token, sentiment in counts
                 if sentiment == Sentiment.negative))

    D = len(train_x)

    # D_pos is number of positive documents
    D_pos = train_y.sum()

    # D_neg is the number of negative documents
    D_neg = D - D_pos

    # the log odds ratio
    logprior = numpy.log(D_pos) - numpy.log(D_neg)

    for word in vocabulary:
        freq_pos = counts[(word, Sentiment.positive)]
        freq_neg = counts[(word, Sentiment.negative)]

        # the probability that the word is positive, and negative
        p_w_pos = (freq_pos + 1)/(N_pos + V)
        p_w_neg = (freq_neg + 1)/(N_neg + V)

        loglikelihood[word] = numpy.log(p_w_pos) - numpy.log(p_w_neg)
    return logprior, loglikelihood

logprior, loglikelihood = train_naive_bayes(counter.counts, train_raw.tweet, train_raw.label)
print(f"Log Prior: {logprior}")
print(f"Words in Log Likelihood: {len(loglikelihood):,}")

print(f"Positive Tweets: {len(train_raw[train_raw.label==Sentiment.positive]):,}")
print(f"Negative Tweets: {len(train_raw[train_raw.label==Sentiment.negative]):,}")

all_raw = pandas.concat([train_raw, test_raw])
check = pandas.concat([
    all_raw[all_raw.label==1].iloc[:4000], all_raw[all_raw.label==0].iloc[:4000]])
logprior, loglikelihood = train_naive_bayes(counter.counts, check.tweet, check.label)
print(f"Log Prior: {logprior}")
print(f"Log Likelihood: {len(loglikelihood)}")

def test_naive_bayes(test_x: pandas.Series, test_y: pandas.Series,
                     logprior: float, loglikelihood: dict) -> float:
    """
    Args:
       test_x: tweets to classify
       test_y: labels for test_x
       logprior: the logprior for the training set
       loglikelihood: a dictionary with the loglikelihoods for each word

    Returns:
       accuracy: (# of tweets classified correctly)/(total # of tweets)
    """
    accuracy = 0

    y_hats = numpy.array([int(naive_bayes_predict(tweet, logprior, loglikelihood) > 0)
              for tweet in test_x])

    # error is the average of the absolute values of the differences between y_hats and test_y
    # error = number wrong/number of tweets
    error = numpy.abs(y_hats - test_y).mean()

    # Accuracy is 1 minus the error
    accuracy = 1 - error
    return accuracy

print("Naive Bayes accuracy = %0.4f" %
      (test_naive_bayes(test_raw.tweet, test_raw.label, logprior, loglikelihood)))

for tweet in ['I am happy', 'I am bad', 'this movie should have been great.',
              'great', 'great great', 'great great great', 'great great great great']:
    p = naive_bayes_predict(tweet, logprior, loglikelihood)
    print(f'{tweet} -> {p:.2f}')

my_tweet = "the answer is nicht in the umwelt"
print(naive_bayes_predict(my_tweet, logprior, loglikelihood))

for word in "the answer is nicht in the umwelt".split():
    print(f"{word}:\t{naive_bayes_predict(word, logprior, loglikelihood):0.2f}")

def get_ratio(freqs: Counter, word: str) -> dict:
    """
    Args:
       freqs: Counter with (word, sentiment) : count
       word: string to lookup

    Returns: 
     dictionary with keys 'positive', 'negative', and 'ratio'.
       Example: {'positive': 10, 'negative': 20, 'ratio': 0.5}
    """
    pos_neg_ratio = dict(
        positive = freqs[(word, Sentiment.positive)],
        negative = freqs[(word, Sentiment.negative)],
    )

    # calculate the ratio of positive to negative counts for the word
    pos_neg_ratio['ratio'] = (pos_neg_ratio["positive"] + 1)/(
        pos_neg_ratio["negative"] + 1)
    return pos_neg_ratio

print(get_ratio(counter.counts, 'happi'))

{'happi':
     {'positive': 10, 'negative': 20, 'ratio': 0.5}
 }

def get_words_by_threshold(freqs: Counter, label: int, threshold: float) -> dict:
    """
    Args:
       freqs: Counter of (word, sentiment): word count
       label: 1 for positive, 0 for negative
       threshold: ratio that will be used as the cutoff for including a word in the returned dictionary

    Returns:
       words: dictionary containing the word and information on its positive count, negative count, and ratio of positive to negative counts.
       example of a key value pair:
       {'happi':
           {'positive': 10, 'negative': 20, 'ratio': 0.5}
       }
    """
    words = {}

    for word, _ in freqs:
        pos_neg_ratio = get_ratio(freqs, word)

        if ((label == Sentiment.positive and pos_neg_ratio["ratio"] >= threshold) or
            (label == Sentiment.negative and pos_neg_ratio["ratio"] <= threshold)):
            words[word] = pos_neg_ratio

    return words

passed = get_words_by_threshold(counter.counts, label=Sentiment.negative, threshold=0.05)
count = 1
for word, info in passed.items():
    print(f"{count}\tword: {word}\t{info}")
    count += 1

passed = get_words_by_threshold(counter.counts, label=Sentiment.positive, threshold=10)
count = 1
for word, info in passed.items():
    print(f"{count}\tword: {word}\t{info}")
    count += 1