vaccine_tweets_sentiment_analysis.py

# -*- coding: utf-8 -*-
"""Vaccine Tweets Sentiment Analysis.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1VbR-PBOOeDkgMMMfmgoalA1cuzImpJ2v

## Installation

Added GPU to speed up the training process
"""

!nvidia-smi

"""All the required libraries and dependiencies can be installed using:"""

pip install -r requirements.txt

"""Imported all the neccessary modules and set the formating of the graphs

Also checks to make sure that the device is using GPU if it is available 

Two key modules used in this project is the TextBlob library and the transformers library.

The TextBlob library is used in the intial anaylsis of the sentiment of the the tweets

Then the transformers library from huggingface is used to access the BERT model used to train a model to analyze the sentiments of any new tweets
"""

# Commented out IPython magic to ensure Python compatibility.
#@title Setup & Config
import transformers
from transformers import BertModel, BertTokenizer, AdamW, get_linear_schedule_with_warmup
import torch
import re
from textblob import TextBlob

import numpy as np
import pandas as pd
import seaborn as sns
from pylab import rcParams
import matplotlib.pyplot as plt
from matplotlib import rc
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix, classification_report
from collections import defaultdict
from textwrap import wrap

from torch import nn, optim
from torch.utils.data import Dataset, DataLoader
import torch.nn.functional as F

# %matplotlib inline
# %config InlineBackend.figure_format='retina'

sns.set(style='whitegrid', palette='muted', font_scale=1.2)

HAPPY_COLORS_PALETTE = ["#01BEFE", "#FFDD00", "#FF7D00", "#FF006D", "#ADFF02", "#8F00FF"]

sns.set_palette(sns.color_palette(HAPPY_COLORS_PALETTE))

rcParams['figure.figsize'] = 12, 8

RANDOM_SEED = 42
np.random.seed(RANDOM_SEED)
torch.manual_seed(RANDOM_SEED)

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
device

"""# Data

The dataset used is located in a csv file. Here we are formating all the column names so that data visualization can be done more easily in the future, and checking the fist few values to make sure everything looks good.
"""

DATASET_ENCODING = "ISO-8859-1"
df = pd.read_csv("covid_vaccine_tweets.csv",encoding=DATASET_ENCODING)
df.head()

"""The intial dataset is a collection of over 800,000 COVID-19 vaccine tweets. This is a small sample of that and currently holds 60 thousand tweets, and after removing any null values the size reduces down to around 31 thousand."""

df.shape

df = df.dropna() #Removes null values

df.shape

"""The tweets are cleaned up to remove any @ symbols, hyper links,punctuation and retweets, so that the sentiment analysis can be more accurate.

"""

from string import punctuation
def cleanTxt(text):
    text = re.sub(r'@[A-Za-z0-9]+','',text) #Removes @mentions
    text = re.sub(r'#','', text) #Removing the '#' symbol
    text = re.sub(r'RT[\s]+', '', text) #Removing RT
    text = re.sub(r'https?:\/\/\S+','', text) #Remove the hyper link
    text = "".join([ch for ch in text if ch not in punctuation])
    return text

#Cleaning text
df['text'] = df['text'].apply(cleanTxt)

#Show the cleaned text
df.head(10)

"""Using the TextBlob library, the subjectivity and polarity of the text can be analyzed."""

# Create a function to get the subjectivity 
def getSubjectivity(text):
    return TextBlob(text).sentiment.subjectivity

#Create a function to get the polarity
def getPolarity(text):
    return TextBlob(text).sentiment.polarity

#Create two new columns
df['Subjectivity'] = df['text'].apply(getSubjectivity)
df['Polarity'] = df['text'].apply(getPolarity)

#Show the new dataframe with the new columns
df.head(10)

"""The TextBlob polarity scales ranges from -1 to 1. If the score is less than 0, it's an negative sentiment, if it's zero it's a neutral sentiment, and if it's more than 0, it's a positve sentiment"""

def getAnalysis(score):
    if score < 0:
        return 0
    elif score == 0:
        return 1
    else:
        return 2

df['sentiment'] = df['Polarity'].apply(getAnalysis)

df.head(20)

df.shape

df.info()

