Skip to content

Pipelines & Preprocessing

Composition

online_cp.pipeline.Transformer

Bases: ABC

Base class for conformal-safe feature transformers.

Subclasses must implement :meth:transform (batch) and :meth:transform_one (single vector) and set the mode attribute. Stateful transformers should also override :meth:fit to compute parameters from a batch of training examples.

The | operator composes a transformer with another transformer or a conformal predictor into a :class:Pipeline. The + operator composes two transformers into a :class:TransformerUnion.

Source code in src/online_cp/pipeline.py 
 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
class Transformer(ABC):
    """Base class for conformal-safe feature transformers.

    Subclasses must implement :meth:`transform` (batch) and
    :meth:`transform_one` (single vector) and set the ``mode`` attribute.
    Stateful transformers should also override :meth:`fit` to compute
    parameters from a batch of training examples.

    The ``|`` operator composes a transformer with another transformer or a
    conformal predictor into a :class:`Pipeline`.  The ``+`` operator
    composes two transformers into a :class:`TransformerUnion`.
    """

    mode: str = "fixed"

    def fit(self, X: NDArray) -> None:
        """Fit the transformer on a batch of training examples.

        The default implementation marks the transformer as fitted and returns
        (suitable for stateless ``mode="fixed"`` transformers).  Stateful
        transformers such as :class:`~online_cp.preprocessing.StandardScaler`
        override this method to compute and store their parameters; they should
        call ``super().fit(X)`` to keep the ``_fitted`` flag up to date.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Training batch *after* preceding transformers have been applied.
        """
        self._fitted: bool = True

    @abstractmethod
    def transform(self, X: NDArray) -> NDArray:
        """Apply the transformation to a batch of examples.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Input matrix.

        Returns
        -------
        ndarray
            Transformed matrix, same number of rows as *X*.
        """

    @abstractmethod
    def transform_one(self, x: NDArray) -> NDArray:
        """Apply the transformation to a single example.

        Parameters
        ----------
        x : ndarray of shape (d,)
            Input vector.

        Returns
        -------
        ndarray
            Transformed vector.
        """

    def __or__(self, other: Any) -> Pipeline:
        """Build a :class:`Pipeline` via the ``|`` operator.

        Parameters
        ----------
        other : Transformer or conformal predictor
            The next step.  If *other* is already a :class:`Pipeline`, the
            current transformer is prepended to its steps; otherwise a new
            two-step pipeline is created.

        Returns
        -------
        Pipeline
        """
        if isinstance(other, Pipeline):
            return Pipeline(self, *other.steps)
        return Pipeline(self, other)

    def __add__(self, other: Transformer) -> TransformerUnion:
        """Build a :class:`TransformerUnion` via the ``+`` operator.

        Parameters
        ----------
        other : Transformer
            The transformer whose output will be concatenated with this one.

        Returns
        -------
        TransformerUnion
        """
        return TransformerUnion(self, other)

fit(X: NDArray) -> None

Fit the transformer on a batch of training examples.

The default implementation marks the transformer as fitted and returns (suitable for stateless mode="fixed" transformers). Stateful transformers such as :class:~online_cp.preprocessing.StandardScaler override this method to compute and store their parameters; they should call super().fit(X) to keep the _fitted flag up to date.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Training batch after preceding transformers have been applied.

 required
Source code in src/online_cp/pipeline.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def fit(self, X: NDArray) -> None:
    """Fit the transformer on a batch of training examples.

    The default implementation marks the transformer as fitted and returns
    (suitable for stateless ``mode="fixed"`` transformers).  Stateful
    transformers such as :class:`~online_cp.preprocessing.StandardScaler`
    override this method to compute and store their parameters; they should
    call ``super().fit(X)`` to keep the ``_fitted`` flag up to date.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Training batch *after* preceding transformers have been applied.
    """
    self._fitted: bool = True

transform(X: NDArray) -> NDArray abstractmethod

Apply the transformation to a batch of examples.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Input matrix.

 required

Returns:

Type Description
ndarray

Transformed matrix, same number of rows as X.
Source code in src/online_cp/pipeline.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@abstractmethod
def transform(self, X: NDArray) -> NDArray:
    """Apply the transformation to a batch of examples.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Input matrix.

    Returns
    -------
    ndarray
        Transformed matrix, same number of rows as *X*.
    """

transform_one(x: NDArray) -> NDArray abstractmethod

Apply the transformation to a single example.

Parameters:

Name Type Description Default
x ndarray of shape (d,)

Input vector.

 required

Returns:

Type Description
ndarray

Transformed vector.
Source code in src/online_cp/pipeline.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@abstractmethod
def transform_one(self, x: NDArray) -> NDArray:
    """Apply the transformation to a single example.

    Parameters
    ----------
    x : ndarray of shape (d,)
        Input vector.

    Returns
    -------
    ndarray
        Transformed vector.
    """

__or__(other: Any) -> Pipeline

Build a :class:Pipeline via the | operator.

Parameters:

Name Type Description Default
other Transformer or conformal predictor

The next step. If other is already a :class:Pipeline, the current transformer is prepended to its steps; otherwise a new two-step pipeline is created.

 required

Returns:

Type Description
Pipeline
Source code in src/online_cp/pipeline.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def __or__(self, other: Any) -> Pipeline:
    """Build a :class:`Pipeline` via the ``|`` operator.

    Parameters
    ----------
    other : Transformer or conformal predictor
        The next step.  If *other* is already a :class:`Pipeline`, the
        current transformer is prepended to its steps; otherwise a new
        two-step pipeline is created.

    Returns
    -------
    Pipeline
    """
    if isinstance(other, Pipeline):
        return Pipeline(self, *other.steps)
    return Pipeline(self, other)

__add__(other: Transformer) -> TransformerUnion

Build a :class:TransformerUnion via the + operator.

Parameters:

Name Type Description Default
other Transformer

The transformer whose output will be concatenated with this one.

 required

Returns:

Type Description
TransformerUnion
Source code in src/online_cp/pipeline.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
def __add__(self, other: Transformer) -> TransformerUnion:
    """Build a :class:`TransformerUnion` via the ``+`` operator.

    Parameters
    ----------
    other : Transformer
        The transformer whose output will be concatenated with this one.

    Returns
    -------
    TransformerUnion
    """
    return TransformerUnion(self, other)

online_cp.pipeline.FuncTransformer

Bases: Transformer

Apply a fixed, stateless callable to every example.

Because the function is data-independent, it is always conformally safe: exchangeability of the example sequence is preserved.

The :meth:fit method is a no-op (inherited from :class:Transformer).

Parameters:

Name Type Description Default
fn callable

A callable that accepts an ndarray and returns an ndarray of the same or different shape. It must work on both a 2-D batch (n, d) and a 1-D vector (d,) — numpy ufuncs (e.g. np.log, np.sqrt) and lambda functions satisfy this automatically.

 required

Examples:

>>> import numpy as np
>>> from online_cp import FuncTransformer
>>> ft = FuncTransformer(np.log1p)
>>> ft.transform_one(np.array([0.0, 1.0, 2.0]))
array([0.        , 0.69314718, 1.09861229])
Source code in src/online_cp/pipeline.py 
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
class FuncTransformer(Transformer):
    """Apply a fixed, stateless callable to every example.

    Because the function is data-independent, it is always conformally safe:
    exchangeability of the example sequence is preserved.

    The :meth:`fit` method is a no-op (inherited from :class:`Transformer`).

    Parameters
    ----------
    fn : callable
        A callable that accepts an ndarray and returns an ndarray of the same
        or different shape.  It must work on both a 2-D batch ``(n, d)`` and a
        1-D vector ``(d,)`` — ``numpy`` ufuncs (e.g. ``np.log``, ``np.sqrt``)
        and lambda functions satisfy this automatically.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import FuncTransformer
    >>> ft = FuncTransformer(np.log1p)
    >>> ft.transform_one(np.array([0.0, 1.0, 2.0]))
    array([0.        , 0.69314718, 1.09861229])
    """

    mode: str = "fixed"

    def __init__(self, fn: Any) -> None:
        self.fn = fn

    def transform(self, X: NDArray) -> NDArray:
        return self.fn(X)

    def transform_one(self, x: NDArray) -> NDArray:
        return self.fn(x)

online_cp.pipeline.TransformerUnion

Bases: Transformer

Concatenate the outputs of several transformers along the feature axis.

Each constituent transformer is applied to the same input; their outputs are column-stacked to produce a wider feature matrix. All constituent transformers are fitted independently during :meth:~Pipeline.learn_initial_training_set.

The union's mode is "fixed" when all constituents are fixed, and "frozen" otherwise (most restrictive member wins).

Parameters:

Name Type Description Default
*transformers Transformer

Two or more transformers whose outputs will be concatenated.

 ()

Examples:

