Skip to content

dqm_ml_core.metrics.representativeness

Representativeness metric processor for evaluating distribution fit.

This module contains the RepresentativenessProcessor class that evaluates how well a dataset represents a target statistical distribution using various statistical tests.

logger = logging.getLogger(__name__) module-attribute

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