Skip to content

dqm_ml_core.metrics

Metric implementations for DQM ML Core.

This package contains concrete implementations of data quality metrics: - CompletenessProcessor: Evaluates data completeness (non-null ratios) - RepresentativenessProcessor: Evaluates distribution representativeness using statistical tests (chi-square, KS, entropy, GRTE)

__all__ = ['CompletenessProcessor', '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

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