"""Next important thing to check if there is a imbalance between the different classes. As seen there seems to be significantly more neutral and positive values for our dataset"""

sns.countplot(df.sentiment)
plt.xlabel('sentiment score');

"""To balance the classes I will be undersampling the data, and reduce the neutral and postive so they have the same amount as the negative tweets."""

#Number of negative tweets
ntweets = df[df.sentiment == 0]
print(ntweets.shape[0]) #we will make it so we undersample

#Number of neutral tweets
netweet = df[df.sentiment == 1]
print(netweet.shape[0])

#Number of positive tweets
ptweet = df[df.sentiment == 2]
print(ptweet.shape[0])

#Randomly remove tweets from the other classes so they have the same number of tweets as the negative one
remove_pos = ptweet.shape[0] - ntweets.shape[0]
remove_neut = netweet.shape[0] - ntweets.shape[0]

neg_df = df[df["sentiment"] == 0] 

pos_df = df[df["sentiment"] == 2]
neut_df = df[df["sentiment"] == 1]

pos_drop_indices = np.random.choice(pos_df.index, remove_pos, replace=False)
neut_drop_indices = np.random.choice(neut_df.index, remove_neut, replace=False)

pos_undersampled = pos_df.drop(pos_drop_indices)
neut_undersampled = neut_df.drop(neut_drop_indices)

df= pd.concat([neg_df, pos_undersampled, neut_undersampled])

class_names = ['negative', 'neutral', 'positive']

"""Now all of the sets are the same size, and the final size of our set is around a thousand."""

ax = sns.countplot(df.sentiment)
plt.xlabel('tweet sentiment')
ax.set_xticklabels(class_names);
df.head(10)
print(df.shape[0])

"""#Data Preprocessing

The tokenizer used is from the transformers library that is the pre-trained model from huggingface.
"""

PRE_TRAINED_MODEL_NAME = 'bert-base-cased'

tokenizer = BertTokenizer.from_pretrained(PRE_TRAINED_MODEL_NAME)

"""The BERT model requires special tokens to work on texts. Some of these prebuilt bodels include:

`[CLS]` - required at the start of each text to know that the text needs to be classified

`[SEP]` - marker for ending of a sentence

`[PAD]` - token is used to pad the texts up to a given length

Attention mask, a binary tensor, is also used so that the model knows to pay attention to the tokens that include text or markers, but not to pay attention to the padding

###Sample:
This is all the preprocessing steps shown on an example text
"""

sample_txt = 'When was I last outside? I am stuck at home for 2 weeks.'

tokens = tokenizer.tokenize(sample_txt)
token_ids = tokenizer.convert_tokens_to_ids(tokens)

print(f' Sentence: {sample_txt}')
print(f'   Tokens: {tokens}')
print(f'Token IDs: {token_ids}')

encoding = tokenizer.encode_plus(
  sample_txt,
  max_length=32,
  add_special_tokens=True, # Add '[CLS]' and '[SEP]'
  return_token_type_ids=False,
  padding='max_length',
  return_attention_mask=True,
  return_tensors='pt',  # Return PyTorch tensors
)

encoding.keys()

print(len(encoding['input_ids'][0]))
encoding['input_ids'][0]

print(len(encoding['attention_mask'][0]))
encoding['attention_mask']

tokenizer.convert_ids_to_tokens(encoding['input_ids'][0])

"""# Actual process"""

token_lens = []

for txt in df.text:
  tokens = tokenizer.encode(txt, max_length=512)
  token_lens.append(len(tokens))

sns.distplot(token_lens)
plt.xlim([0, 256]);
plt.xlabel('Token count');

"""As most of the text are around 100 tokens long, the maximum length we chose was a little bit bigger at 150"""

MAX_LEN = 150

"""Here we created the Pytorch dataset. In the encoding section, it encodes each indivdual tweet, adds the special tokens and pads it to the maximum length. It also uses attention mask to emphasive the text portion of the tweets, as mentioned before"""