>>> import numpy as np
>>> from online_cp import FuncTransformer, Select, TransformerUnion
>>> # Build polynomial-like features: [x, x**2]
>>> union = FuncTransformer(lambda x: x) + FuncTransformer(lambda x: x ** 2)
>>> X = np.arange(6, dtype=float).reshape(3, 2)
>>> union.fit(X)
>>> union.transform(X).shape
(3, 4)
Source code in src/online_cp/pipeline.py 
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
class TransformerUnion(Transformer):
    """Concatenate the outputs of several transformers along the feature axis.

    Each constituent transformer is applied to the same input; their outputs
    are column-stacked to produce a wider feature matrix.  All constituent
    transformers are fitted independently during
    :meth:`~Pipeline.learn_initial_training_set`.

    The union's ``mode`` is ``"fixed"`` when all constituents are fixed, and
    ``"frozen"`` otherwise (most restrictive member wins).

    Parameters
    ----------
    *transformers : Transformer
        Two or more transformers whose outputs will be concatenated.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import FuncTransformer, Select, TransformerUnion
    >>> # Build polynomial-like features: [x, x**2]
    >>> union = FuncTransformer(lambda x: x) + FuncTransformer(lambda x: x ** 2)
    >>> X = np.arange(6, dtype=float).reshape(3, 2)
    >>> union.fit(X)
    >>> union.transform(X).shape
    (3, 4)
    """

    def __init__(self, *transformers: Transformer) -> None:
        if len(transformers) < 2:
            raise ValueError("TransformerUnion requires at least two transformers.")
        if any(t.mode == "bag" for t in transformers):
            raise ValueError(
                "TransformerUnion does not support mode='bag' transformers.  "
                "Bag-mode scalers refit on the augmented bag at each prediction "
                "and cannot be parallelised inside a union."
            )
        self._transformers = list(transformers)
        modes = {t.mode for t in self._transformers}
        self.mode = "frozen" if "frozen" in modes else "fixed"

    def fit(self, X: NDArray) -> None:
        for t in self._transformers:
            t.fit(X)

    def transform(self, X: NDArray) -> NDArray:
        return np.hstack([t.transform(X) for t in self._transformers])

    def transform_one(self, x: NDArray) -> NDArray:
        return np.concatenate([t.transform_one(x) for t in self._transformers])

    def __repr__(self) -> str:
        parts = " + ".join(repr(t) for t in self._transformers)
        return f"TransformerUnion({parts})"

online_cp.pipeline.Select

Bases: Transformer

Keep a subset of input columns.

This is a stateless (mode="fixed"), always-safe transform.

Parameters:

Name Type Description Default
indices array-like of int

Column indices to retain, in the desired output order.

 required

Examples:

>>> import numpy as np
>>> from online_cp import Select
>>> sel = Select([0, 2])
>>> X = np.arange(12, dtype=float).reshape(4, 3)
>>> sel.fit(X)
>>> sel.transform(X)
array([[ 0.,  2.],
       [ 3.,  5.],
       [ 6.,  8.],
       [ 9., 11.]])
Source code in src/online_cp/pipeline.py 
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
745
746
747
748
749
750
751
752
753
class Select(Transformer):
    """Keep a subset of input columns.

    This is a stateless (``mode="fixed"``), always-safe transform.

    Parameters
    ----------
    indices : array-like of int
        Column indices to retain, in the desired output order.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import Select
    >>> sel = Select([0, 2])
    >>> X = np.arange(12, dtype=float).reshape(4, 3)
    >>> sel.fit(X)
    >>> sel.transform(X)
    array([[ 0.,  2.],
           [ 3.,  5.],
           [ 6.,  8.],
           [ 9., 11.]])
    """

    mode: str = "fixed"

    def __init__(self, indices: Any) -> None:
        self._indices = list(indices)

    def transform(self, X: NDArray) -> NDArray:
        return X[:, self._indices]

    def transform_one(self, x: NDArray) -> NDArray:
        return x[self._indices]

    def __repr__(self) -> str:
        return f"Select({self._indices!r})"

online_cp.pipeline.Discard

Bases: Transformer

Drop a subset of input columns, keeping all others.

This is a stateless (mode="fixed"), always-safe transform. The remaining columns are returned in their original order.

Parameters:

Name Type Description Default
indices array-like of int

Column indices to discard.

 required

Examples:

>>> import numpy as np
>>> from online_cp import Discard
>>> drop = Discard([1])
>>> X = np.arange(12, dtype=float).reshape(4, 3)
>>> drop.fit(X)
>>> drop.transform(X)
array([[ 0.,  2.],
       [ 3.,  5.],
       [ 6.,  8.],
       [ 9., 11.]])
Source code in src/online_cp/pipeline.py 
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
class Discard(Transformer):
    """Drop a subset of input columns, keeping all others.

    This is a stateless (``mode="fixed"``), always-safe transform.
    The remaining columns are returned in their original order.

    Parameters
    ----------
    indices : array-like of int
        Column indices to discard.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import Discard
    >>> drop = Discard([1])
    >>> X = np.arange(12, dtype=float).reshape(4, 3)
    >>> drop.fit(X)
    >>> drop.transform(X)
    array([[ 0.,  2.],
           [ 3.,  5.],
           [ 6.,  8.],
           [ 9., 11.]])
    """

    mode: str = "fixed"

    def __init__(self, indices: Any) -> None:
        self._discard = set(indices)

    def _keep(self, d: int) -> list[int]:
        return [i for i in range(d) if i not in self._discard]

    def transform(self, X: NDArray) -> NDArray:
        return X[:, self._keep(X.shape[1])]

    def transform_one(self, x: NDArray) -> NDArray:
        keep = self._keep(x.shape[0])
        return x[keep]

    def __repr__(self) -> str:
        return f"Discard({sorted(self._discard)!r})"

online_cp.pipeline.Pipeline

A conformal-safe pipeline of transformers ending in a conformal predictor.

The pipeline re-exposes the full online-cp predictor API so that a composed object is a drop-in replacement for any bare predictor in :func:~online_cp.evaluate.progressive_val or :func:~online_cp.evaluate.iter_progressive_val.

Validity guarantee. By default the pipeline only accepts transformers whose mode is "fixed" or "frozen"; both regimes preserve conformal validity (ALRW2 §4.5 / §4.7). Transformers with any other mode (e.g. "incremental") break the exchangeability of the example sequence and are rejected at construction time unless you explicitly pass unsafe_incremental=True — which opts out of the finite-sample validity guarantee.

Bare-callable auto-wrap. If a transformer step is a plain callable (rather than a :class:Transformer instance), it is silently wrapped in a :class:FuncTransformer. Non-callable, non-Transformer steps raise a :exc:TypeError.

Parameters:

Name Type Description Default
*steps Any

One or more transformer steps followed by exactly one conformal predictor. At least two steps are required. Each transformer step may be a :class:Transformer instance or a bare callable.

 ()
unsafe_incremental bool

Set to True to allow transformers with modes outside {"fixed", "frozen"}. This voids the conformal validity guarantee.

 False

Examples:

