Coverage for src/tinycta/_kernel.py: 100%
37 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-09-15 05:23 +0000
1"""Pure-NumPy numeric kernel for the correlation-aware position optimizer.
3Every function here operates on NumPy arrays only — no Polars, no :class:`~tinycta.config.Config`
4— so the Polars-facing orchestration in :mod:`tinycta.engine` stays a thin adapter that
5prepares arrays, delegates the timestamp walk to :func:`forward_walk`, and writes the result
6back into a DataFrame.
8This module is on the core import path (``tinycta.engine`` imports it at module level), so it
9must import nothing outside ``[project].dependencies``. In particular it does **not** log:
10``loguru`` ships only with the optional ``hyper`` extra, and importing it here would break
11``pip install tinycta`` for every user who never asked for that extra. Degenerate cases are
12made observable through return values instead — see :func:`_risk_position`.
13"""
15from __future__ import annotations
17from collections.abc import Hashable
19import numpy as np
21from .linalg import inv_a_norm as _inv_a_norm
22from .linalg import solve as _solve
23from .signal import shrink2id as _shrink2id
26def _denominator_is_degenerate(denom: float) -> bool:
27 """Return True when the correlation-norm denominator is effectively zero.
29 ``1e-12`` is an arbitrary epsilon floor: any nearby threshold value or a
30 ``<`` vs ``<=`` boundary behaves identically for realistic denominators, so
31 this comparison is intentionally excluded from mutation.
32 """
33 return denom <= 1e-12 # pragma: no mutate
36def _risk_position(corr: np.ndarray, mu_row: np.ndarray, mask: np.ndarray, shrink: float) -> np.ndarray:
37 """Solve the shrunk correlation system for one timestamp's tradable assets.
39 Shrinks ``corr`` towards the identity by ``shrink`` (via
40 :func:`~tinycta.signal.shrink2id`), restricts it to the masked assets, solves
41 for the expected returns ``mu_row`` and normalises by ``inv_a_norm`` so the raw
42 risk position has unit norm under the correlation metric. Returns zeros when the
43 normaliser is non-finite/degenerate or ``mu_row`` is all-zero.
45 Args:
46 corr: Full EWMA correlation matrix for the timestamp.
47 mu_row: Expected returns for every asset at the timestamp (NaNs tolerated).
48 mask: Boolean mask of currently-tradable assets.
49 shrink: Identity-shrinkage weight in ``[0, 1]``.
51 Returns:
52 np.ndarray: The normalised risk position over the masked assets.
54 Example:
55 >>> import numpy as np
56 >>> from tinycta._kernel import _risk_position
57 >>> both = np.array([True, True])
59 With an identity correlation the solve is trivial: the asset carrying the
60 expected return takes the whole (unit-norm) position and the other takes none.
62 >>> _risk_position(np.eye(2), np.array([1.0, 0.0]), both, shrink=1.0)
63 array([1., 0.])
65 A correlated pair tilts against the correlation — holding the second asset
66 short hedges the first, and the norm stays unit under the shrunk metric:
68 >>> corr = np.array([[1.0, 0.8], [0.8, 1.0]])
69 >>> _risk_position(corr, np.array([1.0, 0.0]), both, shrink=0.5).round(4)
70 array([ 1.0911, -0.4364])
72 An all-zero ``mu`` is degenerate and falls back to zeros rather than
73 dividing by a vanishing normaliser:
75 >>> _risk_position(np.eye(2), np.array([0.0, 0.0]), both, shrink=1.0)
76 array([0., 0.])
78 ``NaN`` expected returns are tolerated — they are zeroed, not propagated:
80 >>> _risk_position(np.eye(2), np.array([1.0, np.nan]), both, shrink=1.0)
81 array([1., 0.])
83 The result spans only the masked assets, so an untradable asset is absent
84 from the output rather than present as a zero:
86 >>> _risk_position(corr, np.array([1.0, 2.0]), np.array([True, False]), shrink=1.0)
87 array([1.])
88 """
89 matrix = _shrink2id(corr, lamb=shrink)[np.ix_(mask, mask)]
90 expected_mu = np.nan_to_num(mu_row[mask])
91 denom = _inv_a_norm(expected_mu, matrix)
92 if denom is None or not np.isfinite(denom) or _denominator_is_degenerate(denom) or np.allclose(expected_mu, 0.0):
93 return np.zeros_like(expected_mu)
94 return _solve(matrix, expected_mu) / denom
97def _update_profit_variance(
98 profit_variance: float,
99 cash_pos_prev: np.ndarray,
100 returns_row: np.ndarray,
101 ret_mask: np.ndarray,
102 lamb: float,
103) -> float:
104 """EWMA-update the running profit-variance estimate with one period's P&L.
106 Realised profit is the previous cash position dotted with the current returns
107 over the jointly-finite assets; the variance decays towards the new squared
108 profit by ``1 - lamb``.
110 Args:
111 profit_variance: Previous running profit-variance estimate.
112 cash_pos_prev: Previous timestamp's cash position (NaNs tolerated).
113 returns_row: Current timestamp's simple returns (NaNs tolerated).
114 ret_mask: Boolean mask of assets finite in both rows.
115 lamb: EWMA decay factor.
117 Returns:
118 float: The updated profit-variance estimate.
120 Example:
121 >>> import numpy as np
122 >>> from tinycta._kernel import _update_profit_variance
123 >>> both = np.array([True, True])
125 A position of 2.0 against a 10% return realises a profit of 0.2, so the
126 estimate decays towards ``0.2 ** 2`` by ``1 - lamb``:
128 >>> _update_profit_variance(1.0, np.array([2.0, 0.0]), np.array([0.1, 0.0]), both, lamb=0.99)
129 0.9904
131 A flat book realises nothing, so the estimate simply decays:
133 >>> _update_profit_variance(1.0, np.array([0.0, 0.0]), np.array([0.1, 0.2]), both, lamb=0.99)
134 0.99
136 ``NaN`` on either side is zeroed rather than poisoning the estimate — here
137 the second asset contributes nothing and the result matches the first case:
139 >>> _update_profit_variance(1.0, np.array([2.0, np.nan]), np.array([0.1, 0.5]), both, lamb=0.99)
140 0.9904
141 """
142 lhs = np.nan_to_num(cash_pos_prev[ret_mask], nan=0.0)
143 rhs = np.nan_to_num(returns_row[ret_mask], nan=0.0)
144 profit = lhs @ rhs
145 return float(lamb * profit_variance + (1 - lamb) * profit**2)
148def forward_walk(
149 cor: dict[Hashable, np.ndarray],
150 prices_num: np.ndarray,
151 returns_num: np.ndarray,
152 mu: np.ndarray,
153 vola_np: np.ndarray,
154 risk_pos_np: np.ndarray,
155 cash_pos_np: np.ndarray,
156 row_of: dict[Hashable, int],
157 shrink: float,
158) -> None:
159 """Walk forward through the post-warmup timestamps, filling positions in place.
161 Mutates ``risk_pos_np`` and ``cash_pos_np`` row-by-row. At each timestamp the
162 previous period's realised P&L EWMA-updates a running profit-variance estimate
163 (decay ``lamb=0.99``), which scales the freshly-solved risk position before it is
164 divided by per-asset volatility to yield the cash position.
166 Args:
167 cor: Per-timestamp correlation matrices, keyed by ``date`` value.
168 prices_num: Asset prices as a ``(rows, assets)`` array (NaNs tolerated).
169 returns_num: Simple returns aligned to ``prices_num``.
170 mu: Expected returns aligned to ``prices_num``.
171 vola_np: Per-asset EWMA volatility aligned to ``prices_num``.
172 risk_pos_np: Output risk-position buffer, mutated in place.
173 cash_pos_np: Output cash-position buffer, mutated in place.
174 row_of: Map from a ``cor`` key back to its row index.
175 shrink: Identity-shrinkage weight in ``[0, 1]`` passed to :func:`_risk_position`.
177 Example:
178 >>> import numpy as np
179 >>> from tinycta._kernel import forward_walk
180 >>> prices = np.array([[100.0, 50.0], [101.0, 50.5], [102.0, 50.0]])
181 >>> returns = np.zeros_like(prices)
182 >>> returns[1:] = prices[1:] / prices[:-1] - 1.0
183 >>> mu = np.array([[1.0, 0.0], [1.0, 0.0], [1.0, 0.0]])
184 >>> vola = np.full((3, 2), 0.1)
185 >>> risk_pos = np.full((3, 2), np.nan)
186 >>> cash_pos = np.full((3, 2), np.nan)
188 Only rows named by ``cor`` are walked; here the first row is warmup and is
189 left untouched. The function returns nothing and writes into the buffers:
191 >>> forward_walk(
192 ... {1: np.eye(2), 2: np.eye(2)},
193 ... prices, returns, mu, vola, risk_pos, cash_pos,
194 ... row_of={1: 1, 2: 2},
195 ... shrink=1.0,
196 ... ) is None
197 True
198 >>> risk_pos[0]
199 array([nan, nan])
201 The first walked row starts from a profit variance of 1.0, so its risk
202 position is the raw solve, and the cash position divides it by volatility:
204 >>> risk_pos[1]
205 array([1., 0.])
206 >>> cash_pos[1]
207 array([10., 0.])
209 By the next row the realised P&L has updated the profit-variance estimate,
210 which rescales the position:
212 >>> risk_pos[2].round(4)
213 array([1.01, 0. ])
214 """
215 profit_variance = 1.0
216 lamb = 0.99
218 prev_row: int | None = None
219 for t in cor:
220 row = row_of[t]
221 mask = np.isfinite(prices_num[row])
223 if prev_row is not None:
224 ret_mask = np.isfinite(returns_num[row]) & mask
225 if ret_mask.any():
226 cash_pos_np[prev_row] = risk_pos_np[prev_row] / vola_np[prev_row]
227 profit_variance = _update_profit_variance(
228 profit_variance, cash_pos_np[prev_row], returns_num[row], ret_mask, lamb
229 )
231 if mask.any():
232 pos = _risk_position(cor[t], mu[row], mask, shrink)
233 risk_pos_np[row, mask] = pos / profit_variance
234 cash_pos_np[row, mask] = risk_pos_np[row, mask] / vola_np[row, mask]
236 prev_row = row