class GPTweetDataset(Dataset):

  def __init__(self, tweet, targets, tokenizer, max_len):
    self.tweet = tweet
    self.targets = targets
    self.tokenizer = tokenizer
    self.max_len = max_len
  
  def __len__(self):
    return len(self.tweet)
  
  def __getitem__(self, item):
    tweet = str(self.tweet[item])
    target = self.targets[item]

    encoding = self.tokenizer.encode_plus(
      tweet,
      add_special_tokens=True,
      max_length=self.max_len,
      return_token_type_ids=False,
      padding='max_length',
      return_attention_mask=True,
      return_tensors='pt',
    )

    return {
      'tweet_text': tweet,
      'input_ids': encoding['input_ids'].flatten(),
      'attention_mask': encoding['attention_mask'].flatten(),
      'targets': torch.tensor(target, dtype=torch.long)
    }

"""Here we are splitting the train, test and validation sets, so that the training set makes up the majority of the data, with validation and testing being a small section of the rest"""

df_train, df_test = train_test_split(df, test_size=0.1, random_state=RANDOM_SEED)
df_val, df_test = train_test_split(df_test, test_size=0.5, random_state=RANDOM_SEED)

df_train.shape, df_val.shape, df_test.shape

"""This function is used to create some data loaders using the GPTweet class defined above, using Pytorch's built in primitives"""

def create_data_loader(df, tokenizer, max_len, batch_size):
  ds = GPTweetDataset(
    tweet=df.text.to_numpy(),
    targets=df.sentiment.to_numpy(),
    tokenizer=tokenizer,
    max_len=max_len
  )

  return DataLoader(
    ds,
    batch_size=batch_size,
    num_workers=2
  )

"""We used 16 mini-batches for the dataloader"""

BATCH_SIZE = 16

train_data_loader = create_data_loader(df_train, tokenizer, MAX_LEN, BATCH_SIZE)
val_data_loader = create_data_loader(df_val, tokenizer, MAX_LEN, BATCH_SIZE)
test_data_loader = create_data_loader(df_test, tokenizer, MAX_LEN, BATCH_SIZE)

"""The data iterates over the data_loader and is comprised of the batch size and the sequence length """

data = next(iter(train_data_loader))
data.keys()

print(data['input_ids'].shape)
print(data['attention_mask'].shape)
print(data['targets'].shape)

"""# Sentiment Classification using the BERT model

For the sentiment classification, we used the base version of the pretrained BERT model with 12 encoders, and it is also set to 'cased' to be case-sensitive, as capitalization can provide more emphasive, and can show more sentiment
"""

bert_model = BertModel.from_pretrained(PRE_TRAINED_MODEL_NAME)

last_hidden_state, pooled_output = bert_model(
  input_ids=encoding['input_ids'],
  attention_mask=encoding['attention_mask'],
  return_dict = False   # this is needed to get a tensor as result
)

"""The last hidden state is a sequence of hidden states from the last layer of the model:"""

last_hidden_state.shape

pooled_output.shape

bert_model.config.hidden_size

"""This class is the Sentiment Classifer that uses the pretrained model for the classification, and the dropout layer for regulization. All of this is then put into the model, which is moved into the GPU"""

class SentimentClassifier(nn.Module):

  def __init__(self, n_classes):
    super(SentimentClassifier, self).__init__()
    self.bert = BertModel.from_pretrained(PRE_TRAINED_MODEL_NAME)
    self.drop = nn.Dropout(p=0.3)
    self.out = nn.Linear(self.bert.config.hidden_size, n_classes)
  
  def forward(self, input_ids, attention_mask,return_dict ):
    _, pooled_output = self.bert(
      input_ids=input_ids,
      attention_mask=attention_mask,
      return_dict=False
    )
    output = self.drop(pooled_output)
    return self.out(output)

model = SentimentClassifier(len(class_names))
model = model.to(device)

"""The input ids and the attention mask is also moved into the GPU for the softmax model that is applied later"""

input_ids = data['input_ids'].to(device)
attention_mask = data['attention_mask'].to(device)

print(input_ids.shape) # batch size, seq length
print(attention_mask.shape) # batch size, seq length

"""This shows the untrained version of the model and the probabilites of each of the three sentiments"""

return_dict = False
F.softmax(model(input_ids, attention_mask=attention_mask, return_dict=return_dict) , dim=1)

