BasanosEngine¶

The main portfolio optimisation engine. Accepts a price history and a signal matrix and exposes positions, diagnostics, and performance metrics as read-only properties.

`BasanosEngine` `dataclass` ¶

Bases: _DiagnosticsMixin, _SignalEvaluatorMixin, _SolveMixin

Engine to compute correlation matrices and optimize risk positions.

Encapsulates price data and configuration to build EWM-based correlations, apply shrinkage, and solve for normalized positions.

Public methods are organised into clearly delimited sections (some inherited from the private mixin classes):

Core data access — :attr:assets, :attr:ret_adj, :attr:vola, :attr:cor, :attr:cor_tensor
Solve / position logic — :attr:cash_position, :attr:position_status, :attr:risk_position, :attr:position_leverage, :meth:warmup_state (solve helpers inherited from :class:~._engine_solve._SolveMixin)
Portfolio and performance — :attr:portfolio, :attr:naive_sharpe, :meth:sharpe_at_shrink, :meth:sharpe_at_window_factors
Matrix diagnostics — :attr:condition_number, :attr:effective_rank, :attr:solver_residual, :attr:signal_utilisation (inherited from :class:~._engine_diagnostics._DiagnosticsMixin)
Signal evaluation — :attr:ic, :attr:rank_ic, :attr:ic_mean, :attr:ic_std, :attr:icir, :attr:rank_ic_mean, :attr:rank_ic_std (inherited from :class:~._engine_ic._SignalEvaluatorMixin)
Reporting — :attr:config_report

Data-flow diagram¶

.. code-block:: text

prices (pl.DataFrame)
  │
  ├─ vol_adj ──► ret_adj (volatility-adjusted log returns)
  │                │
  │                ├─ ewm_corr ──► cor / cor_tensor
  │                │                │
  │                │                └─ shrink2id / FactorModel
  │                │                        │
  │              vola                 covariance matrix
  │                │                        │
  └── mu ──────────┴── _iter_solve ──────────┘
                            │
                      cash_position
                            │
                   ┌────────┴────────┐
               portfolio          diagnostics
              (Portfolio)    (condition_number,
                              effective_rank,
                              solver_residual,
                              signal_utilisation,
                              ic, rank_ic, …)

Attributes:

Name	Type	Description
`prices`	`DataFrame`	Polars DataFrame of price levels per asset over time. Must contain a `'date'` column and at least one numeric asset column with strictly positive values that are not monotonically non-decreasing or non-increasing (i.e. they must vary in sign).
`mu`	`DataFrame`	Polars DataFrame of expected-return signals aligned with prices. Must share the same shape and column names as prices.
`cfg`	`BasanosConfig`	Immutable :class:`BasanosConfig` controlling EWMA half-lives, clipping, shrinkage intensity, and AUM.

Examples:

Build an engine with two synthetic assets over 30 days and inspect the optimized positions and diagnostic properties.

>>> import numpy as np
>>> import polars as pl
>>> from basanos.math import BasanosConfig, BasanosEngine
>>> dates = list(range(30))
>>> rng = np.random.default_rng(42)
>>> prices = pl.DataFrame({
...     "date": dates,
...     "A": np.cumprod(1 + rng.normal(0.001, 0.02, 30)) * 100.0,
...     "B": np.cumprod(1 + rng.normal(0.001, 0.02, 30)) * 150.0,
... })
>>> mu = pl.DataFrame({
...     "date": dates,
...     "A": rng.normal(0.0, 0.5, 30),
...     "B": rng.normal(0.0, 0.5, 30),
... })
>>> cfg = BasanosConfig(vola=5, corr=10, clip=2.0, shrink=0.5, aum=1_000_000)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> engine.assets
['A', 'B']
>>> engine.cash_position.shape
(30, 3)
>>> engine.position_leverage.columns
['date', 'leverage']

`prices: pl.DataFrame` `instance-attribute` ¶

`mu: pl.DataFrame` `instance-attribute` ¶

`cfg: BasanosConfig` `instance-attribute` ¶

`assets: list[str]` `property` ¶

List asset column names (numeric columns excluding 'date').

`ret_adj: pl.DataFrame` `property` ¶

Return per-asset volatility-adjusted log returns clipped by cfg.clip.

Uses an EWMA volatility estimate with lookback cfg.vola to standardize log returns for each numeric asset column.

`vola: pl.DataFrame` `property` ¶

Per-asset EWMA volatility of percentage returns.

Computes percent changes for each numeric asset column and applies an exponentially weighted standard deviation using the lookback specified by cfg.vola. The result is a DataFrame aligned with self.prices whose numeric columns hold per-asset volatility estimates.

`cor: dict[datetime.date, np.ndarray]` `property` ¶

Compute per-timestamp EWM correlation matrices.

