Skip to content

dqm_ml_core.metrics.diversity

Diversity metric processor for evaluating categorical data diversity.

This module contains the DiversityProcessor class that computes diversity indices (Simpson, Gini-Simpson, Shannon Entropy, Richness) for categorical or discretized data columns.

logger = logging.getLogger(__name__) module-attribute

DiversityProcessor

Bases: MetricsProcessor

Computes diversity indices for categorical data columns.

Simpson Index (1 - Σn(n-1) / N(N-1)): Unbiased estimator — probability two random samples belong to different categories. Ranges [0, 1]; high = diverse.

Gini-Simpson Index (1 - Σp²): Gini impurity — probability of incorrect classification. Ranges [0, 1]; high = diverse.

Shannon Entropy (-Σp·log(p)): Information content / uncertainty. Lower bound 0; no upper bound (grows with category count and evenness).

Richness

Simple count of unique categories.

The processor uses a streaming architecture: - Batch level: Computes value counts per column. - Dataset level: Merges all batch counts and computes final indices.

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
class DiversityProcessor(MetricsProcessor):
    """Computes diversity indices for categorical data columns.

    Simpson Index (1 - Σn(n-1) / N(N-1)):
        Unbiased estimator — probability two random samples belong to
        different categories. Ranges [0, 1]; high = diverse.

    Gini-Simpson Index (1 - Σp²):
        Gini impurity — probability of incorrect classification.
        Ranges [0, 1]; high = diverse.

    Shannon Entropy (-Σp·log(p)):
        Information content / uncertainty. Lower bound 0; no upper bound
        (grows with category count and evenness).

    Richness:
        Simple count of unique categories.

    The processor uses a streaming architecture:
    - Batch level: Computes value counts per column.
    - Dataset level: Merges all batch counts and computes final indices.
    """

    SUPPORTED_METRICS = {"simpson", "gini", "shannon", "richness"}

    def __init__(self, name: str = "diversity", config: dict[str, Any] | None = None) -> None:
        """Initialize the diversity processor.

        Args:
            name: Name of the processor.
            config: Configuration dictionary containing:
                - input_columns: List of columns to analyze.
                - metrics: List of metrics to compute (default: all supported).
        """
        super().__init__(name, config)

        cfg = DiversityProcessorConfig.model_validate({**self.config, "name": self.name})

        self.metrics: list[str] = list(cfg.metrics)

    @override
    def generated_metrics(self) -> list[str]:
        """Return the list of metric columns that will be generated.

        Returns:
            List of output metric column names.
        """
        metrics = []
        for col in self.input_columns:
            if "simpson" in self.metrics:
                metrics.append(f"{col}_simpson")
            if "gini" in self.metrics:
                metrics.append(f"{col}_gini")
            if "shannon" in self.metrics:
                metrics.append(f"{col}_shannon")
            if "richness" in self.metrics:
                metrics.append(f"{col}_richness")
        return metrics

    @override
    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """Compute per-batch value counts for streaming aggregation.

        Args:
            features: Dictionary of column arrays from this batch.

        Returns:
            Dictionary of batch-level value-count pairs.
        """
        batch_metrics: dict[str, pa.Array] = {}

        for col in self.input_columns:
            if col not in features:
                logger.warning("[%s] column '%s' not found in batch", self.name, col)
                continue

            col_array = features[col]
            vc = pa.compute.value_counts(col_array)
            if len(vc) == 0:
                continue

            values = vc.field("values")
            counts = vc.field("counts")

            values_str = pa.compute.cast(values, pa.string())
            batch_metrics[f"{col}_values"] = values_str
            batch_metrics[f"{col}_counts"] = counts

        return batch_metrics

    @override
    def compute(self, batch_metrics: dict[str, pa.Array] | None = None) -> dict[str, Any]:
        """Compute final dataset-level diversity indices.

        Merges all batch-level value counts and computes diversity
        indices for each column.

        Args:
            batch_metrics: Dictionary of batch-level value-count arrays.

        Returns:
            Dictionary containing final diversity scores.
        """
        if not batch_metrics:
            return {"_metadata": {"error": "No batch metrics provided"}}

        results: dict[str, Any] = {}
        for col in self.input_columns:
            col_res = self._compute_column_diversity(col, batch_metrics)
            results.update(col_res)
        return results

    def _compute_column_diversity(self, col: str, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:
        """Compute diversity indices for a single column.

        Args:
            col: Column name to compute diversity for.
            batch_metrics: Dictionary of batch-level value-count arrays.

        Returns:
            Dictionary mapping column-specific metric names to values.
        """
        values_key = f"{col}_values"
        counts_key = f"{col}_counts"

        if values_key not in batch_metrics or counts_key not in batch_metrics:
            logger.warning("[%s] no batch metrics for column '%s'", self.name, col)
            return {}

        values_arr = batch_metrics[values_key]
        counts_arr = batch_metrics[counts_key]

        if len(values_arr) == 0:
            return {}

        counter: Counter[str] = Counter()
        total = 0
        for i in range(len(values_arr)):
            val = values_arr[i].as_py()
            cnt = int(counts_arr[i])
            counter[val] += cnt
            total += cnt

        if total == 0:
            return {}

        col_res: dict[str, Any] = {}

        if "richness" in self.metrics:
            col_res["richness"] = len(counter)

        if "shannon" in self.metrics:
            probs = np.array(list(counter.values()), dtype=np.float64) / total
            probs = probs[probs > 0]
            entropy = float(-(probs * np.log(probs)).sum())
            col_res["shannon"] = entropy

        if "gini" in self.metrics:
            probs = np.array(list(counter.values()), dtype=np.float64) / total
            gini = float(1.0 - (probs**2).sum())
            col_res["gini"] = gini

        if "simpson" in self.metrics:
            counts_np = np.array(list(counter.values()), dtype=np.float64)
            simpson = float(1.0 - (counts_np * (counts_np - 1)).sum() / (total * (total - 1))) if total > 1 else 0.0
            col_res["simpson"] = simpson

        return {f"{col}_{metric_name}": value for metric_name, value in col_res.items()}

    @override
    def reset(self) -> None:
        """Reset processor state for new processing run."""

SUPPORTED_METRICS = {'simpson', 'gini', 'shannon', 'richness'} class-attribute instance-attribute

metrics: list[str] = list(cfg.metrics) instance-attribute

__init__(name: str = 'diversity', config: dict[str, Any] | None = None) -> None

Initialize the diversity processor.

Parameters:

Name Type Description Default
name str

Name of the processor.

 'diversity'
config dict[str, Any] | None

Configuration dictionary containing: - input_columns: List of columns to analyze. - metrics: List of metrics to compute (default: all supported).

 None
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(self, name: str = "diversity", config: dict[str, Any] | None = None) -> None:
    """Initialize the diversity processor.

    Args:
        name: Name of the processor.
        config: Configuration dictionary containing:
            - input_columns: List of columns to analyze.
            - metrics: List of metrics to compute (default: all supported).
    """
    super().__init__(name, config)

    cfg = DiversityProcessorConfig.model_validate({**self.config, "name": self.name})

    self.metrics: list[str] = list(cfg.metrics)

compute(batch_metrics: dict[str, pa.Array] | None = None) -> dict[str, Any]

Compute final dataset-level diversity indices.

Merges all batch-level value counts and computes diversity indices for each column.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array] | None

