Skip to content

dqm_ml_core

DQM ML Core package for data quality metrics processing.

This package provides core components for computing data quality metrics on datasets using a streaming architecture. It includes base classes for metric processors and implementations for common metrics like completeness and representativeness.

Main components: - Processor: Base class for all processors - FeaturesProcessor: Base class for feature extraction processors - MetricsProcessor: Base class for metric computation processors - GapProcessor: Base class for domain-gap processors - CompletenessProcessor: Computes data completeness scores - RepresentativenessProcessor: Evaluates distribution representativeness - ProcessorRunner: Orchestrator for running metrics on DataFrames - PluginLoadedRegistry: Registry for dynamically loaded metric plugins

__all__ = ['CompletenessProcessor', 'DiversityProcessor', 'FeaturesProcessor', 'GapProcessor', 'MetricsProcessor', 'PluginLoadedRegistry', 'Processor', 'ProcessorRunner', 'RepresentativenessProcessor'] module-attribute

CompletenessProcessor

Bases: MetricsProcessor

Data completeness processor that evaluates the completeness of tabular data.

This processor calculates completeness scores (ratio of non-null to total values) for specified columns and provides overall dataset completeness metrics.

The processor operates at multiple levels: - Batch level: Aggregated counts for streaming processing - Dataset level: Final completeness scores and statistics

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
 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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class CompletenessProcessor(MetricsProcessor):
    """
    Data completeness processor that evaluates the completeness of tabular data.

    This processor calculates completeness scores (ratio of non-null to
    total values) for specified columns and provides overall dataset
    completeness metrics.

    The processor operates at multiple levels:
    - Batch level: Aggregated counts for streaming processing
    - Dataset level: Final completeness scores and statistics
    """

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

        Args:
            name: Name of the processor.
            config: Configuration dictionary containing:
                - input_columns: List of columns to analyze.
                - output_metrics: Mapping of metric names to column names.
                - include_per_column: Include per-column completeness scores.
                - include_overall: Include overall completeness score.
        """
        super().__init__(name, config)

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

        # Which completeness levels to compute: per-column, overall, and metadata
        self.include_per_column: bool = cfg.include_per_column
        self.include_overall: bool = cfg.include_overall
        self.include_metadata: bool = cfg.include_metadata

        # Output column mappings
        self.output_metrics: dict[str, str] = {}

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

        Returns:
            List of output metric column names
        """
        # TODO : manage output metrics names with configuration
        metrics = []

        if self.include_overall:
            overall_key = self.output_metrics.get("overall_completeness", "completeness_overall")
            metrics.append(overall_key)

        if self.include_per_column:
            for col in self.input_columns:
                col_key = self.output_metrics.get(f"completeness_{col}", f"completeness_{col}")
                metrics.append(col_key)

        return metrics

    @override
    def select_columns(
        self,
        batch: pa.RecordBatch,
        prev_features: dict[str, pa.Array] | None = None,
    ) -> dict[str, pa.Array]:
        """
        Extract the needed columns from the batch for completeness analysis.

        This method simply passes through the columns we need to analyze,
        as completeness calculation is done at batch and dataset levels.

        Args:
            batch: Input batch of data
            prev_features: Previous features (not used in this processor)

        Returns:
            Dictionary containing the columns to analyze
        """
        features = {}

        columns_to_analyze = self.input_columns if self.input_columns else batch.column_names

        for col in columns_to_analyze:
            if col not in batch.schema.names:
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue

            # Simply pass through the column data for batch-level processing
            features[col] = batch.column(col)

        return features

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

        This counts total and non-null values per column in this batch,
        which will be aggregated across all batches for final dataset completeness.

        Args:
            features: Dictionary of column arrays from this batch

        Returns:
            Dictionary of batch-level completeness counts
        """
        batch_metrics = {}

        for col, col_array in features.items():
            total_count = len(col_array)

            # Count complete (non-null, non-NaN) samples in this batch.
            # pa.compute.is_valid() returns True for non-null values, but
            # float NaN from numpy is preserved as a valid float in Arrow
            # (not as a null), so we subtract NaN positions for float cols.
            valid_count = pa.compute.sum(pa.compute.is_valid(col_array)).as_py()
            if pa.types.is_floating(col_array.type):
                nan_count = pa.compute.sum(pa.compute.is_nan(col_array)).as_py()
                complete_count = valid_count - nan_count
            else:
                complete_count = valid_count

            # store counts for aggregation across batches
            batch_metrics[f"{col}_total_count"] = pa.array([total_count], type=pa.int64())
            batch_metrics[f"{col}_complete_count"] = pa.array([complete_count], type=pa.int64())

        return batch_metrics

    @staticmethod
    def _select_columns_from_metrics(
        batch_metrics: dict[str, pa.Array],
    ) -> list[str]:
        """Extract column names from batch metrics keys ending in _total_count.

        Args:
            batch_metrics: Dictionary of batch-level metrics.

        Returns:
            List of column names.
        """
        columns = []
        for key in batch_metrics:
            if key.endswith("_total_count"):
                columns.append(key.replace("_total_count", ""))
        return columns

    @staticmethod
    def _compute_column_completeness(col: str, batch_metrics: dict[str, pa.Array]) -> float | None:
        """Compute completeness score for a single column.

        Args:
            col: Column name.
            batch_metrics: Dictionary of batch-level metrics.

        Returns:
            Completeness score (0.0-1.0) or None if metrics are missing.
        """
        total_key = f"{col}_total_count"
        complete_key = f"{col}_complete_count"
        if total_key not in batch_metrics or complete_key not in batch_metrics:
            return None

        col_total = int(np.sum(batch_metrics[total_key].to_numpy()))
        col_complete = int(np.sum(batch_metrics[complete_key].to_numpy()))
        return col_complete / col_total if col_total > 0 else 0.0

    def _write_output_metrics(
        self,
        results: dict[str, Any],
        per_column_completeness: dict[str, float],
    ) -> None:
        """Write per-column and overall completeness metrics to results.

        Args:
            results: Output dict to populate.
            per_column_completeness: Dict of column -> completeness score.
        """
        if self.include_per_column:
            for col, score in per_column_completeness.items():
                output_key = self.output_metrics.get(f"completeness_{col}", f"completeness_{col}")
                results[output_key] = score

        if self.include_overall:
            overall = (
                sum(per_column_completeness.values()) / len(per_column_completeness) if per_column_completeness else 0.0
            )
            output_key = self.output_metrics.get("overall_completeness", "completeness_overall")
            results[output_key] = overall

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

        This aggregates the batch-level counts to compute final completeness scores
        for each column and overall dataset completeness.

        Args:
            batch_metrics: Dictionary of batch-level metrics to aggregate

        Returns:
            Dictionary of final completeness metrics
        """
        if not batch_metrics:
            return {"_metadata": {"error": "No batch metrics provided"}}

        columns_analyzed = self._select_columns_from_metrics(batch_metrics)
        if not columns_analyzed:
            logger.warning(f"[{self.name}] No columns found in batch metrics")
            return {"_metadata": {"error": "No columns found in batch metrics"}}

        per_column_completeness: dict[str, float] = {}
        total_samples = 0

        for col in columns_analyzed:
            score = self._compute_column_completeness(col, batch_metrics)
            if score is None:
                logger.warning(f"[{self.name}] Missing batch metrics for column '{col}'")
                continue
            per_column_completeness[col] = score
            total_key = f"{col}_total_count"
            total_samples += int(np.sum(batch_metrics[total_key].to_numpy()))

        results: dict[str, Any] = {}
        self._write_output_metrics(results, per_column_completeness)

        if self.include_metadata:
            metadata = {
                "columns_analyzed": columns_analyzed,
                "total_samples_per_column": total_samples // len(columns_analyzed) if columns_analyzed else 0,
                "per_column_scores": per_column_completeness,
                "overall_score": sum(per_column_completeness.values()) / len(per_column_completeness)
                if per_column_completeness
                else 0.0,
            }
            results["_metadata"] = json.dumps(metadata)

        return results

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

        The completeness processor has no persistent state across runs,
        so this is a no-op. Provided for interface compliance.
        """

include_metadata: bool = cfg.include_metadata instance-attribute

include_overall: bool = cfg.include_overall instance-attribute

include_per_column: bool = cfg.include_per_column instance-attribute

output_metrics: dict[str, str] = {} instance-attribute

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

Initialize the completeness processor.

Parameters:

Name Type Description Default
name str

Name of the processor.

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

Configuration dictionary containing: - input_columns: List of columns to analyze. - output_metrics: Mapping of metric names to column names. - include_per_column: Include per-column completeness scores. - include_overall: Include overall completeness score.

 None
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, name: str = "completeness", config: dict[str, Any] | None = None) -> None:
    """
    Initialize the completeness processor.

    Args:
        name: Name of the processor.
        config: Configuration dictionary containing:
            - input_columns: List of columns to analyze.
            - output_metrics: Mapping of metric names to column names.
            - include_per_column: Include per-column completeness scores.
            - include_overall: Include overall completeness score.
    """
    super().__init__(name, config)

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

    # Which completeness levels to compute: per-column, overall, and metadata
    self.include_per_column: bool = cfg.include_per_column
    self.include_overall: bool = cfg.include_overall
    self.include_metadata: bool = cfg.include_metadata

    # Output column mappings
    self.output_metrics: dict[str, str] = {}

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

Compute final dataset-level completeness metrics.

This aggregates the batch-level counts to compute final completeness scores for each column and overall dataset completeness.

Parameters:

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

Dictionary of batch-level metrics to aggregate

 None

Returns:

Type Description
dict[str, Any]

Dictionary of final completeness metrics
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@override
def compute(self, batch_metrics: dict[str, pa.Array] | None = None) -> dict[str, Any]:
    """
    Compute final dataset-level completeness metrics.

    This aggregates the batch-level counts to compute final completeness scores
    for each column and overall dataset completeness.

    Args:
        batch_metrics: Dictionary of batch-level metrics to aggregate

    Returns:
        Dictionary of final completeness metrics
    """
    if not batch_metrics:
        return {"_metadata": {"error": "No batch metrics provided"}}

    columns_analyzed = self._select_columns_from_metrics(batch_metrics)
    if not columns_analyzed:
        logger.warning(f"[{self.name}] No columns found in batch metrics")
        return {"_metadata": {"error": "No columns found in batch metrics"}}

    per_column_completeness: dict[str, float] = {}
    total_samples = 0

    for col in columns_analyzed:
        score = self._compute_column_completeness(col, batch_metrics)
        if score is None:
            logger.warning(f"[{self.name}] Missing batch metrics for column '{col}'")
            continue
        per_column_completeness[col] = score
        total_key = f"{col}_total_count"
        total_samples += int(np.sum(batch_metrics[total_key].to_numpy()))

    results: dict[str, Any] = {}
    self._write_output_metrics(results, per_column_completeness)

    if self.include_metadata:
        metadata = {
            "columns_analyzed": columns_analyzed,
            "total_samples_per_column": total_samples // len(columns_analyzed) if columns_analyzed else 0,
            "per_column_scores": per_column_completeness,
            "overall_score": sum(per_column_completeness.values()) / len(per_column_completeness)
            if per_column_completeness
            else 0.0,
        }
        results["_metadata"] = json.dumps(metadata)

    return results

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

Compute batch-level completeness counts for streaming aggregation.

This counts total and non-null values per column in this batch, which will be aggregated across all batches for final dataset completeness.

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 completeness counts
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
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
@override
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Compute batch-level completeness counts for streaming aggregation.

    This counts total and non-null values per column in this batch,
    which will be aggregated across all batches for final dataset completeness.

    Args:
        features: Dictionary of column arrays from this batch

    Returns:
        Dictionary of batch-level completeness counts
    """
    batch_metrics = {}

    for col, col_array in features.items():
        total_count = len(col_array)

        # Count complete (non-null, non-NaN) samples in this batch.
        # pa.compute.is_valid() returns True for non-null values, but
        # float NaN from numpy is preserved as a valid float in Arrow
        # (not as a null), so we subtract NaN positions for float cols.
        valid_count = pa.compute.sum(pa.compute.is_valid(col_array)).as_py()
        if pa.types.is_floating(col_array.type):
            nan_count = pa.compute.sum(pa.compute.is_nan(col_array)).as_py()
            complete_count = valid_count - nan_count
        else:
            complete_count = valid_count

        # store counts for aggregation across batches
        batch_metrics[f"{col}_total_count"] = pa.array([total_count], type=pa.int64())
        batch_metrics[f"{col}_complete_count"] = pa.array([complete_count], type=pa.int64())

    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/completeness.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@override
