<jupyter_start><jupyter_text>(1) [Mix of Gaussian](https://pythonmachinelearning.pro/clustering-with-gaussian-mixture-models/).(2) [GMM](https://towardsdatascience.com/how-to-code-gaussian-mixture-models-from-scratch-in-python-9e7975df5252).(3) [Python Abstracted Classes](https://www.python-course.eu/python3_abstract_classes.php)<jupyter_code>#import libraries
import pandas as pd
import numpy as np
from collections import Counter
from itertools import groupby
from operator import itemgetter
from tqdm import tqdm_notebook
import matplotlib.pyplot as plt
from numpy.linalg import norm
from abc import ABC, abstractmethod 
import time
#Data fetcher 
class DataFetcher:
    """
    DataFetcher: Grabs all csv data as pandas dataframes.
    """
    def __init__(self, directory, X_name, Y_name):
        self.directory = directory
        self.X_name = X_name
        self.Y_name = Y_name
    
    # get training X path
    # return path of data traning
    def _getTrainingXPath(self, subset_num):
        # 0 < subset <= 9
        return "./%s/train%s%d.csv" % (self.directory, self.X_name, (subset_num+1))

    # get training Y path
    # return path of traning labels
    def _getTrainingYPath(self, subset_num):
        # 0 < subset <= 9
        return "./%s/train%s%d.csv" % (self.directory, self.Y_name, (subset_num+1))

    # get all traning X Y 
    def get_all_training_xy(self):
        trainingX_dfs = []
        trainingY_dfs = []
        for i in range(NUM_SUBSETS):
            trainingX_dfs.append(pd.read_csv(self._getTrainingXPath(i), header=None))
            trainingY_dfs.append(pd.read_csv(self._getTrainingYPath(i), header=None))
        return trainingX_dfs, trainingY_dfs


    # #get test X Y
    def get_all_test_xy(self):
        testX_path = "./%s/test%s.csv" % (self.directory, self.X_name)
        testY_path = "./%s/test%s.csv" % (self.directory, self.Y_name)
        testX_df = pd.read_csv(testX_path, header=None)
        testY_df = pd.read_csv(testY_path, header=None)
        return testX_df, testY_df

#data = DataFetcher('Knn-dataset', 'Data', 'Labels')
# Cross Validation
class CrossValidation:
    def __init__(self, x_dfs, y_dfs):
        self.X = x_dfs
        self.Y = y_dfs
        self.subsets_num = len(x_dfs)

        #split training validation
        #pick one subset for validation and the rest for training(used for both X, and Y)
    def _splitTrainingValidation(self, dfs, subset_num):

        validation_df = dfs[subset_num]

        training_df = dfs[:subset_num] + dfs[subset_num+1:]
        training_df = pd.concat(training_df, ignore_index=True)

        return training_df, validation_df


        #get training validation X
    def _getTrainingValidationX(self, subset_num):
        training_X, validation_X = self._splitTrainingValidation(self.X, subset_num)
        return training_X, validation_X

        #get training validation Y
    def _getTrainingValidationY(self, subset_num):
        training_Y, validation_Y = self._splitTrainingValidation(self.Y, subset_num)
        return training_Y, validation_Y

        #get training validation XY 
    def getTrainingValidationXY(self, subset_num):
        training_X, validation_X = self._getTrainingValidationX(subset_num)
        training_Y, validation_Y = self._getTrainingValidationY(subset_num)

        return [training_X, training_Y, validation_X, validation_Y]
            

        #get all X Y
    def getAllXY(self):
        training_X = pd.concat(self.X, ignore_index=True)
        training_Y = pd.concat(self.Y, ignore_index=True)

        return [training_X, training_Y]

def getAccuracy(actual_y, predicted_y):
    assert(actual_y.__len__()==predicted_y.__len__())
    accuracy = sum(1 for y, y_hat in zip(actual_y, predicted_y) if y == y_hat)/len(actual_y)
    return accuracy
#model class
class Model(ABC):
    def __init__(self, train_x, train_y):
        self.X = train_x
        self.Y = train_y
    
    @classmethod
    def predict(self, x):
        pass

    def predict_df(self, X_df):
        predictions = X_df.apply(lambda row: self.predict(row), raw=True, axis=1)
        return predictions
#importing data 
DIR_NAME = 'knn-dataset'
X_NAME = 'Data'
Y_NAME = 'Labels'
NUM_SUBSETS = 10
POSITIVE_LABEL = 6
NEGATIVE_LABEL = 5
LAMBDAS = [1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000, 10000, 11000, 12000, 13000, 14000, 15000]
#test data
DF = DataFetcher(DIR_NAME, X_NAME, Y_NAME)
test_X, test_Y = DF.get_all_test_xy()
test_X.head()
test_Y.head()
#training validation split
training_X_dfs, training_Y_dfs = DF.get_all_training_xy()
CVData = CrossValidation(training_X_dfs, training_Y_dfs)
print("Training X shape:"+ str(CVData.getTrainingValidationXY(0)[0].shape))
print("Tarining Y shape:"+ str(CVData.getTrainingValidationXY(0)[1].shape))
print("Validation X shape:"+ str(CVData.getTrainingValidationXY(0)[2].shape))
print("Validation Y shape:"+ str(CVData.getTrainingValidationXY(0)[3].shape))
test_Y_np_array = test_Y.to_numpy().flatten()
test_Y_list = test_Y.values.flatten().tolist()
%timeit sum([0 if y == 5 else 1 for y in test_Y_np_array ])/len(test_Y_np_array)
%timeit sum([0 if y == 5 else 1 for y in test_Y_list ])/len(test_Y_list)
test_Y_np_array.data
# # Mixture of Gaussian 
# class MixtureOfGaussian(Model):
#     def __init__(self, train_x, train_y, positive_label, negative_label):
#         super().__init__(train_x, train_y)
#         self.positive_label = positive_label
#         self.negative_label = negative_label
        
    
#         # MLE Estimation for parameters
        
#         y_list = train_y.values.flatten().tolist() #convert df to seq list

#         pi = sum([1 if y == self.positive_label else 0 for y in y_list])/len(y_list)   # pi = summation-n(yn/N)


#         #calculate mu1, mu2, s1, s2, cov, inv_cov
        
#         positive_indices = [ind for ind, y in enumerate(y_list) if y == positive_label]
#         negative_indices = [ind for ind, y in enumerate(y_list) if y != positive_label] 

#         n1 = len(positive_indices)
#         n2 = len(negative_indices)

#         N = n1 + n2

#         positive_X_df = train_x.iloc[positive_indices]
#         negative_X_df = train_x.iloc[negative_indices]

#         mu1 = positive_X_df.mean(axis=0)
#         mu2 = negative_X_df.mean(axis=0)

#         positive_X = positive_X_df.to_numpy()
#         negative_X = negative_X_df.to_numpy()

#         positive_dists = positive_X_df.to_numpy() - mu1.to_numpy()
#         negative_dists = negative_X_df.to_numpy() - mu2.to_numpy()

#         s1 = positive_dists.T.dot(positive_dists) / n1
#         s2 = positive_dists.T.dot(positive_dists) / n2

#         cov = (n1/N) * s1 + (n2/N) * s2
#         cov_inv = np.linalg.inv(cov)

#         w = cov_inv.dot(mu1 - mu2)
#         w_0 = -(1/2) * mu1.T.dot(cov_inv).dot(mu1) + (1/2) * mu2.T.dot(cov_inv).dot(mu2) + np.log(pi/ (1-pi))

#         self.pi = pi
#         self.n1 = n1
#         self.n2 = n2
#         self.mu1 = mu1
#         self.mu2 = mu2
#         self.cov = cov
#         self.w = w
#         self.w_0 = w_0 

#     def predict_prob(self, x):
#         logit_odds = self.w.dot(x) + self.w_0
#         prob = 1 / (1 + np.exp(-logit_odds))        
#         return prob
    
#     def predict(self, x):
#         prob = self.predict_prob(x)
#         if prob > 0.5:
#             return self.positive_label
#         else:
#             return self.negative_label


class GaussianMixture(Model):
    
    def __init__(self, train_X, train_y, positive_label, negative_label):
        super().__init__(train_X, train_y)
        self.positive_label = positive_label
        self.negative_label = negative_label
        
        # MLE Estimates for parameters
        y_list = train_y.values.flatten().tolist()
        pi = sum([1 if y == positive_label else 0 for y in y_list]) / len(y_list)
        
        # for mew 1, mew 2, split train X into positive negative groups and take average vector
        positive_indices = [ind for ind, y in enumerate(y_list) if y == positive_label]
        negative_indices = [ind for ind, y in enumerate(y_list) if y != positive_label]
        
        n1 = len(positive_indices)
        n2 = len(negative_indices)
        
        N = n1 + n2
        
        positive_X_df = train_X.iloc[positive_indices]
        negative_X_df = train_X.iloc[negative_indices]
        
        mu_1 = positive_X_df.mean()
        mu_2 = negative_X_df.mean()
        
        positive_X = positive_X_df.to_numpy()
        negative_X = negative_X_df.to_numpy()

        positive_X_dists = positive_X_df.to_numpy() - mu_1.to_numpy()
        negative_X_dists = negative_X_df.to_numpy() - mu_2.to_numpy()

        s1 = positive_X_dists.T.dot(positive_X_dists) / n1
        s2 = negative_X_dists.T.dot(negative_X_dists) / n2

        cov = (n1/N) * s1 + (n2/N) * s2
        
        cov_inv = np.linalg.inv(cov)
        
        w = cov_inv.dot(mu_1 - mu_2)
        w_0 = -(1/2) * mu_1.T.dot(cov_inv).dot(mu_1) + (1/2) * mu_2.T.dot(cov_inv).dot(mu_2) + np.log(pi/ (1-pi))

        self.pi = pi
        self.n1 = n1
        self.n2 = n2
        self.mu_1 = mu_1
        self.mu_2 = mu_2
        self.cov = cov
        self.w = w
        self.w_0 = w_0
        
    def predict_prob(self, x):
        logit_odds = self.w.dot(x) + self.w_0
        prob = 1 / (1 + np.exp(-logit_odds))        
        return prob
    
    def predict(self, x):
        prob = self.predict_prob(x)
        if prob > 0.5:
            return self.positive_label
        else:
            return self.negative_label
#train and test 
train_x, train_y = CVData.getAllXY()

# Create hypothesis
t1 = time.time()
model = MixtureOfGaussian(train_x, train_y, POSITIVE_LABEL, NEGATIVE_LABEL)
t2 = time.time()

# Train Accuracy 
predicted_train_y = model.predict_df(train_x)
train_accuracy = getAccuracy(train_y.values.flatten(), predicted_train_y.values.flatten())

# Test Accuracy
predicted_test_y = model.predict_df(test_X)
test_accuracy = getAccuracy(test_Y.values.flatten(), predicted_test_y.values.flatten())


print("Train Accuracy: %f" % train_accuracy)
print("Test Accuracy: %f" % test_accuracy)
print("Training Time: " + str(t2-t1))<jupyter_output>Train Accuracy: 0.836000
Test Accuracy: 0.854545
Training Time: 0.00500035285949707
{'fraction_non_alphanumeric': 0.11196572875085191, 'fraction_numerical': 0.023366760782786485, 'mean_word_length': 2.8143334571110286, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 6, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '371356', 'license': 'MIT', 'n_tokens_mistral': 3823, 'n_tokens_neox': 3630, 'n_words': 770, 'path': 'Assignment2/code/Mixture_of_gaussian.ipynb', 'repo_name': 'Kottoz/CS480-680_introduction_to_MachineLearning', 'size': '10271'}