Reference

Clients

class pyoptimum.Client(username: str | None = None, password: str | None = None, token: str | None = None, auto_token_renewal: bool = True, base_url: str = 'https://optimize.vicbee.net', api: str = 'optimize', prefix: str = 'api', auth_url: str = '')

Client object to facilitate connection to the optimize.vicbee.net Optimize and Models API

Calls will be made to a URL of the form:

base_url/api/prefix/entry_point

in which entry_point is set in pyoptimum.Client.call().

Parameters:

username – the username
password – the password
token – an authentication token
auto_token_renewal – whether to automatically renew an expired token
base_url – the api base url
api – the target api
prefix – the target api prefix
auth_url – the api auth url (optional)

call(entry_point: str, data: Any, method: Literal['get', 'post', 'put', 'patch', 'delete'] = 'post') → Any

Calls the api entry_point with data

Parameters:

entry_point – the api entry point
data – the data
method – ‘get’, ‘post’, ‘put’, ‘patch’, ‘delete’ (default=’post’)

Returns:

dictionary with the response

get_token() → None: Retrieve authentication token

static url_join(*args: str) → str

Sanitizes and join url parts

The main role of this function is to remove slashes at the beginning or end of url parts

Parameters:: args – the parts of the url
Returns:: the sanitized url

class pyoptimum.AsyncClient(username: str | None = None, password: str | None = None, token: str | None = None, auto_token_renewal: bool = True, base_url: str = 'https://optimize.vicbee.net', api: str = 'optimize', prefix: str = 'api', auth_url: str = '')

Async client object to facilitate connection to the optimize.vicbee.net Optimize and Models API

Calls will be made to a URL of the form:

base_url/api/prefix/entry_point

in which entry_point is set in pyoptimum.Client.call().

Parameters:

username – the username
password – the password
token – an authentication token
auto_token_renewal – whether to automatically renew an expired token
base_url – the api base url
api – the target api
prefix – the target api prefix
auth_url – the api auth url (optional)

class ClientSession(base_url: str | ~yarl.URL | None = None, *, connector: ~aiohttp.connector.BaseConnector | None = None, loop: ~asyncio.events.AbstractEventLoop | None = None, cookies: ~typing.Mapping[str, str | BaseCookie[str] | Morsel[Any]] | ~typing.Iterable[~typing.Tuple[str, str | BaseCookie[str] | Morsel[Any]]] | BaseCookie[str] | None = None, headers: ~typing.Mapping[str, str] | ~typing.Mapping[~multidict._multidict.istr, str] | ~multidict._multidict.CIMultiDict | ~multidict._multidict.CIMultiDictProxy | ~typing.Iterable[~typing.Tuple[str | ~multidict._multidict.istr, str]] | None = None, skip_auto_headers: ~typing.Iterable[str] | None = None, auth: ~aiohttp.helpers.BasicAuth | None = None, json_serialize: ~typing.Callable[[~typing.Any], str] = <function dumps>, request_class: ~typing.Type[~aiohttp.client_reqrep.ClientRequest] = <class 'aiohttp.client_reqrep.ClientRequest'>, response_class: ~typing.Type[~aiohttp.client_reqrep.ClientResponse] = <class 'aiohttp.client_reqrep.ClientResponse'>, ws_response_class: ~typing.Type[~aiohttp.client_ws.ClientWebSocketResponse] = <class 'aiohttp.client_ws.ClientWebSocketResponse'>, version: ~aiohttp.http_writer.HttpVersion = (1, 1), cookie_jar: ~aiohttp.abc.AbstractCookieJar | None = None, connector_owner: bool = True, raise_for_status: bool | ~typing.Callable[[~aiohttp.client_reqrep.ClientResponse], ~typing.Awaitable[None]] = False, read_timeout: float | ~aiohttp.helpers._SENTINEL = _SENTINEL.sentinel, conn_timeout: float | None = None, timeout: object | ~aiohttp.client.ClientTimeout = _SENTINEL.sentinel, auto_decompress: bool = True, trust_env: bool = False, requote_redirect_url: bool = True, trace_configs: ~typing.List[~aiohttp.tracing.TraceConfig] | None = None, read_bufsize: int = 65536, max_line_size: int = 8190, max_field_size: int = 8190, fallback_charset_resolver: ~typing.Callable[[~aiohttp.client_reqrep.ClientResponse, bytes], str] = <function ClientSession.<lambda>>)

First-class interface for making HTTP requests.

property auth: BasicAuth | None: An object that represents HTTP Basic Authorization

property auto_decompress: bool: Should the body response be automatically decompressed.

async close() → None

Close underlying connector.

Release all acquired resources.

property closed: bool

Is client session closed.

A readonly property.

property connector: BaseConnector | None: Connector instance used for the session.

property connector_owner: bool: Should connector be closed on session closing