def generated_metrics(self) -> list[str]:
    """
    Return the list of metric columns that will be generated.

    Returns:
        List of output metric column names
    """
    # TODO : manage output metrics names with configuration
    metrics = []

    if self.include_overall:
        overall_key = self.output_metrics.get("overall_completeness", "completeness_overall")
        metrics.append(overall_key)

    if self.include_per_column:
        for col in self.input_columns:
            col_key = self.output_metrics.get(f"completeness_{col}", f"completeness_{col}")
            metrics.append(col_key)

    return metrics

reset() -> None

Reset processor state for new processing run.

The completeness processor has no persistent state across runs, so this is a no-op. Provided for interface compliance.

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
262
263
264
265
266
267
268
@override
def reset(self) -> None:
    """Reset processor state for new processing run.

    The completeness processor has no persistent state across runs,
    so this is a no-op. Provided for interface compliance.
    """

select_columns(batch: pa.RecordBatch, prev_features: dict[str, pa.Array] | None = None) -> dict[str, pa.Array]

Extract the needed columns from the batch for completeness analysis.

This method simply passes through the columns we need to analyze, as completeness calculation is done at batch and dataset levels.

Parameters:

Name Type Description Default
batch RecordBatch

Input batch of data

 required
prev_features dict[str, Array] | None

Previous features (not used in this processor)

 None

Returns:

Type Description
dict[str, Array]

Dictionary containing the columns to analyze
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/completeness.py 
 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
@override
def select_columns(
    self,
    batch: pa.RecordBatch,
    prev_features: dict[str, pa.Array] | None = None,
) -> dict[str, pa.Array]:
    """
    Extract the needed columns from the batch for completeness analysis.

    This method simply passes through the columns we need to analyze,
    as completeness calculation is done at batch and dataset levels.

    Args:
        batch: Input batch of data
        prev_features: Previous features (not used in this processor)

    Returns:
        Dictionary containing the columns to analyze
    """
    features = {}

    columns_to_analyze = self.input_columns if self.input_columns else batch.column_names

    for col in columns_to_analyze:
        if col not in batch.schema.names:
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue

        # Simply pass through the column data for batch-level processing
        features[col] = batch.column(col)

    return features

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."""

FeaturesProcessor

Bases: Processor

Base class for all feature extraction processors.

Feature processors transform raw data into per-sample features (e.g., image luminosity, embeddings). The primary lifecycle method is compute_features, which produces a dict of feature arrays from each batch of raw data.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
13
14
15
16
17
18
19
20
21
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
class FeaturesProcessor(Processor):
    """
    Base class for all feature extraction processors.

    Feature processors transform raw data into per-sample features
    (e.g., image luminosity, embeddings). The primary lifecycle method
    is ``compute_features``, which produces a dict of feature arrays
    from each batch of raw data.
    """

    def _resolve_output_name(self, col: str, feature_key: str) -> str:
        """Build the output column name with prefix and suffix applied.

        Args:
            col: The input column name.
            feature_key: The feature key to append after ``col_``.

        Returns:
            The fully qualified output column name.
        """
        from dqm_ml_core.models.columns import ColumnsConfig

        cfg: ColumnsConfig | None = getattr(self, "columns_config", None)
        base = f"{col}_{feature_key}"
        p = cfg.prefix if cfg else ""
        s = cfg.suffix if cfg else ""
        return f"{p}{base}{s}"

    def _check_image_fail_fast(self, exc: Exception, *error_attrs: str) -> None:
        if not (self.errors_config and self.errors_config.images):
            return
        image_errors = self.errors_config.images
        for attr in error_attrs:
            if getattr(image_errors, attr, None) == "fail_fast":
                raise exc

    def generated_features(self) -> list[str]:
        """
        Return the list of columns generated by this processor during feature extraction.

        Returns:
            A list of feature names.
        """
        outputs = getattr(self, "output_features", {})
        return list(outputs.values())

    def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Transform a raw data batch into features.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping feature names to pyarrow Arrays.
        """
        features = {}

        available = batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        for col in cols:
            if col in prev_features:
                continue
            features[col] = batch.column(col)

        return features

