BasanosEngine¶

@dataclasses.dataclass(frozen=True)
class BasanosEngine(_CoreDataMixin, _DiagnosticsMixin, _PerformanceMixin, _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** — `assets`, `ret_adj`, `vola`, `cor`, `cor_tensor`
    * **Solve / position logic** — `cash_position`, `position_status`,
      `risk_position`, `position_leverage`, `warmup_state`
    * **Portfolio and performance** — `portfolio`, `naive_sharpe`,
      `sharpe_at_shrink`, `sharpe_at_window_factors`
    * **Matrix diagnostics** — `condition_number`, `effective_rank`,
      `solver_residual`, `signal_utilisation`
    * **Signal evaluation** — `ic(h)`, `rank_ic(h)`, `ic_mean(h)`, `ic_std(h)`,
      `icir(h)`, `rank_ic_mean(h)`, `rank_ic_std(h)` (``h`` defaults to 1)
    * **Reporting** — `config_report`

    Data-flow diagram
    -----------------

    .. code-block:: text

        prices (pl.DataFrame)
          │
          ├─ vol_adj ──► ret_adj (volatility-adjusted log returns)
          │                │
          │                ├─ ewm_covariance ──► 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:
        prices: 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: Polars DataFrame of expected-return signals aligned with *prices*.
            Must share the same shape and column names as *prices*.
        cfg: Immutable `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
    mu: pl.DataFrame
    cfg: BasanosConfig

    def __post_init__(self) -> None:
        """Validate inputs by delegating to `_validate_inputs`."""
        _validate_inputs(self.prices, self.mu, self.cfg)

    # ------------------------------------------------------------------
    # Core data-access properties — inherited from _CoreDataMixin
    # ------------------------------------------------------------------
    # (assets, ret_adj, vola, cor, cor_tensor)
    # Implementations live in _engine_core.py.

    # ------------------------------------------------------------------
    # Internal solve helpers — inherited from _SolveMixin
    # ------------------------------------------------------------------
    # (_compute_mask, _check_signal, _scale_to_cash, _row_early_check,
    #  _denom_guard_yield, _compute_position, _replay_positions,
    #  _iter_matrices, _iter_solve, warmup_state)
    # Implementations live in _engine_solve.py; patch targets remain in that
    # module's namespace, e.g. ``patch("basanos.math._engine_solve.solve")``.

    # ------------------------------------------------------------------
    # Position properties
    # ------------------------------------------------------------------

    @property
    def cash_position(self) -> pl.DataFrame:
        r"""Optimize correlation-aware risk positions for each timestamp.

        Supports two covariance modes controlled by ``cfg.covariance_config``:

        * `EwmaShrinkConfig` (default): Computes EWMA correlations, applies
          linear shrinkage toward the identity, and solves a normalised linear
          system $C\,x = \mu$ per timestamp via Cholesky / LU.

        * `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 $O(k^3 + kn)$ rather
          than $O(n^3)$ per step.

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

        Returns:
            pl.DataFrame: DataFrame with columns ['date'] + asset names containing
            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).  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.
        """
        assets = self.assets

        # Compute risk positions row-by-row using _replay_positions.
        prices_num = self.prices.select(assets).to_numpy()

        risk_pos_np = np.full_like(prices_num, fill_value=np.nan, dtype=float)
        cash_pos_np = np.full_like(prices_num, fill_value=np.nan, dtype=float)
        vola_np = self.vola.select(assets).to_numpy()

        self._replay_positions(risk_pos_np, cash_pos_np, vola_np)

        # Build Polars DataFrame for cash positions (numeric columns only)
        cash_position = self.prices.with_columns(
            [(pl.lit(cash_pos_np[:, i]).alias(asset)) for i, asset in enumerate(assets)]
        )

        return cash_position

    @property
    def position_status(self) -> pl.DataFrame:
        """Per-timestamp reason code explaining each `cash_position` row.

        Labels every row with exactly one of four `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:
            pl.DataFrame: Two-column DataFrame ``{'date': ..., 'status': ...}``
            with one row per timestamp.  The ``status`` column has
            ``Polars`` dtype ``String``.
        """
        statuses = [status for _i, _t, _mask, _pos, status in self._iter_solve()]
        return pl.DataFrame({"date": self.prices["date"], "status": pl.Series(statuses, dtype=pl.String)})

    @property
    def risk_position(self) -> pl.DataFrame:
        """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:
            pl.DataFrame: DataFrame with columns ``['date'] + assets`` where
            each value is ``cash_position_i * vola_i`` at the given timestamp.
        """
        assets = self.assets
        cp_np = self.cash_position.select(assets).to_numpy()
        vola_np = self.vola.select(assets).to_numpy()
        with np.errstate(invalid="ignore"):
            risk_pos = cp_np * vola_np
        return self.prices.with_columns([pl.lit(risk_pos[:, i]).alias(asset) for i, asset in enumerate(assets)])

    @property
    def position_leverage(self) -> pl.DataFrame:
        """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:
            pl.DataFrame: Two-column DataFrame ``{'date': ..., 'leverage': ...}``
            where ``leverage`` is the L1 norm of the cash-position vector.
        """
        assets = self.assets
        cp_np = self.cash_position.select(assets).to_numpy()
        leverage = np.nansum(np.abs(cp_np), axis=1)
        return pl.DataFrame({"date": self.prices["date"], "leverage": pl.Series(leverage, dtype=pl.Float64)})

    # ------------------------------------------------------------------
    # Portfolio and performance
    # ------------------------------------------------------------------

    @property
    def portfolio(self) -> Portfolio:
        """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 `cfg` is forwarded
        so that `net_cost_nav` and
        `position_delta_costs` work out
        of the box without any further configuration.

        Returns:
            Portfolio: Instance built from cash positions with AUM scaling.
        """
        cp = self.cash_position
        assets = [c for c in cp.columns if c != "date" and cp[c].dtype.is_numeric()]
        scaled = cp.with_columns(pl.col(a) * self.cfg.position_scale for a in assets)
        return Portfolio.from_cash_position(self.prices, scaled, aum=self.cfg.aum, cost_per_unit=self.cfg.cost_per_unit)

    # ------------------------------------------------------------------
    # Performance sweeps — inherited from _PerformanceMixin
    # ------------------------------------------------------------------
    # (sharpe_at_shrink, sharpe_at_window_factors, naive_sharpe)
    # Implementations live in _engine_performance.py.

    # ------------------------------------------------------------------
    # Reporting
    # ------------------------------------------------------------------

    @property
    def config_report(self) -> "ConfigReport":
        """Return a `ConfigReport` facade for this engine.

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

        Returns:
            basanos.math._config_report.ConfigReport: Report facade with
            ``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
        """
        return ConfigReport(config=self.cfg, engine=self)