Coverage for src/jquantstats/_portfolio_constructors.py: 100%
86 statements
« prev ^ index » next coverage.py v7.17.1, created at 2026-11-03 15:52 +0000
1"""Factory classmethods for constructing Portfolio objects.
3`PortfolioConstructorMixin` provides the `from_risk_position`,
4`from_position` and `from_weights` entry points (plus the shared
5`_evaluate_position_expr` helper). All funnel through
6``cls.from_cash_position`` — defined on the `Portfolio` dataclass — so the
7actual instantiation happens in one place.
8"""
10from __future__ import annotations
12import math
13from typing import TYPE_CHECKING, Self
15import numpy as np
16import polars as pl
18from ._cost_model import CostModel
19from ._portfolio_base import _PortfolioMembers
20from ._rebalance import simulate_rebalancing
21from .exceptions import NegativeCostBpsError, PositionExprColumnError, RowCountMismatchError, WeightsColumnError
24def _evaluate_position_expr(prices: pl.DataFrame, expr: pl.Expr, param: str) -> pl.DataFrame:
25 """Evaluate a position expression against *prices* and validate the result.
27 Args:
28 prices: Price levels per asset over time.
29 expr: Polars expression producing positions, evaluated via
30 ``prices.with_columns(expr)``.
31 param: Name of the parameter the expression was passed as (used in
32 the error message).
34 Returns:
35 The evaluated positions frame, guaranteed to have the same columns
36 as *prices*.
38 Raises:
39 PositionExprColumnError: If the expression created columns that do
40 not exist in *prices* — those would leave the original asset
41 columns untouched, silently treating raw prices as positions.
42 """
43 evaluated = prices.with_columns(expr)
44 extra = [c for c in evaluated.columns if c not in prices.columns]
45 if extra:
46 raise PositionExprColumnError(param, extra)
47 return evaluated
50def _validate_cost_bps(cost_bps: float) -> float:
51 """Validate a one-way trading cost in basis points.
53 Args:
54 cost_bps: Candidate cost.
56 Returns:
57 The cost as a float.
59 Raises:
60 TypeError: If *cost_bps* is not a number.
61 ValueError: If *cost_bps* is not finite.
62 NegativeCostBpsError: If *cost_bps* is negative.
63 """
64 if isinstance(cost_bps, bool) or not isinstance(cost_bps, int | float):
65 raise TypeError(f"cost_bps must be a number, got {type(cost_bps).__name__}") # noqa: TRY003
66 if not math.isfinite(cost_bps):
67 raise ValueError(f"cost_bps must be finite, got {cost_bps}") # noqa: TRY003
68 if cost_bps < 0:
69 raise NegativeCostBpsError(cost_bps)
70 return float(cost_bps)
73def _weights_matrix(prices: pl.DataFrame, weights: pl.DataFrame, assets: list[str]) -> np.ndarray:
74 """Align *weights* to the asset columns of *prices* as a ``(t, n)`` array.
76 Args:
77 prices: Price levels per asset over time.
78 weights: Target weights, null on rows that do not rebalance.
79 assets: Numeric asset columns of *prices*, in order.
81 Returns:
82 Float array with NaN wherever no weight is given, including for an
83 asset that has no column in *weights* at all.
85 Raises:
86 RowCountMismatchError: If *weights* and *prices* differ in length.
87 WeightsColumnError: If *weights* has a numeric column that is not an
88 asset of *prices*.
89 """
90 if weights.height != prices.height:
91 raise RowCountMismatchError(prices.height, weights.height, "weights")
92 numeric = [c for c, dtype in weights.schema.items() if dtype.is_numeric()]
93 unknown = [c for c in numeric if c not in assets]
94 if unknown:
95 raise WeightsColumnError(unknown)
96 aligned = pl.select(
97 (weights[a] if a in numeric else pl.repeat(None, prices.height, eager=True)).cast(pl.Float64).alias(a)
98 for a in assets
99 )
100 return aligned.fill_nan(None).to_numpy().astype(float)
103def _validate_vol_cap(vol_cap: float | None) -> None:
104 """Validate the optional ``vol_cap`` lower bound.
106 Args:
107 vol_cap: Candidate lower bound for the EWMA volatility estimate.
109 Raises:
110 ValueError: If *vol_cap* is provided but not strictly positive.
111 """
112 if vol_cap is not None and vol_cap <= 0:
113 raise ValueError(f"vol_cap must be a positive number when provided, got {vol_cap!r}") # noqa: TRY003
116def _validate_vola(vola: int | dict[str, int], assets: list[str]) -> None:
117 """Validate the EWMA ``vola`` span specification against the asset columns.
119 Args:
120 vola: A single span applied to every asset, or a per-asset span dict.
121 assets: Numeric column names available in the price frame.
123 Raises:
124 ValueError: If a dict key matches no numeric column, or any span is
125 not a positive integer.
126 """
127 if isinstance(vola, dict):
128 unknown = set(vola.keys()) - set(assets)
129 if unknown:
130 raise ValueError( # noqa: TRY003
131 f"vola dict contains keys that do not match any numeric column in prices: {sorted(unknown)}"
132 )
133 for asset, span in vola.items():
134 if int(span) <= 0:
135 raise ValueError(f"vola span for '{asset}' must be a positive integer, got {span!r}") # noqa: TRY003
136 elif int(vola) <= 0:
137 raise ValueError(f"vola span must be a positive integer, got {vola!r}") # noqa: TRY003
140class PortfolioConstructorMixin(_PortfolioMembers):
141 """Mixin providing the risk- and notional-position factory classmethods."""
143 if TYPE_CHECKING:
145 @classmethod
146 def from_cash_position(
147 cls,
148 prices: pl.DataFrame,
149 cash_position: pl.DataFrame,
150 aum: float,
151 cost_per_unit: float = 0.0,
152 cost_bps: float = 0.0,
153 cost_model: CostModel | None = None,
154 annual_fee: float = 0.0,
155 compounding: bool = False,
156 realized_costs: pl.DataFrame | None = None,
157 ) -> Self:
158 """Create a Portfolio directly from cash positions aligned with prices."""
159 ...
161 @classmethod
162 def from_risk_position(
163 cls,
164 prices: pl.DataFrame,
165 risk_position: pl.DataFrame | pl.Expr,
166 aum: float,
167 vola: int | dict[str, int] = 32,
168 vol_cap: float | None = None,
169 cost_per_unit: float = 0.0,
170 cost_bps: float = 0.0,
171 cost_model: CostModel | None = None,
172 annual_fee: float = 0.0,
173 ) -> Self:
174 """Create a Portfolio from per-asset risk positions.
176 De-volatizes each risk position using an EWMA volatility estimate
177 derived from the corresponding price series.
179 Args:
180 prices: Price levels per asset over time. A temporal column is
181 normalised to ``'date'`` at construction, whatever it was named.
182 risk_position: Risk units per asset aligned with prices.
183 vola: EWMA lookback (span-equivalent) used to estimate volatility.
184 Pass an ``int`` to apply the same span to every asset, or a
185 ``dict[str, int]`` to set a per-asset span (assets absent from
186 the dict default to ``32``). Every span value must be a
187 positive integer; a ``ValueError`` is raised otherwise. Dict
188 keys that do not correspond to any numeric column in *prices*
189 also raise a ``ValueError``.
190 vol_cap: Optional lower bound for the EWMA volatility estimate.
191 When provided, the vol series is clipped from below at this
192 value before dividing the risk position, preventing
193 position blow-up in calm, low-volatility regimes. For
194 example, ``vol_cap=0.05`` ensures annualised vol is never
195 estimated below 5%. Must be positive when not ``None``.
196 aum: Assets under management used as the base NAV offset.
197 cost_per_unit: One-way trading cost per unit of position change.
198 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
199 cost_bps: One-way trading cost in basis points of AUM turnover.
200 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
201 cost_model: Optional `CostModel`
202 instance. When supplied, its ``cost_per_unit`` and
203 ``cost_bps`` values take precedence over the individual
204 parameters above.
205 annual_fee: Flat annual management fee as a fraction of AUM
206 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
207 Used as the default by `deduct_management_fee`.
209 Returns:
210 A Portfolio instance whose cash positions are risk_position
211 divided by EWMA volatility.
213 Raises:
214 ValueError: If any span value in *vola* is ≤ 0, or if a key in a
215 *vola* dict does not match any numeric column in *prices*, or
216 if *vol_cap* is provided but is not positive.
217 PositionExprColumnError: If *risk_position* is an expression that
218 creates columns not present in *prices*.
219 """
220 if isinstance(risk_position, pl.Expr):
221 risk_position = _evaluate_position_expr(prices, risk_position, "risk_position")
222 if cost_model is not None:
223 cost_per_unit = cost_model.cost_per_unit
224 cost_bps = cost_model.cost_bps
225 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
227 _validate_vol_cap(vol_cap)
228 _validate_vola(vola, assets)
230 def _span(asset: str) -> int:
231 """Return the EWMA span for *asset*, falling back to 32 if not specified."""
232 if isinstance(vola, dict):
233 return int(vola.get(asset, 32))
234 return int(vola)
236 def _vol(asset: str) -> pl.Series:
237 """Return the EWMA volatility series for *asset*, optionally clipped from below."""
238 vol = prices[asset].pct_change().ewm_std(com=_span(asset) - 1, adjust=True, min_samples=_span(asset))
239 if vol_cap is not None:
240 vol = vol.clip(lower_bound=vol_cap)
241 return vol
243 cash_position = risk_position.with_columns((pl.col(asset) / _vol(asset)).alias(asset) for asset in assets)
244 return cls.from_cash_position(
245 prices=prices,
246 cash_position=cash_position,
247 aum=aum,
248 cost_per_unit=cost_per_unit,
249 cost_bps=cost_bps,
250 annual_fee=annual_fee,
251 )
253 @classmethod
254 def from_position(
255 cls,
256 prices: pl.DataFrame,
257 position: pl.DataFrame | pl.Expr,
258 aum: float,
259 cost_per_unit: float = 0.0,
260 cost_bps: float = 0.0,
261 cost_model: CostModel | None = None,
262 annual_fee: float = 0.0,
263 ) -> Self:
264 """Create a Portfolio from share/unit positions.
266 Converts *position* (number of units held per asset) to cash exposure
267 by multiplying element-wise with *prices*, then delegates to
268 :py`from_cash_position`.
270 Args:
271 prices: Price levels per asset over time. A temporal column is
272 normalised to ``'date'`` at construction, whatever it was named.
273 position: Number of units held per asset over time, aligned with
274 *prices*. Non-numeric columns (e.g. ``'date'``) are passed
275 through unchanged.
276 aum: Assets under management used as the base NAV offset.
277 cost_per_unit: One-way trading cost per unit of position change.
278 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
279 cost_bps: One-way trading cost in basis points of AUM turnover.
280 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
281 cost_model: Optional `CostModel` instance.
282 When supplied, its ``cost_per_unit`` and ``cost_bps`` values
283 take precedence over the individual parameters above.
284 annual_fee: Flat annual management fee as a fraction of AUM
285 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
286 Used as the default by `deduct_management_fee`.
288 Returns:
289 A Portfolio instance whose cash positions equal *position* x *prices*.
291 Raises:
292 PositionExprColumnError: If *position* is an expression that
293 creates columns not present in *prices*.
295 Examples:
296 >>> import polars as pl
297 >>> from jquantstats.portfolio import Portfolio
298 >>> prices = pl.DataFrame({"A": [100.0, 110.0, 105.0]})
299 >>> pos = pl.DataFrame({"A": [10.0, 10.0, 10.0]})
300 >>> pf = Portfolio.from_position(prices=prices, position=pos, aum=1e6)
301 >>> pf.cashposition["A"].to_list()
302 [1000.0, 1100.0, 1050.0]
303 """
304 if isinstance(position, pl.Expr):
305 position = _evaluate_position_expr(prices, position, "position")
306 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
307 cash_position = position.with_columns((pl.col(asset) * prices[asset]).alias(asset) for asset in assets)
308 return cls.from_cash_position(
309 prices=prices,
310 cash_position=cash_position,
311 aum=aum,
312 cost_per_unit=cost_per_unit,
313 cost_bps=cost_bps,
314 cost_model=cost_model,
315 annual_fee=annual_fee,
316 )
318 @classmethod
319 def from_weights(
320 cls,
321 prices: pl.DataFrame,
322 weights: pl.DataFrame | pl.Expr,
323 aum: float,
324 cost_bps: float = 0.0,
325 compounding: bool = True,
326 annual_fee: float = 0.0,
327 ) -> Self:
328 """Create a Portfolio that rebalances to target weights on sparse dates.
330 *weights* gives the target fraction of NAV per asset, but only on the
331 rows where the portfolio rebalances — every other row is null. Between
332 two rebalancings the book holds its **units**, so the realised weights
333 drift with prices rather than staying at the targets. On a rebalancing
334 row the units are reset to ``NAV * w / price``, where NAV is the value
335 the drifted book has reached, and the trade pays ``cost_bps`` on its
336 traded notional out of cash. Because each rebalancing is sized off a
337 NAV that every earlier cost has already lowered, the book is simulated
338 row by row rather than vectorised.
340 Weights are applied at the price of the row they appear on, matching
341 the convention that a position set on row *t* earns the return from
342 *t* to *t + 1*. Shift the weights first (``weights.shift(n)``) to
343 trade them *n* rows later. Weights that sum to less than one leave the
344 remainder in cash, which earns nothing; weights above one borrow it.
346 The trading costs are realised inside the simulation and stored as
347 ``realized_costs``, so `profits`, NAV and `returns` are already net of
348 them. The portfolio's own ``cost_bps`` is therefore left at zero:
349 `cost_adjusted_returns` would otherwise charge the same trades twice.
351 Args:
352 prices: Price levels per asset over time. A temporal column is
353 normalised to ``'date'`` at construction, whatever it was named.
354 weights: Target weights per asset, aligned row for row with
355 *prices*, as a DataFrame or as an expression evaluated against
356 *prices*. A row with at least one non-null weight is a
357 rebalancing; within it, a null weight or a missing asset
358 column means a target of zero. Non-numeric columns (e.g.
359 ``'date'``) are ignored.
360 aum: Starting capital, held in cash until the first rebalancing.
361 cost_bps: One-way trading cost in basis points of traded notional.
362 Defaults to 0.0 (no cost).
363 compounding: When True (the default) each rebalancing is sized off
364 the current NAV and `returns` are measured against the
365 previous period's NAV. When False every rebalancing is sized
366 off the constant *aum*, as a fixed-notional book, and
367 `returns` are measured against *aum*.
368 annual_fee: Flat annual management fee as a fraction of AUM, used
369 as the default by `deduct_management_fee`. Defaults to 0.0.
371 Returns:
372 A Portfolio whose cash positions are the simulated units times
373 prices, with ``compounding`` set as requested and the costs paid
374 on every trade in ``realized_costs``.
376 Raises:
377 RowCountMismatchError: If *weights* and *prices* differ in length.
378 WeightsColumnError: If *weights* has a numeric column that is not
379 an asset of *prices*.
380 PositionExprColumnError: If *weights* is an expression that creates
381 columns not present in *prices*.
382 RebalancePriceError: If an asset that is held or targeted has no
383 finite, positive price on a rebalancing row.
384 NegativeCostBpsError: If *cost_bps* is negative.
386 Examples:
387 >>> import polars as pl
388 >>> from jquantstats.portfolio import Portfolio
389 >>> prices = pl.DataFrame({"A": [100.0, 110.0, 110.0], "B": [100.0, 90.0, 90.0]})
390 >>> weights = pl.DataFrame({"A": [0.5, None, 0.5], "B": [0.5, None, 0.5]})
391 >>> pf = Portfolio.from_weights(prices=prices, weights=weights, aum=1000.0)
392 >>> [round(u, 4) for u in pf.units["A"].to_list()] # held through row 1, reset on row 2
393 [5.0, 5.0, 4.5455]
394 >>> [round(w, 2) for w in pf.weights["A"].to_list()] # drifted on row 1
395 [0.5, 0.55, 0.5]
396 """
397 if isinstance(weights, pl.Expr):
398 weights = _evaluate_position_expr(prices, weights, "weights")
399 cost_bps = _validate_cost_bps(cost_bps)
400 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
401 target = _weights_matrix(prices, weights, assets)
402 price_matrix = prices.select(pl.col(a).cast(pl.Float64) for a in assets).fill_nan(None).to_numpy().astype(float)
404 path = simulate_rebalancing(
405 prices=price_matrix,
406 weights=np.nan_to_num(target),
407 is_rebalancing=np.asarray(~np.isnan(target).all(axis=1)),
408 aum=aum,
409 assets=assets,
410 cost_bps=cost_bps,
411 compounding=compounding,
412 )
413 cash_position = prices.with_columns(pl.Series(a, path.units[:, i]) * pl.col(a) for i, a in enumerate(assets))
414 realized_costs = prices.with_columns(pl.Series(a, path.costs[:, i]) for i, a in enumerate(assets))
415 return cls.from_cash_position(
416 prices=prices,
417 cash_position=cash_position,
418 aum=aum,
419 annual_fee=annual_fee,
420 compounding=compounding,
421 realized_costs=realized_costs,
422 )