Builds volatility-adjusted returns for all assets, computes an exponentially weighted correlation using a pure NumPy implementation (with window cfg.corr), and returns a mapping from each timestamp to the corresponding correlation matrix as a NumPy array.

Returns:

Name	Type	Description
`dict`	`dict[date, ndarray]`	Mapping `date -> np.ndarray` of shape (n_assets, n_assets).

Performance

Delegates to :func:ewm_corr, which is O(T·N²) in both time and memory. The returned dict holds T references into the result tensor (one NN view per date); no extra copies are made. For large N or T*, prefer cor_tensor to keep a single contiguous array rather than building a Python dict.

`cor_tensor: np.ndarray` `property` ¶

Return all correlation matrices stacked as a 3-D tensor.

Converts the per-timestamp correlation dict (see :py:attr:cor) into a single contiguous NumPy array so that the full history can be saved to a flat .npy file with :func:numpy.save and reloaded with :func:numpy.load.

Returns:

Type	Description
`ndarray`	np.ndarray: Array of shape `(T, N, N)` where T is the number of
`ndarray`	timestamps and N the number of assets. `tensor[t]` is the
`ndarray`	correlation matrix for the t-th date (same ordering as
`ndarray`	`self.prices["date"]`).

Examples:

>>> import tempfile, pathlib
>>> import numpy as np
>>> import polars as pl
>>> from basanos.math.optimizer import BasanosConfig, BasanosEngine
>>> dates = pl.Series("date", list(range(100)))
>>> rng0 = np.random.default_rng(0).lognormal(size=100)
>>> rng1 = np.random.default_rng(1).lognormal(size=100)
>>> prices = pl.DataFrame({"date": dates, "A": rng0, "B": rng1})
>>> rng2 = np.random.default_rng(2).normal(size=100)
>>> rng3 = np.random.default_rng(3).normal(size=100)
>>> mu = pl.DataFrame({"date": dates, "A": rng2, "B": rng3})
>>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> tensor = engine.cor_tensor
>>> with tempfile.TemporaryDirectory() as td:
...     path = pathlib.Path(td) / "cor.npy"
...     np.save(path, tensor)
...     loaded = np.load(path)
>>> np.testing.assert_array_equal(tensor, loaded)

`cash_position: pl.DataFrame` `property` ¶

Optimize correlation-aware risk positions for each timestamp.

Supports two covariance modes controlled by cfg.covariance_config:

:class:EwmaShrinkConfig (default): Computes EWMA correlations, applies linear shrinkage toward the identity, and solves a normalised linear system :math:C\,x = \mu per timestamp via Cholesky / LU.
:class:SlidingWindowConfig: At each timestamp uses the cfg.covariance_config.window most recent vol-adjusted returns to fit a rank-cfg.covariance_config.n_factors factor model via truncated SVD and solves the system via the Woodbury identity at :math:O(k^3 + kn) rather than :math:O(n^3) per step.

Non-finite or ill-posed cases yield zero positions for safety.

Returns:

Type	Description
`DataFrame`	pl.DataFrame: DataFrame with columns ['date'] + asset names containing
`DataFrame`	the per-timestamp cash positions (risk divided by EWMA volatility).

Performance

For ewma_shrink: dominant cost is self.cor (O(T·N²) time, O(T·N²) memory — see :func:ewm_corr). The per-timestamp linear solve adds O(N³) per row.

For sliding_window: O(T·W·N·k) for sliding SVDs plus O(T·(k³ + kN)) for Woodbury solves. Memory is O(W·N) per step, independent of T.

`position_status: pl.DataFrame` `property` ¶

Per-timestamp reason code explaining each :attr:cash_position row.

Labels every row with exactly one of four :class:~basanos.math.SolveStatus codes (which compare equal to their string equivalents):

'warmup': Insufficient history for the sliding-window covariance mode (i + 1 < cfg.covariance_config.window). Positions are NaN for all assets at this timestamp.
'zero_signal': The expected-return vector mu was all-zeros (or all-NaN) at this timestamp; the optimizer short-circuited and returned zero positions without solving.
'degenerate': The normalisation denominator was non-finite or below cfg.denom_tol, the Cholesky / Woodbury solve failed, or no asset had a finite price; positions were zeroed for safety.
'valid': The linear system was solved successfully and positions are non-trivially non-zero.

The codes map one-to-one onto the three NaN / zero cases described in the issue and allow downstream consumers (backtests, risk monitors) to distinguish data gaps from signal silence from numerical ill-conditioning without re-inspecting mu or the engine configuration.

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Two-column DataFrame `{'date': ..., 'status': ...}`
`DataFrame`	with one row per timestamp. The `status` column has
`DataFrame`	`Polars` dtype `String`.

`risk_position: pl.DataFrame` `property` ¶