>>> import numpy as np
>>> from online_cp import Pipeline, FuncTransformer, ConformalRidgeRegressor
>>> pipe = Pipeline(FuncTransformer(np.log1p), ConformalRidgeRegressor(a=1.0))
>>> # Equivalent — bare callable is auto-wrapped:
>>> pipe2 = Pipeline(np.log1p, ConformalRidgeRegressor(a=1.0))
>>> # Operator sugar:
>>> pipe3 = FuncTransformer(np.log1p) | ConformalRidgeRegressor(a=1.0)
Source code in src/online_cp/pipeline.py 
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
class Pipeline:
    """A conformal-safe pipeline of transformers ending in a conformal predictor.

    The pipeline re-exposes the full ``online-cp`` predictor API so that a
    composed object is a drop-in replacement for any bare predictor in
    :func:`~online_cp.evaluate.progressive_val` or
    :func:`~online_cp.evaluate.iter_progressive_val`.

    **Validity guarantee.** By default the pipeline only accepts transformers
    whose ``mode`` is ``"fixed"`` or ``"frozen"``; both regimes preserve
    conformal validity (ALRW2 §4.5 / §4.7).  Transformers with any other mode
    (e.g. ``"incremental"``) break the exchangeability of the example sequence
    and are **rejected at construction time** unless you explicitly pass
    ``unsafe_incremental=True`` — which opts out of the finite-sample validity
    guarantee.

    **Bare-callable auto-wrap.** If a transformer step is a plain callable
    (rather than a :class:`Transformer` instance), it is silently wrapped in a
    :class:`FuncTransformer`.  Non-callable, non-Transformer steps raise a
    :exc:`TypeError`.

    Parameters
    ----------
    *steps :
        One or more transformer steps followed by exactly one conformal
        predictor.  At least two steps are required.  Each transformer step
        may be a :class:`Transformer` instance or a bare callable.
    unsafe_incremental : bool, default False
        Set to ``True`` to allow transformers with modes outside
        ``{"fixed", "frozen"}``.  This voids the conformal validity guarantee.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import Pipeline, FuncTransformer, ConformalRidgeRegressor
    >>> pipe = Pipeline(FuncTransformer(np.log1p), ConformalRidgeRegressor(a=1.0))
    >>> # Equivalent — bare callable is auto-wrapped:
    >>> pipe2 = Pipeline(np.log1p, ConformalRidgeRegressor(a=1.0))
    >>> # Operator sugar:
    >>> pipe3 = FuncTransformer(np.log1p) | ConformalRidgeRegressor(a=1.0)
    """

    def __init__(self, *steps: Any, unsafe_incremental: bool = False) -> None:
        if len(steps) < 2:
            raise ValueError(
                "Pipeline requires at least two steps: one or more transformers "
                "followed by a conformal predictor."
            )
        # Auto-wrap bare callables; reject non-callable non-Transformers.
        transformer_steps = []
        for i, step in enumerate(steps[:-1]):
            if isinstance(step, Transformer):
                transformer_steps.append(step)
            elif callable(step):
                transformer_steps.append(FuncTransformer(step))
            else:
                raise TypeError(
                    f"Pipeline step {i} is neither a Transformer nor a callable "
                    f"(got {type(step).__name__!r}).  Wrap it in a FuncTransformer "
                    "or provide a callable."
                )
        # Validity guard: reject non-sound modes unless opted in.
        if not unsafe_incremental:
            for t in transformer_steps:
                if t.mode not in _SOUND_MODES:
                    raise ValueError(
                        f"Transformer {t!r} has mode={t.mode!r}, which is not in "
                        f"the set of sound modes {set(_SOUND_MODES)}.  "
                        "Such transformers may break the exchangeability of the "
                        "example sequence and invalidate the conformal coverage "
                        "guarantee (ALRW2 §4.5).  "
                        "Pass unsafe_incremental=True to Pipeline to opt in, "
                        "accepting that finite-sample validity is no longer assured."
                    )
        self.transformers: list[Transformer] = transformer_steps
        self.estimator: Any = steps[-1]
        self.steps: tuple[Any, ...] = (tuple(transformer_steps) + (steps[-1],))
        self._unsafe_incremental = unsafe_incremental

        # Bag-mode support.
        self._bag_mode: bool = any(t.mode == "bag" for t in transformer_steps)
        if self._bag_mode:
            frozen_names = [
                repr(t) for t in transformer_steps if t.mode == "frozen"
            ]
            if frozen_names:
                raise ValueError(
                    "Mixing mode='frozen' and mode='bag' transformers in a Pipeline "
                    "is not supported (v1).  Frozen transformers are fitted once on the "
                    "training set while bag transformers refit on the augmented object "
                    "bag at each prediction.  Use only fixed + bag transformers, or "
                    "only fixed + frozen transformers.  Offending frozen transformers: "
                    + ", ".join(frozen_names)
                )
            # Pristine template — never trained; deep-copied on each predict so
            # predict() is read-only and does not mutate persistent estimator state.
            self._template: Any = copy.deepcopy(steps[-1])
            self._X_raw: NDArray | None = None
            self._y_raw: NDArray | None = None

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _transform_batch(self, X: NDArray) -> NDArray:
        """Apply all transformers to a batch matrix."""
        Xt = X
        for t in self.transformers:
            Xt = t.transform(Xt)
        return Xt

    def _transform_one(self, x: NDArray) -> NDArray:
        """Apply all transformers to a single vector."""
        xt = x
        for t in self.transformers:
            xt = t.transform_one(xt)
        return xt

    def _transform_bag(self, X_aug: NDArray) -> NDArray:
        """Apply the transformer chain to the augmented bag *X_aug*.

        For ``mode="bag"`` transformers, :meth:`~Transformer.fit` is called on
        the current (possibly partially-transformed) augmented matrix before
        :meth:`~Transformer.transform` is applied.  Stateless ``mode="fixed"``
        transformers simply call :meth:`~Transformer.transform` directly.

        Parameters
        ----------
        X_aug : ndarray of shape (n+1, d)
            Augmented matrix — training rows followed by the test object.

        Returns
        -------
        ndarray of shape (n+1, d')
            Fully-transformed augmented matrix.
        """
        Xt = X_aug
        for t in self.transformers:
            if t.mode == "bag":
                t.fit(Xt)
            Xt = t.transform(Xt)
        return Xt

    def _predict_bag(self, x: NDArray, method: str, *args: Any, **kwargs: Any) -> Any:
        """Bag-fit predict path.

        Builds the augmented object bag ``[X_raw, x]``, refits all bag-mode
        transformers symmetrically, then predicts with a fresh deep-copy of the
        pristine estimator template.  The persistent estimator (``self.estimator``)
        and raw bag (``self._X_raw``) are never mutated, so ``predict`` is
        read-only and safe inside :func:`~online_cp.evaluate.progressive_val`.

        When no training data has been accumulated yet, the estimator is left
        untrained, which causes it to return the default CP full-label-space
        prediction (``(-inf, +inf)`` for regressors; full ``label_space`` for
        classifiers) — preserving validity with no data.

        Parameters
        ----------
        x : ndarray of shape (d,)
            Test object.
        method : str
            Name of the estimator method to call (``"predict"``,
            ``"predict_cpd"``, or ``"compute_p_value"``).
        *args
            Positional arguments forwarded to ``estimator.<method>``.
        **kwargs
            Keyword arguments forwarded to ``estimator.<method>``.
        """
        x2d = np.atleast_2d(x)
        has_data = self._X_raw is not None and len(self._X_raw) > 0
        if has_data:
            X_aug = np.vstack([self._X_raw, x2d])
        else:
            X_aug = x2d
        Xt = self._transform_bag(X_aug)
        est = copy.deepcopy(self._template)
        if has_data:
            X_scaled, x_scaled = Xt[:-1], Xt[-1]
            est.learn_initial_training_set(X_scaled, self._y_raw)
        else:
            x_scaled = Xt[0]
        return getattr(est, method)(x_scaled, *args, **kwargs)

    # ------------------------------------------------------------------
    # Training API
    # ------------------------------------------------------------------

    def learn_initial_training_set(self, X: NDArray, y: NDArray) -> None:
        """Fit all transformers then delegate to the estimator's initial fit.

        For each transformer in order: ``fit(Xt)`` is called to let stateful
        transformers (e.g. frozen scalers) compute their parameters, then
        ``transform(Xt)`` is applied to produce the input for the next step.
        The estimator receives the fully-transformed ``Xt``.

        Parameters
        ----------
        X : ndarray of shape (n, d)
        y : ndarray of shape (n,)
        """
        if self._bag_mode:
            # Store the raw training bag; transformers will refit at predict time.
            # Still call fit() on non-bag (fixed) transformers so that _fitted=True
            # is set for summary() and shape validation happens early.
            self._X_raw = np.atleast_2d(X).copy()
            self._y_raw = np.asarray(y).copy()
            Xt = X
            for t in self.transformers:
                if t.mode != "bag":
                    t.fit(Xt)
                    Xt = t.transform(Xt)
            # Do NOT fit the estimator here — it is refitted fresh on every predict.
        else:
            Xt = X
            for t in self.transformers:
                t.fit(Xt)
                Xt = t.transform(Xt)
            self.estimator.learn_initial_training_set(Xt, y)

    def learn_one(self, x: NDArray, y: Any, precomputed: Any = None) -> None:
        """Transform *x* and delegate to the estimator's online update.

        ``precomputed`` is always dropped at the pipeline boundary: the
        transform changes the geometry so any cached intermediate is invalid.

        Parameters
        ----------
        x : ndarray of shape (d,)
        y : label / target value
        precomputed : ignored
            Accepted for API compatibility; always discarded.
        """
        if self._bag_mode:
            # Append raw (x, y) to the stored bag; estimator is NOT updated here —
            # it will be refitted from scratch on the next predict call.
            x2d = np.atleast_2d(x)
            if self._X_raw is None or len(self._X_raw) == 0:
                self._X_raw = x2d.copy()
                self._y_raw = np.array([y])
            else:
                self._X_raw = np.vstack([self._X_raw, x2d])
                self._y_raw = np.append(self._y_raw, y)
        else:
            xt = self._transform_one(x)
            self.estimator.learn_one(xt, y)

    # ------------------------------------------------------------------
    # Prediction API
    # ------------------------------------------------------------------

    def predict(self, x: NDArray, **kwargs: Any) -> Any:
        """Transform *x* then call ``estimator.predict``.

        All keyword arguments (``epsilon``, ``return_p_values``, ``bounds``,
        ``return_update``, …) are forwarded unchanged.

        Parameters
        ----------
        x : ndarray of shape (d,)
        **kwargs
            Forwarded to ``estimator.predict``.
        """
        if self._bag_mode:
            return self._predict_bag(x, "predict", **kwargs)
        xt = self._transform_one(x)
        return self.estimator.predict(xt, **kwargs)

    def predict_cpd(self, x: NDArray, **kwargs: Any) -> Any:
        """Transform *x* then call ``estimator.predict_cpd``.

        Parameters
        ----------
        x : ndarray of shape (d,)
        **kwargs
            Forwarded to ``estimator.predict_cpd``.
        """
        if self._bag_mode:
            return self._predict_bag(x, "predict_cpd", **kwargs)
        xt = self._transform_one(x)
        return self.estimator.predict_cpd(xt, **kwargs)

    def compute_p_value(self, x: NDArray, y: Any, **kwargs: Any) -> Any:
        """Transform *x* then call ``estimator.compute_p_value``.

        Parameters
        ----------
        x : ndarray of shape (d,)
        y : label / target value
        **kwargs
            Forwarded to ``estimator.compute_p_value``.
        """
        if self._bag_mode:
            return self._predict_bag(x, "compute_p_value", y, **kwargs)
        xt = self._transform_one(x)
        return self.estimator.compute_p_value(xt, y, **kwargs)

    # ------------------------------------------------------------------
    # Attribute forwarding
    # ------------------------------------------------------------------

    def __getattr__(self, name: str) -> Any:
        # Only called when normal attribute lookup fails, so self.transformers
        # and self.estimator are always found via the normal path.
        try:
            return getattr(self.__dict__["estimator"], name)
        except KeyError:
            raise AttributeError(name) from None

    # ------------------------------------------------------------------
    # Operator sugar
    # ------------------------------------------------------------------

    def summary(self) -> dict:
        """Return a human-readable summary of the pipeline structure.

        Returns
        -------
        dict
            A dict with the following keys:

            ``n_steps`` : int
                Total number of steps (transformers + estimator).
            ``transformers`` : list of dict
                One entry per transformer with keys ``type`` (class name),
                ``mode`` (e.g. ``"fixed"``, ``"frozen"``, or ``"bag"``), and
                ``fitted`` (``True`` after :meth:`learn_initial_training_set`
                has run).
            ``estimator`` : dict
                ``{"type": <class name>}``.
            ``unsafe_incremental`` : bool
                Whether the validity guard was disabled at construction.
            ``bag_mode`` : bool
                ``True`` when at least one transformer uses ``mode="bag"``;
                in that case the pipeline operates in bag-fit mode.

        Examples
        --------
        >>> import numpy as np
        >>> from online_cp import Pipeline, FuncTransformer, ConformalRidgeRegressor
        >>> pipe = Pipeline(FuncTransformer(np.abs), ConformalRidgeRegressor(a=1.0))
        >>> pipe.summary()["n_steps"]
        2
        """
        def _is_fitted(t: Transformer) -> bool:
            # A transformer is fitted if the base fit() set _fitted,
            # OR if it has frozen-scaler attributes (mean_, data_min_).
            return bool(
                getattr(t, "_fitted", False)
                or getattr(t, "mean_", None) is not None
                or getattr(t, "data_min_", None) is not None
            )

        return {
            "n_steps": len(self.transformers) + 1,
            "transformers": [
                {
                    "type": type(t).__name__,
                    "mode": t.mode,
                    "fitted": _is_fitted(t),
                }
                for t in self.transformers
            ],
            "estimator": {"type": type(self.estimator).__name__},
            "unsafe_incremental": self._unsafe_incremental,
            "bag_mode": self._bag_mode,
        }

    def __or__(self, other: Any) -> Pipeline:
        """Append a step to this pipeline via the ``|`` operator."""
        return Pipeline(*self.steps, other,
                        unsafe_incremental=self._unsafe_incremental)

    def save(self, filepath: str | os.PathLike, *, compress: int = 3) -> None:
        """Save this pipeline to *filepath*.

        All transformers and the estimator are serialised with joblib/pickle.
        :class:`FuncTransformer` functions **must** be module-level named
        functions (not lambdas); lambdas will raise
        :class:`~online_cp.SerializationError` at pickle time.

        .. warning::
            Only load files from **trusted sources**.
        """
        import joblib

        try:
            from online_cp import __version__ as _lib_version
        except Exception:
            _lib_version = "unknown"

        envelope = {
            "format_version": 1,
            "library_version": _lib_version,
            "class": f"{type(self).__module__}.{type(self).__qualname__}",
            "transformers": self.transformers,
            "estimator": self.estimator,
            "unsafe_incremental": self._unsafe_incremental,
            "_X_raw": getattr(self, "_X_raw", None),
            "_y_raw": getattr(self, "_y_raw", None),
            "_template": getattr(self, "_template", None),
        }
        try:
            joblib.dump(envelope, filepath, compress=compress)
        except Exception as exc:
            raise SerializationError(
                f"Failed to save pipeline to {filepath!r}: {exc}. "
                "Ensure all FuncTransformer functions are module-level named functions, "
                "not lambdas."
            ) from exc

    @classmethod
    def load(cls, filepath: str | os.PathLike) -> Pipeline:
        """Load a pipeline from *filepath*.

        .. warning::
            Only load files from **trusted sources**.
        """
        import warnings as _warnings

        import joblib

        try:
            from online_cp import __version__ as _lib_version
        except Exception:
            _lib_version = "unknown"

        try:
            envelope = joblib.load(filepath)
        except Exception as exc:
            raise SerializationError(
                f"Failed to read pipeline file {filepath!r}: {exc}"
            ) from exc

        fmt_ver = envelope.get("format_version", 0)
        if fmt_ver > 1:
            raise SerializationError(
                f"Unsupported format_version {fmt_ver}. Update online-cp to load this file."
            )
        lib_ver = envelope.get("library_version", "unknown")
        if lib_ver != _lib_version:
            _warnings.warn(
                f"Pipeline was saved with online-cp {lib_ver!r}, "
                f"but you are using {_lib_version!r}. Predictions may differ.",
                UserWarning,
                stacklevel=2,
            )
        expected = f"{cls.__module__}.{cls.__qualname__}"
        if envelope.get("class", "") != expected:
            raise SerializationError(
                f"Class mismatch: file contains '{envelope.get('class')}', "
                f"expected '{expected}'."
            )

        steps = list(envelope["transformers"]) + [envelope["estimator"]]
        pipe = cls(*steps, unsafe_incremental=envelope["unsafe_incremental"])
        # Restore training bag for bag-mode pipelines
        if envelope.get("_X_raw") is not None:
            pipe._X_raw = envelope["_X_raw"]
            pipe._y_raw = envelope["_y_raw"]
        if envelope.get("_template") is not None:
            pipe._template = envelope["_template"]
        return pipe

    def __repr__(self) -> str:
        step_reprs = " | ".join(repr(s) for s in self.steps)
        return f"Pipeline({step_reprs})"

