rolling_conformal

Defined in fynance.models.conformal

rolling_conformal(model_factory, X, y, *, train=252, cal=63, test=63, alpha=0.1)[source]

Walk-forward split-conformal prediction intervals.

Rolls a strictly causal train / calibrate / test window over (X, y): at each step, a fresh model (from model_factory()) is fit on train bars, calibrated (à la ConformalWrapper) on the next cal bars, and used to emit point predictions and intervals on the next test bars; the window then advances by test bars (the step size), so consecutive test windows never overlap. Every prediction and interval reported for bar t depends only on bars strictly before the test window containing t – perturbing X/y at or after a test window leaves every earlier output bit-for-bit unchanged.

fynance.data.split.walk_forward is not reused here because it only generates a 2-way (train/test) split; the 3-way train/calibrate/test arithmetic needed for conformal calibration is implemented directly below.

Parameters:
model_factorycallable

No-argument callable returning a fresh model exposing fit(X, y) / predict(X) (the SignalModel contract) each time it is called – one fresh model per window, so no state (and no information) leaks across windows.

Xarray-like

Feature matrix, shape (T, N), time-ordered.

yarray-like

Target, shape (T,) or (T, 1), aligned with X.

train, cal, testint, optional

Train, calibration and test window lengths in bars. Defaults 252/63/63 (roughly one trading year of training, one quarter of calibration, one quarter of test).

alphafloat, optional

Miscoverage level in (0, 1); each window’s interval targets marginal coverage 1 - alpha. Default 0.1.

Returns:
dict
  • 'pred', 'lo', 'hi' : numpy.ndarray, shape (T,) – point prediction and interval bounds, NaN outside any test window (i.e. before the first window’s test slice, and in any trailing bars too short to form one more full window).

  • 'covered' : numpy.ndarray, shape (T,)1.0 where lo <= y <= hi, 0.0 where not covered, NaN outside any test window (NaN-aware boolean mask).

  • 'coverage' : float – nanmean of 'covered', the realized marginal coverage over all test bars; NaN if no test window was ever produced (train + cal + test > T).

Raises:
ValueError

If train, cal or test is not strictly positive, or if alpha is not in (0, 1).

See also

ConformalWrapper, fynance.data.split.walk_forward

Examples

>>> import numpy as np
>>> from fynance.models.conformal import rolling_conformal
>>> class LinearModel:
...     ''' Closed-form ordinary least squares. '''
...     def fit(self, X, y):
...         self.coef_ = np.linalg.lstsq(X, y, rcond=None)[0]
...         return self
...     def predict(self, X):
...         return X @ self.coef_
>>> rng = np.random.default_rng(0)
>>> T, F = 600, 2
>>> X = rng.standard_normal((T, F))
>>> w_true = np.array([1.0, -1.0])
>>> y = X @ w_true + 0.1 * rng.standard_normal(T)
>>> result = rolling_conformal(
...     LinearModel, X, y, train=200, cal=50, test=50, alpha=0.1,
... )
>>> sorted(result.keys())
['coverage', 'covered', 'hi', 'lo', 'pred']
>>> result['pred'].shape
(600,)
>>> 0.7 < result['coverage'] <= 1.0
True