Coverage for src/jquantstats/_portfolio_constructors.py: 100%
54 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +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 (may include a date column).
122 risk_position: Risk units per asset aligned with prices.
123 vola: EWMA lookback (span-equivalent) used to estimate volatility.
124 Pass an ``int`` to apply the same span to every asset, or a
125 ``dict[str, int]`` to set a per-asset span (assets absent from
126 the dict default to ``32``). Every span value must be a
127 positive integer; a ``ValueError`` is raised otherwise. Dict
128 keys that do not correspond to any numeric column in *prices*
129 also raise a ``ValueError``.
130 vol_cap: Optional lower bound for the EWMA volatility estimate.
131 When provided, the vol series is clipped from below at this
132 value before dividing the risk position, preventing
133 position blow-up in calm, low-volatility regimes. For
134 example, ``vol_cap=0.05`` ensures annualised vol is never
135 estimated below 5%. Must be positive when not ``None``.
136 aum: Assets under management used as the base NAV offset.
137 cost_per_unit: One-way trading cost per unit of position change.
138 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
139 cost_bps: One-way trading cost in basis points of AUM turnover.
140 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
141 cost_model: Optional `CostModel`
142 instance. When supplied, its ``cost_per_unit`` and
143 ``cost_bps`` values take precedence over the individual
144 parameters above.
145 annual_fee: Flat annual management fee as a fraction of AUM
146 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
147 Used as the default by `deduct_management_fee`.
149 Returns:
150 A Portfolio instance whose cash positions are risk_position
151 divided by EWMA volatility.
153 Raises:
154 ValueError: If any span value in *vola* is ≤ 0, or if a key in a
155 *vola* dict does not match any numeric column in *prices*, or
156 if *vol_cap* is provided but is not positive.
157 PositionExprColumnError: If *risk_position* is an expression that
158 creates columns not present in *prices*.
159 """
160 if isinstance(risk_position, pl.Expr):
161 risk_position = _evaluate_position_expr(prices, risk_position, "risk_position")
162 if cost_model is not None:
163 cost_per_unit = cost_model.cost_per_unit
164 cost_bps = cost_model.cost_bps
165 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
167 _validate_vol_cap(vol_cap)
168 _validate_vola(vola, assets)
170 def _span(asset: str) -> int:
171 """Return the EWMA span for *asset*, falling back to 32 if not specified."""
172 if isinstance(vola, dict):
173 return int(vola.get(asset, 32))
174 return int(vola)
176 def _vol(asset: str) -> pl.Series:
177 """Return the EWMA volatility series for *asset*, optionally clipped from below."""
178 vol = prices[asset].pct_change().ewm_std(com=_span(asset) - 1, adjust=True, min_samples=_span(asset))
179 if vol_cap is not None:
180 vol = vol.clip(lower_bound=vol_cap)
181 return vol
183 cash_position = risk_position.with_columns((pl.col(asset) / _vol(asset)).alias(asset) for asset in assets)
184 return cls.from_cash_position(
185 prices=prices,
186 cash_position=cash_position,
187 aum=aum,
188 cost_per_unit=cost_per_unit,
189 cost_bps=cost_bps,
190 annual_fee=annual_fee,
191 )
193 @classmethod
194 def from_position(
195 cls,
196 prices: pl.DataFrame,
197 position: pl.DataFrame | pl.Expr,
198 aum: float,
199 cost_per_unit: float = 0.0,
200 cost_bps: float = 0.0,
201 cost_model: CostModel | None = None,
202 annual_fee: float = 0.0,
203 ) -> Self:
204 """Create a Portfolio from share/unit positions.
206 Converts *position* (number of units held per asset) to cash exposure
207 by multiplying element-wise with *prices*, then delegates to
208 :py`from_cash_position`.
210 Args:
211 prices: Price levels per asset over time (may include a date column).
212 position: Number of units held per asset over time, aligned with
213 *prices*. Non-numeric columns (e.g. ``'date'``) are passed
214 through unchanged.
215 aum: Assets under management used as the base NAV offset.
216 cost_per_unit: One-way trading cost per unit of position change.
217 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
218 cost_bps: One-way trading cost in basis points of AUM turnover.
219 Defaults to 0.0 (no cost). Ignored when *cost_model* is given.
220 cost_model: Optional `CostModel` instance.
221 When supplied, its ``cost_per_unit`` and ``cost_bps`` values
222 take precedence over the individual parameters above.
223 annual_fee: Flat annual management fee as a fraction of AUM
224 (e.g. 0.0085 for 85 bps p.a.). Defaults to 0.0 (no fee).
225 Used as the default by `deduct_management_fee`.
227 Returns:
228 A Portfolio instance whose cash positions equal *position* x *prices*.
230 Raises:
231 PositionExprColumnError: If *position* is an expression that
232 creates columns not present in *prices*.
234 Examples:
235 >>> import polars as pl
236 >>> from jquantstats.portfolio import Portfolio
237 >>> prices = pl.DataFrame({"A": [100.0, 110.0, 105.0]})
238 >>> pos = pl.DataFrame({"A": [10.0, 10.0, 10.0]})
239 >>> pf = Portfolio.from_position(prices=prices, position=pos, aum=1e6)
240 >>> pf.cashposition["A"].to_list()
241 [1000.0, 1100.0, 1050.0]
242 """
243 if isinstance(position, pl.Expr):
244 position = _evaluate_position_expr(prices, position, "position")
245 assets = [col for col, dtype in prices.schema.items() if dtype.is_numeric()]
246 cash_position = position.with_columns((pl.col(asset) * prices[asset]).alias(asset) for asset in assets)
247 return cls.from_cash_position(
248 prices=prices,
249 cash_position=cash_position,
250 aum=aum,
251 cost_per_unit=cost_per_unit,
252 cost_bps=cost_bps,
253 cost_model=cost_model,
254 annual_fee=annual_fee,
255 )