learn_initial_training_set(X: NDArray, y: NDArray) -> None

Fit all transformers then delegate to the estimator's initial fit.

For each transformer in order: fit(Xt) is called to let stateful transformers (e.g. frozen scalers) compute their parameters, then transform(Xt) is applied to produce the input for the next step. The estimator receives the fully-transformed Xt.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)
required
y ndarray of shape (n,)
required
Source code in src/online_cp/pipeline.py 
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
def learn_initial_training_set(self, X: NDArray, y: NDArray) -> None:
    """Fit all transformers then delegate to the estimator's initial fit.

    For each transformer in order: ``fit(Xt)`` is called to let stateful
    transformers (e.g. frozen scalers) compute their parameters, then
    ``transform(Xt)`` is applied to produce the input for the next step.
    The estimator receives the fully-transformed ``Xt``.

    Parameters
    ----------
    X : ndarray of shape (n, d)
    y : ndarray of shape (n,)
    """
    if self._bag_mode:
        # Store the raw training bag; transformers will refit at predict time.
        # Still call fit() on non-bag (fixed) transformers so that _fitted=True
        # is set for summary() and shape validation happens early.
        self._X_raw = np.atleast_2d(X).copy()
        self._y_raw = np.asarray(y).copy()
        Xt = X
        for t in self.transformers:
            if t.mode != "bag":
                t.fit(Xt)
                Xt = t.transform(Xt)
        # Do NOT fit the estimator here — it is refitted fresh on every predict.
    else:
        Xt = X
        for t in self.transformers:
            t.fit(Xt)
            Xt = t.transform(Xt)
        self.estimator.learn_initial_training_set(Xt, y)

learn_one(x: NDArray, y: Any, precomputed: Any = None) -> None

Transform x and delegate to the estimator's online update.

precomputed is always dropped at the pipeline boundary: the transform changes the geometry so any cached intermediate is invalid.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required
y label / target value
required
precomputed ignored

Accepted for API compatibility; always discarded.

 None
Source code in src/online_cp/pipeline.py 
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
def learn_one(self, x: NDArray, y: Any, precomputed: Any = None) -> None:
    """Transform *x* and delegate to the estimator's online update.

    ``precomputed`` is always dropped at the pipeline boundary: the
    transform changes the geometry so any cached intermediate is invalid.

    Parameters
    ----------
    x : ndarray of shape (d,)
    y : label / target value
    precomputed : ignored
        Accepted for API compatibility; always discarded.
    """
    if self._bag_mode:
        # Append raw (x, y) to the stored bag; estimator is NOT updated here —
        # it will be refitted from scratch on the next predict call.
        x2d = np.atleast_2d(x)
        if self._X_raw is None or len(self._X_raw) == 0:
            self._X_raw = x2d.copy()
            self._y_raw = np.array([y])
        else:
            self._X_raw = np.vstack([self._X_raw, x2d])
            self._y_raw = np.append(self._y_raw, y)
    else:
        xt = self._transform_one(x)
        self.estimator.learn_one(xt, y)

predict(x: NDArray, **kwargs: Any) -> Any

Transform x then call estimator.predict.

All keyword arguments (epsilon, return_p_values, bounds, return_update, …) are forwarded unchanged.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required
**kwargs Any

Forwarded to estimator.predict.

 {}
Source code in src/online_cp/pipeline.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def predict(self, x: NDArray, **kwargs: Any) -> Any:
    """Transform *x* then call ``estimator.predict``.

    All keyword arguments (``epsilon``, ``return_p_values``, ``bounds``,
    ``return_update``, …) are forwarded unchanged.

    Parameters
    ----------
    x : ndarray of shape (d,)
    **kwargs
        Forwarded to ``estimator.predict``.
    """
    if self._bag_mode:
        return self._predict_bag(x, "predict", **kwargs)
    xt = self._transform_one(x)
    return self.estimator.predict(xt, **kwargs)

predict_cpd(x: NDArray, **kwargs: Any) -> Any

Transform x then call estimator.predict_cpd.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required
**kwargs Any

Forwarded to estimator.predict_cpd.

 {}
