coding utf8 gaussian mixture model clusters with gaussian prior finds clusters by repeatedly performing expectationmaximization em algorithm on the dataset gmm assumes the datasets is distributed in multivariate gaussian and tries to find the underlying structure of the gaussian i e mean and covariance estep computes the responsibility of the data to each cluster given the mean and covariance mstep computes the mean covariance and weights prior of each cluster given the responsibilities it iterates until the total likelihood changes less than the tolerance parameters k int the number of clusters into which the dataset is partitioned maxiters int the maximum iterations of assigning points to the perform em shortcircuited by the assignments converging on their own init str default random the name of the method used to initialize the first clustering random randomly select values from the dataset as the k centroids kmeans initialize the centroids covariances weights with kmeams s clusters tolerance float default 1e3 the tolerance of difference of the two latest likelihood for convergence perform expectationmaximization em until converged self setupinputx y self initialize for in rangeself maxiters self estep self mstep if self isconverged break def initializeself self weights np onesself k if self init random self means self xx for x in random samplerangeself nsamples self k self covs np covself x t for in rangeself k elif self init kmeans kmeans kmeanskself k maxitersself maxiters 3 init kmeans fitself x self assignments kmeans predict self means kmeans centroids self covs for i in np uniqueself assignments self weightsinti self assignments i sum self covs appendnp covself xself assignments i t else raise valueerrorunknown type of init parameter self weights self weights sum def estepself maximization mstep for gaussian mixture weights self responsibilities sumaxis0 for assignment in rangeself k resp self responsibilities assignment np newaxis self meansassignment resp self x sumaxis0 resp sum self covsassignment self x self meansassignment t dot self x self meansassignment resp weightsassignment self weights weights weights sum def isconvergedself get the assignments for x with gmm clusters if not x shape return self assignments likelihoods self getlikelihoodx weightedlikelihoods self getweightedlikelihoodlikelihoods assignments weightedlikelihoods argmaxaxis1 return assignments def getlikelihoodself data ndata data shape0 likelihoods np zerosndata self k for c in rangeself k likelihoods c multivariatenormal pdfdata self meansc self covsc return likelihoods def getweightedlikelihoodself likelihood return self weights likelihood def plotself datanone axnone holdonfalse generate grid plot scatters plot contours coding utf 8 gaussian mixture model clusters with gaussian prior finds clusters by repeatedly performing expectation maximization em algorithm on the dataset gmm assumes the datasets is distributed in multivariate gaussian and tries to find the underlying structure of the gaussian i e mean and covariance e step computes the responsibility of the data to each cluster given the mean and covariance m step computes the mean covariance and weights prior of each cluster given the responsibilities it iterates until the total likelihood changes less than the tolerance parameters k int the number of clusters into which the dataset is partitioned max_iters int the maximum iterations of assigning points to the perform em short circuited by the assignments converging on their own init str default random the name of the method used to initialize the first clustering random randomly select values from the dataset as the k centroids kmeans initialize the centroids covariances weights with kmeams s clusters tolerance float default 1e 3 the tolerance of difference of the two latest likelihood for convergence perform expectation maximization em until converged set the initial weights means and covs with full covariance matrix weights the prior of the clusters what percentage of data does a cluster have means the mean points of the clusters covs the covariance matrix of the clusters expectation e step for gaussian mixture maximization m step for gaussian mixture check if the difference of the latest two likelihood is less than the tolerance get the assignments for x with gmm clusters plot contour for 2d data generate grid plot scatters plot contours
import random

import matplotlib.pyplot as plt
import numpy as np
from scipy.stats import multivariate_normal

from mla.base import BaseEstimator
from mla.kmeans import KMeans


