Coverage for src/jquantstats/_portfolio_constructors.py: 100%
54 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 04:11 +0000
1"""Factory classmethods for constructing Portfolio objects.
3`PortfolioConstructorMixin` provides the `from_risk_position` and
4`from_position` entry points (plus the shared `_evaluate_position_expr`
5helper). Both funnel through ``cls.from_cash_position`` — defined on the
6`Portfolio` dataclass — so the actual instantiation happens in one place.
7"""
9from __future__ import annotations
11from typing import TYPE_CHECKING, Self
13import polars as pl
15from ._cost_model import CostModel
16from ._portfolio_base import _PortfolioMembers
17from .exceptions import PositionExprColumnError
20def _evaluate_position_expr(prices: pl.DataFrame, expr: pl.Expr, param: str) -> pl.DataFrame:
21 """Evaluate a position expression against *prices* and validate the result.
23 Args:
24 prices: Price levels per asset over time.
25 expr: Polars expression producing positions, evaluated via
26 ``prices.with_columns(expr)``.
27 param: Name of the parameter the expression was passed as (used in
28 the error message).
30 Returns:
31 The evaluated positions frame, guaranteed to have the same columns
32 as *prices*.
34 Raises:
35 PositionExprColumnError: If the expression created columns that do
36 not exist in *prices* — those would leave the original asset
37 columns untouched, silently treating raw prices as positions.
38 """
39 evaluated = prices.with_columns(expr)
40 extra = [c for c in evaluated.columns if c not in prices.columns]
41 if extra:
42 raise PositionExprColumnError(param, extra)
43 return evaluated
46def _validate_vol_cap(vol_cap: float | None) -> None:
47 """Validate the optional ``vol_cap`` lower bound.
49 Args:
50 vol_cap: Candidate lower bound for the EWMA volatility estimate.
52 Raises:
53 ValueError: If *vol_cap* is provided but not strictly positive.
54 """
55 if vol_cap is not None and vol_cap <= 0:
56 raise ValueError(f"vol_cap must be a positive number when provided, got {vol_cap!r}") # noqa: TRY003
59def _validate_vola(vola: int | dict[str, int], assets: list[str]) -> None:
60 """Validate the EWMA ``vola`` span specification against the asset columns.
62 Args:
63 vola: A single span applied to every asset, or a per-asset span dict.
64 assets: Numeric column names available in the price frame.
66 Raises:
67 ValueError: If a dict key matches no numeric column, or any span is
68 not a positive integer.
69 """
70 if isinstance(vola, dict):
71 unknown = set(vola.keys()) - set(assets)
72 if unknown:
73 raise ValueError( # noqa: TRY003
74 f"vola dict contains keys that do not match any numeric column in prices: {sorted(unknown)}"
75 )
76 for asset, span in vola.items():
77 if int(span) <= 0:
78 raise ValueError(f"vola span for '{asset}' must be a positive integer, got {span!r}") # noqa: TRY003
79 elif int(vola) <= 0:
80 raise ValueError(f"vola span must be a positive integer, got {vola!r}") # noqa: TRY003
83class PortfolioConstructorMixin(_PortfolioMembers):
84 """Mixin providing the risk- and notional-position factory classmethods."""
86 if TYPE_CHECKING:
88 @classmethod
89 def from_cash_position(
90 cls,
91 prices: pl.DataFrame,
92 cash_position: pl.DataFrame,
93 aum: float,
94 cost_per_unit: float = 0.0,
95 cost_bps: float = 0.0,
96 cost_model: CostModel | None = None,
97 annual_fee: float = 0.0,
98 ) -> Self:
99 """Create a Portfolio directly from cash positions aligned with prices."""
100 ...
102 @classmethod
103 def from_risk_position(
104 cls,
105 prices: pl.DataFrame,
106 risk_position: pl.DataFrame | pl.Expr,
107 aum: float,
108 vola: int | dict[str, int] = 32,
109 vol_cap: float | None = None,
110 cost_per_unit: float = 0.0,
111 cost_bps: float = 0.0,
112 cost_model: CostModel | None = None,
113 annual_fee: float = 0.0,
114 ) -> Self:
115 """Create a Portfolio from per-asset risk positions.
117 De-volatizes each risk position using an EWMA volatility estimate
118 derived from the corresponding price series.
120 Args:
121 prices: Price levels per asset over time. A temporal column is
122 normalised to ``'date'`` at construction, whatever it was named.
123 risk_position: Risk units per asset aligned with prices.
124 vola: EWMA lookback (span-equivalent) used to estimate volatility.
125 Pass an ``int`` to apply the same span to every asset, or a
126 ``dict[str, int]`` to set a per-asset span (assets absent from
127 the dict default to ``32``). Every span value must be a
128 positive integer; a ``ValueError`` is raised otherwise. Dict
129 keys that do not correspond to any numeric column in *prices*
130 also raise a ``ValueError``.
131 vol_cap: Optional lower bound for the EWMA volatility estimate.
132 When provided, the vol series is clipped from below at this
133 value before dividing the risk position, preventing
134 position blow-up in calm, low-volatility regimes. For
135 example, ``vol_cap=0.05`` ensures annualised vol is never
136 estimated below 5%. Must be positive when not ``None``.
137 aum: Assets under management used as the base NAV offset.
138 cost_per_unit: One-way trading cost per unit of position change.
139 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
140 cost_bps: One-way trading cost in basis points of AUM turnover.
141 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
142 cost_model: Optional `CostModel`
143 instance. When supplied, its ``cost_per_unit`` and
144 ``cost_bps`` values take precedence over the individual
145 parameters above.
146 annual_fee: Flat annual management fee as a fraction of AUM
147 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
148 Used as the default by `deduct_management_fee`.
150 Returns:
151 A Portfolio instance whose cash positions are risk_position
152 divided by EWMA volatility.
154 Raises:
155 ValueError: If any span value in *vola* is ≤ 0, or if a key in a
156 *vola* dict does not match any numeric column in *prices*, or
157 if *vol_cap* is provided but is not positive.
158 PositionExprColumnError: If *risk_position* is an expression that
159 creates columns not present in *prices*.
160 """
161 if isinstance(risk_position, pl.Expr):
162 risk_position = _evaluate_position_expr(prices, risk_position, "risk_position")
163 if cost_model is not None:
164 cost_per_unit = cost_model.cost_per_unit
165 cost_bps = cost_model.cost_bps
166 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
168 _validate_vol_cap(vol_cap)
169 _validate_vola(vola, assets)
171 def _span(asset: str) -> int:
172 """Return the EWMA span for *asset*, falling back to 32 if not specified."""
173 if isinstance(vola, dict):
174 return int(vola.get(asset, 32))
175 return int(vola)
177 def _vol(asset: str) -> pl.Series:
178 """Return the EWMA volatility series for *asset*, optionally clipped from below."""
179 vol = prices[asset].pct_change().ewm_std(com=_span(asset) - 1, adjust=True, min_samples=_span(asset))
180 if vol_cap is not None:
181 vol = vol.clip(lower_bound=vol_cap)
182 return vol
184 cash_position = risk_position.with_columns((pl.col(asset) / _vol(asset)).alias(asset) for asset in assets)
185 return cls.from_cash_position(
186 prices=prices,
187 cash_position=cash_position,
188 aum=aum,
189 cost_per_unit=cost_per_unit,
190 cost_bps=cost_bps,
191 annual_fee=annual_fee,
192 )
194 @classmethod
195 def from_position(
196 cls,
197 prices: pl.DataFrame,
198 position: pl.DataFrame | pl.Expr,
199 aum: float,
200 cost_per_unit: float = 0.0,
201 cost_bps: float = 0.0,
202 cost_model: CostModel | None = None,
203 annual_fee: float = 0.0,
204 ) -> Self:
205 """Create a Portfolio from share/unit positions.
207 Converts *position* (number of units held per asset) to cash exposure
208 by multiplying element-wise with *prices*, then delegates to
209 :py`from_cash_position`.
211 Args:
212 prices: Price levels per asset over time. A temporal column is
213 normalised to ``'date'`` at construction, whatever it was named.
214 position: Number of units held per asset over time, aligned with
215 *prices*. Non-numeric columns (e.g. ``'date'``) are passed
216 through unchanged.
217 aum: Assets under management used as the base NAV offset.
218 cost_per_unit: One-way trading cost per unit of position change.
219 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
220 cost_bps: One-way trading cost in basis points of AUM turnover.
221 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
222 cost_model: Optional `CostModel` instance.
223 When supplied, its ``cost_per_unit`` and ``cost_bps`` values
224 take precedence over the individual parameters above.
225 annual_fee: Flat annual management fee as a fraction of AUM
226 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
227 Used as the default by `deduct_management_fee`.
229 Returns:
230 A Portfolio instance whose cash positions equal *position* x *prices*.
232 Raises:
233 PositionExprColumnError: If *position* is an expression that
234 creates columns not present in *prices*.
236 Examples:
237 >>> import polars as pl
238 >>> from jquantstats.portfolio import Portfolio
239 >>> prices = pl.DataFrame({"A": [100.0, 110.0, 105.0]})
240 >>> pos = pl.DataFrame({"A": [10.0, 10.0, 10.0]})
241 >>> pf = Portfolio.from_position(prices=prices, position=pos, aum=1e6)
242 >>> pf.cashposition["A"].to_list()
243 [1000.0, 1100.0, 1050.0]
244 """
245 if isinstance(position, pl.Expr):
246 position = _evaluate_position_expr(prices, position, "position")
247 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
248 cash_position = position.with_columns((pl.col(asset) * prices[asset]).alias(asset) for asset in assets)
249 return cls.from_cash_position(
250 prices=prices,
251 cash_position=cash_position,
252 aum=aum,
253 cost_per_unit=cost_per_unit,
254 cost_bps=cost_bps,
255 cost_model=cost_model,
256 annual_fee=annual_fee,
257 )