Source code in src/online_cp/pipeline.py 
455
456
457
458
459
460
461
462
463
464
465
466
467
def predict_cpd(self, x: NDArray, **kwargs: Any) -> Any:
    """Transform *x* then call ``estimator.predict_cpd``.

    Parameters
    ----------
    x : ndarray of shape (d,)
    **kwargs
        Forwarded to ``estimator.predict_cpd``.
    """
    if self._bag_mode:
        return self._predict_bag(x, "predict_cpd", **kwargs)
    xt = self._transform_one(x)
    return self.estimator.predict_cpd(xt, **kwargs)

compute_p_value(x: NDArray, y: Any, **kwargs: Any) -> Any

Transform x then call estimator.compute_p_value.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required
y label / target value
required
**kwargs Any

Forwarded to estimator.compute_p_value.

 {}
Source code in src/online_cp/pipeline.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
def compute_p_value(self, x: NDArray, y: Any, **kwargs: Any) -> Any:
    """Transform *x* then call ``estimator.compute_p_value``.

    Parameters
    ----------
    x : ndarray of shape (d,)
    y : label / target value
    **kwargs
        Forwarded to ``estimator.compute_p_value``.
    """
    if self._bag_mode:
        return self._predict_bag(x, "compute_p_value", y, **kwargs)
    xt = self._transform_one(x)
    return self.estimator.compute_p_value(xt, y, **kwargs)

summary() -> dict

Return a human-readable summary of the pipeline structure.

Returns:

Type Description
dict

A dict with the following keys:

n_steps : int Total number of steps (transformers + estimator). transformers : list of dict One entry per transformer with keys type (class name), mode (e.g. "fixed", "frozen", or "bag"), and fitted (True after :meth:learn_initial_training_set has run). estimator : dict {"type": <class name>}. unsafe_incremental : bool Whether the validity guard was disabled at construction. bag_mode : bool True when at least one transformer uses mode="bag"; in that case the pipeline operates in bag-fit mode.

Examples:

>>> import numpy as np
>>> from online_cp import Pipeline, FuncTransformer, ConformalRidgeRegressor
>>> pipe = Pipeline(FuncTransformer(np.abs), ConformalRidgeRegressor(a=1.0))
>>> pipe.summary()["n_steps"]
2
Source code in src/online_cp/pipeline.py 
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
def summary(self) -> dict:
    """Return a human-readable summary of the pipeline structure.

    Returns
    -------
    dict
        A dict with the following keys:

        ``n_steps`` : int
            Total number of steps (transformers + estimator).
        ``transformers`` : list of dict
            One entry per transformer with keys ``type`` (class name),
            ``mode`` (e.g. ``"fixed"``, ``"frozen"``, or ``"bag"``), and
            ``fitted`` (``True`` after :meth:`learn_initial_training_set`
            has run).
        ``estimator`` : dict
            ``{"type": <class name>}``.
        ``unsafe_incremental`` : bool
            Whether the validity guard was disabled at construction.
        ``bag_mode`` : bool
            ``True`` when at least one transformer uses ``mode="bag"``;
            in that case the pipeline operates in bag-fit mode.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import Pipeline, FuncTransformer, ConformalRidgeRegressor
    >>> pipe = Pipeline(FuncTransformer(np.abs), ConformalRidgeRegressor(a=1.0))
    >>> pipe.summary()["n_steps"]
    2
    """
    def _is_fitted(t: Transformer) -> bool:
        # A transformer is fitted if the base fit() set _fitted,
        # OR if it has frozen-scaler attributes (mean_, data_min_).
        return bool(
            getattr(t, "_fitted", False)
            or getattr(t, "mean_", None) is not None
            or getattr(t, "data_min_", None) is not None
        )

    return {
        "n_steps": len(self.transformers) + 1,
        "transformers": [
            {
                "type": type(t).__name__,
                "mode": t.mode,
                "fitted": _is_fitted(t),
            }
            for t in self.transformers
        ],
        "estimator": {"type": type(self.estimator).__name__},
        "unsafe_incremental": self._unsafe_incremental,
        "bag_mode": self._bag_mode,
    }

__or__(other: Any) -> Pipeline

Append a step to this pipeline via the | operator.

Source code in src/online_cp/pipeline.py 
555
556
557
558
def __or__(self, other: Any) -> Pipeline:
    """Append a step to this pipeline via the ``|`` operator."""
    return Pipeline(*self.steps, other,
                    unsafe_incremental=self._unsafe_incremental)

save(filepath: str | os.PathLike, *, compress: int = 3) -> None

Save this pipeline to filepath.

All transformers and the estimator are serialised with joblib/pickle. :class:FuncTransformer functions must be module-level named functions (not lambdas); lambdas will raise :class:~online_cp.SerializationError at pickle time.

.. warning:: Only load files from trusted sources.

Source code in src/online_cp/pipeline.py 
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
def save(self, filepath: str | os.PathLike, *, compress: int = 3) -> None:
    """Save this pipeline to *filepath*.

    All transformers and the estimator are serialised with joblib/pickle.
    :class:`FuncTransformer` functions **must** be module-level named
    functions (not lambdas); lambdas will raise
    :class:`~online_cp.SerializationError` at pickle time.

    .. warning::
        Only load files from **trusted sources**.
    """
    import joblib

    try:
        from online_cp import __version__ as _lib_version
    except Exception:
        _lib_version = "unknown"

    envelope = {
        "format_version": 1,
        "library_version": _lib_version,
        "class": f"{type(self).__module__}.{type(self).__qualname__}",
        "transformers": self.transformers,
        "estimator": self.estimator,
        "unsafe_incremental": self._unsafe_incremental,
        "_X_raw": getattr(self, "_X_raw", None),
        "_y_raw": getattr(self, "_y_raw", None),
        "_template": getattr(self, "_template", None),
    }
    try:
        joblib.dump(envelope, filepath, compress=compress)
    except Exception as exc:
        raise SerializationError(
            f"Failed to save pipeline to {filepath!r}: {exc}. "
            "Ensure all FuncTransformer functions are module-level named functions, "
            "not lambdas."
        ) from exc

load(filepath: str | os.PathLike) -> Pipeline classmethod

Load a pipeline from filepath.

.. warning:: Only load files from trusted sources.

Source code in src/online_cp/pipeline.py 
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
@classmethod
def load(cls, filepath: str | os.PathLike) -> Pipeline:
    """Load a pipeline from *filepath*.

    .. warning::
        Only load files from **trusted sources**.
    """
    import warnings as _warnings

    import joblib

    try:
        from online_cp import __version__ as _lib_version
    except Exception:
        _lib_version = "unknown"

    try:
        envelope = joblib.load(filepath)
    except Exception as exc:
        raise SerializationError(
            f"Failed to read pipeline file {filepath!r}: {exc}"
        ) from exc

    fmt_ver = envelope.get("format_version", 0)
    if fmt_ver > 1:
        raise SerializationError(
            f"Unsupported format_version {fmt_ver}. Update online-cp to load this file."
        )
    lib_ver = envelope.get("library_version", "unknown")
    if lib_ver != _lib_version:
        _warnings.warn(
            f"Pipeline was saved with online-cp {lib_ver!r}, "
            f"but you are using {_lib_version!r}. Predictions may differ.",
            UserWarning,
            stacklevel=2,
        )
    expected = f"{cls.__module__}.{cls.__qualname__}"
    if envelope.get("class", "") != expected:
        raise SerializationError(
            f"Class mismatch: file contains '{envelope.get('class')}', "
            f"expected '{expected}'."
        )

    steps = list(envelope["transformers"]) + [envelope["estimator"]]
    pipe = cls(*steps, unsafe_incremental=envelope["unsafe_incremental"])
    # Restore training bag for bag-mode pipelines
    if envelope.get("_X_raw") is not None:
        pipe._X_raw = envelope["_X_raw"]
        pipe._y_raw = envelope["_y_raw"]
    if envelope.get("_template") is not None:
        pipe._template = envelope["_template"]
    return pipe

Preprocessing

online_cp.preprocessing.StandardScaler

Bases: Transformer

Standardise features to zero mean and unit variance.

Parameters are computed once from the initial training batch (via :meth:fit) and held constant thereafter (mode="frozen"). This preserves training-conditional conformal validity.

Zero-variance features are left unchanged (their scale is set to 1 to avoid division by zero).

Parameters:

Name Type Description Default
with_mean bool

Subtract the training mean from each feature.

 True
with_std bool

Divide each feature by its training standard deviation.

 True

Attributes:

Name Type Description
mean_ ndarray of shape (d,) or None

Per-feature training mean. None before :meth:fit is called.
scale_ ndarray of shape (d,) or None

Per-feature training standard deviation (with zero-variance guard). None before :meth:fit is called.

Examples:

>>> import numpy as np
>>> from online_cp import StandardScaler, Pipeline, ConformalRidgeRegressor
>>> rng = np.random.default_rng(42)
>>> X = rng.normal(loc=5.0, scale=10.0, size=(50, 3))
>>> y = X[:, 0] + rng.normal(scale=0.5, size=50)
>>> pipe = StandardScaler() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
>>> pipe.learn_initial_training_set(X, y)
>>> interval = pipe.predict(X[0], epsilon=0.1)
Source code in src/online_cp/preprocessing.py 
 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
