Coverage for src/jquantstats/_portfolio_cost.py: 100%
72 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +0000
1"""Cost analysis mixin for Portfolio."""
3from __future__ import annotations
5import math
7import numpy as np
8import polars as pl
10from ._portfolio_base import _PortfolioMembers
11from ._stats._core import _std_is_negligible
12from .exceptions import InvalidMaxBpsError, NegativeAnnualFeeError, NegativeCostBpsError
15class PortfolioCostMixin(_PortfolioMembers):
16 """Mixin providing cost analysis methods for Portfolio."""
18 @property
19 def position_delta_costs(self) -> pl.DataFrame:
20 """Daily trading cost using the position-delta model.
22 Computes the per-period cost as::
24 cost_t = sum_i( |x_{i,t} - x_{i,t-1}| ) * cost_per_unit
26 where ``x_{i,t}`` is the cash position in asset *i* at time *t* and
27 ``cost_per_unit`` is the one-way cost per unit of traded notional.
28 The first row is always zero because there is no prior position to
29 form a difference against.
31 Returns:
32 pl.DataFrame: Frame with an optional ``'date'`` column and a
33 ``'cost'`` column (absolute cash cost per period).
35 Examples:
36 >>> from jquantstats.portfolio import Portfolio
37 >>> import polars as pl
38 >>> from datetime import date
39 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
40 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
41 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
42 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5, cost_per_unit=0.01)
43 >>> pf.position_delta_costs["cost"].to_list()
44 [0.0, 2.0, 3.0]
45 """
46 assets = [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()]
47 abs_position_changes = pl.sum_horizontal(pl.col(c).diff().abs().fill_null(0.0).fill_nan(0.0) for c in assets)
48 daily_cost = (abs_position_changes * self.cost_per_unit).alias("cost")
49 cols: list[str | pl.Expr] = []
50 if "date" in self.cashposition.columns:
51 cols.append("date")
52 cols.append(daily_cost)
53 return self.cashposition.select(cols)
55 @property
56 def net_cost_nav(self) -> pl.DataFrame:
57 """Net-of-cost cumulative additive NAV using the position-delta cost model.
59 Deducts `position_delta_costs` from daily portfolio profit and
60 computes the running cumulative sum offset by AUM. The result
61 represents the realised NAV path a strategy would achieve after paying
62 ``cost_per_unit`` on every unit of position change.
64 When ``cost_per_unit`` is zero the result equals `nav_accumulated`.
66 Returns:
67 pl.DataFrame: Frame with an optional ``'date'`` column,
68 ``'profit'``, ``'cost'``, and ``'NAV_accumulated_net'`` columns.
70 Examples:
71 >>> from jquantstats.portfolio import Portfolio
72 >>> import polars as pl
73 >>> from datetime import date
74 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
75 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
76 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
77 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5, cost_per_unit=0.0)
78 >>> net = pf.net_cost_nav
79 >>> list(net.columns)
80 ['date', 'profit', 'cost', 'NAV_accumulated_net']
81 """
82 profit_df = self.profit
83 cost_df = self.position_delta_costs
84 if "date" in profit_df.columns:
85 df = profit_df.join(cost_df, on="date", how="left")
86 else:
87 df = profit_df.hstack(cost_df.select(["cost"]))
88 return df.with_columns(((pl.col("profit") - pl.col("cost")).cum_sum() + self.aum).alias("NAV_accumulated_net"))
90 def cost_adjusted_returns(self, cost_bps: float | None = None) -> pl.DataFrame:
91 """Return daily portfolio returns net of estimated one-way trading costs.
93 Trading costs are modelled as a linear function of daily one-way
94 turnover: for every unit of AUM traded, the strategy incurs
95 ``cost_bps`` basis points (i.e. ``cost_bps / 10_000`` fractional
96 cost). The daily cost deduction is therefore::
98 daily_cost = turnover * (cost_bps / 10_000)
100 where ``turnover`` is the fraction-of-AUM one-way turnover already
101 computed by `turnover`. The deduction is applied to the
102 ``returns`` column of `returns`, leaving all other columns
103 (including ``date``) untouched.
105 Args:
106 cost_bps: One-way trading cost in basis points per unit of AUM
107 traded. Must be non-negative. Defaults to ``self.cost_bps``
108 set at construction time.
110 Returns:
111 pl.DataFrame: Same schema as `returns` but with the
112 ``returns`` column reduced by the per-period trading cost.
114 Raises:
115 TypeError: If ``cost_bps`` is not a number.
116 ValueError: If ``cost_bps`` is not finite (NaN or infinity).
117 NegativeCostBpsError: If ``cost_bps`` is negative.
119 Examples:
120 >>> from jquantstats.portfolio import Portfolio
121 >>> import polars as pl
122 >>> from datetime import date
123 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
124 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
125 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
126 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
127 >>> adj = pf.cost_adjusted_returns(0.0)
128 >>> float(adj["returns"][1]) == float(pf.returns["returns"][1])
129 True
130 """
131 effective_bps = cost_bps if cost_bps is not None else self.cost_bps
132 if isinstance(effective_bps, bool) or not isinstance(effective_bps, int | float):
133 raise TypeError(f"cost_bps must be a number, got {type(effective_bps).__name__}") # noqa: TRY003
134 effective_bps = float(effective_bps)
135 if not math.isfinite(effective_bps):
136 raise ValueError(f"cost_bps must be finite, got {effective_bps}") # noqa: TRY003
137 if effective_bps < 0:
138 raise NegativeCostBpsError(effective_bps)
139 base = self.returns
140 daily_cost = self.turnover["turnover"] * (effective_bps / 10_000.0)
141 return base.with_columns((pl.col("returns") - daily_cost).alias("returns"))
143 def deduct_management_fee(
144 self,
145 annual_fee: float | None = None,
146 base: pl.DataFrame | None = None,
147 ) -> pl.DataFrame:
148 """Return daily portfolio returns net of a flat annual management fee.
150 Management fees accrue per calendar day on total AUM regardless of
151 trading activity. The per-period deduction is::
153 daily_deduction_t = annual_fee * days_elapsed_t / 365
155 where ``days_elapsed_t`` is the number of calendar days between row
156 *t* and the previous row. Weekends and holidays are therefore charged
157 to the next trading day. The first row always accrues zero (no prior
158 date), consistent with the turnover convention.
160 The deduction is linear and composes naturally with
161 `cost_adjusted_returns`::
163 adj = pf.cost_adjusted_returns(cost_bps=5)
164 net = pf.deduct_management_fee(annual_fee=0.0085, base=adj)
166 When no ``date`` column is present every period is assumed to span
167 exactly one calendar day.
169 Args:
170 annual_fee: Flat annual management fee as a fraction (e.g. 0.0085
171 for 85 bps). Must be non-negative. Defaults to
172 ``self.annual_fee`` set at construction time.
173 base: Returns DataFrame to deduct the fee from. Must have the
174 same schema as `returns` (``'returns'`` column, optional
175 ``'date'`` column). Defaults to ``self.returns``.
177 Returns:
178 pl.DataFrame: Same schema as *base* (or `returns`) but with the
179 ``returns`` column reduced by the pro-rata daily fee.
181 Raises:
182 TypeError: If ``annual_fee`` is not a number.
183 ValueError: If ``annual_fee`` is not finite (NaN or infinity).
184 NegativeAnnualFeeError: If ``annual_fee`` is negative.
186 Examples:
187 >>> from jquantstats.portfolio import Portfolio
188 >>> import polars as pl
189 >>> from datetime import date
190 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
191 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
192 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1000.0, 1000.0]})
193 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
194 >>> net = pf.deduct_management_fee(annual_fee=0.0)
195 >>> float(net["returns"][1]) == float(pf.returns["returns"][1])
196 True
197 """
198 effective_fee = annual_fee if annual_fee is not None else self.annual_fee
199 if isinstance(effective_fee, bool) or not isinstance(effective_fee, int | float):
200 raise TypeError(f"annual_fee must be a number, got {type(effective_fee).__name__}") # noqa: TRY003
201 effective_fee = float(effective_fee)
202 if not math.isfinite(effective_fee):
203 raise ValueError(f"annual_fee must be finite, got {effective_fee}") # noqa: TRY003
204 if effective_fee < 0:
205 raise NegativeAnnualFeeError(effective_fee)
206 if base is None:
207 base = self.returns
208 if "date" in base.columns and base["date"].dtype.is_temporal():
209 days_elapsed = base["date"].diff().dt.total_days().fill_null(0).cast(pl.Float64)
210 else:
211 days_elapsed = pl.Series([0.0] + [1.0] * (base.height - 1))
212 daily_deduction = days_elapsed * (effective_fee / 365.0)
213 return base.with_columns((pl.col("returns") - daily_deduction).alias("returns"))
215 def trading_cost_impact(self, max_bps: int = 20) -> pl.DataFrame:
216 """Estimate the impact of trading costs on the Sharpe ratio.
218 Computes the annualised Sharpe ratio of cost-adjusted returns for
219 each integer cost level from 0 up to and including ``max_bps`` basis
220 points (1 bp = 0.01 %). The result lets you quickly assess at what
221 cost level the strategy's edge is eroded.
223 Args:
224 max_bps: Maximum one-way trading cost to evaluate, in basis
225 points. Defaults to 20 (i.e., evaluates 0, 1, 2, …, 20
226 bps). Must be a positive integer.
228 Returns:
229 pl.DataFrame: Frame with columns ``'cost_bps'`` (Int64) and
230 ``'sharpe'`` (Float64), one row per cost level from 0 to
231 ``max_bps`` inclusive.
233 Raises:
234 InvalidMaxBpsError: If ``max_bps`` is not a positive integer.
236 Examples:
237 >>> from jquantstats.portfolio import Portfolio
238 >>> import polars as pl
239 >>> from datetime import date, timedelta
240 >>> import numpy as np
241 >>> start = date(2020, 1, 1)
242 >>> dates = pl.date_range(
243 ... start=start, end=start + timedelta(days=99), interval="1d", eager=True
244 ... )
245 >>> rng = np.random.default_rng(0)
246 >>> prices = pl.DataFrame({
247 ... "date": dates,
248 ... "A": pl.Series(np.cumprod(1 + rng.normal(0.001, 0.01, 100)) * 100),
249 ... })
250 >>> pos = pl.DataFrame({"date": dates, "A": pl.Series(np.ones(100) * 1000.0)})
251 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
252 >>> impact = pf.trading_cost_impact(max_bps=5)
253 >>> list(impact["cost_bps"])
254 [0, 1, 2, 3, 4, 5]
255 """
256 if not isinstance(max_bps, int) or max_bps < 1:
257 raise InvalidMaxBpsError(max_bps)
258 periods = self.data._periods_per_year # one Data object, outside the loop
259 sqrt_periods = float(np.sqrt(periods))
260 cost_levels = list(range(max_bps + 1))
262 # Extract base returns and turnover once — O(1) allocations regardless of max_bps
263 base_rets = self.returns["returns"]
264 turnover_s = self.turnover["turnover"]
266 # Build all cost-adjusted return columns in one vectorised DataFrame construction,
267 # then compute means and stds in a single aggregate pass (no per-iteration allocation).
268 sweep = pl.DataFrame({str(bps): base_rets - turnover_s * (bps / 10_000.0) for bps in cost_levels})
269 means_row = sweep.mean().row(0)
270 stds_row = sweep.std(ddof=1).row(0)
272 sharpe_values: list[float] = []
273 for mean_raw, std_raw in zip(means_row, stds_row, strict=False):
274 mean_val = 0.0 if mean_raw is None else float(mean_raw)
275 if _std_is_negligible(std_raw, mean_val):
276 sharpe_values.append(float("nan"))
277 else:
278 sharpe_values.append(mean_val / float(std_raw) * sqrt_periods)
279 return pl.DataFrame({"cost_bps": pl.Series(cost_levels, dtype=pl.Int64), "sharpe": pl.Series(sharpe_values)})