"""# Training
The optimizer used is the AdamW algorithm, which is based on the one used in the original BERT paper
The number of epochs are set to 10 and is used to calculate the total number of steps. 
The loss function is also defined and the built in CrossEntropyLoss is used

"""

EPOCHS = 10

optimizer = AdamW(model.parameters(), lr=2e-5, correct_bias=False)
total_steps = len(train_data_loader) * EPOCHS

scheduler = get_linear_schedule_with_warmup(
  optimizer,
  num_warmup_steps=0,
  num_training_steps=total_steps
)

loss_fn = nn.CrossEntropyLoss().to(device)

"""This is the training function, that takes in all the neccessary values. It iterates over the dataloader and takes the outputs of the model, to make a prediction on the most likely sentiment and the loss. The correct predictions are incremented, and the loss is backproporated and gradient clipping is applied to the model.

The entire function returns the number of correct predictions and the mean loss 
"""

def train_epoch(
  model, 
  data_loader, 
  loss_fn, 
  optimizer, 
  device, 
  scheduler, 
  n_examples
):
  model = model.train()

  losses = []
  correct_predictions = 0
  
  for d in data_loader:
    input_ids = d["input_ids"].to(device)
    attention_mask = d["attention_mask"].to(device)
    targets = d["targets"].to(device)

    outputs = model(
      input_ids=input_ids,
      attention_mask=attention_mask,
      return_dict=False
    )

    _, preds = torch.max(outputs, dim=1)
    loss = loss_fn(outputs, targets)

    correct_predictions += torch.sum(preds == targets).detach().cpu().numpy()
    losses.append(loss.item())

    loss.backward()
    nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
    optimizer.step()
    scheduler.step()
    optimizer.zero_grad()

  return float(correct_predictions) / n_examples , np.mean(losses)

"""This function evaluates the model, by disabiling the gradient function and dropout, and doing the same steps as in the train function by iterating over the dataloader, collectiong the loss and correct predictions"""

def eval_model(model, data_loader, loss_fn, device, n_examples):
  model = model.eval()

  losses = []
  correct_predictions = 0

  with torch.no_grad():
    for d in data_loader:
      input_ids = d["input_ids"].to(device)
      attention_mask = d["attention_mask"].to(device)
      targets = d["targets"].to(device)

      outputs = model(
        input_ids=input_ids,
        attention_mask=attention_mask,
        return_dict=False
      )
      _, preds = torch.max(outputs, dim=1)

      loss = loss_fn(outputs, targets)

      correct_predictions += torch.sum(preds == targets).detach().cpu().numpy()
      losses.append(loss.item())

  return float(correct_predictions) / n_examples, np.mean(losses)

"""This section creates the training group and shows the training loss and accuracy and the validation losss and accuracy. It runs through all the epochs, and the training data goes into the train function, and the validation data goes into the evaluation function. It also saves the model with the best accuracy and improves future accuracy using that."""

# Commented out IPython magic to ensure Python compatibility.
# %%time
# 
# history = defaultdict(list)
# best_accuracy = 0
# 
# for epoch in range(EPOCHS):
# 
#   print(f'Epoch {epoch + 1}/{EPOCHS}')
#   print('-' * 10)
# 
#   train_acc, train_loss = train_epoch(
#     model,
#     train_data_loader,    
#     loss_fn, 
#     optimizer, 
#     device, 
#     scheduler, 
#     len(df_train)
#   )
# 
#   print(f'Train loss {train_loss} accuracy {train_acc}')
# 
#   val_acc, val_loss = eval_model(
#     model,
#     val_data_loader,
#     loss_fn, 
#     device, 
#     len(df_val)
#   )
# 
#   print(f'Val   loss {val_loss} accuracy {val_acc}')
#   print()
# 
#   history['train_acc'].append(train_acc)
#   history['train_loss'].append(train_loss)
#   history['val_acc'].append(val_acc)
#   history['val_loss'].append(val_loss)
# 
#   if val_acc > best_accuracy:
#     torch.save(model.state_dict(), 'best_model_state.bin')
#     best_accuracy = val_acc

plt.plot(history['train_acc'], label='train accuracy')
plt.plot(history['val_acc'], label='validation accuracy')