class StandardScaler(Transformer):
    """Standardise features to zero mean and unit variance.

    Parameters are computed once from the initial training batch (via
    :meth:`fit`) and held constant thereafter (``mode="frozen"``).  This
    preserves training-conditional conformal validity.

    Zero-variance features are left unchanged (their scale is set to 1 to
    avoid division by zero).

    Parameters
    ----------
    with_mean : bool, default True
        Subtract the training mean from each feature.
    with_std : bool, default True
        Divide each feature by its training standard deviation.

    Attributes
    ----------
    mean_ : ndarray of shape (d,) or None
        Per-feature training mean.  ``None`` before :meth:`fit` is called.
    scale_ : ndarray of shape (d,) or None
        Per-feature training standard deviation (with zero-variance guard).
        ``None`` before :meth:`fit` is called.

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import StandardScaler, Pipeline, ConformalRidgeRegressor
    >>> rng = np.random.default_rng(42)
    >>> X = rng.normal(loc=5.0, scale=10.0, size=(50, 3))
    >>> y = X[:, 0] + rng.normal(scale=0.5, size=50)
    >>> pipe = StandardScaler() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
    >>> pipe.learn_initial_training_set(X, y)
    >>> interval = pipe.predict(X[0], epsilon=0.1)
    """

    mode: str = "frozen"  # class-level default; overridden by instance in __init__

    def __init__(self, with_mean: bool = True, with_std: bool = True, mode: str = "frozen") -> None:
        if mode not in ("frozen", "bag"):
            raise ValueError(
                f"StandardScaler mode must be 'frozen' or 'bag', got {mode!r}."
            )
        self.mode = mode  # instance attribute
        self.with_mean = with_mean
        self.with_std = with_std
        self.mean_: NDArray | None = None
        self.scale_: NDArray | None = None

    def fit(self, X: NDArray) -> None:
        """Compute mean and standard deviation from *X*.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Training batch.
        """
        self.mean_ = X.mean(axis=0) if self.with_mean else np.zeros(X.shape[1])
        if self.with_std:
            std = X.std(axis=0)
            std[std == 0.0] = 1.0  # guard: zero-variance feature → no scaling
            self.scale_ = std
        else:
            self.scale_ = np.ones(X.shape[1])

    def _check_fitted(self) -> None:
        if self.mean_ is None or self.scale_ is None:
            raise RuntimeError(
                "StandardScaler has not been fitted yet. "
                "Call Pipeline.learn_initial_training_set() before predict()."
            )

    def transform(self, X: NDArray) -> NDArray:
        self._check_fitted()
        return (X - self.mean_) / self.scale_

    def transform_one(self, x: NDArray) -> NDArray:
        self._check_fitted()
        return (x - self.mean_) / self.scale_

    def __repr__(self) -> str:
        return f"StandardScaler(with_mean={self.with_mean}, with_std={self.with_std}, mode={self.mode!r})"

fit(X: NDArray) -> None

Compute mean and standard deviation from X.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Training batch.

 required
Source code in src/online_cp/preprocessing.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def fit(self, X: NDArray) -> None:
    """Compute mean and standard deviation from *X*.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Training batch.
    """
    self.mean_ = X.mean(axis=0) if self.with_mean else np.zeros(X.shape[1])
    if self.with_std:
        std = X.std(axis=0)
        std[std == 0.0] = 1.0  # guard: zero-variance feature → no scaling
        self.scale_ = std
    else:
        self.scale_ = np.ones(X.shape[1])

online_cp.preprocessing.MinMaxScaler

Bases: Transformer

Scale features to a fixed range [feature_range[0], feature_range[1]].

Parameters are computed once from the initial training batch (via :meth:fit) and held constant thereafter (mode="frozen").

Features whose training range is zero (constant column) are mapped to the lower bound of feature_range without raising an error.

Parameters:

Name Type Description Default
feature_range tuple of (float, float)

Target range for the scaled features.

 (0, 1)

Attributes:

Name Type Description
data_min_ ndarray of shape (d,) or None

Per-feature minimum over the training set.
data_range_ ndarray of shape (d,) or None

Per-feature max - min over the training set (zero-range guarded).

Examples:

>>> import numpy as np
>>> from online_cp import MinMaxScaler, Pipeline, ConformalRidgeRegressor
>>> rng = np.random.default_rng(0)
>>> X = rng.uniform(low=-100.0, high=100.0, size=(50, 2))
>>> y = X[:, 0] * 0.5 + rng.normal(scale=1.0, size=50)
>>> pipe = MinMaxScaler() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
>>> pipe.learn_initial_training_set(X, y)
>>> interval = pipe.predict(X[0], epsilon=0.1)
Source code in src/online_cp/preprocessing.py 
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
class MinMaxScaler(Transformer):
    """Scale features to a fixed range ``[feature_range[0], feature_range[1]]``.

    Parameters are computed once from the initial training batch (via
    :meth:`fit`) and held constant thereafter (``mode="frozen"``).

    Features whose training range is zero (constant column) are mapped to the
    lower bound of ``feature_range`` without raising an error.

    Parameters
    ----------
    feature_range : tuple of (float, float), default (0, 1)
        Target range for the scaled features.

    Attributes
    ----------
    data_min_ : ndarray of shape (d,) or None
        Per-feature minimum over the training set.
    data_range_ : ndarray of shape (d,) or None
        Per-feature ``max - min`` over the training set (zero-range guarded).

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import MinMaxScaler, Pipeline, ConformalRidgeRegressor
    >>> rng = np.random.default_rng(0)
    >>> X = rng.uniform(low=-100.0, high=100.0, size=(50, 2))
    >>> y = X[:, 0] * 0.5 + rng.normal(scale=1.0, size=50)
    >>> pipe = MinMaxScaler() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
    >>> pipe.learn_initial_training_set(X, y)
    >>> interval = pipe.predict(X[0], epsilon=0.1)
    """

    mode: str = "frozen"  # class-level default; overridden by instance in __init__

    def __init__(self, feature_range: tuple[float, float] = (0.0, 1.0), mode: str = "frozen") -> None:
        lo, hi = feature_range
        if lo >= hi:
            raise ValueError(
                f"feature_range must satisfy lo < hi, got ({lo}, {hi})."
            )
        if mode not in ("frozen", "bag"):
            raise ValueError(
                f"MinMaxScaler mode must be 'frozen' or 'bag', got {mode!r}."
            )
        self.mode = mode  # instance attribute
        self.feature_range = feature_range
        self.data_min_: NDArray | None = None
        self.data_range_: NDArray | None = None

    def fit(self, X: NDArray) -> None:
        """Compute min and range from *X*.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Training batch.
        """
        self.data_min_ = X.min(axis=0)
        data_range = X.max(axis=0) - self.data_min_
        data_range[data_range == 0.0] = 1.0  # guard: constant feature
        self.data_range_ = data_range

    def _check_fitted(self) -> None:
        if self.data_min_ is None or self.data_range_ is None:
            raise RuntimeError(
                "MinMaxScaler has not been fitted yet. "
                "Call Pipeline.learn_initial_training_set() before predict()."
            )

    def _scale(self, v: NDArray) -> NDArray:
        lo, hi = self.feature_range
        return lo + (v - self.data_min_) / self.data_range_ * (hi - lo)

    def transform(self, X: NDArray) -> NDArray:
        self._check_fitted()
        return self._scale(X)

    def transform_one(self, x: NDArray) -> NDArray:
        self._check_fitted()
        return self._scale(x)

    def __repr__(self) -> str:
        return f"MinMaxScaler(feature_range={self.feature_range!r}, mode={self.mode!r})"

fit(X: NDArray) -> None

Compute min and range from X.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Training batch.

 required
Source code in src/online_cp/preprocessing.py 
177
178
179
180
181
182
183
184
185
186
187
188
def fit(self, X: NDArray) -> None:
    """Compute min and range from *X*.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Training batch.
    """
    self.data_min_ = X.min(axis=0)
    data_range = X.max(axis=0) - self.data_min_
    data_range[data_range == 0.0] = 1.0  # guard: constant feature
    self.data_range_ = data_range

online_cp.preprocessing.PCA

Bases: Transformer, SerializableMixin

Rotate features into their principal-component basis.

Parameters are computed once from the initial training batch (via :meth:fit) and held constant thereafter (mode="frozen"), or recomputed at each prediction from the augmented bag (mode="bag").

mode="frozen" preserves training-conditional conformal validity (parameters are a fixed function of the training data). mode="bag" recomputes the rotation from the label-free augmented bag [X_train, x_test] at each prediction, preserving exact finite-sample conformal validity at the cost of O(n·d²+d³) per predict.

The transformation is the standard PCA projection: subtract the training mean, then project onto the top-k eigenvectors of the (unbiased) sample covariance matrix. Eigenvectors are ordered by descending explained variance and given a deterministic sign (largest-magnitude element positive).

Intended use-case: axis-aligning features before :class:~online_cp.mondrian.MondrianConformalRegressor / :class:~online_cp.mondrian.MondrianConformalClassifier. Mondrian methods partition the feature space by hyperplanes; after PCA rotation the axes align with the directions of greatest variance, yielding tighter and more balanced partitions.

Parameters:

Name Type Description Default
n_components int or None

Number of principal components to retain. None keeps all components (min(n-1, d) after fitting n samples with d features).

 None
mode ('frozen', 'bag')

See module docstring for the two operation modes.

 "frozen"

Attributes:

Name Type Description
n_ int or None

Number of samples seen during :meth:fit.
mean_ ndarray of shape (d,) or None

Per-feature training mean.
components_ ndarray of shape (k, d) or None