property cookie_jar: AbstractCookieJar: The session cookies.

delete(url: str | URL, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP DELETE request.

detach() → None

Detach connector from session without closing the former.

Session is switched to closed state anyway.

get(url: str | URL, *, allow_redirects: bool = True, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP GET request.

head(url: str | URL, *, allow_redirects: bool = False, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP HEAD request.

property headers: CIMultiDict[str]: The default headers of the client session.

property json_serialize: Callable[[Any], str]: Json serializer callable

property loop: AbstractEventLoop: Session’s loop.

options(url: str | URL, *, allow_redirects: bool = True, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP OPTIONS request.

patch(url: str | URL, *, data: Any = None, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP PATCH request.

post(url: str | URL, *, data: Any = None, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP POST request.

put(url: str | URL, *, data: Any = None, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP PUT request.

property raise_for_status: bool | Callable[[ClientResponse], Awaitable[None]]: Should ClientResponse.raise_for_status() be called for each response.

request(method: str, url: str | URL, **kwargs: Any) → _BaseRequestContextManager[ClientResponse]: Perform HTTP request.

property requote_redirect_url: bool: Do URL requoting on redirection handling.

property skip_auto_headers: FrozenSet[istr]: Headers for which autogeneration should be skipped

property timeout: ClientTimeout: Timeout for the session.

property trace_configs: List[TraceConfig]: A list of TraceConfig instances used for client tracing

property trust_env: bool

Should proxies information from environment or netrc be trusted.

Information is from HTTP_PROXY / HTTPS_PROXY environment variables or ~/.netrc file if present.

property version: Tuple[int, int]: The session HTTP protocol version.

ws_connect(url: str | URL, *, method: str = 'GET', protocols: Iterable[str] = (), timeout: float = 10.0, receive_timeout: float | None = None, autoclose: bool = True, autoping: bool = True, heartbeat: float | None = None, auth: BasicAuth | None = None, origin: str | None = None, params: None | str | Mapping[str, str | SupportsInt | float | Sequence[str | SupportsInt | float]] | Sequence[tuple[str, str | SupportsInt | float | Sequence[str | SupportsInt | float]]] = None, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, proxy: str | URL | None = None, proxy_auth: BasicAuth | None = None, ssl: None | bool | Fingerprint = True, verify_ssl: bool | None = None, fingerprint: bytes | None = None, ssl_context: None = None, proxy_headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, compress: int = 0, max_msg_size: int = 4194304) → _BaseRequestContextManager[ClientWebSocketResponse]: Initiate websocket connection.

async call(entry_point: str, data: dict | List[dict] | None = None, params: dict | None = None, method: Literal['get', 'post', 'put', 'patch', 'delete'] = 'post', follow_resource: bool = False, wait_time: float = 10, max_retries: int = 18) → Any

Calls the api entry_point with data

Parameters:

params
entry_point – the api entry point
data – the data for post, put, and patch
params – the params for get, and delete
method – ‘get’, ‘post’, ‘put’, ‘patch’, ‘delete’ (default=’post’)
follow_resource – whether to automatically retrieve resource if calculation is deferred
wait_time – how many seconds to wait before pooling resource again
max_retries – maximum number of retries

Returns:

dictionary with the response

async get_token() → None: Retrieve authentication token

Model

class pyoptimum.portfolio.Model(data: dict | Model)

Helper class representing a mean and variance portfolio model

Model parameters are the arrays:

\(r\): a vector with the expected returns of each asset
\(Q\): a vector representing the diagonal component of the portfolio covariance matrix
\(F\): an array representing the systematic part of the return
\(D\): the covariance matrix of the model factors

If \(x\) are portfolio weights, that is \(x\) sums to one, then the expected return \(\bar{r}\) and variance \(\sigma^2\) calculated by Model are of the form:

\begin{align} \bar{r} &= r^T x, & \sigma^2 &= x^T (\operatorname{diag}(Q) + F D F^T) x \end{align}

Parameters:: data – a dictionary with model parameters r, Q, F, and D, or a Model instance; F and D can be omitted, in which case the covarance is diagonal.

property D: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None

Returns:: the covariance of the factors; may be None

property Di: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None

Returns:: the inverse of the covariance of the factors

get_mean_and_variance() → Tuple[ndarray[tuple[Any, ...], dtype[_ScalarT]], ndarray[tuple[Any, ...], dtype[_ScalarT]]]: Returns vectors with the mean and variance :return: a tuple of vectors

property has_factors: bool

Returns:: True if F and D are present

return_and_variance(x: ndarray[tuple[Any, ...], dtype[_ScalarT]]) → Tuple[float, float]

Calculate the expected return and standard deviation of the portoflio holdings x

Parameters:: x – the portfolio holdings
Returns:: a tuple with the return and the standard deviation

property std: ndarray[tuple[Any, ...], dtype[_ScalarT]]

Returns:: the standard deviation of the individual assets in the model

to_dict(fields: Iterable | None = None, as_list: bool = False, normalize_variance=False) → Dict[str, ndarray[tuple[Any, ...], dtype[_ScalarT]]]

Returns a dictionary with the model parameters

Parameters:

fields – list of fields to be present (default: [r, Q, F, D] or [r, Q] if F and D are None)
as_list – return parameters are lists if True
normalize_variance – normalize the parameters so that the covariance matrix has one as its largest diagonal entry

Returns:

a dictionary with the parameters

unconstrained_frontier(x_bar: float = 1.0) → Tuple[float, float, float]

Calculates the parameters of the unconstrained optimal frontier

The unconstrained optimal frontier is the solution to the quadratic optimization problem

\[\sigma^2(\bar{\mu}) = \min_{x} \left \{ \sigma^2(x) : \quad r^T x = \bar{\mu}, \quad \sum_{i} x_i = \bar{x} \right \}\]

which has as solution

\[\sigma^2(\bar{\mu}) = a (\bar{\mu} - \mu^*)^2 + \sigma^2_0\]

This function calculates and returns the parameters \(a\), \(\mu^*\), and \(\sigma_0\).

Parameters:: x_bar – the portfolio sum (defaul: 1)
Returns:: tuple with a, mu_star, and sigma_0

Portfolio

class pyoptimum.portfolio.Portfolio(portfolio_client: AsyncClient, model_client: AsyncClient, model_method: Literal['linear', 'linear-fractional', 'diagonal'] = 'linear', follow_resource: bool = True, max_retries: int = 18, wait_time: float | None = None)

Helper class to facilitate portfolio calculations using the Optimize and Models API.

Portfolio objects are constructed from two AsyncClient s, e.g.

from pyoptimum import AsyncClient
from pyoptimum.portfolio import Portfolio
optimize_client = pyoptimum.AsyncClient(username=username, password=password, api='optimize')
models_client = pyoptimum.AsyncClient(username=username, password=password, api='models')
portfolio = Portfolio(self.portfolio_client, self.model_client)

The portfolio object can now be used to coordinate calls to both clients in order to build and manipulate portfolios.

Optional arguments are:

Parameters:

model_method – one of [‘linear’, ‘linear-fractional’, ‘diagonal’]
follow_resource – if True, will handle polling asynchronous resources
wait_time – how long to wait before pooling again (in seconds)
max_retries – how many times to retry before timing out

add_to_frontier(mu: float, std: float, x: ndarray[tuple[Any, ...], dtype[_ScalarT]]) → None

Add point to current frontier

Parameters:

mu – the return
std – the standard deviation
x – the portfolio values

apply_constraint(tickers: List[str], function: Literal['purchases', 'sales', 'holdings', 'short sales'], sign: Literal['≤', '≥', '='], value: List[float] | ndarray[tuple[Any, ...], dtype[_ScalarT]] | float | int, unit: Literal['shares', 'value', 'percent value'], short_sales: bool = True, buy: bool = True, sell: bool = True) → None

Apply constraints to list of tickers

Parameters:

tickers – the list of tickers
function – the function to apply on the left-hand side of the inequality
sign – the sign of the inequality
value – the value of the right-hand side of the inequality
unit – the unit in which the constraint is expressed
short_sales – whether to allow short sales
buy – whether to allow buying
sell – whether to allow selling

apply_group_constraint(group: str, function: Literal['purchases', 'sales', 'holdings', 'short sales', 'return'], sign: Literal['≤', '≥', '='], value: float | int, unit: Literal['value', 'percent value'], short_sales: bool = True, buy: bool = True, sell: bool = True) → None

Add group constraint

Parameters:

group – the group label
function – the function to apply on the left-hand side of the inequality
sign – the sign of the inequality
value – the value of the right-hand side of the inequality
unit – the unit in which the constraint is expressed
short_sales – whether to allow short sales
buy – whether to allow buying
sell – whether to allow selling

create_group(label: str, tickers: List[str]) → None

Create group

Parameters:

label – the group label
tickers – the group tickers

get_follow_resource() → dict

Get parameters for pooling asynchronous resources

Returns:: dictionary with parameters

get_frontier_range()

Returns:: the range of frontier values

get_group_dataframe(x: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None, cashflow: float = 0)

Parameters:

x – the portfolio weights (default=None)
cashflow – the cashflow (default=0)

Returns:

the portfolio groups as a dataframe

get_model() → Model

Returns:: the portfolio model for the current model and weights

get_portfolio_dataframe() → DataFrame

Returns:: the portfolio as a dataframe

get_range() → Tuple[List[float], List[float]]

Returns:: the range of the current model

get_recommendation_dataframe(x: ndarray[tuple[Any, ...], dtype[_ScalarT]], cashflow: float) → DataFrame

Parameters:

x – the portfolio weights
cashflow – the cashflow

Returns:

the portfolio recommendation as a dataframe

get_return_and_variance() → Tuple[float, float]

Returns:: the return and standard deviation of the current portfolio

get_tickers() → List[str]

Returns:: the portfolio tickers

get_unconstrained_frontier(x_bar: float = 1.0)

Returns:: the unconstrained frontier parameters

get_value() → float

Returns:: the total portfolio value

has_frontier() → bool

Returns:: True if a frontier is available

has_models() → bool

Returns:: True if models have been retrieved

has_prices() → bool

Returns:: True if prices have been retrieved

import_csv(filepath: str | bytes | BytesIO | Path) → None

Import portfolio from csv file

Parameters:: filepath – the file path

import_dataframe(portfolio: DataFrame) → None

Import portfolio from dataframe

Parameters:: portfolio – the dataframe

invalidate_frontier(): Invalidate the current frontier

invalidate_model(): Invalidate the current portfolio models

remove_constraints(tickers: List[str]) → None

Remove all individual constraints on the listed tickers

Parameters:: tickers – the list of tickers

remove_group(label: str) → None

Remove group and all its constraints

Parameters:: label – the group label

remove_group_constraint(group: str, function: Literal['purchases', 'sales', 'holdings', 'short sales', 'return']) → dict | None

Remove group constraints

Parameters:

group – the group
function – the function

async retrieve_basic_models(end: date = datetime.date(2025, 7, 2), model_weights: Dict[str, float] | None = None, include_prices: bool = False, horizon: int = 1) → Tuple[List[str], List[str], List[str]]

Retrieve basic portfolio models based on market tickers

If market_tickers is empty then returns a diagonal model in which all correlations are ignored.

Parameters:

end – the last day to retrieve models
model_weights – the model weights (default: None, which is the same as equal weights)
include_prices – whether to include prices on results (default: False)
horizon – the investment horizon in days (default: 1)

Returns:

a list of messages

async retrieve_custom_models(market_tickers: List[str], ranges: str | List[str], end: date = datetime.date(2025, 7, 2), model_weights: Dict[str, float] | None = None, common_factors: bool = False, include_prices: bool = False, trim_weekends: bool = True, horizon: int = 1, weights: Literal['linear', 'quadratic'] = 'linear') → Tuple[List[str], List[str], List[str]]

Retrieve custom portfolio models based on market tickers

If market_tickers is empty then returns a diagonal model in which all correlations are ignored.

Parameters:

market_tickers – the market tickers
ranges – the ranges to retrieve the portfolio models
end – the last day to retrieve models
model_weights – the model weights (default: None, which is the same as equal weights)
common_factors – whether to keep factors common (default: False)
include_prices – whether to include prices on results (default: False)
trim_weekends – whether to trim weekends (default: True)
horizon – the investment horizon in days (default: 1)
weights – the model weights (default: linear)

Returns:

a list of messages

async retrieve_frontier(cashflow: float, max_sales: float, short_sales: bool, buy: bool, sell: bool, rho: float = 0.0) → None

Retrieve the portfolio frontier

Parameters:

cashflow – the cashflow
max_sales – the max sales
short_sales – whether to allow short sales
buy – whether to allow buys
sell – whether to allow sells
rho – regularization factor (default: 0.0)

async retrieve_prices() → float

Retrieve the latest prices of all portfolio assets

Returns:: the total portfolio value

async retrieve_recommendation(mu: float | None = None, method: Literal['approximate', 'optimal'] = 'approximate') → dict

Retrieve or calculate recommendations

Parameters:

mu – the expected return
method – if approximate calculates approximate recommendations using the current frontier; if exact retrieve exact recommendation from the Optimize API

Returns:

set_follow_resource(follow_resource: bool, max_retries: int = 18, wait_time: float | None = None) → None

Set parameters for pooling asynchronous resources

Parameters:

follow_resource – if True, will handle polling asynchronous resources
wait_time – how long to wait before pooling again (in seconds)
max_retries – how many times to retry before timing out

set_model_weights(model_weights: Dict[str, float]) → None

Set weights for the current portfolio models

Parameters:: model_weights – the model weights

set_models(models: Dict[str, dict | Model], model_weights: Dict[str, float] | None = None, tickers: List[str] | None = None) → None

Set portfolio models

Parameters:

models – a dictionary with the models per range
model_weights – the model weights (default: None, which is the same as equal weights)
tickers – list of model tickers

split(tickers: List[str]) → None

Split inactive portfolio

Parameters:: tickers – the list of active tickers in the portfolio

Others

class pyoptimum.PyOptimumException: Base calls for pyoptimum exceptions