"""Collection of benchmarks."""
from abc import ABC, abstractmethod
import cvxpy as cp
from cvxpylayers.torch import CvxpyLayer
import torch
from .layers import CovarianceMatrix
[docs]class Benchmark(ABC):
"""Abstract benchmark class.
The idea is to create some benchmarks that we can use for comparison to our neural networks. Note that we
assume that benchmarks are not trainable - one can only use them for inference.
"""
[docs] @abstractmethod
def __call__(self, x):
"""Prediction of the model."""
@property
def hparams(self):
"""Hyperparamters relevant to construction of the model."""
return {}
[docs]class InverseVolatility(Benchmark):
"""Allocation only considering volatility of individual assets.
Parameters
----------
use_std : bool
If True, then we use standard deviation as a measure of volatility. Otherwise variance is used.
returns_channel : int
Which channel in the `x` feature matrix to consider (the 2nd dimension) as returns.
"""
def __init__(self, use_std=False, returns_channel=0):
self.use_std = use_std
self.returns_channel = returns_channel
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
eps = 1e-6
x_rets = x[:, self.returns_channel, ...]
vols = x_rets.std(dim=1) if self.use_std else x_rets.var(dim=1)
ivols = 1 / (vols + eps)
weights = ivols / ivols.sum(dim=1, keepdim=True)
return weights
@property
def hparams(self):
"""Hyperparamters relevant to construction of the model."""
return {'use_std': self.use_std,
'returns_channel': self.returns_channel}
[docs]class MaximumReturn(Benchmark):
"""Markowitz portfolio optimization - maximum return.
Parameters
----------
max_weight : float
A number in (0, 1] representing the maximum weight per asset.
n_assets : None or int
If specifed the benchmark will always have to be provided with `n_assets` of assets in the `__call__`.
This way one can achieve major speedups since the optimization problem is canonicalized only once in the
constructor. However, when `n_assets` is None the optimization problem is canonicalized before each
inside of `__call__` which results in overhead but allows for variable number of assets.
returns_channel : int
Which channel in the `x` feature matrix to consider (the 2nd dimension) as returns.
Attributes
----------
optlayer : cvxpylayers.torch.CvxpyLayer or None
Equal to None if `n_assets` not provided in the constructor. In this case optimization problem is constructed
with each forward pass. This allows for variable number of assets but is slower. If `n_assets` provided than
constructed once and for all in the constructor.
"""
def __init__(self, max_weight=1, n_assets=None, returns_channel=0):
self.max_weight = max_weight
self.n_assets = n_assets
self.returns_channel = returns_channel
self.optlayer = self._construct_problem(n_assets, max_weight) if self.n_assets is not None else None
@staticmethod
def _construct_problem(n_assets, max_weight):
"""Construct cvxpylayers problem."""
rets = cp.Parameter(n_assets)
w = cp.Variable(n_assets)
ret = rets @ w
prob = cp.Problem(cp.Maximize(ret), [cp.sum(w) == 1,
w >= 0,
w <= max_weight])
return CvxpyLayer(prob, parameters=[rets], variables=[w])
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
n_samples, _, lookback, n_assets = x.shape
# Problem setup
if self.optlayer is not None:
if self.n_assets != n_assets:
raise ValueError('Incorrect number of assets: {}, expected: {}'.format(n_assets, self.n_assets))
optlayer = self.optlayer
else:
optlayer = self._construct_problem(n_assets, self.max_weight)
rets_estimate = x[:, self.returns_channel, :, :].mean(dim=1) # (n_samples, n_assets)
return optlayer(rets_estimate)[0]
@property
def hparams(self):
"""Hyperparamters relevant to construction of the model."""
return {'max_weight': self.max_weight,
'returns_channel': self.returns_channel,
'n_assets': self.n_assets}
[docs]class MinimumVariance(Benchmark):
"""Markowitz portfolio optimization - minimum variance.
Parameters
----------
max_weight : float
A number in (0, 1] representing the maximum weight per asset.
n_assets : None or int
If specifed the benchmark will always have to be provided with `n_assets` of assets in the `__call__`.
This way one can achieve major speedups since the optimization problem is canonicalized only once in the
constructor. However, when `n_assets` is None the optimization problem is canonicalized before each
inside of `__call__` which results in overhead but allows for variable number of assets.
returns_channel : int
Which channel in the `x` feature matrix to consider (the 2nd dimension) as returns.
Attributes
----------
optlayer : cvxpylayers.torch.CvxpyLayer or None
Equal to None if `n_assets` not provided in the constructor. In this case optimization problem is constructed
with each forward pass. This allows for variable number of assets but is slower. If `n_assets` provided than
constructed once and for all in the constructor.
"""
def __init__(self, max_weight=1, returns_channel=0, n_assets=None):
self.n_assets = n_assets
self.returns_channel = returns_channel
self.max_weight = max_weight
self.optlayer = self._construct_problem(n_assets, max_weight) if self.n_assets is not None else None
@staticmethod
def _construct_problem(n_assets, max_weight):
"""Construct cvxpylayers problem."""
covmat_sqrt = cp.Parameter((n_assets, n_assets))
w = cp.Variable(n_assets)
risk = cp.sum_squares(covmat_sqrt @ w)
prob = cp.Problem(cp.Minimize(risk), [cp.sum(w) == 1,
w >= 0,
w <= max_weight])
return CvxpyLayer(prob, parameters=[covmat_sqrt], variables=[w])
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
n_samples, _, lookback, n_assets = x.shape
# Problem setup
if self.optlayer is not None:
if self.n_assets != n_assets:
raise ValueError('Incorrect number of assets: {}, expected: {}'.format(n_assets, self.n_assets))
optlayer = self.optlayer
else:
optlayer = self._construct_problem(n_assets, self.max_weight)
# problem solver
covmat_sqrt_estimates = CovarianceMatrix(sqrt=True)(x[:, self.returns_channel, :, :])
return optlayer(covmat_sqrt_estimates)[0]
@property
def hparams(self):
"""Hyperparamters relevant to construction of the model."""
return {'max_weight': self.max_weight,
'returns_channel': self.returns_channel,
'n_assets': self.n_assets}
[docs]class OneOverN(Benchmark):
"""Equally weighted portfolio."""
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
n_samples, n_channels, lookback, n_assets = x.shape
return torch.ones((n_samples, n_assets), dtype=x.dtype, device=x.device) / n_assets
[docs]class Random(Benchmark):
"""Random allocation for each prediction."""
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
n_samples, n_channels, lookback, n_assets = x.shape
weights_unscaled = torch.rand((n_samples, n_assets), dtype=x.dtype, device=x.device)
weights_sums = weights_unscaled.sum(dim=1, keepdim=True).repeat(1, n_assets)
return weights_unscaled / weights_sums
[docs]class Singleton(Benchmark):
"""Predict a single asset.
Parameters
----------
asset_ix : int
Index of the asset to predict.
"""
def __init__(self, asset_ix):
self.asset_ix = asset_ix
[docs] def __call__(self, x):
"""Predict weights.
Parameters
----------
x : torch.Tensor
Tensor of shape `(n_samples, n_channels, lookback, n_assets)`.
Returns
-------
weights : torch.Tensor
Tensor of shape `(n_samples, n_assets)` representing the predicted weights.
"""
n_samples, n_channels, lookback, n_assets = x.shape
if self.asset_ix not in set(range(n_assets)):
raise IndexError('The selected asset index is out of range.')
weights = torch.zeros((n_samples, n_assets), dtype=x.dtype, device=x.device)
weights[:, self.asset_ix] = 1
return weights
@property
def hparams(self):
"""Hyperparamters relevant to construction of the model."""
return {'asset_ix': self.asset_ix}