Principal components as row vectors (sorted by descending variance).
singular_values_ ndarray of shape (k,) or None

Square roots of the retained eigenvalues (≈ singular values of the centred data matrix, up to the n-1 denominator).

Examples:

>>> import numpy as np
>>> from online_cp import PCA, Pipeline, ConformalRidgeRegressor
>>> rng = np.random.default_rng(0)
>>> X = rng.normal(loc=[0.0, 0.0], scale=[10.0, 0.1], size=(60, 2))
>>> y = X[:, 0] * 0.5 + rng.normal(scale=0.1, size=60)
>>> pipe = PCA() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
>>> pipe.learn_initial_training_set(X, y)
>>> interval = pipe.predict(X[0], epsilon=0.1)
Source code in src/online_cp/preprocessing.py 
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
class PCA(Transformer, SerializableMixin):
    """Rotate features into their principal-component basis.

    Parameters are computed once from the initial training batch (via
    :meth:`fit`) and held constant thereafter (``mode="frozen"``), or
    recomputed at each prediction from the augmented bag (``mode="bag"``).

    ``mode="frozen"`` preserves *training-conditional conformal validity*
    (parameters are a fixed function of the training data).  ``mode="bag"``
    recomputes the rotation from the label-free augmented bag ``[X_train,
    x_test]`` at each prediction, preserving *exact finite-sample conformal
    validity* at the cost of O(n·d²+d³) per predict.

    The transformation is the standard PCA projection: subtract the training
    mean, then project onto the top-*k* eigenvectors of the (unbiased) sample
    covariance matrix.  Eigenvectors are ordered by descending explained
    variance and given a deterministic sign (largest-magnitude element
    positive).

    **Intended use-case**: axis-aligning features before
    :class:`~online_cp.mondrian.MondrianConformalRegressor` /
    :class:`~online_cp.mondrian.MondrianConformalClassifier`.  Mondrian
    methods partition the feature space by hyperplanes; after PCA rotation the
    axes align with the directions of greatest variance, yielding tighter and
    more balanced partitions.

    Parameters
    ----------
    n_components : int or None, default None
        Number of principal components to retain.  ``None`` keeps all
        components (``min(n-1, d)`` after fitting *n* samples with *d*
        features).
    mode : {"frozen", "bag"}, default "frozen"
        See module docstring for the two operation modes.

    Attributes
    ----------
    n_ : int or None
        Number of samples seen during :meth:`fit`.
    mean_ : ndarray of shape (d,) or None
        Per-feature training mean.
    components_ : ndarray of shape (k, d) or None
        Principal components as row vectors (sorted by descending variance).
    singular_values_ : ndarray of shape (k,) or None
        Square roots of the retained eigenvalues (≈ singular values of the
        centred data matrix, up to the ``n-1`` denominator).

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import PCA, Pipeline, ConformalRidgeRegressor
    >>> rng = np.random.default_rng(0)
    >>> X = rng.normal(loc=[0.0, 0.0], scale=[10.0, 0.1], size=(60, 2))
    >>> y = X[:, 0] * 0.5 + rng.normal(scale=0.1, size=60)
    >>> pipe = PCA() | ConformalRidgeRegressor(a=1.0, epsilon=0.1)
    >>> pipe.learn_initial_training_set(X, y)
    >>> interval = pipe.predict(X[0], epsilon=0.1)
    """

    mode: str = "frozen"

    _SAVE_PARAMS: tuple[str, ...] = ("n_components", "mode")
    _SAVE_STATE: tuple[str, ...] = ("n_", "mean_", "components_", "singular_values_")

    def __init__(
        self,
        n_components: int | None = None,
        mode: str = "frozen",
    ) -> None:
        if mode not in ("frozen", "bag"):
            raise ValueError(
                f"PCA mode must be 'frozen' or 'bag', got {mode!r}."
            )
        if n_components is not None and n_components < 1:
            raise ValueError(
                f"n_components must be a positive integer or None, got {n_components!r}."
            )
        self.mode = mode
        self.n_components = n_components
        self.n_: int | None = None
        self.mean_: NDArray | None = None
        self.components_: NDArray | None = None
        self.singular_values_: NDArray | None = None

    def fit(self, X: NDArray) -> None:
        """Compute principal components from *X*.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Training batch.  Requires ``n ≥ 2``.
        """
        n, d = X.shape
        if n < 2:
            raise ValueError(
                f"PCA requires at least 2 samples to fit, got {n}."
            )
        self.mean_ = X.mean(axis=0)
        X_c = X - self.mean_
        cov = X_c.T @ X_c / (n - 1)  # (d, d) unbiased sample covariance

        vals, vecs = np.linalg.eigh(cov)  # ascending order, real symmetric
        idx = np.argsort(vals)[::-1]      # descending variance order

        k_max = min(n - 1, d)
        k = min(self.n_components, k_max) if self.n_components is not None else k_max

        components = vecs[:, idx[:k]].T   # (k, d)
        self.components_ = _sign_flip_components(components)
        self.singular_values_ = np.sqrt(np.maximum(vals[idx[:k]], 0.0))
        self.n_ = n

    def _check_fitted(self) -> None:
        if self.components_ is None:
            raise RuntimeError(
                "PCA has not been fitted yet. "
                "Call Pipeline.learn_initial_training_set() before predict()."
            )

    def transform(self, X: NDArray) -> NDArray:
        """Project *X* onto the principal-component basis.

        Parameters
        ----------
        X : ndarray of shape (n, d)

        Returns
        -------
        ndarray of shape (n, k)
        """
        self._check_fitted()
        return (X - self.mean_) @ self.components_.T

    def transform_one(self, x: NDArray) -> NDArray:
        """Project a single feature vector *x*.

        Parameters
        ----------
        x : ndarray of shape (d,)

        Returns
        -------
        ndarray of shape (k,)
        """
        self._check_fitted()
        return (x - self.mean_) @ self.components_.T

    def __repr__(self) -> str:
        return f"PCA(n_components={self.n_components!r}, mode={self.mode!r})"

fit(X: NDArray) -> None

Compute principal components from X.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Training batch. Requires n ≥ 2.

 required
Source code in src/online_cp/preprocessing.py 
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
def fit(self, X: NDArray) -> None:
    """Compute principal components from *X*.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Training batch.  Requires ``n ≥ 2``.
    """
    n, d = X.shape
    if n < 2:
        raise ValueError(
            f"PCA requires at least 2 samples to fit, got {n}."
        )
    self.mean_ = X.mean(axis=0)
    X_c = X - self.mean_
    cov = X_c.T @ X_c / (n - 1)  # (d, d) unbiased sample covariance

    vals, vecs = np.linalg.eigh(cov)  # ascending order, real symmetric
    idx = np.argsort(vals)[::-1]      # descending variance order

    k_max = min(n - 1, d)
    k = min(self.n_components, k_max) if self.n_components is not None else k_max

    components = vecs[:, idx[:k]].T   # (k, d)
    self.components_ = _sign_flip_components(components)
    self.singular_values_ = np.sqrt(np.maximum(vals[idx[:k]], 0.0))
    self.n_ = n

transform(X: NDArray) -> NDArray

Project X onto the principal-component basis.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)
required

Returns:

Type Description
ndarray of shape (n, k)
Source code in src/online_cp/preprocessing.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
def transform(self, X: NDArray) -> NDArray:
    """Project *X* onto the principal-component basis.

    Parameters
    ----------
    X : ndarray of shape (n, d)

    Returns
    -------
    ndarray of shape (n, k)
    """
    self._check_fitted()
    return (X - self.mean_) @ self.components_.T

transform_one(x: NDArray) -> NDArray

Project a single feature vector x.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required

Returns:

Type Description
ndarray of shape (k,)
Source code in src/online_cp/preprocessing.py 
375
376
377
378
379
380
381
382
383
384
385
386
387
def transform_one(self, x: NDArray) -> NDArray:
    """Project a single feature vector *x*.

    Parameters
    ----------
    x : ndarray of shape (d,)

    Returns
    -------
    ndarray of shape (k,)
    """
    self._check_fitted()
    return (x - self.mean_) @ self.components_.T

online_cp.preprocessing.SVD

Bases: Transformer, SerializableMixin

Project features onto the top-k right singular vectors of X.

Computes the eigen-decomposition of the (optionally centred) Gram matrix X^T X and projects data onto the leading eigenvectors. This is equivalent to truncated SVD on the data matrix and is the standard approach for dimensionality reduction and multicollinearity removal before ridge-based methods.

When center=True the Gram matrix is built from the mean-centred data X - mean(X), making this identical to :class:PCA in the projected subspace. Use center=False when the data is already mean-zero or when you want the uncentred factorisation (e.g. for non-negative data).

Intended use-case: dimensionality reduction and multicollinearity removal before :class:~online_cp.cps.RidgePredictionMachine and :class:~online_cp.regressors.ConformalRidgeRegressor. Reducing d to k < d speeds up the O(d³) matrix inversion and stabilises it when features are near-collinear.

Parameters:

Name Type Description Default
n_components int or None

Number of right singular vectors to retain. None keeps all (min(n-1, d) when center=True, or min(n, d) when center=False).

 None
mode ('frozen', 'bag')

See module docstring for the two operation modes.

 "frozen"
center bool

If True, subtract the training mean before computing the Gram matrix (uses unbiased n-1 denominator). If False, use the raw data (uses n denominator).

 True