compute_features(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Transform a raw data batch into features.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping feature names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
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
def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Transform a raw data batch into features.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping feature names to pyarrow Arrays.
    """
    features = {}

    available = batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    for col in cols:
        if col in prev_features:
            continue
        features[col] = batch.column(col)

    return features

generated_features() -> list[str]

Return the list of columns generated by this processor during feature extraction.

Returns:

Type Description
list[str]

A list of feature names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
49
50
51
52
53
54
55
56
57
def generated_features(self) -> list[str]:
    """
    Return the list of columns generated by this processor during feature extraction.

    Returns:
        A list of feature names.
    """
    outputs = getattr(self, "output_features", {})
    return list(outputs.values())

GapProcessor

Bases: Processor

Base class for all domain gap processors.

Gap processors compute distribution shift between two datasets (e.g., MMD, FID, KL divergence). The primary lifecycle methods are select_features (per-batch column selection aware of previous features), compute_batch_metric (batch aggregation), compute (final dataset-level statistics), and compute_delta (pairwise comparison).

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
18
19
20
21
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
class GapProcessor(Processor):
    """
    Base class for all domain gap processors.

    Gap processors compute distribution shift between two datasets
    (e.g., MMD, FID, KL divergence). The primary lifecycle methods are
    ``select_features`` (per-batch column selection aware of previous features),
    ``compute_batch_metric`` (batch aggregation), ``compute`` (final dataset-level
    statistics), and ``compute_delta`` (pairwise comparison).
    """

    def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Extract relevant columns from a batch, resolving patterns against
        both batch columns and previously computed upstream features.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping column names to pyarrow Arrays.
        """
        available = list(prev_features.keys()) + batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        features: dict[str, pa.Array] = {}
        for col in cols:
            if col in prev_features:
                continue
            if col not in batch.schema.names:
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue
            features[col] = batch.column(col)

        return features

    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Aggregate features into intermediate statistics for the current batch.

        Args:
            features: Dictionary of feature arrays computed on the batch.

        Returns:
            A dictionary of aggregated statistics.
        """
        return {}

    def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
        """
        Perform the final dataset-level aggregation of batch statistics.

        Args:
            batch_metrics: The aggregated intermediate statistics from all batches.

        Returns:
            A dictionary containing the final dataset-level statistics.
        """
        # SonarQube raises a warning because batch_metrics is not used.
        # It is irrelevant because compute is implemented in child classes which use batch_metrics.
        return {}

    def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
        """
        Compare metrics between two different dataselections.

        Args:
            source: Final metrics from the source dataselection.
            target: Final metrics from the target dataselection.

        Returns:
            A dictionary containing distance or difference scores.
        """
        return {}

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

Perform the final dataset-level aggregation of batch statistics.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array]

The aggregated intermediate statistics from all batches.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the final dataset-level statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
    """
    Perform the final dataset-level aggregation of batch statistics.

    Args:
        batch_metrics: The aggregated intermediate statistics from all batches.

    Returns:
        A dictionary containing the final dataset-level statistics.
    """
    # SonarQube raises a warning because batch_metrics is not used.
    # It is irrelevant because compute is implemented in child classes which use batch_metrics.
    return {}

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

Aggregate features into intermediate statistics for the current batch.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of feature arrays computed on the batch.

 required

Returns:

Type Description
dict[str, Array]

A dictionary of aggregated statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
59
60
61
62
63
64
65
66
67
68
69
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Aggregate features into intermediate statistics for the current batch.

    Args:
        features: Dictionary of feature arrays computed on the batch.

    Returns:
        A dictionary of aggregated statistics.
    """
    return {}

compute_delta(source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]

Compare metrics between two different dataselections.

Parameters:

Name Type Description Default
source dict[str, Any]

Final metrics from the source dataselection.

 required
target dict[str, Any]

Final metrics from the target dataselection.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing distance or difference scores.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
85
86
87
88
89
90
91
92
93
94
95
96
def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
    """
    Compare metrics between two different dataselections.

    Args:
        source: Final metrics from the source dataselection.
        target: Final metrics from the target dataselection.

    Returns:
        A dictionary containing distance or difference scores.
    """
    return {}

select_features(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Extract relevant columns from a batch, resolving patterns against both batch columns and previously computed upstream features.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping column names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
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
def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Extract relevant columns from a batch, resolving patterns against
    both batch columns and previously computed upstream features.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping column names to pyarrow Arrays.
    """
    available = list(prev_features.keys()) + batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    features: dict[str, pa.Array] = {}
    for col in cols:
        if col in prev_features:
            continue
        if col not in batch.schema.names:
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue
        features[col] = batch.column(col)

    return features

MetricsProcessor

Bases: Processor

Base class for all metric computation processors.

Metric processors compute dataset-level scores (e.g., completeness, diversity, representativeness). The primary lifecycle methods are select_columns (per-batch column selection), compute_batch_metric (batch aggregation), and compute (final dataset-level computation).

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
 18
 19
 20
 21
 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
class MetricsProcessor(Processor):
    """
    Base class for all metric computation processors.

    Metric processors compute dataset-level scores (e.g., completeness,
    diversity, representativeness). The primary lifecycle methods are
    ``select_columns`` (per-batch column selection), ``compute_batch_metric``
    (batch aggregation), and ``compute`` (final dataset-level computation).
    """

    def generated_metrics(self) -> list[str]:
        """
        Return the names of the final metrics produced by this processor.

        Returns:
            A list of metric names.
        """
        outputs = getattr(self, "output_metrics", {})
        return list(outputs.values())

    def select_columns(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Select relevant columns from a raw batch for metric computation.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping column names to pyarrow Arrays.
        """
        features = {}

        available = batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        for col in cols:
            if col in prev_features:
                continue

            if col not in available:
                if (
                    self.errors_config
                    and self.errors_config.tabular
                    and self.errors_config.tabular.on_missing_column == "fail_fast"
                ):
                    raise KeyError(f"Column '{col}' not found in batch")
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue
            features[col] = batch.column(col)

        return features

    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Aggregate features into intermediate statistics for the current batch.

        This method is critical for scalability. It should return a compact
        representation of the data (e.g., partial sums) that can be
        efficiently combined later.

        Args:
            features: Dictionary of feature arrays computed on the batch.

        Returns:
            A dictionary of aggregated statistics.
        """
        return {}

    def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
        """
        Perform the final dataset-level metric calculation.

        Args:
            batch_metrics: The aggregated intermediate statistics from all batches.

        Returns:
            A dictionary containing the final metrics.
        """
        # SonarQube raises a warning because batch_metrics is not used.
        # It is irrelevant because compute is implemented in child classes which use batch_metrics.
        return {}

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

Perform the final dataset-level metric calculation.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array]

The aggregated intermediate statistics from all batches.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the final metrics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
    """
    Perform the final dataset-level metric calculation.

    Args:
        batch_metrics: The aggregated intermediate statistics from all batches.

    Returns:
        A dictionary containing the final metrics.
    """
    # SonarQube raises a warning because batch_metrics is not used.
    # It is irrelevant because compute is implemented in child classes which use batch_metrics.
    return {}

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

Aggregate features into intermediate statistics for the current batch.

This method is critical for scalability. It should return a compact representation of the data (e.g., partial sums) that can be efficiently combined later.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of feature arrays computed on the batch.

 required

Returns:

Type Description
dict[str, Array]

A dictionary of aggregated statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Aggregate features into intermediate statistics for the current batch.

    This method is critical for scalability. It should return a compact
    representation of the data (e.g., partial sums) that can be
    efficiently combined later.

    Args:
        features: Dictionary of feature arrays computed on the batch.

    Returns:
        A dictionary of aggregated statistics.
    """
    return {}

generated_metrics() -> list[str]

Return the names of the final metrics produced by this processor.

Returns:

Type Description
list[str]

A list of metric names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
28
29
30
31
32
33
34
35
36
def generated_metrics(self) -> list[str]:
    """
    Return the names of the final metrics produced by this processor.

    Returns:
        A list of metric names.
    """
    outputs = getattr(self, "output_metrics", {})
    return list(outputs.values())

select_columns(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Select relevant columns from a raw batch for metric computation.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping column names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
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
def select_columns(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Select relevant columns from a raw batch for metric computation.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping column names to pyarrow Arrays.
    """
    features = {}

    available = batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    for col in cols:
        if col in prev_features:
            continue

        if col not in available:
            if (
                self.errors_config
                and self.errors_config.tabular
                and self.errors_config.tabular.on_missing_column == "fail_fast"
            ):
                raise KeyError(f"Column '{col}' not found in batch")
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue
        features[col] = batch.column(col)

    return features

PluginLoadedRegistry

Singleton registry that provides lazy access to all registered DQM components.

Components include: - Metrics (Processor) - DataLoaders - OutputWriters

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
 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
class PluginLoadedRegistry:
    """
    Singleton registry that provides lazy access to all registered DQM components.

    Components include:
    - Metrics (Processor)
    - DataLoaders
    - OutputWriters
    """

    _metrics_registry: dict[str, type[Processor]] | None = None
    _features_registry: dict[str, type[Processor]] | None = None
    _gap_registry: dict[str, type[Processor]] | None = None
    _dataloaders_registry: dict[str, Any] | None = None
    _outputwriter_registry: dict[str, Any] | None = None

    @classmethod
    def get_metrics_registry(cls) -> dict[str, type[Processor]]:
        """Return the registry of available metric processors.

        Returns:
            A dictionary mapping metric processor names to their classes.
        """
        if not cls._metrics_registry:
            from dqm_ml_core.api.metrics_processor import MetricsProcessor

            cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", MetricsProcessor)

        return cls._metrics_registry

    @classmethod
    def get_features_registry(cls) -> dict[str, type[Processor]]:
        """Return the registry of available feature extraction processors.

        Returns:
            A dictionary mapping feature processor names to their classes.
        """
        if not cls._features_registry:
            from dqm_ml_core.api.features_processor import FeaturesProcessor

            cls._features_registry = load_registered_plugins("dqm_ml.features", FeaturesProcessor)

        return cls._features_registry

    @classmethod
    def get_gap_registry(cls) -> dict[str, type[Processor]]:
        """Return the registry of available gap processors.

        Returns:
            A dictionary mapping gap processor names to their classes.
        """
        if not cls._gap_registry:
            from dqm_ml_core.api.gap_processor import GapProcessor

            cls._gap_registry = load_registered_plugins("dqm_ml.gap", GapProcessor)

        return cls._gap_registry

    @classmethod
    def get_dataloaders_registry(cls) -> dict[str, Any]:
        """Return the registry of available data loaders.

        Returns:
            A dictionary mapping data loader names to their classes.
        """
        if not cls._dataloaders_registry:
            cls._dataloaders_registry = load_registered_plugins("dqm_ml.dataloaders", None)  # TODO add base class
        return cls._dataloaders_registry

    @classmethod
    def get_outputwriter_registry(cls) -> dict[str, Any]:
        """Return the registry of available output writers.

        Returns:
            A dictionary mapping output writer names to their classes.
        """
        if not cls._outputwriter_registry:
            cls._outputwriter_registry = load_registered_plugins("dqm_ml.outputwriter", None)  # TODO add base class

        return cls._outputwriter_registry

get_dataloaders_registry() -> dict[str, Any] classmethod

Return the registry of available data loaders.

Returns:

Type Description
dict[str, Any]

A dictionary mapping data loader names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
113
114
115
116
117
118
119
120
121
122
@classmethod
def get_dataloaders_registry(cls) -> dict[str, Any]:
    """Return the registry of available data loaders.

    Returns:
        A dictionary mapping data loader names to their classes.
    """
    if not cls._dataloaders_registry:
        cls._dataloaders_registry = load_registered_plugins("dqm_ml.dataloaders", None)  # TODO add base class
    return cls._dataloaders_registry

get_features_registry() -> dict[str, type[Processor]] classmethod

Return the registry of available feature extraction processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping feature processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def get_features_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available feature extraction processors.

    Returns:
        A dictionary mapping feature processor names to their classes.
    """
    if not cls._features_registry:
        from dqm_ml_core.api.features_processor import FeaturesProcessor

        cls._features_registry = load_registered_plugins("dqm_ml.features", FeaturesProcessor)

    return cls._features_registry

get_gap_registry() -> dict[str, type[Processor]] classmethod

Return the registry of available gap processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping gap processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
@classmethod
def get_gap_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available gap processors.

    Returns:
        A dictionary mapping gap processor names to their classes.
    """
    if not cls._gap_registry:
        from dqm_ml_core.api.gap_processor import GapProcessor

        cls._gap_registry = load_registered_plugins("dqm_ml.gap", GapProcessor)

    return cls._gap_registry

get_metrics_registry() -> dict[str, type[Processor]] classmethod

Return the registry of available metric processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping metric processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
@classmethod
def get_metrics_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available metric processors.

    Returns:
        A dictionary mapping metric processor names to their classes.
    """
    if not cls._metrics_registry:
        from dqm_ml_core.api.metrics_processor import MetricsProcessor

        cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", MetricsProcessor)

    return cls._metrics_registry

get_outputwriter_registry() -> dict[str, Any] classmethod

Return the registry of available output writers.

Returns:

Type Description
dict[str, Any]

A dictionary mapping output writer names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def get_outputwriter_registry(cls) -> dict[str, Any]:
    """Return the registry of available output writers.

    Returns:
        A dictionary mapping output writer names to their classes.
    """
    if not cls._outputwriter_registry:
        cls._outputwriter_registry = load_registered_plugins("dqm_ml.outputwriter", None)  # TODO add base class

    return cls._outputwriter_registry

Processor

Base class for all Data Quality metrics, feature extractors, and gap processors.

Provides shared initialization, failure-rate checking, and column resolution. Lifecycle methods are defined in the appropriate subclass: :class:FeaturesProcessor, :class:MetricsProcessor, or :class:GapProcessor.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
13
14
15
16
17
18
19
20
21
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
class Processor:
    """
    Base class for all Data Quality metrics, feature extractors, and gap processors.

    Provides shared initialization, failure-rate checking, and column resolution.
    Lifecycle methods are defined in the appropriate subclass:
    :class:`FeaturesProcessor`, :class:`MetricsProcessor`, or :class:`GapProcessor`.
    """

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

        Args:
            name: Unique name of the processor instance.
            config: Configuration dictionary (optional).
        """
        self.name = name
        config = config or {}
        self.storage_raw = config.pop("storage", None)
        self.config = config
        self.errors_config = None
        self.compute_device: str = "cpu"
        self.compute_seed: int | None = None
        self.current_path_prefix: dict[str, str] = {}
        self._failure_count = 0
        self._total_count = 0

        self.input_columns: list[str] = []
        self.exclude_columns: list[str] | None = None
        self.columns_config = None
        if "columns" in self.config and isinstance(self.config["columns"], dict):
            from dqm_ml_core.models.columns import ColumnsConfig

            self.columns_config = ColumnsConfig.model_validate(self.config["columns"])
            self.input_columns = self.columns_config.input
            self.exclude_columns = self.columns_config.exclude

    def _check_failure_rate(self) -> None:
        if self.errors_config and self.errors_config.max_failure_rate is not None and self._total_count > 0:
            failure_rate = self._failure_count / self._total_count
            if failure_rate > self.errors_config.max_failure_rate:
                raise RuntimeError(
                    f"Failure rate {failure_rate:.1%} exceeds max {self.errors_config.max_failure_rate:.1%}"
                )

    def needed_columns(self) -> list[str]:
        """
        Return the list of raw input columns required for processing.

        Returns:
            A list of column names.
        """
        return self.input_columns

    def reset(self) -> None:
        """Reset per-selection state between dataselections.

        Processors that cache per-selection state (e.g. histogram bin
        edges in RepresentativenessProcessor) MUST override this to
        clear that state.  Called by DatasetJob after each selection.
        """

columns_config = None instance-attribute

compute_device: str = 'cpu' instance-attribute

compute_seed: int | None = None instance-attribute

config = config instance-attribute

current_path_prefix: dict[str, str] = {} instance-attribute

errors_config = None instance-attribute

exclude_columns: list[str] | None = None instance-attribute

input_columns: list[str] = [] instance-attribute

name = name instance-attribute

storage_raw = config.pop('storage', None) instance-attribute

__init__(name: str, config: dict[str, Any] | None)

Initialize the processor.

Parameters:

Name Type Description Default
name str

Unique name of the processor instance.

 required
config dict[str, Any] | None

Configuration dictionary (optional).

 required
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.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
def __init__(self, name: str, config: dict[str, Any] | None):
    """
    Initialize the processor.

    Args:
        name: Unique name of the processor instance.
        config: Configuration dictionary (optional).
    """
    self.name = name
    config = config or {}
    self.storage_raw = config.pop("storage", None)
    self.config = config
    self.errors_config = None
    self.compute_device: str = "cpu"
    self.compute_seed: int | None = None
    self.current_path_prefix: dict[str, str] = {}
    self._failure_count = 0
    self._total_count = 0

    self.input_columns: list[str] = []
    self.exclude_columns: list[str] | None = None
    self.columns_config = None
    if "columns" in self.config and isinstance(self.config["columns"], dict):
        from dqm_ml_core.models.columns import ColumnsConfig

        self.columns_config = ColumnsConfig.model_validate(self.config["columns"])
        self.input_columns = self.columns_config.input
        self.exclude_columns = self.columns_config.exclude

needed_columns() -> list[str]

Return the list of raw input columns required for processing.

Returns:

Type Description
list[str]

A list of column names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
59
60
61
62
63
64
65
66
def needed_columns(self) -> list[str]:
    """
    Return the list of raw input columns required for processing.

    Returns:
        A list of column names.
    """
    return self.input_columns

reset() -> None

Reset per-selection state between dataselections.

Processors that cache per-selection state (e.g. histogram bin edges in RepresentativenessProcessor) MUST override this to clear that state. Called by DatasetJob after each selection.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
68
69
70
71
72
73
74
def reset(self) -> None:
    """Reset per-selection state between dataselections.

    Processors that cache per-selection state (e.g. histogram bin
    edges in RepresentativenessProcessor) MUST override this to
    clear that state.  Called by DatasetJob after each selection.
    """

ProcessorRunner

Orchestrator for executing metric processors on in-memory Pandas DataFrames.

This class provides a high-level API for users who want to compute metrics directly on DataFrames without using the full YAML-driven pipeline.

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/processor_runner.py 
 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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class ProcessorRunner:
    """
    Orchestrator for executing metric processors on in-memory Pandas DataFrames.

    This class provides a high-level API for users who want to compute metrics
    directly on DataFrames without using the full YAML-driven pipeline.
    """

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

        Args:
            config: Optional configuration for metric default behaviors.
        """
        self.config = config or {}

    @staticmethod
    def _df_to_record_batch(df: DataFrame) -> pa.RecordBatch:
        """Convert a pandas DataFrame to a pyarrow RecordBatch.

        Handles conversion of object columns containing uniform-size numpy
        arrays (e.g. embeddings) to FixedSizeListArray, which is required
        by processors like DomainGapProcessor.
        """
        arrays: list[pa.Array] = []
        for col_name in df.columns:
            series = df[col_name]
            if series.dtype == object and len(series) > 0:
                first = series.iloc[0]
                if isinstance(first, np.ndarray):
                    dim = first.shape[0]
                    if all(isinstance(v, np.ndarray) and v.shape == (dim,) for v in series):
                        flat = np.vstack(series.values).astype(np.float64)
                        arr = pa.FixedSizeListArray.from_arrays(pa.array(flat.flatten().tolist()), dim)
                        arrays.append(arr)
                        continue
            arrays.append(pa.Array.from_pandas(series))
        table = pa.table(dict(zip(df.columns, arrays, strict=True)))
        return table.to_batches()[0]

    def _compute_batch_level(self, df: DataFrame, processors: list[Processor]) -> tuple[dict[str, Any], dict[str, Any]]:
        """Compute features and batch-level metrics from non-Gap processors.

        Returns:
            Tuple of (batch_features, batch_metrics) merged dicts.
        """
        from dqm_ml_core.api.features_processor import FeaturesProcessor

        batch = self._df_to_record_batch(df)
        batch_features: dict[str, Any] = {}
        batch_metrics: dict[str, Any] = {}

        for metric in processors:
            logger.debug(f"Processing metric {metric.__class__.__name__}")
            if isinstance(metric, FeaturesProcessor):
                batch_features |= metric.compute_features(batch, prev_features=batch_features)
            elif isinstance(metric, MetricsProcessor):
                # select_columns returns intermediate columns for metric computation,
                # not final features - use only for compute_batch_metric
                intermediate = metric.select_columns(batch, prev_features=batch_features)

            if isinstance(metric, MetricsProcessor):
                batch_metrics |= metric.compute_batch_metric(intermediate)

        return batch_features, batch_metrics

    def _compute_dataset_level(self, batch_metrics: dict[str, Any], processors: list[Processor]) -> dict[str, Any]:
        """Compute dataset-level metrics from MetricsProcessor instances."""
        dataset_metrics: dict[str, Any] = {}
        for metric in processors:
            if not isinstance(metric, MetricsProcessor):
                continue
            logger.debug(f"Computing final score for {metric.__class__.__name__}")
            dataset_metrics |= metric.compute(batch_metrics=batch_metrics)
        return dataset_metrics

    def run(self, df: DataFrame, processors: list[Processor]) -> dict[str, Any]:
        """
        Execute the provided processors on a DataFrame.

        Args:
            df: The input Pandas DataFrame.
            processors: List of initialized Processor instances.

        Returns:
            A dictionary containing the aggregated dataset-level metrics
            and/or per-sample features. Features from FeaturesProcessor
            instances are included alongside metrics from MetricsProcessor instances.
        """
        if df.empty or not processors:
            logger.warning("Empty DataFrame or no processors provided to ProcessorRunner")
            return {}

        active = [p for p in processors if not isinstance(p, GapProcessor)]
        batch_features, batch_metrics = self._compute_batch_level(df, active)
        dataset_metrics = self._compute_dataset_level(batch_metrics, active)

        # Merge features (from FeaturesProcessor) and metrics (from MetricsProcessor)
        result = {}
        result.update(batch_features)  # Per-sample features
        result.update(dataset_metrics)  # Aggregated metrics
        return result

    def _compute_features(self, df: DataFrame, features: list[Processor]) -> DataFrame:
        """Run FeaturesProcessors on a DataFrame and return a new DataFrame with computed columns.

        Args:
            df: The input DataFrame.
            features: List of FeaturesProcessor instances.

        Returns:
            A new DataFrame with the original columns plus any new feature columns.
        """
        result = self.run(df, features)
        if not result:
            return df
        out = df.copy()
        for col, values in result.items():
            out[col] = values
        return out

    def run_gap(
        self,
        source_df: DataFrame,
        target_df: DataFrame,
        processor: GapProcessor,
        features: list[Processor] | None = None,
        source_selection_name: str = "source",
        target_selection_name: str = "target",
    ) -> dict[str, Any]:
        """
        Execute a GapProcessor on two DataFrames and compute the domain gap.

        This method handles the two-dataset execution pattern required by GapProcessor:
        1. Optionally compute features (e.g. embeddings) on both DataFrames
        2. Process source DataFrame to compute source statistics
        3. Process target DataFrame to compute target statistics
        4. Compute the domain gap delta between source and target

        Args:
            source_df: The source dataset DataFrame.
            target_df: The target dataset DataFrame.
            processor: An initialized GapProcessor instance.
            features: Optional list of FeaturesProcessor instances to run on both
                DataFrames before computing the gap. For example, pass an
                ImageEmbeddingProcessor to compute embeddings from raw images.
            source_selection_name: Name for the source selection (default: "source").
            target_selection_name: Name for the target selection (default: "target").

        Returns:
            A dictionary containing:
            - {metric_name}: The domain gap metric value
            - "selection_source": Name of the source selection
            - "selection_target": Name of the target selection
        """
        if source_df.empty or target_df.empty:
            logger.warning("Empty DataFrame provided to run_gap")
            return {}

        # Compute features if provided (e.g. embeddings from raw images)
        if features:
            source_df = self._compute_features(source_df, features)
            target_df = self._compute_features(target_df, features)

        # Process source dataset
        source_batch = self._df_to_record_batch(source_df)
        source_features = processor.select_features(source_batch, prev_features={})
        source_batch_metrics = processor.compute_batch_metric(source_features)
        source_stats = processor.compute(source_batch_metrics)

        # Process target dataset
        target_batch = self._df_to_record_batch(target_df)
        target_features = processor.select_features(target_batch, prev_features={})
        target_batch_metrics = processor.compute_batch_metric(target_features)
        target_stats = processor.compute(target_batch_metrics)

        # Compute domain gap delta
        result = processor.compute_delta(source_stats, target_stats)

        # Add selection metadata
        result["selection_source"] = pa.array([source_selection_name])
        result["selection_target"] = pa.array([target_selection_name])

        return result

config = config or {} instance-attribute

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

Initialize the runner.

Parameters:

Name Type Description Default
config dict[str, Any] | None

Optional configuration for metric default behaviors.

 None
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/processor_runner.py 
31
32
33
34
35
36
37
38
def __init__(self, config: dict[str, Any] | None = None) -> None:
    """
    Initialize the runner.

    Args:
        config: Optional configuration for metric default behaviors.
    """
    self.config = config or {}

run(df: DataFrame, processors: list[Processor]) -> dict[str, Any]

Execute the provided processors on a DataFrame.

Parameters:

Name Type Description Default
df DataFrame

The input Pandas DataFrame.

 required
processors list[Processor]

List of initialized Processor instances.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the aggregated dataset-level metrics
dict[str, Any]

and/or per-sample features. Features from FeaturesProcessor
dict[str, Any]

instances are included alongside metrics from MetricsProcessor instances.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/processor_runner.py 
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
def run(self, df: DataFrame, processors: list[Processor]) -> dict[str, Any]:
    """
    Execute the provided processors on a DataFrame.

    Args:
        df: The input Pandas DataFrame.
        processors: List of initialized Processor instances.

    Returns:
        A dictionary containing the aggregated dataset-level metrics
        and/or per-sample features. Features from FeaturesProcessor
        instances are included alongside metrics from MetricsProcessor instances.
    """
    if df.empty or not processors:
        logger.warning("Empty DataFrame or no processors provided to ProcessorRunner")
        return {}

    active = [p for p in processors if not isinstance(p, GapProcessor)]
    batch_features, batch_metrics = self._compute_batch_level(df, active)
    dataset_metrics = self._compute_dataset_level(batch_metrics, active)

    # Merge features (from FeaturesProcessor) and metrics (from MetricsProcessor)
    result = {}
    result.update(batch_features)  # Per-sample features
    result.update(dataset_metrics)  # Aggregated metrics
    return result

run_gap(source_df: DataFrame, target_df: DataFrame, processor: GapProcessor, features: list[Processor] | None = None, source_selection_name: str = 'source', target_selection_name: str = 'target') -> dict[str, Any]

Execute a GapProcessor on two DataFrames and compute the domain gap.

This method handles the two-dataset execution pattern required by GapProcessor: 1. Optionally compute features (e.g. embeddings) on both DataFrames 2. Process source DataFrame to compute source statistics 3. Process target DataFrame to compute target statistics 4. Compute the domain gap delta between source and target

Parameters:

Name Type Description Default
source_df DataFrame

The source dataset DataFrame.

 required
target_df DataFrame

The target dataset DataFrame.

 required
processor GapProcessor

An initialized GapProcessor instance.

 required
features list[Processor] | None

Optional list of FeaturesProcessor instances to run on both DataFrames before computing the gap. For example, pass an ImageEmbeddingProcessor to compute embeddings from raw images.

 None
source_selection_name str

Name for the source selection (default: "source").

 'source'
target_selection_name str

Name for the target selection (default: "target").

 'target'

Returns:

Type Description
dict[str, Any]

A dictionary containing:
dict[str, Any]
  • {metric_name}: The domain gap metric value
dict[str, Any]
  • "selection_source": Name of the source selection
dict[str, Any]
  • "selection_target": Name of the target selection
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/processor_runner.py 
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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def run_gap(
    self,
    source_df: DataFrame,
    target_df: DataFrame,
    processor: GapProcessor,
    features: list[Processor] | None = None,
    source_selection_name: str = "source",
    target_selection_name: str = "target",
) -> dict[str, Any]:
    """
    Execute a GapProcessor on two DataFrames and compute the domain gap.

    This method handles the two-dataset execution pattern required by GapProcessor:
    1. Optionally compute features (e.g. embeddings) on both DataFrames
    2. Process source DataFrame to compute source statistics
    3. Process target DataFrame to compute target statistics
    4. Compute the domain gap delta between source and target

    Args:
        source_df: The source dataset DataFrame.
        target_df: The target dataset DataFrame.
        processor: An initialized GapProcessor instance.
        features: Optional list of FeaturesProcessor instances to run on both
            DataFrames before computing the gap. For example, pass an
            ImageEmbeddingProcessor to compute embeddings from raw images.
        source_selection_name: Name for the source selection (default: "source").
        target_selection_name: Name for the target selection (default: "target").

    Returns:
        A dictionary containing:
        - {metric_name}: The domain gap metric value
        - "selection_source": Name of the source selection
        - "selection_target": Name of the target selection
    """
    if source_df.empty or target_df.empty:
        logger.warning("Empty DataFrame provided to run_gap")
        return {}

    # Compute features if provided (e.g. embeddings from raw images)
    if features:
        source_df = self._compute_features(source_df, features)
        target_df = self._compute_features(target_df, features)

    # Process source dataset
    source_batch = self._df_to_record_batch(source_df)
    source_features = processor.select_features(source_batch, prev_features={})
    source_batch_metrics = processor.compute_batch_metric(source_features)
    source_stats = processor.compute(source_batch_metrics)

    # Process target dataset
    target_batch = self._df_to_record_batch(target_df)
    target_features = processor.select_features(target_batch, prev_features={})
    target_batch_metrics = processor.compute_batch_metric(target_features)
    target_stats = processor.compute(target_batch_metrics)

    # Compute domain gap delta
    result = processor.compute_delta(source_stats, target_stats)

    # Add selection metadata
    result["selection_source"] = pa.array([source_selection_name])
    result["selection_target"] = pa.array([target_selection_name])

    return result

RepresentativenessProcessor

Bases: MetricsProcessor

Evaluates how well the dataset represents a target statistical distribution.

This processor performs on samples discretisation statistical tests to compare the observed distribution of numerical columns against a theoretical target distribution (Normal or Uniform).

Supported Metrics
  • Chi-square: Goodness-of-fit test for categorical/binned data.
  • Kolmogorov-Smirnov (KS): Non-parametric test for continuous distributions (approximated via sampling).
  • Shannon Entropy: Measures the information diversity of the binned data.
  • GRTE (Geometric Representativeness Trajectory Error): Measures the exponential gap between observed and theoretical entropy.

The processor uses a streaming architecture: - Batch level: Computes partial calculus. - Dataset level: Aggregates histograms and performs final statistical tests.

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/representativeness.py 
 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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
class RepresentativenessProcessor(MetricsProcessor):
    """
    Evaluates how well the dataset represents a target statistical distribution.

    This processor performs on samples discretisation statistical tests to compare the observed
    distribution of numerical columns against a theoretical target distribution
    (Normal or Uniform).

    Supported Metrics:
      - Chi-square: Goodness-of-fit test for categorical/binned data.
      - Kolmogorov-Smirnov (KS): Non-parametric test for continuous distributions (approximated via sampling).
      - Shannon Entropy: Measures the information diversity of the binned data.
      - GRTE (Geometric Representativeness Trajectory Error): Measures the exponential gap
        between observed and theoretical entropy.

    The processor uses a streaming architecture:
    - Batch level: Computes partial calculus.
    - Dataset level: Aggregates histograms and performs final statistical tests.
    """

    SUPPORTED_METRICS = {
        "chi-square",
        "grte",
        "shannon-entropy",
        "kolmogorov-smirnov",
    }
    SUPPORTED_DISTS = {"normal", "uniform"}

    # Configuration constants - can be overridden in config
    DEFAULT_ALPHA = 0.05  # Significance level for statistical tests
    DEFAULT_SHANNON_ENTROPY_THRESHOLD = 2.0  # Threshold for high/low diversity interpretation
    DEFAULT_GRTE_THRESHOLD = 0.5  # Threshold for high/low representativeness interpretation
    DEFAULT_KS_SAMPLE_SIZE = 500  # Maximum sample size for KS test
    DEFAULT_KS_MIN_SAMPLE_SIZE = 50  # Minimum sample size for KS test
    DEFAULT_KS_SAMPLE_DIVISOR = 20  # Divisor for calculating sample size per batch
    DEFAULT_EPSILON = 1e-9  # Small value to avoid division by zero
    DEFAULT_INTERPRETATION_THRESHOLDS = {
        "follows_distribution": "follows_distribution",
        "does_not_follow_distribution": "does_not_follow_distribution",
        "high_diversity": "high_diversity",
        "low_diversity": "low_diversity",
        "high_representativeness": "high_representativeness",
        "low_representativeness": "low_representativeness",
    }

    def __init__(
        self,
        name: str = "representativeness",
        config: dict[str, Any] | None = None,
    ) -> None:
        """
        Initialize the representativeness 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).
                - bins: Number of bins for histograms (default: 10).
                - distribution: Target distribution ("normal" or "uniform").
                - alpha: Significance level (default: 0.05).
                - distribution_params: Dictionary of params (e.g., mean, std, min, max).
        """
        super().__init__(name, config)
        self.name = name
        cfg = RepresentativenessProcessorConfig.model_validate({**self.config, "name": self.name})
        self._init_from_cfg(cfg)

        # Use device from compute config if available
        compute_device = getattr(self, "compute_device", None)
        self.device = self._resolve_device(compute_device) if compute_device else self._resolve_device("cpu")

        # Use seed from compute config if available
        compute_seed = getattr(self, "compute_seed", None)
        if compute_seed is not None:
            self._rng = np.random.default_rng(compute_seed)
        else:
            self._rng = np.random.default_rng(42)

        self._bin_edges: dict[str, np.ndarray] = {}
        self._bin_params: dict[str, dict[str, float]] = {}  # caching for _compute_expected_counts
        self._initialized: bool = False

    def _set_optional_subconfig(self, cfg: RepresentativenessProcessorConfig) -> None:
        self.bins = cfg.histogram.bins if cfg.histogram else 10
        self.shannon_entropy_threshold = (
            cfg.shannon.threshold if cfg.shannon else self.DEFAULT_SHANNON_ENTROPY_THRESHOLD
        )
        self.grte_threshold = cfg.grte.threshold if cfg.grte else self.DEFAULT_GRTE_THRESHOLD
        self.grte_scaling_factor = cfg.grte.scaling_factor if cfg.grte else -2.0
        self.ks_sample_size = cfg.ks.sample_size if cfg.ks else self.DEFAULT_KS_SAMPLE_SIZE
        self.ks_min_sample_size = cfg.ks.min_sample_size if cfg.ks else self.DEFAULT_KS_MIN_SAMPLE_SIZE
        self.ks_sample_divisor = cfg.ks.sample_divisor if cfg.ks else self.DEFAULT_KS_SAMPLE_DIVISOR
        self.interpretation_thresholds = (
            dict(cfg.interpretation.model_dump()) if cfg.interpretation else self.DEFAULT_INTERPRETATION_THRESHOLDS
        )

    @staticmethod
    def _parse_distribution_params(
        cfg: RepresentativenessProcessorConfig,
    ) -> dict[str, dict[str, float]]:
        dist_params: dict[str, dict[str, float]] = {}
        if not cfg.distribution_params:
            return dist_params
        for p in cfg.distribution_params:
            dd: dict[str, float] = {}
            if p.mean is not None:
                dd["mean"] = p.mean
            if p.std is not None:
                dd["std"] = p.std
            if p.min is not None:
                dd["min"] = p.min
            if p.max is not None:
                dd["max"] = p.max
            dist_params[p.column] = dd
        return dist_params

    def _init_from_cfg(self, cfg: RepresentativenessProcessorConfig) -> None:
        self.metrics = list(cfg.metrics)
        self._set_optional_subconfig(cfg)
        self.distribution = cfg.distribution
        self.mean_std_estimation = cfg.mean_std_estimation
        self.expected_counts_method = cfg.expected_counts_method
        self.alpha = cfg.alpha
        self.epsilon = cfg.epsilon
        self.dist_params = self._parse_distribution_params(cfg)

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

        Returns:
            List of output metric column names
        """
        # TODO : manage output metrics names with configuration
        # for now we follow a fixed naming convention
        metrics = []
        for col in self.input_columns:
            if "chi-square" in self.metrics:
                metrics.append(f"{col}_chi-square_p_value")
                metrics.append(f"{col}_chi-square_statistic")
                metrics.append(f"{col}_chi-square_interpretation")
            if "kolmogorov-smirnov" in self.metrics:
                metrics.append(f"{col}_kolmogorov-smirnov_p_value")
                metrics.append(f"{col}_kolmogorov-smirnov_statistic")
                metrics.append(f"{col}_kolmogorov-smirnov_interpretation")
            if "shannon-entropy" in self.metrics:
                metrics.append(f"{col}_shannon-entropy_entropy")
                metrics.append(f"{col}_shannon-entropy_interpretation")
            if "grte" in self.metrics:
                metrics.append(f"{col}_grte_grte_value")
                metrics.append(f"{col}_grte_interpretation")

        return metrics

    # utils functions
    @staticmethod
    def _resolve_device(device: str) -> str:
        """Resolve ``"auto"`` to CUDA if available, else CPU."""
        if device == "auto":
            try:
                import torch

                return "cuda" if torch.cuda.is_available() else "cpu"
            except ImportError:
                return "cpu"
        return device

    @staticmethod
    def _convert_column_to_numeric(feature_array: pa.Array) -> pd.Series | None:
        """Convert a PyArrow column array to a numeric pandas Series with NaN handling.

        Args:
            feature_array: PyArrow array from the batch.

        Returns:
            Numeric pandas Series with NaN dropped, or None if conversion fails.
        """
        try:
            np_col = np.asarray(feature_array.to_numpy(zero_copy_only=False))
        except Exception:
            np_col = pd.Series(feature_array.to_pylist()).to_numpy(copy=True)
        numeric_values = pd.to_numeric(pd.Series(np_col), errors="coerce").dropna()
        return numeric_values if not numeric_values.empty else None

    def _compute_batch_ks_sample(self, numeric_values: pd.Series) -> np.ndarray | None:
        """Compute a random KS sample from numeric values if KS or chi-square is enabled.

        Args:
            numeric_values: Numeric pandas Series from a batch.

        Returns:
            Sampled numpy array, or None if no sampling is needed.
        """
        if "kolmogorov-smirnov" not in self.metrics and "chi-square" not in self.metrics:
            return None

        sample_per_batch = min(
            self.ks_sample_size,
            max(
                self.ks_min_sample_size,
                len(numeric_values) // self.ks_sample_divisor,
            ),
        )
        if len(numeric_values) > sample_per_batch:
            sample_indices = self._rng.choice(len(numeric_values), sample_per_batch, replace=False)
            return np.asarray(numeric_values.iloc[sample_indices])
        return np.asarray(numeric_values)

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

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

        Returns:
            Dictionary containing:
                - {col}_count: Total valid numeric samples.
                - {col}_hist: Histogram counts.
                - {col}_ks_sample: Random subset of data for KS test.
        """
        batch_metrics = {}

        for col in self.input_columns:
            if col not in features:
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue

            numeric_values = self._convert_column_to_numeric(features[col])
            if numeric_values is None:
                logger.warning(f"[{self.name}] column '{col}' has no valid numeric values in this batch")
                continue

            if not self._initialized or col not in self._bin_edges:
                self._initialize_bin_edges(numeric_values.to_numpy(), col)

            edges = self._bin_edges[col]
            hist_counts = np.histogram(numeric_values, bins=edges)[0].astype(np.int64)

            batch_metrics[f"{col}_count"] = pa.array([len(numeric_values)], type=pa.int64())
            batch_metrics[f"{col}_hist"] = pa.FixedSizeListArray.from_arrays(
                hist_counts, list_size=hist_counts.shape[0]
            )

            ks_sample = self._compute_batch_ks_sample(numeric_values)
            if ks_sample is not None:
                batch_metrics[f"{col}_ks_sample"] = pa.array(ks_sample.tolist(), type=pa.float64())

        if not self._initialized and batch_metrics:
            self._initialized = True

        return batch_metrics

    def _get_normal_init_params(self, col: str, sample_data: np.ndarray) -> tuple[float, float]:
        col_params = self.dist_params.get(col, {})
        if self.mean_std_estimation == "user_provided":
            if "mean" not in col_params or "std" not in col_params:
                raise ValueError(
                    f"[{self.name}] user_provided strategy requires 'mean' and 'std' "
                    f"in distribution_params for column '{col}'"
                )
            mean = float(col_params["mean"])
            std = float(col_params["std"])
        else:
            mean = float(col_params.get("mean", np.mean(sample_data)))
            std = float(col_params.get("std", np.std(sample_data, ddof=0)))
        std = std if std > 0.0 else self.epsilon
        return mean, std

    def _get_uniform_init_params(self, col: str, sample_data: np.ndarray) -> tuple[float, float]:
        col_params = self.dist_params.get(col, {})
        if self.mean_std_estimation == "user_provided":
            if "min" not in col_params or "max" not in col_params:
                raise ValueError(
                    f"[{self.name}] user_provided strategy requires 'min' and 'max' "
                    f"in distribution_params for column '{col}'"
                )
            min_val = float(col_params["min"])
            max_val = float(col_params["max"])
        else:
            min_val = float(col_params.get("min", np.min(sample_data)))
            max_val = float(col_params.get("max", np.max(sample_data)))
        if max_val <= min_val:
            max_val = min_val + self.epsilon
        return min_val, max_val

    def _initialize_bin_edges(self, sample_data: np.ndarray, col: str) -> None:
        sample_data = sample_data[np.isfinite(sample_data)]
        if len(sample_data) == 0:
            sample_data = np.array([0.0])
        if self.distribution == "normal":
            mean, std = self._get_normal_init_params(col, sample_data)
            edges = self._bin_edges_normal(mean, std, self.bins)
            self._bin_params[col] = {"mean": mean, "std": std}
        else:
            min_val, max_val = self._get_uniform_init_params(col, sample_data)
            edges = self._bin_edges_uniform(min_val, max_val, self.bins, sample_data)
        self._bin_edges[col] = edges

    def _aggregate_column_metrics(
        self, batch_metrics: dict[str, pa.Array], col: str
    ) -> tuple[int, np.ndarray, np.ndarray] | None:
        """Aggregate histogram and count for a single column across all batches.

        Args:
            batch_metrics: Dictionary of batch-level metrics.
            col: Column name.

        Returns:
            Tuple of (total_count, obs_counts, edges) or None if aggregation fails.
        """
        count_key = f"{col}_count"
        hist_key = f"{col}_hist"
        if count_key not in batch_metrics or hist_key not in batch_metrics:
            logger.warning(f"[{self.name}] no batch metrics for column '{col}'")
            return None

        hist_batch_arrays = np.asarray(batch_metrics[hist_key].to_numpy(zero_copy_only=False))
        if hist_batch_arrays.shape[0] == 0:
            logger.warning(f"[{self.name}] no histogram batch for '{col}'")
            return None

        total_count = int(np.sum(batch_metrics[count_key].to_numpy()))
        hist_arrays = hist_batch_arrays[0].copy()
        for batch_hist in hist_batch_arrays[1:]:
            hist_arrays += batch_hist
        obs_counts = hist_arrays.astype(float)

        if total_count <= 0 or obs_counts.sum() <= 0:
            logger.warning(f"[{self.name}] no valid data for column '{col}'")
            return None

        if col not in self._bin_edges:
            logger.warning(f"[{self.name}] no bin edges for column '{col}' - skipping")
            return None

        return total_count, obs_counts, self._bin_edges[col]

    def _resolve_normal_params(self, col: str, batch_metrics: dict[str, pa.Array]) -> tuple[float, float]:
        if self.mean_std_estimation == "from_first_batch":
            params = self._bin_params.get(col)
            if params is not None:
                return params["mean"], params["std"]
            return self._estimate_normal_params(col, batch_metrics)
        if self.mean_std_estimation == "per_batch":
            return self._estimate_normal_params(col, batch_metrics)
        if self.mean_std_estimation == "user_provided":
            col_params = self.dist_params.get(col, {})
            if "mean" not in col_params or "std" not in col_params:
                raise ValueError(
                    f"[{self.name}] user_provided strategy requires 'mean' and 'std' "
                    f"in distribution_params for column '{col}'"
                )
            std = float(col_params["std"])
            return float(col_params["mean"]), std if std > 0.0 else self.epsilon
        raise NotImplementedError(
            f"[{self.name}] from_all_data strategy requires two-pass processing and is not yet implemented"
        )

    def _compute_expected_counts(
        self,
        col: str,
        batch_metrics: dict[str, pa.Array],
        total_count: int,
        edges: np.ndarray,
    ) -> np.ndarray:
        if self.distribution == "normal":
            mean, std = self._resolve_normal_params(col, batch_metrics)
            if self.expected_counts_method == "cdf":
                expected_probs = stats.norm.cdf(edges[1:], mean, std) - stats.norm.cdf(edges[:-1], mean, std)
                return np.asarray((expected_probs * total_count).astype(np.float64))
            expected_values = self._rng.normal(mean, std, total_count)
        else:
            col_params = self.dist_params.get(col, {})
            min_val = float(col_params.get("min", edges[0]))
            max_val = float(col_params.get("max", edges[-1]))
            if self.expected_counts_method == "cdf":
                expected_probs = stats.uniform.cdf(edges[1:], min_val, max_val - min_val) - stats.uniform.cdf(
                    edges[:-1], min_val, max_val - min_val
                )
                return np.asarray((expected_probs * total_count).astype(np.float64))
            expected_values = self._rng.uniform(min_val, max_val, total_count)
        return np.histogram(expected_values, bins=edges)[0].astype(np.float64)

    def _estimate_from_samples(
        self,
        col_params: dict[str, Any],
        sample_key: str,
        batch_metrics: dict[str, pa.Array],
    ) -> tuple[float, float]:
        """Estimate normal params from KS sample data or fall back to defaults."""
        if sample_key in batch_metrics:
            sample_arrays = batch_metrics[sample_key].to_numpy()
            if sample_arrays.ndim > 1:
                sample_arrays = sample_arrays.flatten()
            mean = float(col_params.get("mean", np.mean(sample_arrays)))
            std = float(col_params.get("std", np.std(sample_arrays, ddof=0)))
        else:
            mean = float(col_params.get("mean", 0.0))
            std = float(col_params.get("std", 1.0))
        return mean, std

    def _estimate_normal_params(self, col: str, batch_metrics: dict[str, pa.Array]) -> tuple[float, float]:
        """Estimate normal distribution parameters from config or KS samples.

        Args:
            col: Column name.
            batch_metrics: Batch-level metrics dict.

        Returns:
            Tuple of (mean, std).
        """
        col_params = self.dist_params.get(col, {})
        if self.mean_std_estimation == "user_provided":
            if "mean" not in col_params or "std" not in col_params:
                raise ValueError(
                    f"[{self.name}] user_provided strategy requires 'mean' and 'std' "
                    f"in distribution_params for column '{col}'"
                )
            mean = float(col_params["mean"])
            std = float(col_params["std"])
        else:
            sample_key = f"{col}_ks_sample"
            mean, std = self._estimate_from_samples(col_params, sample_key, batch_metrics)
        std = std if std > 0.0 else self.epsilon
        return mean, std

    def _compute_chi_square_metric(self, obs_counts: np.ndarray, exp_counts: np.ndarray) -> dict[str, Any]:
        """Compute chi-square goodness-of-fit test between observed and expected counts.

        Args:
            obs_counts: Observed frequency counts per bin.
            exp_counts: Expected frequency counts per bin.

        Returns:
            Dict with p_value, statistic, interpretation keys.
        """
        mask = exp_counts > 0
        if mask.sum() < 2:
            return {
                "p_value": float("nan"),
                "statistic": float("nan"),
                "interpretation": "insufficient_bins",
            }

        obs_sum = obs_counts[mask].sum()
        exp_sum = exp_counts[mask].sum()

        exp_counts_normalized = exp_counts[mask] * (obs_sum / exp_sum)
        try:
            chi = stats.chisquare(f_obs=obs_counts[mask], f_exp=exp_counts_normalized)
            return {
                "p_value": float(chi.pvalue),
                "statistic": float(chi.statistic),
                "interpretation": self.interpretation_thresholds.get(
                    "follows_distribution" if chi.pvalue >= self.alpha else "does_not_follow_distribution",
                    "follows_distribution",
                ),
            }
        except ValueError as e:
            return {
                "p_value": float("nan"),
                "statistic": float("nan"),
                "interpretation": f"chi_square_failed: {e!s}",
                "note": "using observed counts only due to statistical constraints",
            }

    def _compute_ks_metric(self, col: str, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:
        """Compute Kolmogorov-Smirnov test using sampled data.

        Args:
            col: Column name.
            batch_metrics: Batch-level metrics dict containing KS samples.

        Returns:
            Dict with p_value, statistic, interpretation keys.
        """
        sample_key = f"{col}_ks_sample"
        if sample_key not in batch_metrics:
            return {
                "p_value": float("nan"),
                "statistic": float("nan"),
                "interpretation": "no_sample_data_found",
            }

        sample_arrays = batch_metrics[sample_key].to_numpy()
        ks_samples = sample_arrays if sample_arrays.ndim == 1 else sample_arrays.flatten()

        if len(ks_samples) == 0:
            return {
                "p_value": float("nan"),
                "statistic": float("nan"),
                "interpretation": "no_samples_available",
            }

        if self.distribution == "normal":
            mean, std = self._estimate_normal_params(col, batch_metrics)
            ks = stats.kstest(ks_samples, stats.norm.cdf, args=(mean, std))
        else:
            col_params = self.dist_params.get(col, {})
            min_val = float(col_params.get("min", np.min(ks_samples)))
            max_val = float(col_params.get("max", np.max(ks_samples)))
            if max_val <= min_val:
                max_val = min_val + self.epsilon
            ks = stats.kstest(ks_samples, stats.uniform.cdf, args=(min_val, max_val - min_val))

        return {
            "p_value": float(ks.pvalue),
            "statistic": float(ks.statistic),
            "interpretation": self.interpretation_thresholds.get(
                "follows_distribution" if ks.pvalue >= self.alpha else "does_not_follow_distribution",
                "follows_distribution",
            ),
            "sample_size": len(ks_samples),
            "note": "approximated_from_random_samples",
        }

    def _compute_shannon_entropy_metric(self, exp_counts: np.ndarray) -> dict[str, Any]:
        """Compute Shannon entropy from expected frequency counts.

        Args:
            exp_counts: Expected frequency counts per bin.

        Returns:
            Dict with entropy and interpretation.
        """
        p_exp = exp_counts / exp_counts.sum()
        h_exp = float(stats.entropy(p_exp))
        is_high = h_exp > self.shannon_entropy_threshold
        return {
            "entropy": h_exp,
            "interpretation": self.interpretation_thresholds.get(
                "high_diversity" if is_high else "low_diversity",
                "high_diversity",
            ),
        }

    def _compute_grte_metric(self, obs_counts: np.ndarray, exp_counts: np.ndarray) -> dict[str, Any]:
        """Compute GRTE (exponential gap between observed and theoretical entropy).

        Args:
            obs_counts: Observed frequency counts per bin.
            exp_counts: Expected frequency counts per bin.

        Returns:
            Dict with grte_value and interpretation.
        """
        p_obs = obs_counts / obs_counts.sum()
        p_exp = exp_counts / exp_counts.sum()
        h_obs = float(stats.entropy(p_obs))
        h_exp = float(stats.entropy(p_exp))
        grte = float(np.exp(self.grte_scaling_factor * abs(h_exp - h_obs)))
        is_high = grte > self.grte_threshold
        return {
            "grte_value": grte,
            "interpretation": self.interpretation_thresholds.get(
                "high_representativeness" if is_high else "low_representativeness",
                "high_representativeness",
            ),
        }

    @staticmethod
    def _build_compute_metadata(
        total_samples: int,
        batch_metrics: dict[str, pa.Array],
        input_columns: list[str],
        distribution: str,
        metrics: list[str],
        bins: int,
    ) -> str:
        """Build metadata JSON string for compute results.

        Args:
            total_samples: Total number of samples processed.
            batch_metrics: Batch-level metrics dict.
            input_columns: Input column names.
            distribution: Target distribution name.
            metrics: List of computed metric names.
            bins: Number of histogram bins.

        Returns:
            JSON-encoded metadata string.
        """
        return json.dumps(
            {
                "bins": bins,
                "distribution": distribution,
                "metrics_computed": metrics,
                "total_samples": total_samples,
                "columns_analyzed": [c for c in input_columns if f"{c}_count" in batch_metrics],
                "ks_sampling_enabled": "kolmogorov-smirnov" in metrics,
                "note": "KS test uses random sampling approximation for scalability",
            }
        )

    @staticmethod
    def _flatten_col_results(col_res: dict[str, Any], col: str, results: dict[str, Any]) -> None:
        """Flatten nested metric dicts into flat output keys.

        Args:
            col_res: Per-column metric results (potentially nested).
            col: Column name.
            results: Output dict to populate.
        """
        for key, value in col_res.items():
            if isinstance(value, dict):
                for prop, content in value.items():
                    results[f"{col}_{key}_{prop}"] = content
            else:
                results[f"{col}_{key}"] = value

    def _compute_column_results(
        self,
        col: str,
        batch_metrics: dict[str, pa.Array],
        results: dict[str, Any],
    ) -> int:
        """Compute all metrics for a single column and write them to results.

        Args:
            col: Column name.
            batch_metrics: Batch-level metrics dict.
            results: Output dict to populate (mutated in place).

        Returns:
            Number of samples processed (0 if column has no valid data).
        """
        agg = self._aggregate_column_metrics(batch_metrics, col)
        if agg is None:
            return 0

        total_count, obs_counts, edges = agg
        exp_counts = self._compute_expected_counts(col, batch_metrics, total_count, edges)

        col_res: dict[str, Any] = {}

        if "chi-square" in self.metrics:
            col_res["chi-square"] = self._compute_chi_square_metric(obs_counts, exp_counts)

        if "kolmogorov-smirnov" in self.metrics:
            col_res["kolmogorov-smirnov"] = self._compute_ks_metric(col, batch_metrics)

        if "shannon-entropy" in self.metrics:
            col_res["shannon-entropy"] = self._compute_shannon_entropy_metric(exp_counts)

        if "grte" in self.metrics:
            col_res["grte"] = self._compute_grte_metric(obs_counts, exp_counts)

        if col_res:
            self._flatten_col_results(col_res, col, results)

        return total_count

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

        Args:
            batch_metrics: Dictionary of batch-level metrics collected during processing.

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

        results: dict[str, Any] = {}
        total_samples = 0

        for col in self.input_columns:
            total_samples += self._compute_column_results(col, batch_metrics, results)

        results["_metadata"] = self._build_compute_metadata(
            total_samples,
            batch_metrics,
            self.input_columns or [],
            self.distribution,
            self.metrics,
            self.bins,
        )
        return results

    @override
    def reset(self) -> None:
        """Reset processor state for new processing run."""
        self._bin_edges = {}
        self._bin_params = {}
        self._initialized = False

    # utils methods for bin edge calculation

    def _bin_edges_normal(self, mean: float, std: float, bins: int) -> np.ndarray:
        """Calculate bin edges using the PPF of a Normal distribution.

        This ensures bins represent equal probability mass under the
        theoretical distribution. The first and last bins are extended
        to -inf and +inf respectively.
        """
        # logic from dqm-ml v1: use stats.norm.ppf with linspace(1/bins, 1, bins)
        bin_edges_list = [stats.norm.ppf(i / bins, mean, std) for i in range(1, bins)]
        return np.array([-np.inf] + bin_edges_list + [np.inf])

    def _bin_edges_uniform(self, param_min: float, param_max: float, bins: int, data: np.ndarray) -> np.ndarray:
        """
        Calculate linearly spaced bin edges for a Uniform distribution.

        The range is determined by the minimum/maximum of both the configured
        parameters and the actual observed data.
        """
        low_edge = min(param_min, float(np.min(data)))
        high_edge = max(param_max, float(np.max(data)))
        # Handle degenerate case where all data is identical
        if high_edge <= low_edge:
            high_edge = low_edge + self.epsilon
        return np.linspace(low_edge, high_edge, bins + 1)

DEFAULT_ALPHA = 0.05 class-attribute instance-attribute

DEFAULT_EPSILON = 1e-09 class-attribute instance-attribute

DEFAULT_GRTE_THRESHOLD = 0.5 class-attribute instance-attribute

DEFAULT_INTERPRETATION_THRESHOLDS = {'follows_distribution': 'follows_distribution', 'does_not_follow_distribution': 'does_not_follow_distribution', 'high_diversity': 'high_diversity', 'low_diversity': 'low_diversity', 'high_representativeness': 'high_representativeness', 'low_representativeness': 'low_representativeness'} class-attribute instance-attribute

DEFAULT_KS_MIN_SAMPLE_SIZE = 50 class-attribute instance-attribute

DEFAULT_KS_SAMPLE_DIVISOR = 20 class-attribute instance-attribute

DEFAULT_KS_SAMPLE_SIZE = 500 class-attribute instance-attribute

DEFAULT_SHANNON_ENTROPY_THRESHOLD = 2.0 class-attribute instance-attribute

SUPPORTED_DISTS = {'normal', 'uniform'} class-attribute instance-attribute

SUPPORTED_METRICS = {'chi-square', 'grte', 'shannon-entropy', 'kolmogorov-smirnov'} class-attribute instance-attribute

device = self._resolve_device(compute_device) if compute_device else self._resolve_device('cpu') instance-attribute

name = name instance-attribute

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

Initialize the representativeness processor.

Parameters:

Name Type Description Default
name str

Name of the processor.

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

Configuration dictionary containing: - input_columns: List of columns to analyze. - metrics: List of metrics to compute (default: all supported). - bins: Number of bins for histograms (default: 10). - distribution: Target distribution ("normal" or "uniform"). - alpha: Significance level (default: 0.05). - distribution_params: Dictionary of params (e.g., mean, std, min, max).

 None
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/representativeness.py 
 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
def __init__(
    self,
    name: str = "representativeness",
    config: dict[str, Any] | None = None,
) -> None:
    """
    Initialize the representativeness 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).
            - bins: Number of bins for histograms (default: 10).
            - distribution: Target distribution ("normal" or "uniform").
            - alpha: Significance level (default: 0.05).
            - distribution_params: Dictionary of params (e.g., mean, std, min, max).
    """
    super().__init__(name, config)
    self.name = name
    cfg = RepresentativenessProcessorConfig.model_validate({**self.config, "name": self.name})
    self._init_from_cfg(cfg)

    # Use device from compute config if available
    compute_device = getattr(self, "compute_device", None)
    self.device = self._resolve_device(compute_device) if compute_device else self._resolve_device("cpu")

    # Use seed from compute config if available
    compute_seed = getattr(self, "compute_seed", None)
    if compute_seed is not None:
        self._rng = np.random.default_rng(compute_seed)
    else:
        self._rng = np.random.default_rng(42)

    self._bin_edges: dict[str, np.ndarray] = {}
    self._bin_params: dict[str, dict[str, float]] = {}  # caching for _compute_expected_counts
    self._initialized: bool = False

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

Compute final dataset-level metrics by aggregating batch histograms.

Parameters:

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

Dictionary of batch-level metrics collected during processing.

 None

Returns:

Type Description
dict[str, Any]

Dictionary containing final scores and interpretations.
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/representativeness.py 
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
@override
def compute(self, batch_metrics: dict[str, pa.Array] | None = None) -> dict[str, Any]:
    """
    Compute final dataset-level metrics by aggregating batch histograms.

    Args:
        batch_metrics: Dictionary of batch-level metrics collected during processing.

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

    results: dict[str, Any] = {}
    total_samples = 0

    for col in self.input_columns:
        total_samples += self._compute_column_results(col, batch_metrics, results)

    results["_metadata"] = self._build_compute_metadata(
        total_samples,
        batch_metrics,
        self.input_columns or [],
        self.distribution,
        self.metrics,
        self.bins,
    )
    return results

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

Compute partial histogram statistics per batch 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 containing: - {col}_count: Total valid numeric samples. - {col}_hist: Histogram counts. - {col}_ks_sample: Random subset of data for KS test.
Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/representativeness.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@override
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Compute partial histogram statistics per batch for streaming aggregation.

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

    Returns:
        Dictionary containing:
            - {col}_count: Total valid numeric samples.
            - {col}_hist: Histogram counts.
            - {col}_ks_sample: Random subset of data for KS test.
    """
    batch_metrics = {}

    for col in self.input_columns:
        if col not in features:
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue

        numeric_values = self._convert_column_to_numeric(features[col])
        if numeric_values is None:
            logger.warning(f"[{self.name}] column '{col}' has no valid numeric values in this batch")
            continue

        if not self._initialized or col not in self._bin_edges:
            self._initialize_bin_edges(numeric_values.to_numpy(), col)

        edges = self._bin_edges[col]
        hist_counts = np.histogram(numeric_values, bins=edges)[0].astype(np.int64)

        batch_metrics[f"{col}_count"] = pa.array([len(numeric_values)], type=pa.int64())
        batch_metrics[f"{col}_hist"] = pa.FixedSizeListArray.from_arrays(
            hist_counts, list_size=hist_counts.shape[0]
        )

        ks_sample = self._compute_batch_ks_sample(numeric_values)
        if ks_sample is not None:
            batch_metrics[f"{col}_ks_sample"] = pa.array(ks_sample.tolist(), type=pa.float64())

    if not self._initialized and batch_metrics:
        self._initialized = True

    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/representativeness.py 
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
@override
def generated_metrics(self) -> list[str]:
    """
    Return the list of metric columns that will be generated.

    Returns:
        List of output metric column names
    """
    # TODO : manage output metrics names with configuration
    # for now we follow a fixed naming convention
    metrics = []
    for col in self.input_columns:
        if "chi-square" in self.metrics:
            metrics.append(f"{col}_chi-square_p_value")
            metrics.append(f"{col}_chi-square_statistic")
            metrics.append(f"{col}_chi-square_interpretation")
        if "kolmogorov-smirnov" in self.metrics:
            metrics.append(f"{col}_kolmogorov-smirnov_p_value")
            metrics.append(f"{col}_kolmogorov-smirnov_statistic")
            metrics.append(f"{col}_kolmogorov-smirnov_interpretation")
        if "shannon-entropy" in self.metrics:
            metrics.append(f"{col}_shannon-entropy_entropy")
            metrics.append(f"{col}_shannon-entropy_interpretation")
        if "grte" in self.metrics:
            metrics.append(f"{col}_grte_grte_value")
            metrics.append(f"{col}_grte_interpretation")

    return metrics

reset() -> None

Reset processor state for new processing run.

Source code in packages/dqm-ml-core/src/dqm_ml_core/metrics/representativeness.py 
712
713
714
715
716
717
@override
def reset(self) -> None:
    """Reset processor state for new processing run."""
    self._bin_edges = {}
    self._bin_params = {}
    self._initialized = False