API Reference

EHDBSCAN

Exhaustive HDBSCAN performs iterative clustering. The noise points left over from each iteration are remapped to a new density structure. At each step the clusters from the current iteration are assigned parents from the previous iteration. This results in a cluster of linked clusters.

This implementation uses HDBSCAN from the sci-kit library.

Overall, this lib relies heavily on sci-kit.

Initialize the EHDBSCAN to perform iterative clustering.

Parameters:

metric (str, default: 'cosine' ) –

The metric to be used by HDBSCAN. It must be one acceptable by sci-kit pairwise_distances method.

If precomputed is passed, the fit method will accept a square matrix of pairwise distances.
tree_metric (str, default: 'cosine' ) –

The metric to be used for forming parent-child links at each clustering step.

It must be one acceptable by pairwise_distances method.

If precomputed, this will not apply.
min_clustersize_multiplier (float, default: 0.05 ) –

The mutliplier applied to the full length of the input at each clustering step.
max_clustersize_multiplier (float, default: 5.0 ) –

The multiplier applied to the minimum cluster size at each clustering step.
min_samples (int, default: 5 ) –

The starting number of minimum samples to be considered by HDBSCAN as neighbors.

This will automatically be reduced by 1 to a minimum of 2 through each clustering step.
n_jobs (int, default: None ) –

Number of threads. Application is same as sci-kit.
log_level (int, default: DEBUG ) –

Set the logging level. Same values apply as default Python logging, DEBUG, INFO, WARNING, ERROR
cluster_core_k (int, default: -1 ) –

This identifies nearest neighbors to be considered for forming parent-child links.

The core distance between clusters.
reducer (Reducer, default: None ) –

The dimension reduction object. The Reducer must be subclassed with desired reduction method.

See examples for implementation.

While dimension reduction is not required for HDBSCAN, it is highly recommended. Typically, UMAP is very effective with HDBSCAN. But UMAP implementations may introduce non-determinism.
encoder (Encoder, default: None ) –

The text embedder object. The Encoder must be subclassed with desired encoding method/model.

If no encoder is used, text embeddings or pairwise distances must be provided.

See examples for implementation.

Returns:	`self`( `object` ) – Returns self.

Attributes:

cluster_feats (ClusterFeatures) –

An instance of ClusterFeatures. Only callable after fit is called.
reducer (Reducer) –

The instance of Reducer passed to the EHDBSCAN object.
encoder (Encoder) –

The instance of Encoder passed to the EHDBSCAN object.
models (dict) –

Dictionary object of HDBSCAN models at each iteration.

If a reducer is used, it will also store the instance of Reducer after each fit is called. In this case, use key "hdb" for HDBSCAN model and "reducer" for the Reducer instance.

Source code in src\exhaustive_hdbscan\ehdbscan.py

def __init__(self, *,
             metric: str = 'cosine',
             tree_metric: str = 'cosine',
             min_clustersize_multiplier: float = 0.05,
             max_clustersize_multiplier: float = 5.0,
             min_samples: int = 5,
             n_jobs: int = None,
             log_level: int = logging.DEBUG,
             cluster_core_k: int = -1,
             reducer: Reducer = None,
             encoder: Encoder = None
             ):

    """
    Initialize the EHDBSCAN to perform iterative clustering.

    Parameters
    ----------

    metric : str
        The metric to be used by HDBSCAN. It must be one acceptable by sci-kit pairwise_distances method.

        If precomputed is passed, the fit method will accept a square matrix of pairwise distances.

    tree_metric : str
        The metric to be used for forming parent-child links at each clustering step.

        It must be one acceptable by pairwise_distances method.

        If precomputed, this will not apply.

    min_clustersize_multiplier : float
        The mutliplier applied to the full length of the input at each clustering step.

    max_clustersize_multiplier : float
        The multiplier applied to the minimum cluster size at each clustering step.

    min_samples : int
        The starting number of minimum samples to be considered by HDBSCAN as neighbors.

        This will automatically be reduced by 1 to a minimum of 2 through each clustering step.

    n_jobs : int
        Number of threads. Application is same as sci-kit.

    log_level : int
        Set the logging level. Same values apply as default Python logging, DEBUG, INFO, WARNING, ERROR

    cluster_core_k : int
        This identifies nearest neighbors to be considered for forming parent-child links.

        The core distance between clusters.

    reducer : Reducer
        The dimension reduction object. The Reducer must be subclassed with desired reduction method.

        See examples for implementation.

        While dimension reduction is not required for HDBSCAN, it is highly recommended. Typically,
        UMAP is very effective with HDBSCAN. But UMAP implementations may introduce non-determinism.

    encoder : Encoder
        The text embedder object. The Encoder must be subclassed with desired encoding method/model.

        If no encoder is used, text embeddings or pairwise distances must be provided.

        See examples for implementation.

    Returns
    -------

    self : object
        Returns self.

    Attributes
    ----------

    cluster_feats : ClusterFeatures
        An instance of ClusterFeatures. Only callable after fit is called.

    reducer : Reducer
        The instance of Reducer passed to the EHDBSCAN object.

    encoder : Encoder
        The instance of Encoder passed to the EHDBSCAN object.

    models : dict
        Dictionary object of HDBSCAN models at each iteration. 

        If a reducer is used, it will also store the instance of Reducer after each fit is called.
        In this case, use key "hdb" for HDBSCAN model and "reducer" for the Reducer instance.


    """
    self.metric = validate_metric(metric)
    self.tree_metric = validate_tree_metric(tree_metric)
    self.precomputed = True if self.metric == 'precomputed' else False
    self.min_clustersize_multiplier = min_clustersize_multiplier
    self.max_clustersize_multiplier = max_clustersize_multiplier
    self.min_samples = min_samples
    self.n_jobs = n_jobs
    logger.setLevel(log_level)
    streamer.setLevel(log_level)
    self.cluster_core_k = cluster_core_k
    self.reducer = reducer
    self.encoder = encoder
    self.models = {}
    self.isfit = False

`fit(X, y=None)`

The fit method, performs the iterative clustering.

Parameters:	`X` (`array - like`, default: `array-like` ) – A feature array of text, embeddings or pairwise distances. `y` (`None`, default: `None` ) – Ignored

Returns:	`self`( `object` ) – Returns self.

Source code in src\exhaustive_hdbscan\ehdbscan.py

