_RollingBasis¶

Defined in fynance.models.rolling

class _RollingBasis(X, y, f=None, index=None)[source]

Bases: object

Base object to roll a model over a time axis.

At each step the model trains on X[t-n:t] and predicts on X[t:t+s]. Call set_roll_period (or __call__) to configure the window sizes, then iterate with run.

The leading underscore signals that this class is internal — its _train, _get_loss_on and sub_predict hooks are expected to be overridden by a concrete subclass mixed with a BaseNeuralNet descendant. The public, stable entry point is RollMultiLayerPerceptron. Other ready-to-use combinations can be added by following the same pattern (multiple inheritance + set_roll_period instead of __call__, since the latter is captured by torch.nn.Module).

Parameters:

X, yarray_like: Respectively input and output data, shaped (T, N) and (T, M). Strict temporal ordering is required — no shuffling, no future leakage.
fcallable, optional: Function to transform target, e.g. torch.sign.
indexarray_like, optional: Time index of data.

Attributes:

n, s, rint: Respectively size of training, testing and rolling period.
b, e, Tint: Respectively batch size, number of epochs and size of entire dataset.
t, _e, iint: Respectively the current time period, the current epoch and the current iteration.
n_iterint: Total number of iterations.
y_eval, y_testnp.ndarray: Respectively evaluating and testing predictions.
loglist of dict: Per-step record of {step, train_loss, eval_loss, test_loss}, populated by run. Use get_stats to get a DataFrame.

__call__(train_period, test_period, start=0, end=None, roll_period=None, eval_period=None, batch_size=64, epochs=1)[source]

Configure rolling window parameters.

Parameters:

train_period, test_periodint: Size of respectively training and testing sub-periods.
startint, optional: Starting observation, default is first observation.
endint, optional: Ending observation, default is last observation.
roll_periodint, optional: Size of the rolling period, default equals test_period.
eval_periodint, optional: Size of the evaluating period (unused, kept for API compat).
batch_sizeint, optional: Training batch size, default is 64.
epochsint, optional: Number of epochs per sub-period, default is 1.

Returns:

_RollingBasis

cross_validate(model_factory, X, y, metric_fn=None, epochs=1)[source]

Walk-forward cross-validation with out-of-fold predictions.

At each fold a fresh model is created via model_factory(), trained on the rolling training window, and used to predict the next out-of-sample window. Results are accumulated across all folds without any state leaking between them.

Call __call__ (or set_roll_period for RollMultiLayerPerceptron) to configure train_period, test_period, and roll_period before calling this method.

Parameters:

model_factorycallable: Called with no arguments before every fold. Must return an object that exposes train_on(X, y) and predict(X) -> NDArray | Tensor (the BaseNeuralNet interface).
X, yarray_like: Input and target arrays shaped (T, N) and (T, M).
metric_fncallable, optional: metric_fn(y_true, y_pred) -> float evaluated on the test window of each fold. If None, CVResult.fold_metrics is an empty list and the mean/std fields are None.
epochsint, optional: Number of full training passes per fold, default 1.

Returns:

CVResult

Examples

>>> import numpy as np
>>> import torch, torch.nn as nn
>>> from fynance.models.mlp import MultiLayerPerceptron
>>> from fynance.models.rolling import _RollingBasis
>>> rng = np.random.default_rng(0)
>>> X = rng.standard_normal((80, 4)).astype(np.float32)
>>> y = rng.standard_normal((80, 1)).astype(np.float32)
>>> Xt, yt = torch.from_numpy(X), torch.from_numpy(y)
>>> def factory():
...     m = MultiLayerPerceptron(4, 1, layers=[8])
...     m.set_optimizer(nn.MSELoss, torch.optim.Adam, lr=1e-3)
...     return m
>>> rb = _RollingBasis(Xt, yt)
>>> _ = rb(train_period=40, test_period=10, roll_period=10)
>>> result = rb.cross_validate(factory, Xt, yt)
>>> result.oof_predictions.shape
(80, 1)

get_stats()[source]

Return per-step loss history as a DataFrame.

Returns:

pd.DataFrame: Columns: step, train_loss, eval_loss, test_loss.

plot_loss(figsize=(9, 4))[source]

Plot train / eval / test loss curves.

Parameters:

figsizetuple of int, optional

Returns:

matplotlib.figure.Figure

run(backtest_plot=True, backtest_kpi=True, figsize=(9, 6), func=np.sign)[source]

Run the rolling model and collect backtest predictions.

Parameters:

backtest_plotbool, optional: If True, display a live backtest performance plot.
backtest_kpibool, optional: If True, print KPIs to stdout at each step.
figsizetuple of int, optional: Figure size.
funccallable, optional: Function applied to predictions before computing returns.