Coverage for src/basanos/math/_engine_solve_base.py: 100%
66 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-04 07:53 +0000
1"""Foundational types and stateless solve primitives for `_SolveMixin`.
3Holds the `SolveStatus` enum, the `MatrixBundle` / `WarmupState` carriers, the
4`MatrixYield` / `SolveYield` aliases, and `_SolvePrimitivesMixin` — the pure
5(``@staticmethod``) helpers that carry no dependency on the linear-algebra
6backend. Splitting them out of ``_engine_solve`` keeps the solve-orchestration
7module focused on the generators that drive `BasanosEngine`.
9All public names are re-exported from ``_engine_solve`` so existing imports
10(``from basanos.math._engine_solve import MatrixBundle, SolveStatus, ...``) and
11the ``solve`` / ``inv_a_norm`` patch targets that live there are unchanged.
12"""
14from __future__ import annotations
16import dataclasses
17import datetime
18import logging
19from enum import StrEnum
20from typing import TypeAlias, cast
22import numpy as np
24# Solve-step warnings are emitted under the ``_engine_solve`` logger name (not
25# this module's ``__name__``) so the logging contract — log filters and the JSON
26# round-trip in the diagnostics — stays stable regardless of how the solve
27# helpers are split across private modules.
28_logger = logging.getLogger("basanos.math._engine_solve")
31class SolveStatus(StrEnum):
32 """Solver outcome labels for each timestamp.
34 Since `SolveStatus` inherits from `str` via ``StrEnum``,
35 values compare equal to their string equivalents (e.g.
36 ``SolveStatus.VALID == "valid"``), preserving backward compatibility
37 with code that matches on string literals.
39 Attributes:
40 WARMUP: Insufficient history for the sliding-window covariance mode.
41 ZERO_SIGNAL: The expected-return vector was all-zero; positions zeroed.
42 DEGENERATE: Normalisation denominator was non-finite, solve failed, or
43 no asset had a finite price; positions zeroed for safety.
44 VALID: Linear system solved successfully; positions are non-trivially
45 non-zero.
46 """
48 WARMUP = "warmup"
49 ZERO_SIGNAL = "zero_signal"
50 DEGENERATE = "degenerate"
51 VALID = "valid"
54@dataclasses.dataclass(frozen=True)
55class MatrixBundle:
56 """Container for the covariance matrix and any mode-specific auxiliary state.
58 Wrapping the covariance matrix in a dataclass decouples
59 `_compute_position` from the raw array so that future
60 covariance modes (e.g. DCC-GARCH, RMT-cleaned) can carry additional fields
61 through the same interface without changing the method signature.
63 Attributes:
64 matrix: The ``(n_active, n_active)`` covariance sub-matrix for the
65 active assets at a given timestamp.
66 """
68 matrix: np.ndarray
71#: Yield type for `_iter_matrices`:
72#: ``(i, t, mask, bundle)`` where ``bundle`` is ``None`` during warmup/no-data.
73MatrixYield: TypeAlias = tuple[int, datetime.date, np.ndarray, MatrixBundle | None]
75#: Yield type for `_iter_solve`:
76#: ``(i, t, mask, pos_or_none, status)`` where ``pos_or_none`` is ``None`` only for warmup rows.
77SolveYield: TypeAlias = tuple[int, datetime.date, np.ndarray, np.ndarray | None, SolveStatus]
80@dataclasses.dataclass(frozen=True)
81class WarmupState:
82 """Final state produced by a full batch solve; consumed by `from_warmup`.
84 Returned by `warmup_state` and used by
85 `from_warmup` to initialise the streaming state without
86 coupling to the private `_iter_solve` generator.
88 Attributes:
89 prev_cash_pos: Cash positions at the last warmup row, shape
90 ``(n_assets,)``. ``NaN`` for assets that were still in their
91 own warmup period.
92 """
94 prev_cash_pos: np.ndarray
97class _SolvePrimitivesMixin:
98 """Stateless solve helpers shared by `_SolveMixin`.
100 Every method here is a ``@staticmethod`` that carries no dependency on the
101 linear-algebra backend, so it can be reasoned about (and reused) in
102 isolation from the per-timestamp solve generators.
103 """
105 @staticmethod
106 def _compute_mask(prices_row: np.ndarray) -> np.ndarray:
107 """Return boolean mask indicating which assets have finite prices in the given row."""
108 mask: np.ndarray = np.isfinite(prices_row)
109 return mask
111 @staticmethod
112 def _check_signal(mu: np.ndarray, mask: np.ndarray) -> SolveStatus | None:
113 """Return ``ZERO_SIGNAL`` when the masked expected-return vector is all-zero.
115 Returns ``None`` when the signal is non-trivially non-zero, indicating
116 that the caller should proceed to the linear solve.
117 """
118 if np.allclose(np.nan_to_num(mu[mask]), 0.0):
119 return SolveStatus.ZERO_SIGNAL
120 return None
122 @staticmethod
123 def _scale_to_cash(pos: np.ndarray, vola_active: np.ndarray) -> np.ndarray:
124 """Convert raw solver positions to cash-adjusted positions.
126 Divides *pos* by *vola_active* (volatility for the active asset subset)
127 to get cash positions. ``np.errstate(invalid="ignore")`` is applied
128 internally so NaN volatility values propagate quietly.
129 """
130 with np.errstate(invalid="ignore"):
131 return cast("np.ndarray", pos / vola_active)
133 @staticmethod
134 def _row_early_check(
135 i: int,
136 t: datetime.date,
137 mask: np.ndarray,
138 mu_row: np.ndarray,
139 ) -> tuple[np.ndarray, SolveYield | None]:
140 """Validate the price mask and expected-return signal for a single row.
142 Returns an ``(expected_mu, early_yield)`` pair. When ``early_yield``
143 is not ``None``, the caller should ``yield early_yield; continue``
144 immediately — the row is either degenerate (empty mask) or has an
145 all-zero signal. When ``early_yield`` is ``None`` the row is ready
146 for the mode-specific solve step.
148 Args:
149 i: Row index.
150 t: Timestamp.
151 mask: Boolean array of shape ``(n_assets,)`` indicating finite prices.
152 mu_row: Expected-return row of shape ``(n_assets,)``.
154 Returns:
155 tuple: ``(expected_mu, early_yield)`` where ``expected_mu`` is
156 ``np.nan_to_num(mu_row[mask])`` and ``early_yield`` is either a
157 complete `SolveYield` tuple (when the caller should yield
158 and continue) or ``None`` (when the caller should proceed to solve).
159 """
160 if not mask.any():
161 return np.zeros(0), (i, t, mask, np.zeros(0), SolveStatus.DEGENERATE)
162 expected_mu = np.nan_to_num(mu_row[mask])
163 sig_status = _SolvePrimitivesMixin._check_signal(mu_row, mask)
164 if sig_status is not None:
165 return expected_mu, (i, t, mask, np.zeros_like(expected_mu), sig_status)
166 return expected_mu, None
168 @staticmethod
169 def _denom_guard_yield(
170 i: int,
171 t: datetime.date,
172 mask: np.ndarray,
173 expected_mu: np.ndarray,
174 pos_raw: np.ndarray,
175 denom: float,
176 denom_tol: float,
177 ) -> SolveYield:
178 """Apply the normalisation-denominator guard and return the appropriate yield tuple.
180 Emits a `WARNING` and returns a
181 `DEGENERATE` yield when *denom* is non-finite or at
182 or below *denom_tol*; otherwise returns a `VALID`
183 yield with normalised positions ``pos_raw / denom``.
185 Args:
186 i: Row index.
187 t: Timestamp.
188 mask: Boolean asset mask of shape ``(n_assets,)``.
189 expected_mu: Masked expected-return vector of shape ``(n_active,)``.
190 pos_raw: Raw (pre-normalisation) position vector of shape ``(n_active,)``.
191 denom: Computed normalisation denominator.
192 denom_tol: Tolerance threshold below which *denom* is treated as degenerate.
194 Returns:
195 SolveYield: Either a degenerate or valid ``(i, t, mask, pos, status)`` tuple.
196 """
197 n_active = len(expected_mu)
198 if not np.isfinite(denom) or denom <= denom_tol:
199 _logger.warning(
200 "Positions zeroed at t=%s: normalisation denominator is degenerate "
201 "(denom=%s, denom_tol=%s). Check signal magnitude and covariance matrix.",
202 t,
203 denom,
204 denom_tol,
205 extra={
206 "context": {
207 "t": str(t),
208 "denom": denom,
209 "denom_tol": denom_tol,
210 }
211 },
212 )
213 return i, t, mask, np.zeros(n_active), SolveStatus.DEGENERATE
214 return i, t, mask, pos_raw / denom, SolveStatus.VALID
216 @staticmethod
217 def _apply_turnover_constraint(
218 new_cash: np.ndarray,
219 prev_cash: np.ndarray,
220 max_turnover: float,
221 ) -> np.ndarray:
222 """Cap the L1 norm of the position change to *max_turnover*.
224 When ``sum(|new_cash - prev_cash|) > max_turnover``, the delta is
225 scaled back proportionally toward *prev_cash* so that the constraint
226 is exactly met. When the constraint is already satisfied the input is
227 returned unchanged.
229 Args:
230 new_cash: Proposed cash positions after the solve step, shape
231 ``(n_active,)`` — ``NaN`` values treated as zero.
232 prev_cash: Cash positions at the previous step, shape
233 ``(n_active,)`` — ``NaN`` values treated as zero.
234 max_turnover: Maximum allowed L1 norm of the position change.
236 Returns:
237 np.ndarray: The (possibly scaled) new cash positions.
238 """
239 curr = np.nan_to_num(new_cash, nan=0.0)
240 prev = np.nan_to_num(prev_cash, nan=0.0)
241 delta = curr - prev
242 total_delta = float(np.sum(np.abs(delta)))
243 if total_delta > max_turnover:
244 scale = max_turnover / total_delta
245 return cast("np.ndarray", prev + delta * scale)
246 return new_cash
248 @staticmethod
249 def _sliding_warmup_or_degenerate(
250 i: int,
251 t: datetime.date,
252 mask: np.ndarray,
253 win_w: int,
254 ) -> SolveYield:
255 """Classify a no-matrix sliding-window row as WARMUP or DEGENERATE.
257 Distinguishes an insufficient-history warm-up row (mask non-empty but
258 fewer than ``win_w`` rows seen) from a genuine no-data / model-failure
259 row, which is zeroed and marked degenerate.
260 """
261 if mask.any() and i + 1 < win_w:
262 return i, t, mask, None, SolveStatus.WARMUP
263 return i, t, mask, np.zeros(int(mask.sum())), SolveStatus.DEGENERATE