mixture_model.py 文件源码

def fit(self, X):
        # Constants
        max_iter = self.max_iter
        tol = self.tol
        n_samples, n_features = X.shape
        n_components = self.n_components

        # Initialize parameters
        self._init_params(X)

        # Initialize
        responsibility = sp.empty((n_samples, n_components))
        log_likelihood = self.log_likelihood(X)

        for iter in range(max_iter):

            # E step
            for n in range(n_samples):
                for k in range(n_components):
                    responsibility[n][k] = self.pdf(X[n], k) / self.pdf(X[n])

            # M step
            eff = sp.sum(responsibility, axis=0)
            for k in range(n_components):
                # Update mean
                mean = sp.dot(responsibility[:, k], X) / eff[k]

                # Update covariance
                cov = sp.zeros((n_features, n_features))
                for n in range(n_samples):
                    cov += responsibility[n][k] * sp.outer(X[n] - mean, X[n] - mean)
                cov /= eff[k]

                # Update the k component
                self.comps[k] = multivariate_normal(mean, cov, allow_singular=True)

                # Update mixture coefficient
                self.coef[k] = eff[k] / n_samples

            # Convergent test
            log_likelihood_new = self.log_likelihood(X)
            diff = log_likelihood_new - log_likelihood
            print('GMM: {0:5d}: {1:10.5e} {2:10.5e}'.format(
                iter, log_likelihood_new, spla.norm(diff) / spla.norm(log_likelihood)))
            if (spla.norm(diff) < tol * spla.norm(log_likelihood)):
                break

            log_likelihood = log_likelihood_new

        return self