RollMultiLayerPerceptron¶

Defined in fynance.models.rolling

class RollMultiLayerPerceptron(X, y, layers=[], activation=None, drop=None, bias=True, x_type=None, y_type=None, activation_kwargs={}, **kwargs)[source]

Bases: MultiLayerPerceptron, _RollingBasis

Rolling version of the multi-layer perceptron model.

End-to-end walk-forward training pipeline for an MLP: at each step the model is fitted on a sliding window of length n, evaluated on the previous out-of-sample slice, then used to predict the next slice. Losses (train, eval, test) and out-of-sample predictions are accumulated step by step in self.log, self.y_eval and self.y_test for downstream analysis.

Use set_roll_period to configure window sizes and batch options, then run to execute the loop. run can also drive a live fynance.backtest.dynamic_plot_backtest.BacktestNeuralNet figure to monitor convergence.

Combines MultiLayerPerceptron with the walk-forward iterator from _RollingBasis. Use set_roll_period instead of calling the object directly (__call__ is captured by torch.nn.Module).

Methods

`set_roll_period`(train_period, test_period[, ...])	Configure rolling window parameters.
`sub_predict`(X)	Return predictions as a numpy array.

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

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)

forward(x): Forward computation.

get_stats()

Return per-step loss history as a DataFrame.

Returns:

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

load_model(path, load_optimizer=False)

Save the model with this weights and parameters.

Parameters:

pathstr or os.PathLike object: Path to load the model.
load_optimizerbool, optional: If True, then load also the optimizer.

plot_loss(figsize=(9, 4))

Plot train / eval / test loss curves.

Parameters:

figsizetuple of int, optional

Returns:

matplotlib.figure.Figure

predict(X)

Predicts outputs of neural network model.

Runs self.forward(X) under torch.no_grad, so no autograd graph is built. The returned tensor is detached and lives on the same device as the model parameters.

Parameters:

Xtorch.Tensor: Inputs to compute prediction. Same shape and dtype contract as train_on.

Returns:

torch.Tensor: Outputs prediction (detached, gradient-free).

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

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.

save(path)[source]

Save the model weights.

Parameters:

pathstr: Destination path.

save_model(path, save_optimizer=False)

Save the model with this weights and parameters.

Parameters:

pathstr or os.PathLike object: Path to save the model.
save_optimizerbool, optional: If True, then save also the optimizer.

set_data(X, y, x_type=None, y_type=None)

Set data inputs and outputs.

Coerces X and y to torch.Tensor and caches them as self.X / self.y. After the call the attributes self.T (number of observations), self.N (input columns) and self.M (output columns) are set.

Parameters:

X, yarray-like: Respectively input and output data. Accepted types: numpy.ndarray, torch.Tensor, pandas.DataFrame. Shapes must be (T, N) and (T, M) respectively.
x_type, y_typetorch.dtype, optional: Target dtypes for the resulting tensors. Default is None, which preserves the input dtype.

Returns:

BaseNeuralNet: self, to allow chaining.

Raises:

ValueError: If self.N / self.M were already set and X / y do not match, or if X and y have different lengths.

set_lr_scheduler(lr_scheduler, **kwargs)

Set dynamic learning rate.

Parameters:

lr_schedulertorch.optim.lr_scheduler._LRScheduler: Method from torch.optim.lr_scheduler to wrap self.optimizer, cf module torch.optim.lr_scheduler in PyTorch documentation [2].
**kwargs: Keyword arguments to pass to the learning rate scheduler.

References

[2]

https://pytorch.org/docs/stable/optim.html#how-to-adjust-learning-rate

set_optimizer(criterion, optimizer, params=None, **kwargs)

Set the optimizer object.

Set optimizer object with specified criterion as loss function and any kwargs as optional parameters.

Parameters:

criterionCallabletorch.nn.modules.loss: A loss function.
optimizertorch.optim.Optimizer: An optimizer algorithm.
paramsobject or iterable object: Layer of parameters to optimize or dicts defining parameter groups. If set to None then all parameters of model will be optimized. Default is None.
**kwargs: Keyword arguments of optimizer, cf PyTorch documentation [1].

Returns:

BaseNeuralNet: Self object model.

References

[1]

https://pytorch.org/docs/stable/optim.html

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

Configure rolling window parameters.

This is the preferred entry-point for RollMultiLayerPerceptron because __call__ is captured by torch.nn.Module.

Parameters:

train_period, test_periodint: Size of respectively training and testing sub-periods.
startint, optional
endint, optional
roll_periodint, optional
eval_periodint, optional
batch_sizeint, optional
epochsint, optional

Returns:

_RollingBasis

set_seed(seed_torch=None, seed_numpy=None)

Set seed for PyTorch and NumPy random number generator.

Parameters:

seed_torch, seed_numpybool or int, optional: If seed is an int \(0 < seed < 2^32\) set respectively PyTorch and NumPy seed with the number. Otherwise if is True then choose a random number, else doesn’t set seed.

sub_predict(X)[source]: Return predictions as a numpy array.

train_on(X, y)

Trains the neural network model on a single batch.

Runs one forward / backward / optimizer-step cycle on the batch (X, y). As a side effect, gradients of all parameters are zeroed before the forward pass and the optimizer state is advanced afterwards. If a learning-rate scheduler has been registered via set_lr_scheduler, its step is also called.

Parameters:

X, ytorch.Tensor: Respectively inputs and outputs to train model. Shapes must match what self.forward expects (see the class-level “Public API contract” section).

Returns:

torch.Tensor: The loss tensor produced by self.criterion(self(X), y), with gradient already consumed by loss.backward().

Raises:

AttributeError: If set_optimizer has not been called yet.