import numpy as np from .validation import * from .util import * import warnings class DensityCluster: """ Clustering by fast search and find of density peaks. Read more: https://www.science.org/doi/10.1126/science.1242072 Examples -------- >>> from density_cluster import DensityCluster >>> import numpy as np >>> X = np.array([[1, 2], [1, 4], [1, 0], ... [10, 2], [10, 4], [10, 0]]) >>> dc = DensityCluster(percent=2.0).fit(X) >>> dc.labels_ array([1, 1, 1, 0, 0, 0], dtype=int32) >>> dc.predict([[0, 0], [12, 3]]) array([1, 0], dtype=int32) >>> dc.cluster_centers_ array([[10., 2.], [ 1., 2.]]) """ def __init__(self, percent=2.0, min_cluster_centers=2, max_cluster_centers=10): """ Constructor Parameters ---------- percent: float Sort all distances in ascending order, and use 'percent' position as dc. min_cluster_centers: int The minimum number of cluster centers max_cluster_centers: int The maximum number of cluster centers """ self.percent_ = percent self.min_cluster_centers_ = min_cluster_centers self.max_cluster_centers_ = max_cluster_centers def fit(self, X, y=None, sample_weight=None): """Compute density clustering. Parameters ---------- X : {array-like, sparse matrix} of shape (n_samples, n_features) Training instances to cluster. It must be noted that the data will be converted to C ordering, which will cause a memory copy if the given data is not C-contiguous. If a sparse matrix is passed, a copy will be made if it's not in CSR format. y : Ignored Not used, present here for API consistency by convention. sample_weight : array-like of shape (n_samples,), default=None The weights for each observation in X. If None, all observations are assigned equal weight. .. versionadded:: 0.20 Returns ------- self : object Fitted estimator. """ X = self._check_test_data(X) sample_weight = check_sample_weight(sample_weight, X, dtype=X.dtype) distance = cal_distance(X, sample_weight) dc = cal_dc(distance, self.percent_) rho = cal_rho(distance, dc) delta, closest_leader = cal_delta(distance, rho) normal_rho = (rho - np.min(rho)) / (np.max(rho) - np.min(rho)) normal_delta = (delta - np.min(delta)) / (np.max(delta) - np.min(delta)) sigma = np.multiply(normal_rho, normal_delta) self.cluster_centers_, cluster_labels = cal_cluster_centers(X, sample_weight, closest_leader, sigma, self.min_cluster_centers_, self.max_cluster_centers_) self.labels_ = cal_outlier(cluster_labels, rho, delta, self.percent_) return self def predict(self, X, sample_weight=None): """ Predict the closest cluster each sample in X belongs to. In the vector quantization literature, `cluster_centers_` is called the code book and each value returned by `predict` is the index of the closest code in the code book. Parameters ---------- X : {array-like, sparse matrix} of shape (n_samples, n_features) New data to predict. sample_weight : array-like of shape (n_samples,), default=None The weights for each observation in X. If None, all observations are assigned equal weight. Returns ------- labels : ndarray of shape (n_samples,) Index of the cluster each sample belongs to. """ check_is_fitted(self) X = self._check_test_data(X) sample_weight = check_sample_weight(sample_weight, X, dtype=X.dtype) labels = cal_labels(np.multiply(X.T, sample_weight).T, self.cluster_centers_, return_inertia=False) return labels def fit_predict(self, X, y=None, sample_weight=None): """ Compute cluster centers and predict cluster index for each sample. Parameters ---------- X: {array-like, sparse matrix} of shape (n_samples, n_features) New data to transform. y: Ignored Not used, present here for API consistency by convention. sample_weight: array-like of shape (n_samples,), default=None The weights for each observation in X. If None, all observations are assigned equal weight. Returns ------- labels: ndarray of shape (n_samples,) Index of the cluster each sample belongs to. """ return self.fit(X, sample_weight=sample_weight).labels_ def _check_test_data(self, X): X = self._validate_data( X, accept_sparse="csr", reset=False, dtype=[np.float64, np.float32], order="C", accept_large_sparse=False, ) return X def _validate_data( self, X="no_validation", y="no_validation", reset=True, validate_separately=False, **check_params, ): """Validate input data and set or check the `n_features_in_` attribute. Parameters ---------- X : {array-like, sparse matrix, dataframe} of shape \ (n_samples, n_features), default='no validation' The input samples. If `'no_validation'`, no validation is performed on `X`. This is useful for meta-estimator which can delegate input validation to their underlying estimator(s). In that case `y` must be passed and the only accepted `check_params` are `multi_output` and `y_numeric`. y : array-like of shape (n_samples,), default='no_validation' The targets. - If `None`, `check_array` is called on `X`. If the estimator's requires_y tag is True, then an error will be raised. - If `'no_validation'`, `check_array` is called on `X` and the estimator's requires_y tag is ignored. This is a default placeholder and is never meant to be explicitly set. In that case `X` must be passed. - Otherwise, only `y` with `_check_y` or both `X` and `y` are checked with either `check_array` or `check_X_y` depending on `validate_separately`. reset : bool, default=True Whether to reset the `n_features_in_` attribute. If False, the input will be checked for consistency with data provided when reset was last True. .. note:: It is recommended to call reset=True in `fit` and in the first call to `partial_fit`. All other methods that validate `X` should set `reset=False`. validate_separately : False or tuple of dicts, default=False Only used if y is not None. If False, call validate_X_y(). Else, it must be a tuple of kwargs to be used for calling check_array() on X and y respectively. `estimator=self` is automatically added to these dicts to generate more informative error message in case of invalid input data. **check_params : kwargs Ignored if validate_separately is not False. `estimator=self` is automatically added to these params to generate more informative error message in case of invalid input data. Returns ------- out : {ndarray, sparse matrix} or tuple of these The validated input. A tuple is returned if both `X` and `y` are validated. """ self._check_feature_names(X, reset=reset) if y is None and self._get_tags()["requires_y"]: raise ValueError( f"This {self.__class__.__name__} estimator " "requires y to be passed, but the target y is None." ) no_val_X = isinstance(X, str) and X == "no_validation" no_val_y = y is None or isinstance(y, str) and y == "no_validation" default_check_params = {"estimator": self} check_params = {**default_check_params, **check_params} if no_val_X and no_val_y: raise ValueError("Validation should be done on X, y or both.") elif not no_val_X and no_val_y: X = check_array(X, input_name="X", **check_params) out = X elif no_val_X and not no_val_y: y = check_y(y, **check_params) out = y else: if validate_separately: # We need this because some estimators validate X and y # separately, and in general, separately calling check_array() # on X and y isn't equivalent to just calling check_X_y() # :( check_X_params, check_y_params = validate_separately if "estimator" not in check_X_params: check_X_params = {**default_check_params, **check_X_params} X = check_array(X, input_name="X", **check_X_params) if "estimator" not in check_y_params: check_y_params = {**default_check_params, **check_y_params} y = check_array(y, input_name="y", **check_y_params) else: X, y = check_X_y(X, y, **check_params) out = X, y if not no_val_X and check_params.get("ensure_2d", True): self._check_n_features(X, reset=reset) return out def _check_feature_names(self, X, *, reset): """Set or check the `feature_names_in_` attribute. .. versionadded:: 1.0 Parameters ---------- X : {ndarray, dataframe} of shape (n_samples, n_features) The input samples. reset : bool Whether to reset the `feature_names_in_` attribute. If False, the input will be checked for consistency with feature names of data provided when reset was last True. .. note:: It is recommended to call `reset=True` in `fit` and in the first call to `partial_fit`. All other methods that validate `X` should set `reset=False`. """ if reset: feature_names_in = get_feature_names(X) if feature_names_in is not None: self.feature_names_in_ = feature_names_in elif hasattr(self, "feature_names_in_"): # Delete the attribute when the estimator is fitted on a new dataset # that has no feature names. delattr(self, "feature_names_in_") return fitted_feature_names = getattr(self, "feature_names_in_", None) X_feature_names = get_feature_names(X) if fitted_feature_names is None and X_feature_names is None: # no feature names seen in fit and in X return if X_feature_names is not None and fitted_feature_names is None: warnings.warn( f"X has feature names, but {self.__class__.__name__} was fitted without" " feature names" ) return if X_feature_names is None and fitted_feature_names is not None: warnings.warn( "X does not have valid feature names, but" f" {self.__class__.__name__} was fitted with feature names" ) return # validate the feature names against the `feature_names_in_` attribute if len(fitted_feature_names) != len(X_feature_names) or np.any( fitted_feature_names != X_feature_names ): message = ( "The feature names should match those that were passed during fit.\n" ) fitted_feature_names_set = set(fitted_feature_names) X_feature_names_set = set(X_feature_names) unexpected_names = sorted(X_feature_names_set - fitted_feature_names_set) missing_names = sorted(fitted_feature_names_set - X_feature_names_set) def add_names(names): output = "" max_n_names = 5 for i, name in enumerate(names): if i >= max_n_names: output += "- ...\n" break output += f"- {name}\n" return output if unexpected_names: message += "Feature names unseen at fit time:\n" message += add_names(unexpected_names) if missing_names: message += "Feature names seen at fit time, yet now missing:\n" message += add_names(missing_names) if not missing_names and not unexpected_names: message += ( "Feature names must be in the same order as they were in fit.\n" ) raise ValueError(message) def _check_n_features(self, X, reset): """Set the `n_features_in_` attribute, or check against it. Parameters ---------- X : {ndarray, sparse matrix} of shape (n_samples, n_features) The input samples. reset : bool If True, the `n_features_in_` attribute is set to `X.shape[1]`. If False and the attribute exists, then check that it is equal to `X.shape[1]`. If False and the attribute does *not* exist, then the check is skipped. .. note:: It is recommended to call reset=True in `fit` and in the first call to `partial_fit`. All other methods that validate `X` should set `reset=False`. """ try: n_features = num_features(X) except TypeError as e: if not reset and hasattr(self, "n_features_in_"): raise ValueError( "X does not contain any features, but " f"{self.__class__.__name__} is expecting " f"{self.n_features_in_} features" ) from e # If the number of features is not defined and reset=True, # then we skip this check return if reset: self.n_features_in_ = n_features return if not hasattr(self, "n_features_in_"): # Skip this check if the expected number of expected input features # was not recorded by calling fit first. This is typically the case # for stateless transformers. return if n_features != self.n_features_in_: raise ValueError( f"X has {n_features} features, but {self.__class__.__name__} " f"is expecting {self.n_features_in_} features as input." )