Risk positions (before EWMA-volatility scaling) at each timestamp.

Derives the un-volatility-scaled position by multiplying the cash position by the per-asset EWMA volatility. Equivalently, this is the quantity solved by the correlation-adjusted linear system before dividing by vola.

Relationship to other properties::

cash_position = risk_position / vola
risk_position = cash_position * vola

Returns:

Type	Description
`DataFrame`	pl.DataFrame: DataFrame with columns `['date'] + assets` where
`DataFrame`	each value is `cash_position_i * vola_i` at the given timestamp.

`position_leverage: pl.DataFrame` `property` ¶

L1 norm of cash positions (gross leverage) at each timestamp.

Sums the absolute values of all asset cash positions at each row. NaN positions are treated as zero (they contribute nothing to gross leverage).

Returns:

Type	Description
`DataFrame`	pl.DataFrame: Two-column DataFrame `{'date': ..., 'leverage': ...}`
`DataFrame`	where `leverage` is the L1 norm of the cash-position vector.

`portfolio: Portfolio` `property` ¶

Construct a Portfolio from the optimized cash positions.

Converts the computed cash positions into a Portfolio using the configured AUM. The cost_per_unit from :attr:cfg is forwarded so that :attr:~jquantstats.Portfolio.net_cost_nav and :attr:~jquantstats.Portfolio.position_delta_costs work out of the box without any further configuration.

Returns:

Name	Type	Description
`Portfolio`	`Portfolio`	Instance built from cash positions with AUM scaling.

`naive_sharpe: float` `property` ¶

Sharpe ratio of the naïve equal-weight signal (μ = 1 for every asset/timestamp).

Replaces the expected-return signal mu with a constant matrix of ones, then runs the optimiser with the current configuration and returns the annualised Sharpe ratio of the resulting portfolio.

This provides the baseline answer to "does the signal add value?": a real signal should produce a higher Sharpe than the naïve benchmark. Combined with :meth:sharpe_at_shrink, this yields a three-way comparison:

+--------------------+----------------------------------------------+ | Benchmark | What it measures | +====================+==============================================+ | naive_sharpe | No signal skill; pure correlation routing | +--------------------+----------------------------------------------+ | sharpe_at_shrink(0.0) | Signal skill, no correlation adj. | +--------------------+----------------------------------------------+ | sharpe_at_shrink(cfg.shrink) | Signal + correlation adj. | +--------------------+----------------------------------------------+

Returns:

Type	Description
`float`	Annualised Sharpe ratio of the equal-weight portfolio as a `float`.
`float`	Returns `float("nan")` when the Sharpe ratio cannot be computed.

Examples:

>>> import numpy as np
>>> import polars as pl
>>> from basanos.math.optimizer import BasanosConfig, BasanosEngine
>>> dates = pl.Series("date", list(range(200)))
>>> rng = np.random.default_rng(0)
>>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)})
>>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)})
>>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> s = engine.naive_sharpe
>>> isinstance(s, float)
True

`config_report: ConfigReport` `property` ¶

Return a :class:~basanos.math._config_report.ConfigReport facade for this engine.

Returns a :class:~basanos.math._config_report.ConfigReport that includes the full lambda-sweep chart — an interactive plot of the annualised Sharpe ratio as :attr:~BasanosConfig.shrink (λ) is swept across [0, 1] — in addition to the parameter table, shrinkage-guidance table, and theory section available from :attr:BasanosConfig.report.

Returns:

Type	Description
`ConfigReport`	basanos.math._config_report.ConfigReport: Report facade with
`ConfigReport`	`to_html()` and `save()` methods.

Examples:

>>> import numpy as np
>>> import polars as pl
>>> from basanos.math.optimizer import BasanosConfig, BasanosEngine
>>> dates = pl.Series("date", list(range(200)))
>>> rng = np.random.default_rng(0)
>>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)})
>>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)})
>>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> report = engine.config_report
>>> html = report.to_html()
>>> "Lambda" in html
True

`__post_init__() -> None` ¶

Validate inputs by delegating to :func:_validate_inputs.

`sharpe_at_shrink(shrink: float) -> float` ¶

Return the annualised portfolio Sharpe ratio for the given shrinkage weight.

Constructs a new :class:BasanosEngine with all parameters identical to self except that cfg.shrink is replaced by shrink, then returns the annualised Sharpe ratio of the resulting portfolio.

This is the canonical single-argument callable required by the benchmarks specification: f(λ) → Sharpe. Use it to sweep λ across [0, 1] and measure whether correlation adjustment adds value over the signal-proportional baseline (λ = 0) or the unregularised limit (λ = 1).

Corner cases

λ = 0 — the shrunk matrix equals the identity, so the optimiser treats all assets as uncorrelated and positions are purely signal-proportional (no correlation adjustment).
λ = 1 — the raw EWMA correlation matrix is used without shrinkage.