def fit(self, X, y=None):
    """
    The fit method, performs the iterative clustering.

    Parameters
    ----------

    X : {array-like} of shape (n_samples, n_features) or ndarray of shape (n_samples, n_samples)
        A feature array of text, embeddings or pairwise distances.

    y : None
        Ignored

    Returns
    -------

    self : object
        Returns self.

    """

    X_recur, is_embedding = validate_input(X)
    idx = np.arange(0, len(X_recur))
    mask = np.ones(idx.shape, dtype=bool)
    itr_idx = idx[mask]

    self.cluster_feats = ClusterFeatures()

    if self.precomputed:
        self.cluster_feats._add_distances(X_recur)
    else:
        if is_embedding == False:
            self.cluster_feats._add_input(X)
            if self.encoder == None:
                raise ValueError('Expects Encoder to be specified when text is input. OR directly input embeddings or numeric features.')
            X_recur = self.encoder.encode(X_recur)
            self.cluster_feats._add_embeddings(X_recur)

        pdist = pairwise_distances(X_recur, metric=self.tree_metric)
        self.cluster_feats._add_distances(pdist)

    recur = True
    itr = 0        
    min_samps = 5
    min_cluster_size_m = 0.05
    max_cluster_size_m = 5
    min_size = int(round(len(X_recur) * min_cluster_size_m))

    if self.min_clustersize_multiplier:
        min_cluster_size_m = self.min_clustersize_multiplier
        min_size = int(round(len(X_recur) * min_cluster_size_m))

    if self.max_clustersize_multiplier:
        max_cluster_size_m = self.max_clustersize_multiplier

    if self.min_samples:
        min_samps = self.min_samples


    while recur:
        model = {'hdb': None, 'reducer': None}

        X_hdb = X_recur

        if self.reducer:
            redr = self.reducer.fit(X_recur)
            X_hdb = self.reducer.transform(X_recur)
            redr_fill_shape = (idx.shape[0], X_hdb.shape[-1])
            if len(X_hdb.shape) == 1:
                redr_fill_shape = (idx.shape[0],)
            redr_fill = np.full(shape=redr_fill_shape, fill_value=np.nan)
            redr_fill[itr_idx] = X_hdb
            self.cluster_feats._add_reduce_embeddings(redr_fill)
            model['reducer'] = deepcopy(self.reducer)


        hdb = HDBSCAN(metric=self.metric,
                            min_cluster_size=min_size,
                            min_samples=min(min_samps, min_size // 2),
                            n_jobs=self.n_jobs,
                            max_cluster_size=int(min_size * max_cluster_size_m))

        hdb.fit(X_hdb)
        model['hdb'] = deepcopy(hdb)

        labels = hdb.labels_

        if max(labels) < 0:
            recur = False
            logger.info(f'Exiting clustering loop. No more clusters found.')
            break

        current_labels = list(np.unique(labels))

        if -1 in current_labels:
            current_labels.remove(-1)
        else:
            logger.info(f'FINAL ITERATION {itr}. All values consumed.')
            recur = False

        fill_vec = np.full(shape=idx.shape, fill_value=-2)
        fill_vec[itr_idx] = labels
        self.cluster_feats._add_parent(fill_vec)
        self.models[self.cluster_feats.parent_names[-1]] = model

        logger.info(f'Iteration: {itr}, Clusters: {current_labels}, Noise: {np.sum((labels == -1)).item()}')

        if itr > 0:
            child_vec = np.full(shape=idx.shape, fill_value=-2)

            for label in current_labels:
                this_child_mask = (fill_vec == label)
                this_child_dist = []
                for plabel in parent_labels:                    
                    this_parent_mask = (parent_clusters == plabel)
                    this_child_dist.append(self._cluster_sim_score_dst(self.cluster_feats.get_distances()[0], parent=idx[this_parent_mask],
                                                                       child=idx[this_child_mask]))                    
                max_idx = np.argmax(this_child_dist)
                logger.info(f'For child_{itr-1}={label}, parent_{itr-1}={max_idx}')
                child_vec[this_child_mask] = parent_labels[max_idx]

            self.cluster_feats._add_child(child_vec)

        parent_labels = current_labels
        parent_clusters = fill_vec

        mask = (labels == -1)
        itr_idx = itr_idx[mask]

        X_recur = X_recur[mask, :]

        if self.precomputed:
            X_recur = X_recur[:, mask]

        if len(X_recur) <= 5:
            recur = False
            logger.info(f'Exiting clustering loop. Minimum values reached. Remaining length = {len(X_recur)}')
            break

        min_size = int(round(len(X_recur) * min_cluster_size_m))
        if min_size < 5:
            min_size = 5

        max_size = min_size * 5
        if max_size > len(X_recur):
            max_size = len(X_recur)

        min_samps -= 1
        if min_samps <= 2:
            min_samps = 2

        itr += 1


    self.isfit = True

    return self

`approx_predict(X)`

Approximate prediction will assign unseen data to existing clusters.

HDBSCAN is not built for predictions, so this is not a perfect transform, only an approximate prediction.

It expects data to be passed in the same format as the fit method. But if text was passed to fit, it will still work with embeddings.

This method will not work with pairwise distances where metric is precomputed.

If EHDBSCAN fit was executed with pairwise distances, this method will not work, even if text or embeddings are passed for prediction.

Parameters:	`X` (`array - like`, default: `array-like` ) – An array of text or embeddings.

Returns:	`ClusterFeatures`( `object` ) – An instance of ClusterFeatures for the prediction X.

Source code in src\exhaustive_hdbscan\ehdbscan.py

def approx_predict(self, X):
    """
    Approximate prediction will assign unseen data to existing clusters.

    HDBSCAN is not built for predictions, so this is not a perfect transform,
    only an approximate prediction.

    It expects data to be passed in the same format as the fit method. But if text was
    passed to fit, it will still work with embeddings.

    This method will not work with pairwise distances where metric is precomputed.

    If EHDBSCAN fit was executed with pairwise distances, this method will not work,
    even if text or embeddings are passed for prediction.

    Parameters
    ----------
    X : {array-like} of shape (n_samples, n_features)
        An array of text or embeddings.

    Returns
    -------
    ClusterFeatures : object
        An instance of ClusterFeatures for the prediction X.

    """

    if self.isfit == False:
        raise TypeError('Cannot be called on un-fitted model. Please call fit first.')

    if self.precomputed:
        raise ValueError('This method is not compatible with precomputed distances. It requires raw features to calculate pair distances.')

    cluster_feats = ClusterFeatures()

    X_input, is_embedding = validate_input(X)

    if is_embedding == False:
        cluster_feats._add_input(X_input)
        X_input = self.encoder.encode(X_input)
        cluster_feats._add_embeddings(X_input)
        pair_dist = pairwise_distances(X_input, metric=self.tree_metric)
        cluster_feats._add_distances(pair_dist)

    X_recur = deepcopy(X_input)

    idx = np.arange(0, X_input.shape[0])
    itr_idx = np.arange(0, X_input.shape[0])

    min_samps = self.min_samples

    id_mask = 0

    parent_labels = []

    for i, ex_parent in enumerate(self.cluster_feats.parent_names):
        if len(X_recur) <= 5:
            logging.info('Exiting clustering loop. Minimum values reached.')
            break

        ex_embeddings = ''
        ex_ids = ''
        y_dst = ''
        y_nbrs = NearestNeighbors(metric=self.metric, n_neighbors=min_samps)
        ex_nbrs = NearestNeighbors(metric=self.metric, n_neighbors=min_samps)
        exy_dst = ''

        if self.reducer:
            ex_embeddings, ex_ids = self.cluster_feats.get_reduce_embeddings(name=ex_parent)
            y_redr = self.models[ex_parent]['reducer'].transform(X_recur)
            y_redr_shape = (idx.shape[0], y_redr.shape[-1])
            if len(y_redr.shape) == 1:
                y_redr_shape = (idx.shape[0],)
            emb_fill = np.full(shape=y_redr_shape, fill_value=np.nan)
            emb_fill[itr_idx] = y_redr
            cluster_feats._add_reduce_embeddings(emb_fill)
            y_nbrs.fit(y_redr)
            y_dst, _ = y_nbrs.kneighbors(y_redr)
            exy_dst = pairwise_distances(y_redr, ex_embeddings, metric=self.metric)
        else:
            ex_embeddings, ex_ids = self.cluster_feats.get_embeddings(name=ex_parent)
            y_nbrs.fit(X_recur)
            y_dst, _ = y_nbrs.kneighbors(X_recur)
            exy_dst = pairwise_distances(X_recur, ex_embeddings, metric=self.metric)

        ex_labels, _ = self.cluster_feats.get_parent(name=ex_parent)
        ex_nbrs.fit(ex_embeddings)
        ex_dst, _ = ex_nbrs.kneighbors(ex_embeddings)

        this_y_dist = y_dst[:, -1]
        this_y_dist = this_y_dist.reshape(-1, 1)

        this_ex_dist = ex_dst[:, -1]

        ext_p_dist = np.tile(this_ex_dist, this_y_dist.shape[0])
        ext_p_dist = ext_p_dist.reshape(-1, this_ex_dist.shape[0])

        mrd = np.maximum(np.maximum(this_y_dist, ext_p_dist), exy_dst)

        min_mrd = np.argmin(mrd, axis=1)

        ex_core = this_ex_dist[min_mrd]
        this_ex_labels = ex_labels[min_mrd]

        labels = np.where(this_y_dist.ravel() < ex_core, this_ex_labels, -1)

        child_labels = list(np.unique(labels))

        logger.info(f'Iteration {i}, Clusters: {child_labels}')

        if max(child_labels) < 0:
            logger.info('Exiting clustering loop. No more clusters found.')
            break

        if -1 in child_labels:
            child_labels.remove(-1)
        else:
            logging.info(f'FINAL ITERATION {i}. All values consumed.')

        child_labels = [lb for lb in child_labels if lb >= 0]

        fill_vec = np.full(shape=idx.shape[0], fill_value=-2)
        fill_vec[itr_idx] = labels

        cluster_feats._add_parent(fill_vec)

        if i > 0:
            child_vec = np.full(shape=idx.shape[0], fill_value=-2)
            all_dist = cluster_feats.get_distances()[0]
            for label in child_labels:
                this_child_mask = (fill_vec == label)
                this_child_dst = []
                for plabel in parent_labels:
                    _, this_parent_ids = cluster_feats.get_parent(name=cluster_feats.parent_names[i-1], label=plabel)
                    this_child_dst.append(self._cluster_sim_score_dst(all_dist, parent=this_parent_ids, child=idx[this_child_mask]))
                id_max = np.argmax(this_child_dst)
                child_ids = idx[this_child_mask]
                child_vec[child_ids] = parent_labels[id_max]
                logging.info(f'Child = {label}, Parent = {id_max}')
            cluster_feats._add_child(child_vec)

        parent_labels = child_labels
        min_samps -= 1
        id_mask = (labels < 0)
        itr_idx = itr_idx[id_mask]
        X_recur = X_recur[id_mask, :]

    return cluster_feats

`plot_tree(cluster_features=None, labeler=None, figsize=(20, 10), horizontal_spacing=3, vertical_spacing=1, ax=None, label_font_size=10, label_box_dim=100, label_box_style='round,pad=0.3', label_box_color='aliceblue', label_box_border_color='black', label_box_border_width=1)`

It plots the tree of clusters with parent-child links.

Parameters:

cluster_features (ClusterFeatures, default: None ) –

If ClusterFeatures are provided, it will plot the tree of the input. If not provided, it will plot the tree of the cluster features from the fit.
labeler (LabelGeneratorConfig, default: None ) –

If labeler is provided, they will be displayed on the tree nodes.

Labeler must be a subclass of LabelGeneratorConfig.

See examples for implementation
figsize (tuple, default: (20, 10) ) –

A tuple setting the figure size. Same as matplotlib figsize.
horizontal_spacing (int, default: 3 ) –

Set the horizontal spacing between tree nodes.
vertical_spacing (int, default: 1 ) –

Set the vertical spacing between tree levels.
ax (Axes, default: None ) –

Set the axis object on which to draw the tree.
label_font_size (int, default: 10 ) –

Font size for node labels.
label_box_dim (int, default: 100 ) –

Box size for node boxes.
label_box_style (str, default: 'round,pad=0.3' ) –

Box style for label boxes. Matplotlib compatible.
label_box_color (str, default: 'aliceblue' ) –

Box inner color. Matplotlib compatible.
label_box_border_color (str, default: 'black' ) –

Box border line color. Matplotlib compatible.
label_box_border_width (int, default: 1 ) –

Box border line width.

Returns:	`(Figure, Axes)` – The instance of figure and axis on which the tree is drawn.

Source code in src\exhaustive_hdbscan\ehdbscan.py

def plot_tree(self,
              cluster_features: ClusterFeatures = None,
              labeler: LabelGeneratorConfig = None,
             figsize: tuple = (20, 10),
              horizontal_spacing: int = 3,
              vertical_spacing: int = 1,
              ax : matplotlib.pyplot.axis = None,
              label_font_size: int = 10,
              label_box_dim: int = 100,
              label_box_style: str = "round,pad=0.3",
              label_box_color: str = 'aliceblue',
              label_box_border_color: str = 'black',
              label_box_border_width: int = 1
             ):

    """
    It plots the tree of clusters with parent-child links.

    Parameters
    ----------
    cluster_features : ClusterFeatures
        If ClusterFeatures are provided, it will plot the tree of the input. If
        not provided, it will plot the tree of the cluster features from the fit.

    labeler : LabelGeneratorConfig
        If labeler is provided, they will be displayed on the tree nodes.

        Labeler must be a subclass of LabelGeneratorConfig.

        See examples for implementation

    figsize : tuple
        A tuple setting the figure size. Same as matplotlib figsize.

    horizontal_spacing : int
        Set the horizontal spacing between tree nodes.

    vertical_spacing : int
        Set the vertical spacing between tree levels.

    ax : matplotlib.axes.Axes
        Set the axis object on which to draw the tree.

    label_font_size : int
        Font size for node labels.

    label_box_dim : int
        Box size for node boxes.

    label_box_style : str
        Box style for label boxes. Matplotlib compatible.

    label_box_color : str
        Box inner color. Matplotlib compatible.

    label_box_border_color : str
        Box border line color. Matplotlib compatible.

    label_box_border_width : int
        Box border line width.



    Returns
    -------
    matplotlib.figure.Figure, matplotlib.axes.Axes
        The instance of figure and axis on which the tree is drawn.


    """

    cluster_feats = self.cluster_feats

    if cluster_features:
        cluster_feats = cluster_features

    parent_count = 0
    child_count = 0

    vert_spacing = vertical_spacing
    hori_spacing = horizontal_spacing

    annotate_labels = []

    if labeler:
        if not isinstance(labeler, LabelGeneratorConfig):
            raise ValueError('Expects labeler to be of type LabelGeneratorConfig.')
        annotate_labels = labeler.compute(cluster_feats)

    ann_x = 0
    ann_y = parent_count

    x_range = [0, 0]
    annotations = []


    fig, axs = plt.subplots(figsize=figsize)
    if ax:
        axs = ax
    parent_boxes = {}
    total_itr = 0

    for i, parent in enumerate(cluster_feats.parent_names):
        parent_all, _ = cluster_feats.get_parent(name=parent)
        parent_all = parent_all[(parent_all >= 0)]
        parent_labels = list(np.unique(parent_all))
        this_parent = {}
        for k, label in enumerate(parent_labels):
            parent_text = ''
            if ann_x > 0:
                ann_x += k * -hori_spacing
            else:
                ann_x += k * hori_spacing
            ann_label = k
            ann_label = f'Parent{i}_Cluster{k}'
            if labeler:
                ann_label += f'\n{annotate_labels[total_itr]}'

            self._annotate_box(axs, ann_label, (ann_x, ann_y),
                               box_style=label_box_style,
                               box_color=label_box_color,
                               box_border_color=label_box_border_color,
                               box_border_width=label_box_border_width,
                               dim=label_box_dim,
                               fontsize=label_font_size)

            this_parent[label] = {'points': [ann_x, ann_y], 'label': ann_label}
            total_itr += 1

        parent_boxes[i] = this_parent
        ann_y -= vert_spacing
        hori_spacing += 1
        ann_x = 0

    if len(parent_boxes.keys()) == 1:
        box_locs = parent_boxes[0]
        k = list(box_locs.keys())[-1]
        loc = box_locs[k]['points']
        xlim = abs(loc[0])
        axs.set_ylim(-1, 1)
        axs.set_xlim(-xlim, xlim)
    else:
        for parent in parent_boxes.keys():
            if parent < len(parent_boxes.keys()) - 1:
                box_locs = parent_boxes[parent]
                parent_labels, _ = cluster_feats.get_parent(name=f'parent_{parent}')
                child_labels, child_pts = cluster_feats.get_child(name=f'child_{parent}')

                for label in box_locs.keys():
                    parent_loc = box_locs[label]['points']                    
                    child_cluster_mask = (child_labels == label)
                    sub_parent, sub_parent_pts = cluster_feats.get_parent(name=f'parent_{parent+1}')
                    csp_mask = np.isin(sub_parent_pts, child_pts[child_cluster_mask])
                    sub_parent = sub_parent[csp_mask]
                    sub_parents = np.unique(sub_parent)
                    sub_parent_locs = parent_boxes[parent+1]
                    for sub in sub_parents:                       
                        child_loc = sub_parent_locs[sub]['points']
                        axs.plot([parent_loc[0], child_loc[0]], [parent_loc[1], child_loc[1]], 'k-', lw=1)
    plt.axis('off')
    plt.show()

    return fig, axs

ClusterFeatures

The data layer for the EHDBSCAN. It provides storage and easy access of clusters, inputs, distances and embeddings.

The ClusterFeatures object is accessible as an attribute from the EHDBSCAN object after the fit is called. It is returned by the approx_predict method as a separate instance.

It is not intended for populating manually. Storage are automatically encoded with designated name where names are sequentially automatically generated.

Naming convention will assign names as {cluster type}_{iteration index}. For example second iteration will designate the generated labels as "parent_1" and the assigned child labels as "child_0". Here "parent_1" is the current iteration label and "child_0" is the child of the previous iteration "parent_0".

Note that any child labels will have the same unique indexes as its parent indicating which parent that child belongs to.

Attributes:	`parent_names` (`list`) – List of all parent names in the order their labels and embeddings are stored. `child_names` (`list`) – List of all children names in the order their labels and embeddings are stored.

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def __init__(self):
    self.parent_labels = np.array([])
    self.child_labels = np.array([])
    self.parent_names = []
    self.child_names = []
    self.pandas_data = pd.DataFrame()
    self.distances = 0
    self.embeddings = np.array([])
    self.reduce_embeddings = np.array([])
    self.input = np.array([])

`get_parent(name=None, label=None, get_ids=None)`

Get the label of parent by name, label and index ID.

If neither is specified, all labels are returned.

Parameters:

name (str, default: None ) –

Name of parent from parent_names.
label (int, default: None ) –

Label from the parent where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the parent labels.

If no name or label is specified, it will return the IDs for all parent clusters.

Returns:	`(ndarray, ndarray)` – Returns an array of the labels and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_parent(self, name: str = None, label: int = None, get_ids: Any = None):
    """
    Get the label of parent by name, label and index ID.

    If neither is specified, all labels are returned.

    Parameters
    ----------
    name : str
        Name of parent from parent_names.

    label : int
        Label from the parent where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the parent labels.

        If no name or label is specified, it will return the IDs for all parent
        clusters.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the labels and their respective IDs

    """
    parent_out = self.parent_labels
    select_ids = np.arange(0, parent_out.shape[0])
    idx = None

    if name is not None:
        try:
            idx = self.parent_names.index(name)
        except:
            raise ValueError('Parent name not found.')
        parent_out = self.parent_labels[:, idx]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, specific parent name must be provided to retrieve the labels.')
        else:
            label_mask = (parent_out == label)
            select_ids = select_ids[label_mask]
            parent_out = parent_out[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)
        select_ids = select_ids[id_mask]
        parent_out = parent_out[id_mask]

    return parent_out, select_ids

`get_child(name=None, label=None, get_ids=None)`

Get the label of child by name, label and index ID.

If neither is specified, all labels are returned.

Parameters:

name (str, default: None ) –

Name of child from child_names.
label (int, default: None ) –

Label from the child where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the child labels.

If no name or label is specified, it will return the IDs for all child clusters.

Returns:	`(ndarray, ndarray)` – Returns an array of the labels and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_child(self, name=None, label=None, get_ids=None):
    """
    Get the label of child by name, label and index ID.

    If neither is specified, all labels are returned.

    Parameters
    ----------
    name : str
        Name of child from child_names.

    label : int
        Label from the child where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the child labels.

        If no name or label is specified, it will return the IDs for all child
        clusters.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the labels and their respective IDs

    """
    child_out = self.child_labels
    select_ids = np.arange(0, child_out.shape[0])
    idx = None

    if name is not None:
        try:
            idx = self.child_names.index(name)
        except:
            raise ValueError('Child name not found.')
        child_out = self.child_labels[:, idx]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, specific child name must be provided to retrieve the labels.')
        else:
            label_mask = (child_out == label)
            select_ids = select_ids[label_mask]
            child_out = child_out[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)
        select_ids = select_ids[id_mask]
        child_out = child_out[id_mask]

    return child_out, select_ids

`get_input(name=None, label=None, get_ids=None)`

Get the input text by parent or child name, label and index ID.

If neither is specified, all input is returned.

Parameters:

name (str, default: None ) –

Name of parent or child from parent_names or child_names.
label (int, default: None ) –

Label from the parent or child where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the input.

If no name or label is specified, it will return the IDs for all clusters.

Returns:	`(ndarray, ndarray)` – Returns an array of the input text and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_input(self, name=None, label=None, get_ids=None):
    """
    Get the input text by parent or child name, label and index ID.

    If neither is specified, all input is returned.

    Parameters
    ----------
    name : str
        Name of parent or child from parent_names or child_names.

    label : int
        Label from the parent or child where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the input.

        If no name or label is specified, it will return the IDs for all clusters.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the input text and their respective IDs

    """
    in_out = self.input
    name_array = None
    idx = None
    select_ids = np.arange(0, self.input.shape[0])
    is_child = False

    if name is not None:
        try:
            idx = self.parent_names.index(name)
        except:
            try:
                idx = self.child_names.index(name)
                is_child = True
            except:
                raise ValueError('Name not found in parent and child.')

        if is_child:
            name_array = self.child_labels[:, idx]
        else:
            name_array = self.parent_labels[:, idx]

        name_mask = (name_array >= 0)
        in_out = self.input[name_mask]
        name_array = name_array[name_mask]
        select_ids = select_ids[name_mask]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, specific parent or child name must be provided to retrieve the values.')
        else:
            label_mask = (name_array == label)
            in_out = in_out[label_mask]
            select_ids = select_ids[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)
        select_ids = select_ids[id_mask]
        in_out = in_out[id_mask]

    return in_out, select_ids

`get_distances(name=None, label=None, get_ids=None, row_ids=None, col_ids=None)`

Get the pairwise distances by parent or child name, label and index ID.

If neither is specified, all distances are returned.

Parameters:

name (str, default: None ) –

Name of parent or child from parent_names or child_names.
label (int, default: None ) –

Label from the parent or child where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the input.

If no name or label is specified, it will return the IDs for all clusters.
row_ids (array - like, default: array-like ) –

Specify the row_ids to be extracted from the square matrix of pairwise distances.

This will override name, label and ID specification.
col_ids (array - like, default: array-like ) –

Specify the col_ids to extracted from the square matrix of pairwise distances.

This will override name, label and ID specification, but not row_ids.

If row_ids are specified, the specified columns will be extracted only for those rows.

Returns:	`(ndarray, ndarray)` – Returns an array of the input text and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_distances(self, name=None, label=None, get_ids=None, row_ids=None, col_ids=None):
    """
    Get the pairwise distances by parent or child name, label and index ID.

    If neither is specified, all distances are returned.

    Parameters
    ----------
    name : str
        Name of parent or child from parent_names or child_names.

    label : int
        Label from the parent or child where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the input.

        If no name or label is specified, it will return the IDs for all clusters.

    row_ids : {array-like} of shape (n_samples,)
        Specify the row_ids to be extracted from the square matrix of pairwise
        distances.

        This will override name, label and ID specification.

    col_ids : {array-like} of shape (n_samples,)
        Specify the col_ids to extracted from the square matrix of pairwise
        distances.

        This will override name, label and ID specification, but not row_ids.

        If row_ids are specified, the specified columns will be extracted only
        for those rows.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the input text and their respective IDs

    """
    dst_out = self.distances
    name_array = None
    select_ids = np.arange(0, self.distances.shape[0])
    is_child = False
    idx = None
    row_col = False

    if name is not None:
        try:
            idx = self.parent_names.index(name)
        except:
            try:
                idx = self.child_names.index(name)
                is_child = True
            except:
                raise ValueError('Name not found in parent and child.')

        if is_child:
            name_array = self.child_labels[:, idx]
        else:
            name_array = self.parent_labels[:, idx]
        name_mask = (name_array >= 0)
        dst_out = self.distances[:, name_mask]
        dst_out = dst_out[name_mask, :]
        name_array = name_array[name_mask]
        select_ids = select_ids[name_mask]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, specific parent or child name must be provided to retrieve the values.')
        else:
            label_mask = (parent_labels == label)
            if np.any(label_mask):
                dst_out = dst_out[:, label_mask]
                dst_out = dst_out[label_mask, :]
                select_ids = select_ids[label_mask]
            else:
                logger.info('Label not found.')
                dst_out = dst_out[:, label_mask]
                select_ids = select_ids[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)
        select_ids = select_ids[id_mask]
        dst_out = dst_out[id_mask, :]
        dst_out = dst_out[:, id_mask]

    if row_ids is not None:
        row_col = True
        dst_out = self.distances[row_ids, :]
        select_ids = np.tile(np.arange(0, self.distances.shape[0]), self.distances.shape[0]).reshape(self.distances.shape)  
        select_ids = select_ids[row_ids, :]

    if col_ids is not None:
        if row_col:
            dst_out = dst_out[:, col_ids]
            select_ids = select_ids[:, col_ids]
        else:
            dst_out = self.distances[:, col_ids]
            select_ids = np.tile(np.arange(0, self.distances.shape[0]), self.distances.shape[0]).reshape(self.distances.shape)
            select_ids = select_ids[:, col_ids]


    return dst_out, select_ids

`get_embeddings(name=None, label=None, get_ids=None)`

Get the text embeddings by parent or child name, label and index ID.

If neither is specified, all input is returned.

Parameters:

name (str, default: None ) –

Name of parent or child from parent_names or child_names.
label (int, default: None ) –

Label from the parent or child where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the input.

If no name or label is specified, it will return the IDs for all clusters.

Returns:	`(ndarray, ndarray)` – Returns an array of the text embeddings and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_embeddings(self, name=None, label=None, get_ids=None):
    """
    Get the text embeddings by parent or child name, label and index ID.

    If neither is specified, all input is returned.

    Parameters
    ----------
    name : str
        Name of parent or child from parent_names or child_names.

    label : int
        Label from the parent or child where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the input.

        If no name or label is specified, it will return the IDs for all clusters.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the text embeddings and their respective IDs

    """
    embed_out = self.embeddings
    select_ids = np.arange(0, embed_out.shape[0])
    idx = None
    is_child = False
    name_array = np.array([])

    if name is not None:
        try:
            idx = self.parent_names.index(name)
        except:
            try:
                idx = self.child_names.index(name)
                is_child = True
            except:
                raise ValueError('Name not found in parent and child.')
        if is_child:
            name_array = self.child_labels[:, idx]
        else:
            name_array = self.parent_labels[:, idx]

        name_mask = (name_array >= 0)
        embed_out = embed_out[name_mask, :]
        select_ids = select_ids[name_mask]
        name_array = name_array[name_mask]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, either a parent or a child name must be specified.')
        label_mask = (name_array == label)
        embed_out = embed_out[label_mask, :]
        select_ids = select_ids[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)
        embed_out = embed_out[id_mask, :]
        select_ids = select_ids[id_mask]


    return embed_out, select_ids

`get_reduce_embeddings(name=None, label=None, get_ids=None)`

Get the dimension reduction embeddings by parent or child name, label and index ID.

If neither is specified, all input is returned.

Parameters:

name (str, default: None ) –

Name of parent or child from parent_names or child_names.
label (int, default: None ) –

Label from the parent or child where name must be specified.

If label is specified without name, this method will throw an error.
get_ids (array - like, default: array-like ) –

Give specific index IDs to be extracted from the input.

If no name or label is specified, it will return the IDs for all clusters.

Returns:	`(ndarray, ndarray)` – Returns an array of the dimension reduction embeddings and their respective IDs

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_reduce_embeddings(self, name=None, label=None, get_ids=None):
    """
    Get the dimension reduction embeddings by parent or child name, label and index ID.

    If neither is specified, all input is returned.

    Parameters
    ----------
    name : str
        Name of parent or child from parent_names or child_names.

    label : int
        Label from the parent or child where name must be specified.

        If label is specified without name, this method will throw an error.

    get_ids : {array-like} of shape (n_samples,)
        Give specific index IDs to be extracted from the input.

        If no name or label is specified, it will return the IDs for all clusters.

    Returns
    -------
    ndarray, ndarray
        Returns an array of the dimension reduction embeddings and their respective IDs

    """
    embed_out = self.reduce_embeddings
    select_ids = np.arange(0, embed_out.shape[1])
    idx = None
    is_child = False
    name_array = np.array([])
    embed_array = embed_out

    if name is not None:
        try:
            idx = self.parent_names.index(name)
        except:
            try:
                idx = self.child_names.index(name)
                is_child = True
            except:
                raise ValueError('Name not found in parent and child.')
        if is_child:
            name_array = self.child_labels[:, idx]
            embed_array = self.reduce_embeddings[idx+1]
        else:
            name_array = self.parent_labels[:, idx]
            embed_array = self.reduce_embeddings[idx]

        name_mask = (name_array >= 0)

        embed_out = embed_array[name_mask, :]
        select_ids = select_ids[name_mask]
        name_array = name_array[name_mask]

    if label is not None:
        if idx is None:
            raise ValueError('When label is specified, either a parent or a child name must be specified.')
        label_mask = (name_array == label)
        embed_out = embed_out[label_mask, :]
        select_ids = select_ids[label_mask]

    if get_ids is not None:
        id_mask = np.isin(select_ids, get_ids)

        if idx:
            embed_out = embed_out[id_mask, :]
        else:
            logger.warning('No parent or child was specified, this returns the reduction embeddings at the specified IDs for all iterations '+\
                        'with dimensions (iteration, index, xy). Some values may be Nan.')
            embed_out = embed_out[:, id_mask, :]

        select_ids = select_ids[id_mask]

    return embed_out, select_ids

`get_pandas_data()`

Get all data as a Pandas dataframe.

Returns:	`DataFrame` – A Pandas dataframe object.

Source code in src\exhaustive_hdbscan\clusterfeatures.py

def get_pandas_data(self):
    """
    Get all data as a Pandas dataframe.

    Returns
    -------
    DataFrame
        A Pandas dataframe object.
    """
    return self.pandas_data

Utilities

Reducer

Dimension reduction method passed to EHDBSCAN must subclass this Reducer class.

Source code in src\exhaustive_hdbscan\utilities.py

def __init__(self):
    """
    Dimension reduction method passed to EHDBSCAN must subclass this Reducer class.
    """
    pass

`fit(X)`

All subclasses must implement this method.

Parameters:	`X` (`array - like`, default: `array-like` ) – Fit the input embeddings on the reduction method. Expects encoded embeddings input with numeric dtype.

Returns:	`self`( `object` ) – Expects return self object, but not required.

Source code in src\exhaustive_hdbscan\utilities.py

def fit(self, X):
    """
    All subclasses must implement this method.

    Parameters
    ----------
    X : {array-like} of shape (n_samples, n_features)
        Fit the input embeddings on the reduction method.

        Expects encoded embeddings input with numeric dtype.

    Returns
    -------
    self : object
        Expects return self object, but not required.

    """

    raise NotImplementedError('Subclass must implement the fit method.')

`transform(X)`

All subclasses must implement this method and return an array-like object of shape (n_samples, n_features).

EHDBSCAN will call fit and transform separately. A fit_transform may be written for personal use but is not required.

Parameters:	`X` (`array - like`, default: `array-like` ) – Transform the encoded embeddings with the fitted reduction method. Expects encoded embeddings input with numeric dtype.

Returns:	`{array-like} of shape (n_samples, n_reduced_features)` – Method must return an array-like object that will be used for clustering. These embeddings will be stored in ClusterFeatures for approx_predict of EHDBSCAN.

Source code in src\exhaustive_hdbscan\utilities.py

def transform(self, X):
    """
    All subclasses must implement this method and return an array-like object
    of shape (n_samples, n_features).

    EHDBSCAN will call fit and transform separately. A fit_transform may be
    written for personal use but is not required.

    Parameters
    ---------
    X : {array-like} of shape (n_samples, n_features)
        Transform the encoded embeddings with the fitted reduction method.

        Expects encoded embeddings input with numeric dtype.

    Returns
    -------
    {array-like} of shape (n_samples, n_reduced_features)
        Method must return an array-like object that will be used for clustering.

        These embeddings will be stored in ClusterFeatures for approx_predict of
        EHDBSCAN.

    """

    raise NotImplementedError('Subclass must implement the transform method.')

Encoder

Encoder passed to the EHDBSCAN must subclass this Encoder class.

Source code in src\exhaustive_hdbscan\utilities.py

def __init__(self, encoder):
    """
    Encoder passed to the EHDBSCAN must subclass this Encoder class.
    """
    pass

`encode(X)`

All subclasses must implement the encode method.

Parameters:	`X` (`array - like`, default: `array-like` ) – Text input acceptable by EHDBSCAN.

Returns:	`{array-like} of shape (n_samples, n_features)` – Must output array of embeddings of specified shape.

Source code in src\exhaustive_hdbscan\utilities.py

def encode(self, X):
    """
    All subclasses must implement the encode method.

    Parameters
    ----------
    X : {array-like} of shape (n_samples,)
        Text input acceptable by EHDBSCAN.

    Returns
    -------
    {array-like} of shape (n_samples, n_features)
        Must output array of embeddings of specified shape.
    """

    raise NotImplementedError('Subclass must implement the encode method and return an array-like of shape (n_samples, n_features).')

ClusterVectorOps

This performs some basic vector operations that are useful for interpreting EHDBSCAN outputs.

They are just raw vector operations. However it can be combined with LabelGeneratorConfig to generate labels for drawing the tree of clusters.

Parameters:

cluster_features (ClusterFeatures) –

Input instance of ClusterFeatures that has been fit.
metric (str, default: 'cosine' ) –

The global metric for all operations. Metric may also be defined for individual operations.

Metric is expected to be the same expected by sci-kit DitanceMetric
use_embedding (bool, default: False ) –

Whether to use the input text embeddings stored in the ClusterFeatures. Otherwise it will use the pairwise distances stored in the ClusterFeatures.

Generally, using embeddings is expected to yield better results.

Returns:	`self`( `object` ) – Returns self.

Source code in src\exhaustive_hdbscan\clustervectorops.py

def __init__(self,
            cluster_features: ClusterFeatures,
            metric: str = 'cosine',
            use_embedding: bool = False):
    self.cf = cluster_features
    self.metric = validate_metric(metric)        
    self.use_embedding = use_embedding

`cluster_centroid_neighbors(top_n=3, name=None, label=None, metric=None)`

This will generate the neighbors for a parent or child cluster centroid, where the centroid is taken as the median.

Parameters:

top_n (int, default: 3 ) –

Top N neighbors to return
name (str, default: None ) –

Parent or child name as stored in the ClusterFeatures instance.
label (int, default: None ) –

Parent or child label from the available labels in the ClusterFeatures instance.
metric (str, default: None ) –

The metric to be used for centroid neighbors. It will override the global metric specification.

Expects the same input as accepted by sci-kit DistanceMetric.

Returns:	`IDs of centroid neighbors {array-like} of shape (n_samples,)` – These IDs are aligned with the input to the EHDBSCAN and stored in the ClusterFeatures instance.

Source code in src\exhaustive_hdbscan\clustervectorops.py

def cluster_centroid_neighbors(self, top_n: int = 3,
                               name: str = None,
                               label: int = None,
                               metric: str = None):
    """
    This will generate the neighbors for a parent or child cluster centroid, where the centroid
    is taken as the median.

    Parameters
    ----------
    top_n : int
        Top N neighbors to return

    name : str
        Parent or child name as stored in the ClusterFeatures instance.

    label : int
        Parent or child label from the available labels in the ClusterFeatures instance.

    metric : str
        The metric to be used for centroid neighbors. It will override the global metric specification.

        Expects the same input as accepted by sci-kit DistanceMetric.

    Returns
    -------
    IDs of centroid neighbors {array-like} of shape (n_samples,)
        These IDs are aligned with the input to the EHDBSCAN and stored in the ClusterFeatures instance.
    """
    use_metric = self.metric
    if metric:
        use_metric = metric

    use_metric = validate_metric(use_metric)

    X, ids = self.cf.get_distances(name=name, label=label)
    if self.use_embedding:
        X, ids = self.cf.get_embeddings(name=name, label=label)
        if len(X) <= 0:
            raise ValueError('The ClusterFeatures instance provided has no stored embeddings.')

    X_median = np.median(X, axis=0)
    X_dst = X_median
    if self.use_embedding:
        X_dst = pairwise_distances(X_median.reshape(1, -1), X, metric=use_metric)

    X_dst = X_dst.ravel()
    mask = (X_dst != 0)
    ids = ids[mask]
    X_dst = X_dst[mask]

    args = np.argsort(X_dst)[:top_n]

    if top_n < 0:
        args = np.argsort(X_dst)[top_n:]

    ids = ids[args]
    return ids

`parent_child_transition(*, parent, child, top_n=3, metric=None)`

In EHDBSCAN, parent-child links are based on core distance-type metric that relies on cluster centroids. Therefore, it is valuable to find out what was the common topical link between the two clusters. It helps interpreting the cluster tree.

Similarly it is also helpful to see, which topics are unique to the parent and the child. In experiments, it has been observed that topics unique to the child can be seen as the topical drift subtracted from the parent (parent_median - child_median) and vice versa.

For instance, if the parent cluster's central theme is Policy and Law and child cluster's theme is Malicious practices in the Law profession, the common link between the two is Law & Legal. The topical drift is Ethics, this is the difference unique to the child and will be reflected in the child IDs returned by this method.

This method performs these operations to return IDs unique to the parent and child and IDs common among the two.

It works with embeddings and distances, but it is highly recommended to use embeddings. The only distance estimations are simplification and not entirely representative. A lot of accuracy is lost in trying to work back differences in absolute positions with relative position metrics like distances.

Parameters:

parent (tuple) –

A tuple of parent name and one of its cluster label e.g. ('parent_0', 2)
child (tuple) –

A tuple of child name and one of its cluster label e.g. ('child_0', 3)
top_n (int, default: 3 ) –

Top N IDs of differences and commonalities to return.
metric (str, default: None ) –

The metric to be used for distance estimations. It will override the global metric.

Expects the same input as accepted by sci-kit DistanceMetric.

Returns:	`IDs for Parent Difference, Child Difference, Parent-Child Commmon {array-like} of shape (n_samples,)` – IDs are returned in the sequence of the input stored in the ClusterFeats instance.

Source code in src\exhaustive_hdbscan\clustervectorops.py

def parent_child_transition(self, *, parent: tuple, child: tuple, top_n: int = 3, metric: str = None):
    """
    In EHDBSCAN, parent-child links are based on core distance-type metric that relies on cluster centroids.
    Therefore, it is valuable to find out what was the common topical link between the two clusters. It helps
    interpreting the cluster tree.

    Similarly it is also helpful to see, which topics are unique to the parent and the child. In experiments,
    it has been observed that topics unique to the child can be seen as the topical drift subtracted from the
    parent (parent_median - child_median) and vice versa.

    For instance, if the parent cluster's central theme is Policy and Law and child cluster's theme is
    Malicious practices in the Law profession, the common link between the two is Law & Legal. The topical drift is
    Ethics, this is the difference unique to the child and will be reflected in the child IDs returned by this
    method.

    This method performs these operations to return IDs unique to the parent and child and IDs common
    among the two.

    It works with embeddings and distances, but it is highly recommended to use embeddings. The only distance
    estimations are simplification and not entirely representative. A lot of accuracy is lost in trying to work
    back differences in absolute positions with relative position metrics like distances.

    Parameters
    ----------
    parent : tuple
        A tuple of parent name and one of its cluster label e.g. ('parent_0', 2)

    child : tuple
        A tuple of child name and one of its cluster label e.g. ('child_0', 3)

    top_n : int
        Top N IDs of differences and commonalities to return.

    metric : str
        The metric to be used for distance estimations. It will override the global metric.

        Expects the same input as accepted by sci-kit DistanceMetric.

    Returns
    -------
    IDs for Parent Difference, Child Difference, Parent-Child Commmon {array-like} of shape (n_samples,)
        IDs are returned in the sequence of the input stored in the ClusterFeats instance.

    """
    use_metric = self.metric
    if metric:
        use_metric = metric

    use_metric = validate_metric(use_metric)

    if not isinstance(parent, tuple):
        raise ValueError('Expects a tuple of size 2 with name at first index and label at second.')

    if not isinstance(child, tuple):
        raise ValueError('Expects a tuple of size 2 with name at first index and label at second.')

    p_ids = []
    c_ids = []
    common_ids = []

    if self.use_embedding:
        X_chk, _ = self.cf.get_embeddings()
        if len(X_chk) <= 0:
            raise ValueError('The ClusterFeatures instance provided has no stored embeddings.')
        X_parent, p_ids = self.cf.get_embeddings(name=parent[0], label=parent[1])
        X_child, c_ids = self.cf.get_embeddings(name=child[0], label=child[1])
        all_embed = np.concatenate((X_parent, X_child), axis=0)
        all_ids = np.concatenate((p_ids, c_ids), axis=0)

        X_p_median = np.median(X_parent, axis=0)
        X_c_median = np.median(X_child, axis=0)
        X_c_diff = X_p_median - X_c_median
        X_p_diff = X_c_median - X_p_median
        X_common = X_p_diff + X_c_diff

        X_p_dst = pairwise_distances(X_p_diff.reshape(1, -1), X_parent, metric=use_metric)
        X_p_dst = np.where(X_p_dst != 0., X_p_dst, np.inf)
        X_c_dst = pairwise_distances(X_c_diff.reshape(1, -1), X_child, metric=use_metric)
        X_c_dst = np.where(X_c_dst != 0., X_c_dst, np.inf)
        X_cm_dst = pairwise_distances(X_common.reshape(1, -1), all_embed, metric=use_metric)
        X_cm_dst = np.where(X_cm_dst != 0., X_cm_dst, np.inf)

        p_args = np.argsort(X_p_dst.ravel())[:top_n]
        c_args = np.argsort(X_c_dst.ravel())[:top_n]
        cm_args = np.argsort(X_cm_dst.ravel())[:top_n]

        p_ids = p_ids[p_args]
        c_ids = c_ids[c_args]
        common_ids = all_ids[cm_args]
    else:
        X_parent, p_ids = self.cf.get_distances(name=parent[0], label=parent[1])
        X_child, c_ids = self.cf.get_distances(name=child[0], label=child[1])

        X_all, all_ids = self.cf.get_distances(row_ids=p_ids, col_ids=c_ids)
        X_dst = np.where(X_all != 0., X_all, np.inf)
        X_cm_args = np.argsort(X_dst.ravel())[:top_n]
        common_ids = all_ids.ravel()[X_cm_args]

        p_args = np.argsort(np.median(X_parent, axis=0))
        p_ids = p_ids[p_args][:top_n]

        c_args = np.argsort(np.median(X_child, axis=0))
        c_ids = c_ids[c_args][:top_n]


    return p_ids, c_ids, common_ids

Label Generation

LabelGeneratorConfig

The base class for generating labels compatible with EHDBSCAN.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def __init__(self):
    pass

`compute(cluster_features)`

All subclasses must implement this method.

It expects an instance of ClusterFeatures and returns a list of labels in the same sequence as the clustering operation.

Cluster Iteration > Each Cluster in ascending sequence [0, 1, 2...].

Returned list must be single level, not nested. This is must for compatibility with EHDBSCAN.

See examples for implementation.

Parameters:	`cluster_features` (`ClusterFeatures`) – Instance of ClusterFeatures. This is must for compatibility with EHDBSCAN.

Returns:	`output`( `list` ) – Returns a single level list of generated labels for cluster visualization or other downstream tasks. Labels must be in the sequence of the clustering procedure.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def compute(self, cluster_features: ClusterFeatures):

    """
    All subclasses must implement this method.

    It expects an instance of ClusterFeatures and returns a list of
    labels in the same sequence as the clustering operation. 

    Cluster Iteration > Each Cluster in ascending sequence [0, 1, 2...].

    Returned list must be single level, not nested. This is must for
    compatibility with EHDBSCAN.

    See examples for implementation.

    Parameters
    ----------
    cluster_features : ClusterFeatures
        Instance of ClusterFeatures. This is must for compatibility with EHDBSCAN.

    Returns
    -------
    output : list
        Returns a single level list of generated labels for cluster visualization or other
        downstream tasks. Labels must be in the sequence of the clustering procedure.
    """


    raise NotImplementedError('Subclass must implement compute method.'+\
                              'It must take ClusterFeatures as input and return a list of labels in the sequence cluster iteration, cluster label...')

MMR

Bases: LabelGeneratorConfig

Maximal Marginal Relevance implementation subclassed to LabelGeneratorConfig. It filters the labels for relevance and diversity.

This is a generalized standalone implementation and will work without EHDBSCAN inputs. But the compute method will only work with an instance of ClusterFeatures. Fit and transform may still be called with raw inputs.

Parameters:

lda (float, default: 0.5 ) –

lambda value for MMR. Ranges from 0 to 1.

0 = Only diversity; 1 = Only relevance.
encoder (Encoder {Optional}, default: None ) –

Expects a subclass of Encoder.

If no encoder is passed, it expects embeddings as input. In this case, labels are index of the passed embeddings.
metric (str, default: 'cosine' ) –

Metric to be used for similarity. Expects metric acceptable by sci-kit DistanceMetric.
top_n (int, default: 3 ) –

Top N similarity labels to be output.

Attributes:	`label_sim` (`{array-like} of shape (n_samples, n_samples)`) – Pairwise similarity matrix of input. `label_embed` (`{array-like} of shape (n_samples, n_features)`) – Label embeddings. Only available if Encoder is passed.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def __init__(self, *,
             lda: float = 0.5,
             encoder: Encoder = None,
             metric: str = 'cosine',
            top_n: int = 3):
    self.lda = lda
    self.encoder = encoder
    self.top_n = top_n
    self.metric = metric

`fit(X)`

Creates the similarity matrix from inputs. Expects text input or numeric embeddings.

X : {array-like} of shape (n_samples,) for text OR (n_samples, n_features) for embeddings Expects text or numeric embeddings. If text is provided, an Encoder must also be specified.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def fit(self, X):
    """
    Creates the similarity matrix from inputs. Expects text input or
    numeric embeddings.

    X : {array-like} of shape (n_samples,) for text OR (n_samples, n_features) for embeddings
        Expects text or numeric embeddings. If text is provided, an Encoder must also be specified.

    """
    X, is_embedding = validate_input(X)

    if is_embedding:
        self.labels = np.arange(0, X.shape[0])
        self.label_embed = X
    else:
        if self.encoder:
            self.labels = X
            self.label_embed = self.encoder.encode(X)
        else:
            raise ValueError('Must specify an Encoder when text input is passed.')

    self.label_sim = pairwise_distances(self.label_embed, metric=self.metric)

`transform(y)`

Get the MMR labels.

y : {array-like} of shape (n_sample,) for text OR (n_sample, n_features) for embeddings Expects text or numeric embeddings. If text is provided, an Encoder must also be specified. It expects exactly one entry.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def transform(self, y):
    #TODO transform based only on pairwise distances
    """
    Get the MMR labels.

    y : {array-like} of shape (n_sample,) for text OR (n_sample, n_features) for embeddings
        Expects text or numeric embeddings. If text is provided, an Encoder must also be specified.
        It expects exactly one entry.

    """
    label_idx = []
    y, is_embedding = validate_input(y)
    y_embed = y
    if is_embedding == False:
        if self.encoder:
            y_embed = self.encoder.encode(y)
        else:
            raise ValueError('Must specify an Encoder when text input is passed.')

    if len(y_embed.shape) < 2:
        y_embed = y_embed.reshape(1, -1)

    if y_embed.shape[0] > 1:
        raise ValueError('Expects only single sample input. Input array must be of shape (1,...).')

    target_sim = pairwise_distances(y_embed, self.label_embed, metric=self.metric)
    idx = np.argmin(target_sim, axis=1)
    label_idx.append(idx[0])

    remainder = np.arange(0, self.label_embed.shape[0])
    remainder_mask = np.isin(remainder, label_idx, invert=True)
    remainder = remainder[remainder_mask]

    for i in range(self.top_n):
        t_sim = target_sim[0, remainder]
        d_sim = self.label_sim[np.ix_(label_idx, remainder)]
        d_sim = np.min(d_sim, axis=0)
        mmr_scores = (self.lda * t_sim) - ((1 - self.lda) * d_sim)
        top_score = np.argsort(mmr_scores.ravel())[0]
        label_idx.append(remainder[top_score])
        remainder_mask = np.isin(remainder, label_idx, invert=True)
        remainder = remainder[remainder_mask]

    sorted_labels = self.labels[label_idx]
    return sorted_labels

`compute(cluster_feats)`

Performs the fit and transforms the input labels as MMR labels and outputs list of labels compatible with EHDBSCAN. It expects a ClusterFeatures input.

For ranking, the entire cluster text is clubbed and encoded to compare against individual items in the cluster. If an encoder is provided, the combined cluster text is encoded with this encoder. If an encoder is not provided, the average embeddings of individual cluster items is used.

If the ClusterFeatures instance provided has no embeddings, an encoder must be provided.

Parameters:	`cluster_feats` (`ClusterFeatures`) – An instance of ClusterFeatures.

Returns:	`MMR Ranked Labels : list` – A single level list of MMR ranked labels for each cluster in sequence iteration > cluster.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def compute(self, cluster_feats: ClusterFeatures):
    """

    Performs the fit and transforms the input labels as MMR labels and outputs
    list of labels compatible with EHDBSCAN. It expects a ClusterFeatures input.

    For ranking, the entire cluster text is clubbed and encoded to compare against
    individual items in the cluster. If an encoder is provided, the combined cluster
    text is encoded with this encoder. If an encoder is not provided, the average
    embeddings of individual cluster items is used.

    If the ClusterFeatures instance provided has no embeddings, an encoder must be
    provided.

    Parameters
    ----------
    cluster_feats : ClusterFeatures
        An instance of ClusterFeatures.

    Returns
    -------
    MMR Ranked Labels : list
        A single level list of MMR ranked labels for each cluster in sequence
        iteration > cluster.

    """

    ot, has_embedding = self._preprocess(cluster_feats)
    X = []
    y = []
    all_text = [text for x in ot for text in x['texts']]
    y = [x['class_text'] for x in ot]

    if has_embedding:
        X = np.concatenate([item['embeddings'] for item in ot], axis=0)
        self.fit(X)
        if self.encoder is None:
            y = [np.mean(x['embeddings'], axis=0) for x in ot]
    else:
        if self.encoder:
            self.fit(all_text)
        else:
            raise ValueError('Neither the ClusterFeatures had embeddings, nor an Encoder was provided with MMR. One of the two must be available.')
    labels = []
    for item in y:
        label = self.transform(np.array(item))
        labels.append(' '.join([all_text[x] for x in label]))

    return labels            

ClassTfidf

Bases: LabelGeneratorConfig

Class Term Frequency Inverse Document Frequency implementation subclassed to LabelGeneratorConfig. It assigns descriptive labels for each cluster.

This is a generalized standalone implementation and will work without EHDBSCAN inputs. But the compute method will only work with an instance of ClusterFeatures. Fit and transform may still be called with raw inputs.

It uses the sci-kit CountVectorizer and some of those parameters are exposed here.

Parameters:

top_n (int, default: 3 ) –

Top N labels to output.
min_df (float, default: 0.2 ) –

Minimum frequency threshold to include terms. Same as sci-kit.
max_df (float, default: 0.5 ) –

Maximum frequency threshold to include terms. Same as sci-kit.
ngram_range (tuple, default: (1, 2) ) –

Range of ngrams to include. Same as sci-kit.
stop_words ((str, list), default: 'english' ) –

List of words to exclude. Same as sci-kit.
token_pattern (str, default: '(?u)\\b[a-zA-Z]{4,}\\b' ) –

Regex pattern for including words. Same as sci-kit.
lowercase (bool, default: True ) –

Converts input to lowercase. Same as sci-kit.
max_features (int, default: None ) –

Max length of features to retain. Same as sci-kit.
mmr (MMR, default: None ) –

An instance of MMR.

This will apply on the class labels after the ClassTfidf procedure has run.

This will only work with the compute method. Simple fit and transform will not execute an MMR ranking.

Attributes:	`cW` (`scipy sparse matrix of shape (n_samples, n_features)`) – The class weights generated after the fit method is called. `self.features` (`{array-like} of shape (n_features,)`) – The vectorized vocabulary. Only callable after fit method is called.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def __init__(self, *, 
             top_n: int = 3, 
             min_df: float = 0.20, 
             max_df: float = 0.50, 
             ngram_range: tuple = (1, 2),
            stop_words: Any = 'english',
             token_pattern: str = r'(?u)\b[a-zA-Z]{4,}\b',
             lowercase: bool = True,
             max_features: int = None,
             mmr: MMR = None):
    super().__init__()
    self.top_n = top_n
    self.mmr = mmr
    self.min_df = min_df
    self.max_df = max_df
    self.ngram_range = ngram_range
    self.stop_words = stop_words
    self.token_pattern = token_pattern
    self.isfit = False
    self.lowercase = lowercase
    self.max_features = max_features

`fit(X)`

Fits a corpus of docs by vectorizing ngrams and generating normalized TF-IDF vectors for each doc.

It also caches the fit vectorizer and the Class Weights matrix for transform.

Unlike regular TF-IDF, the Class TF-IDF applies over a class of documents weighting the regular TF-IDF by a class-relative weighting formula.

This is a suitable TF-IDF labelling scheme for clusters, where each cluster is treated as a class. Labels are then generated for each cluster relative to all other clusters.

Parameters:	`X` (`array - like`, default: `array-like` ) – Only text input accepted.

Returns:	`TF-IDF {array-like} of shape (n_samples, n_features)` – The array matrix of Class TF-IDF normalized scores of each document.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def fit(self, X):
    """
    Fits a corpus of docs by vectorizing ngrams and generating normalized TF-IDF vectors
    for each doc.

    It also caches the fit vectorizer and the Class Weights matrix for transform.

    Unlike regular TF-IDF, the Class TF-IDF applies over a class of documents weighting the
    regular TF-IDF by a class-relative weighting formula.

    This is a suitable TF-IDF labelling scheme for clusters, where each cluster is treated as
    a class. Labels are then generated for each cluster relative to all other clusters.

    Parameters
    ----------
    X : {array-like} of shape (n_samples,)
        Only text input accepted.

    Returns
    ----------
    TF-IDF {array-like} of shape (n_samples, n_features)
        The array matrix of Class TF-IDF normalized scores of each document.

    """
    X, _ = validate_input(X, classtfidf=True)

    self.vectorizer = CountVectorizer(stop_words=self.stop_words, token_pattern=self.token_pattern,
                                 ngram_range=self.ngram_range, min_df=self.min_df, max_df=self.max_df,
                                      lowercase=self.lowercase,
                                      max_features=self.max_features)
    self.vectorizer.fit(X)
    X_transform = self.vectorizer.transform(X)
    self.features = self.vectorizer.get_feature_names_out()

    X_norm = normalize(X_transform, norm='l1', axis=1)
    A = np.mean(np.sum(X_transform, axis=1))
    f = np.sum(X_transform.toarray(), axis=0)
    self.cW = sp.diags(np.log(1 + A / f))
    cTf = X_norm @ self.cW
    self.isfit = True
    return cTf

`transform(y)`

Transform unseen data to current Class TF-IDF. It vectorizes the input to the current fit vocabulary and applies the Class TF-IDF weights to obtain a vector of TF-IDF for the current input.

This method does not assign new data to classes by design. That is a downstream task and really upto task and user preferences. Class assignment may be executed on similarity between TF-IDF vectors, label distances or even label embedding vector operations. Class assignment may not require doing this transform at all.

Parameters:	`y` (`array - like`, default: `array-like` ) – Input text array for transforming along the current Class TF-IDF structure.

Returns:	`TF-IDF {array-like} of shape (n_samples, n_features)` –

Source code in src\exhaustive_hdbscan\labelgenerator.py

def transform(self, y):
    """
    Transform unseen data to current Class TF-IDF. It vectorizes the input to the current
    fit vocabulary and applies the Class TF-IDF weights to obtain a vector of TF-IDF for the
    current input.

    This method does not assign new data to classes by design. That is a downstream task and really upto task
    and user preferences. Class assignment may be executed on similarity between TF-IDF vectors, label
    distances or even label embedding vector operations. Class assignment may not require doing this transform
    at all.

    Parameters
    ----------
    y : {array-like} of shape (n_samples,)
        Input text array for transforming along the current Class TF-IDF structure.

    Returns
    -------
    TF-IDF {array-like} of shape (n_samples, n_features)

    """

    if self.isfit == False:
        raise ValueError('Must call fit first.')
    y, _ = validate_input(y, classtfidf=True)
    y_ct = self.vectorizer.transform(y)
    y_norm = normalize(y_ct, norm='l1', axis=1)
    yctf = y_norm @ self.cW

    return yctf

`compute(cluster_feats)`

Extracts data from ClusterFeatures, packs them into classes by their cluster and generates Class TF-IDF labels.

If MMR is passed, it will perform an MMR ranking after class labels have been generated.

Parameters:	`cluster_feats` (`ClusterFeatures`) – An instance of ClusterFeatures.

Returns:	`Class Labels : list` – A single level list of top_n class labels for each cluster of each iteration in sequence iteration > cluster.

Source code in src\exhaustive_hdbscan\labelgenerator.py

def compute(self, cluster_feats: ClusterFeatures):
    """
    Extracts data from ClusterFeatures, packs them into classes by their cluster and generates
    Class TF-IDF labels.

    If MMR is passed, it will perform an MMR ranking after class labels have been generated.

    Parameters
    ----------
    cluster_feats : ClusterFeatures
        An instance of ClusterFeatures.

    Returns
    -------
    Class Labels : list
        A single level list of top_n class labels for each cluster of each iteration in sequence
        iteration > cluster. 
    """
    class_texts = self._preprocess(cluster_feats)
    texts = [x['text'] for x in class_texts]

    cTf = self.fit(texts)

    ctf_labels = []
    ot_labels = []

    for i, score in enumerate(cTf):
        score_arr = score.toarray().ravel()
        arr = np.argsort(score_arr)[::-1]
        labels = self.features[arr]
        score_arr = score_arr[arr]
        mask = (score_arr > 0)
        labels = labels[mask]
        ctf_labels.append(labels)

    if self.mmr:
        self.mmr.fit(self.features)
        for l in ctf_labels:
            lb = [' '.join(l)]
            ot_labels.append(' '.join(self.mmr.transform(lb)))
    else:
        ot_labels = [' '.join(x[:self.top_n]) for x in ctf_labels]

    return ot_labels