plt.title('Training history')
plt.ylabel('Accuracy')
plt.xlabel('Epoch')
plt.legend()
plt.ylim([0, 1]);

"""The accuracy of the test data is very similar to that of the validation data:"""

test_acc, _ = eval_model(
  model,
  test_data_loader,
  loss_fn,
  device,
  len(df_test)
)

print(test_acc)

"""This function evalutes the trained model on the test data to see how accurate it is """

def get_predictions(model, data_loader):
  model = model.eval()
  
  tweet_texts = []
  predictions = []
  prediction_probs = []
  real_values = []

  with torch.no_grad():
    for d in data_loader:

      texts = d["tweet_text"]
      input_ids = d["input_ids"].to(device)
      attention_mask = d["attention_mask"].to(device)
      targets = d["targets"].to(device)

      outputs = model(
        input_ids=input_ids,
        attention_mask=attention_mask,
        return_dict=False
      )
      _, preds = torch.max(outputs, dim=1)

      probs = F.softmax(outputs, dim=1)

      tweet_texts.extend(texts)
      predictions.extend(preds)
      prediction_probs.extend(probs)
      real_values.extend(targets)

  predictions = torch.stack(predictions).cpu()
  prediction_probs = torch.stack(prediction_probs).cpu()
  real_values = torch.stack(real_values).cpu()
  return tweet_texts, predictions, prediction_probs, real_values

y_tweet_texts, y_pred, y_pred_probs, y_test = get_predictions(
  model,
  test_data_loader
)

"""Here is the Classification Report"""

print(classification_report(y_test, y_pred, target_names=class_names))

def show_confusion_matrix(confusion_matrix):
  hmap = sns.heatmap(confusion_matrix, annot=True, fmt="d", cmap="Blues")
  hmap.yaxis.set_ticklabels(hmap.yaxis.get_ticklabels(), rotation=0, ha='right')
  hmap.xaxis.set_ticklabels(hmap.xaxis.get_ticklabels(), rotation=30, ha='right')
  plt.ylabel('True sentiment')
  plt.xlabel('Predicted sentiment');

cm = confusion_matrix(y_test, y_pred)
df_cm = pd.DataFrame(cm, index=class_names, columns=class_names)
show_confusion_matrix(df_cm)

idx = 4

tweet_text = y_tweet_texts[idx]
true_sentiment = y_test[idx]
pred_df = pd.DataFrame({
  'class_names': class_names,
  'values': y_pred_probs[idx]
})

print("\n".join(wrap(tweet_text)))
print()
print(f'True sentiment: {class_names[true_sentiment]}')

sns.barplot(x='values', y='class_names', data=pred_df, orient='h')
plt.ylabel('sentiment')
plt.xlabel('probability')
plt.xlim([0, 1]);

"""The model is also able to identify the sentiment of raw text such as shown below:"""

tweet_text = "The vaccine was great!"

encoded_tweet = tokenizer.encode_plus(
  tweet_text,
  max_length=MAX_LEN,
  add_special_tokens=True,
  return_token_type_ids=False,
  pad_to_max_length=True,
  return_attention_mask=True,
  return_tensors='pt',
)

input_ids = encoded_tweet['input_ids'].to(device)
attention_mask = encoded_tweet['attention_mask'].to(device)

output = model(input_ids, attention_mask,return_dict=False)
_, prediction = torch.max(output, dim=1)

print(f'Tweet text: {tweet_text}')
print(f'Sentiment  : {class_names[prediction]}')

"""# References:
https://towardsdatascience.com/my-absolute-go-to-for-sentiment-analysis-textblob-3ac3a11d524


https://textblob.readthedocs.io/en/dev/advanced_usage.html#sentiment-analyzers

https://curiousily.com/posts/sentiment-analysis-with-bert-and-hugging-face-using-pytorch-and-python/

https://www.nbshare.io/notebook/754493525/Tweet-Sentiment-Analysis-Using-LSTM-With-PyTorch/

# Data visualization:

For the data visualization we put all the sentiment analysis and the new columns added with TextBlob into a new file
"""

file = open("tweets_visual.csv", "a")
df.to_csv('tweets_visual.csv')