class GaussianMixture(BaseEstimator):
    

    y_required = False

    def __init__(self, K=4, init="random", max_iters=500, tolerance=1e-3):
        self.K = K
        self.max_iters = max_iters
        self.init = init
        self.assignments = None
        self.likelihood = []
        self.tolerance = tolerance

    def fit(self, X, y=None):
        
        self._setup_input(X, y)
        self._initialize()
        for _ in range(self.max_iters):
            self._E_step()
            self._M_step()
            if self._is_converged():
                break

    def _initialize(self):
        
        self.weights = np.ones(self.K)
        if self.init == "random":
            self.means = [self.X[x] for x in random.sample(range(self.n_samples), self.K)]
            self.covs = [np.cov(self.X.T) for _ in range(self.K)]

        elif self.init == "kmeans":
            kmeans = KMeans(K=self.K, max_iters=self.max_iters // 3, init="++")
            kmeans.fit(self.X)
            self.assignments = kmeans.predict()
            self.means = kmeans.centroids
            self.covs = []
            for i in np.unique(self.assignments):
                self.weights[int(i)] = (self.assignments == i).sum()
                self.covs.append(np.cov(self.X[self.assignments == i].T))
        else:
            raise ValueError("Unknown type of init parameter")
        self.weights /= self.weights.sum()

    def _E_step(self):
        
        likelihoods = self._get_likelihood(self.X)
        self.likelihood.append(likelihoods.sum())
        weighted_likelihoods = self._get_weighted_likelihood(likelihoods)
        self.assignments = weighted_likelihoods.argmax(axis=1)
        weighted_likelihoods /= weighted_likelihoods.sum(axis=1)[:, np.newaxis]
        self.responsibilities = weighted_likelihoods

    def _M_step(self):
        
        weights = self.responsibilities.sum(axis=0)
        for assignment in range(self.K):
            resp = self.responsibilities[:, assignment][:, np.newaxis]
            self.means[assignment] = (resp * self.X).sum(axis=0) / resp.sum()
            self.covs[assignment] = (self.X - self.means[assignment]).T.dot(
                (self.X - self.means[assignment]) * resp
            ) / weights[assignment]
        self.weights = weights / weights.sum()

    def _is_converged(self):
        
        if (len(self.likelihood) > 1) and (self.likelihood[-1] - self.likelihood[-2] <= self.tolerance):
            return True
        return False

    def _predict(self, X):
        
        if not X.shape:
            return self.assignments
        likelihoods = self._get_likelihood(X)
        weighted_likelihoods = self._get_weighted_likelihood(likelihoods)
        assignments = weighted_likelihoods.argmax(axis=1)
        return assignments

    def _get_likelihood(self, data):
        n_data = data.shape[0]
        likelihoods = np.zeros([n_data, self.K])
        for c in range(self.K):
            likelihoods[:, c] = multivariate_normal.pdf(data, self.means[c], self.covs[c])
        return likelihoods

    def _get_weighted_likelihood(self, likelihood):
        return self.weights * likelihood

    def plot(self, data=None, ax=None, holdon=False):
        
        if not (len(self.X.shape) == 2 and self.X.shape[1] == 2):
            raise AttributeError("Only support for visualizing 2D data.")

        if ax is None:
            _, ax = plt.subplots()

        if data is None:
            data = self.X
            assignments = self.assignments
        else:
            assignments = self.predict(data)

        COLOR = "bgrcmyk"
        cmap = lambda assignment: COLOR[int(assignment) % len(COLOR)]

        
        delta = 0.025
        margin = 0.2
        xmax, ymax = self.X.max(axis=0) + margin
        xmin, ymin = self.X.min(axis=0) - margin
        axis_X, axis_Y = np.meshgrid(np.arange(xmin, xmax, delta), np.arange(ymin, ymax, delta))

        def grid_gaussian_pdf(mean, cov):
            grid_array = np.array(list(zip(axis_X.flatten(), axis_Y.flatten())))
            return multivariate_normal.pdf(grid_array, mean, cov).reshape(axis_X.shape)

        
        if assignments is None:
            c = None
        else:
            c = [cmap(assignment) for assignment in assignments]
        ax.scatter(data[:, 0], data[:, 1], c=c)

        
        for assignment in range(self.K):
            ax.contour(
                axis_X,
                axis_Y,
                grid_gaussian_pdf(self.means[assignment], self.covs[assignment]),
                colors=cmap(assignment),
            )

        if not holdon:
            plt.show()