Dictionary of batch-level value-count arrays.

 None

Returns:

Type Description
dict[str, Any]

Dictionary containing final diversity scores.
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@override
def compute(self, batch_metrics: dict[str, pa.Array] | None = None) -> dict[str, Any]:
    """Compute final dataset-level diversity indices.

    Merges all batch-level value counts and computes diversity
    indices for each column.

    Args:
        batch_metrics: Dictionary of batch-level value-count arrays.

    Returns:
        Dictionary containing final diversity scores.
    """
    if not batch_metrics:
        return {"_metadata": {"error": "No batch metrics provided"}}

    results: dict[str, Any] = {}
    for col in self.input_columns:
        col_res = self._compute_column_diversity(col, batch_metrics)
        results.update(col_res)
    return results

compute_batch_metric(features: dict[str, pa.Array]) -> dict[str, pa.Array]

Compute per-batch value counts for streaming aggregation.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of column arrays from this batch.

 required

Returns:

Type Description
dict[str, Array]

Dictionary of batch-level value-count pairs.
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@override
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """Compute per-batch value counts for streaming aggregation.

    Args:
        features: Dictionary of column arrays from this batch.

    Returns:
        Dictionary of batch-level value-count pairs.
    """
    batch_metrics: dict[str, pa.Array] = {}

    for col in self.input_columns:
        if col not in features:
            logger.warning("[%s] column '%s' not found in batch", self.name, col)
            continue

        col_array = features[col]
        vc = pa.compute.value_counts(col_array)
        if len(vc) == 0:
            continue

        values = vc.field("values")
        counts = vc.field("counts")

        values_str = pa.compute.cast(values, pa.string())
        batch_metrics[f"{col}_values"] = values_str
        batch_metrics[f"{col}_counts"] = counts

    return batch_metrics

generated_metrics() -> list[str]

Return the list of metric columns that will be generated.

Returns:

Type Description
list[str]

List of output metric column names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@override
def generated_metrics(self) -> list[str]:
    """Return the list of metric columns that will be generated.

    Returns:
        List of output metric column names.
    """
    metrics = []
    for col in self.input_columns:
        if "simpson" in self.metrics:
            metrics.append(f"{col}_simpson")
        if "gini" in self.metrics:
            metrics.append(f"{col}_gini")
        if "shannon" in self.metrics:
            metrics.append(f"{col}_shannon")
        if "richness" in self.metrics:
            metrics.append(f"{col}_richness")
    return metrics

reset() -> None

Reset processor state for new processing run.

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/diversity.py 
191
192
193
@override
def reset(self) -> None:
    """Reset processor state for new processing run."""