Attributes:

Name Type Description
n_ int or None

Number of samples seen during :meth:fit.
mean_ ndarray of shape (d,) or None

Per-feature training mean. None when center=False.
components_ ndarray of shape (k, d) or None

Right singular vectors as row vectors (sorted by descending singular value).
singular_values_ ndarray of shape (k,) or None

Retained singular values (square roots of the retained eigenvalues of the Gram matrix).

Examples:

>>> import numpy as np
>>> from online_cp import SVD, Pipeline, ConformalRidgeRegressor
>>> rng = np.random.default_rng(1)
>>> X = rng.normal(size=(50, 5))
>>> X[:, 2] = X[:, 0] + 0.01 * rng.normal(size=50)  # near-collinear
>>> y = X[:, 0] - X[:, 1] + rng.normal(scale=0.1, size=50)
>>> pipe = SVD(n_components=3) | ConformalRidgeRegressor(a=1e-3, epsilon=0.1)
>>> pipe.learn_initial_training_set(X, y)
>>> interval = pipe.predict(X[0], epsilon=0.1)
Source code in src/online_cp/preprocessing.py 
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
class SVD(Transformer, SerializableMixin):
    """Project features onto the top-*k* right singular vectors of *X*.

    Computes the eigen-decomposition of the (optionally centred) Gram matrix
    ``X^T X`` and projects data onto the leading eigenvectors.  This is
    equivalent to truncated SVD on the data matrix and is the standard
    approach for dimensionality reduction and multicollinearity removal before
    ridge-based methods.

    When ``center=True`` the Gram matrix is built from the mean-centred data
    ``X - mean(X)``, making this *identical* to :class:`PCA` in the projected
    subspace.  Use ``center=False`` when the data is already mean-zero or when
    you want the uncentred factorisation (e.g. for non-negative data).

    **Intended use-case**: dimensionality reduction and multicollinearity
    removal before :class:`~online_cp.cps.RidgePredictionMachine` and
    :class:`~online_cp.regressors.ConformalRidgeRegressor`.  Reducing *d* to
    *k < d* speeds up the O(d³) matrix inversion and stabilises it when
    features are near-collinear.

    Parameters
    ----------
    n_components : int or None, default None
        Number of right singular vectors to retain.  ``None`` keeps all
        (``min(n-1, d)`` when ``center=True``, or ``min(n, d)`` when
        ``center=False``).
    mode : {"frozen", "bag"}, default "frozen"
        See module docstring for the two operation modes.
    center : bool, default True
        If ``True``, subtract the training mean before computing the Gram
        matrix (uses unbiased ``n-1`` denominator).  If ``False``, use the raw
        data (uses ``n`` denominator).

    Attributes
    ----------
    n_ : int or None
        Number of samples seen during :meth:`fit`.
    mean_ : ndarray of shape (d,) or None
        Per-feature training mean.  ``None`` when ``center=False``.
    components_ : ndarray of shape (k, d) or None
        Right singular vectors as row vectors (sorted by descending singular
        value).
    singular_values_ : ndarray of shape (k,) or None
        Retained singular values (square roots of the retained eigenvalues of
        the Gram matrix).

    Examples
    --------
    >>> import numpy as np
    >>> from online_cp import SVD, Pipeline, ConformalRidgeRegressor
    >>> rng = np.random.default_rng(1)
    >>> X = rng.normal(size=(50, 5))
    >>> X[:, 2] = X[:, 0] + 0.01 * rng.normal(size=50)  # near-collinear
    >>> y = X[:, 0] - X[:, 1] + rng.normal(scale=0.1, size=50)
    >>> pipe = SVD(n_components=3) | ConformalRidgeRegressor(a=1e-3, epsilon=0.1)
    >>> pipe.learn_initial_training_set(X, y)
    >>> interval = pipe.predict(X[0], epsilon=0.1)
    """

    mode: str = "frozen"

    _SAVE_PARAMS: tuple[str, ...] = ("n_components", "mode", "center")
    _SAVE_STATE: tuple[str, ...] = ("n_", "mean_", "components_", "singular_values_")

    def __init__(
        self,
        n_components: int | None = None,
        mode: str = "frozen",
        center: bool = True,
    ) -> None:
        if mode not in ("frozen", "bag"):
            raise ValueError(
                f"SVD mode must be 'frozen' or 'bag', got {mode!r}."
            )
        if n_components is not None and n_components < 1:
            raise ValueError(
                f"n_components must be a positive integer or None, got {n_components!r}."
            )
        self.mode = mode
        self.n_components = n_components
        self.center = center
        self.n_: int | None = None
        self.mean_: NDArray | None = None
        self.components_: NDArray | None = None
        self.singular_values_: NDArray | None = None

    def fit(self, X: NDArray) -> None:
        """Compute right singular vectors from *X*.

        Parameters
        ----------
        X : ndarray of shape (n, d)
            Training batch.  Requires ``n ≥ 2``.
        """
        n, d = X.shape
        if n < 2:
            raise ValueError(
                f"SVD requires at least 2 samples to fit, got {n}."
            )
        if self.center:
            self.mean_ = X.mean(axis=0)
            X_c = X - self.mean_
            gram = X_c.T @ X_c / (n - 1)  # unbiased
        else:
            self.mean_ = None
            X_c = X
            gram = X_c.T @ X_c / n        # biased (n denominator)

        vals, vecs = np.linalg.eigh(gram)  # ascending order
        idx = np.argsort(vals)[::-1]       # descending singular-value order

        k_max = min(n - 1, d) if self.center else min(n, d)
        k = min(self.n_components, k_max) if self.n_components is not None else k_max

        components = vecs[:, idx[:k]].T    # (k, d)
        self.components_ = _sign_flip_components(components)
        self.singular_values_ = np.sqrt(np.maximum(vals[idx[:k]], 0.0))
        self.n_ = n

    def _check_fitted(self) -> None:
        if self.components_ is None:
            raise RuntimeError(
                "SVD has not been fitted yet. "
                "Call Pipeline.learn_initial_training_set() before predict()."
            )

    def transform(self, X: NDArray) -> NDArray:
        """Project *X* onto the retained right singular vectors.

        Parameters
        ----------
        X : ndarray of shape (n, d)

        Returns
        -------
        ndarray of shape (n, k)
        """
        self._check_fitted()
        if self.center:
            return (X - self.mean_) @ self.components_.T
        return X @ self.components_.T

    def transform_one(self, x: NDArray) -> NDArray:
        """Project a single feature vector *x*.

        Parameters
        ----------
        x : ndarray of shape (d,)

        Returns
        -------
        ndarray of shape (k,)
        """
        self._check_fitted()
        if self.center:
            return (x - self.mean_) @ self.components_.T
        return x @ self.components_.T

    def __repr__(self) -> str:
        return (
            f"SVD(n_components={self.n_components!r}, "
            f"mode={self.mode!r}, center={self.center!r})"
        )

fit(X: NDArray) -> None

Compute right singular vectors from X.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)

Training batch. Requires n ≥ 2.

 required
Source code in src/online_cp/preprocessing.py 
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
def fit(self, X: NDArray) -> None:
    """Compute right singular vectors from *X*.

    Parameters
    ----------
    X : ndarray of shape (n, d)
        Training batch.  Requires ``n ≥ 2``.
    """
    n, d = X.shape
    if n < 2:
        raise ValueError(
            f"SVD requires at least 2 samples to fit, got {n}."
        )
    if self.center:
        self.mean_ = X.mean(axis=0)
        X_c = X - self.mean_
        gram = X_c.T @ X_c / (n - 1)  # unbiased
    else:
        self.mean_ = None
        X_c = X
        gram = X_c.T @ X_c / n        # biased (n denominator)

    vals, vecs = np.linalg.eigh(gram)  # ascending order
    idx = np.argsort(vals)[::-1]       # descending singular-value order

    k_max = min(n - 1, d) if self.center else min(n, d)
    k = min(self.n_components, k_max) if self.n_components is not None else k_max

    components = vecs[:, idx[:k]].T    # (k, d)
    self.components_ = _sign_flip_components(components)
    self.singular_values_ = np.sqrt(np.maximum(vals[idx[:k]], 0.0))
    self.n_ = n

transform(X: NDArray) -> NDArray

Project X onto the retained right singular vectors.

Parameters:

Name Type Description Default
X ndarray of shape (n, d)
required

Returns:

Type Description
ndarray of shape (n, k)
Source code in src/online_cp/preprocessing.py 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
def transform(self, X: NDArray) -> NDArray:
    """Project *X* onto the retained right singular vectors.

    Parameters
    ----------
    X : ndarray of shape (n, d)

    Returns
    -------
    ndarray of shape (n, k)
    """
    self._check_fitted()
    if self.center:
        return (X - self.mean_) @ self.components_.T
    return X @ self.components_.T

transform_one(x: NDArray) -> NDArray

Project a single feature vector x.

Parameters:

Name Type Description Default
x ndarray of shape (d,)
required

Returns:

Type Description
ndarray of shape (k,)
Source code in src/online_cp/preprocessing.py 
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def transform_one(self, x: NDArray) -> NDArray:
    """Project a single feature vector *x*.

    Parameters
    ----------
    x : ndarray of shape (d,)

    Returns
    -------
    ndarray of shape (k,)
    """
    self._check_fitted()
    if self.center:
        return (x - self.mean_) @ self.components_.T
    return x @ self.components_.T