Parameters:

Name	Type	Description	Default
`shrink`	`float`	Retention weight λ ∈ [0, 1]. See :attr:`BasanosConfig.shrink` for full documentation.	required

Returns:

Type	Description
`float`	Annualised Sharpe ratio of the portfolio returns as a `float`.
`float`	Returns `float("nan")` when the Sharpe ratio cannot be computed
`float`	(e.g. zero-variance returns).

Raises:

Type	Description
`ValidationError`	When `shrink` is outside [0, 1] (delegated to :class:`BasanosConfig` field validation).

Examples:

>>> import numpy as np
>>> import polars as pl
>>> from basanos.math.optimizer import BasanosConfig, BasanosEngine
>>> dates = pl.Series("date", list(range(200)))
>>> rng = np.random.default_rng(0)
>>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)})
>>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)})
>>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> s = engine.sharpe_at_shrink(0.5)
>>> isinstance(s, float)
True

`sharpe_at_window_factors(window: int, n_factors: int) -> float` ¶

Return the annualised portfolio Sharpe ratio for the given sliding-window parameters.

Constructs a new :class:BasanosEngine with covariance_mode set to "sliding_window" and the supplied window / n_factors, keeping all other configuration identical to self.

Use this method to sweep (W, k) and compare the sliding-window estimator against the EWMA baseline (via :meth:sharpe_at_shrink).

Parameters:

Name	Type	Description	Default
`window`	`int`	Rolling window length :math:`W \geq 1`. Rule of thumb: :math:`W \geq 2 \cdot n_{\text{assets}}`.	required
`n_factors`	`int`	Number of latent factors :math:`k \geq 1`.	required

Returns:

Type	Description
`float`	Annualised Sharpe ratio of the portfolio returns as a `float`.
`float`	Returns `float("nan")` when the Sharpe ratio cannot be computed
`float`	(e.g. not enough history to fill the first window).

Raises:

Type	Description
`ValidationError`	When `window` or `n_factors` fail field constraints (delegated to :class:`BasanosConfig`).

Examples:

>>> import numpy as np
>>> import polars as pl
>>> from basanos.math.optimizer import BasanosConfig, BasanosEngine
>>> dates = pl.Series("date", list(range(200)))
>>> rng = np.random.default_rng(0)
>>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)})
>>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)})
>>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6)
>>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg)
>>> s = engine.sharpe_at_window_factors(window=40, n_factors=2)
>>> isinstance(s, float)
True

BasanosEngine¶

BasanosEngine dataclass ¶

Data-flow diagram¶

prices: pl.DataFrame instance-attribute ¶

mu: pl.DataFrame instance-attribute ¶

cfg: BasanosConfig instance-attribute ¶

assets: list[str] property ¶

ret_adj: pl.DataFrame property ¶

vola: pl.DataFrame property ¶

cor: dict[datetime.date, np.ndarray] property ¶

cor_tensor: np.ndarray property ¶

cash_position: pl.DataFrame property ¶

position_status: pl.DataFrame property ¶

risk_position: pl.DataFrame property ¶

position_leverage: pl.DataFrame property ¶

portfolio: Portfolio property ¶

naive_sharpe: float property ¶

config_report: ConfigReport property ¶

__post_init__() -> None ¶

sharpe_at_shrink(shrink: float) -> float ¶

sharpe_at_window_factors(window: int, n_factors: int) -> float ¶

__init__(prices: pl.DataFrame, mu: pl.DataFrame, cfg: BasanosConfig) -> None ¶

`BasanosEngine` `dataclass` ¶

`prices: pl.DataFrame` `instance-attribute` ¶

`mu: pl.DataFrame` `instance-attribute` ¶

`cfg: BasanosConfig` `instance-attribute` ¶

`assets: list[str]` `property` ¶

`ret_adj: pl.DataFrame` `property` ¶

`vola: pl.DataFrame` `property` ¶

`cor: dict[datetime.date, np.ndarray]` `property` ¶

`cor_tensor: np.ndarray` `property` ¶

`cash_position: pl.DataFrame` `property` ¶

`position_status: pl.DataFrame` `property` ¶

`risk_position: pl.DataFrame` `property` ¶

`position_leverage: pl.DataFrame` `property` ¶

`portfolio: Portfolio` `property` ¶

`naive_sharpe: float` `property` ¶

`config_report: ConfigReport` `property` ¶

`__post_init__() -> None` ¶

`sharpe_at_shrink(shrink: float) -> float` ¶

`sharpe_at_window_factors(window: int, n_factors: int) -> float` ¶

`init(prices: pl.DataFrame, mu: pl.DataFrame, cfg: BasanosConfig) -> None` ¶