Coverage for src / jquantstats / _portfolio_cost.py: 100%
49 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-07 14:36 +0000
1"""Cost analysis mixin for Portfolio."""
3from __future__ import annotations
5from typing import TYPE_CHECKING
7import polars as pl
9if TYPE_CHECKING:
10 from .data import Data
13class PortfolioCostMixin:
14 """Mixin providing cost analysis methods for Portfolio."""
16 if TYPE_CHECKING:
17 cashposition: pl.DataFrame
18 aum: float
19 cost_per_unit: float
20 cost_bps: float
21 returns: pl.DataFrame
22 turnover: pl.DataFrame
23 profit: pl.DataFrame
24 data: Data
26 @property
27 def position_delta_costs(self) -> pl.DataFrame:
28 """Daily trading cost using the position-delta model.
30 Computes the per-period cost as::
32 cost_t = sum_i( |x_{i,t} - x_{i,t-1}| ) * cost_per_unit
34 where ``x_{i,t}`` is the cash position in asset *i* at time *t* and
35 ``cost_per_unit`` is the one-way cost per unit of traded notional.
36 The first row is always zero because there is no prior position to
37 form a difference against.
39 Returns:
40 pl.DataFrame: Frame with an optional ``'date'`` column and a
41 ``'cost'`` column (absolute cash cost per period).
43 Examples:
44 >>> from jquantstats.portfolio import Portfolio
45 >>> import polars as pl
46 >>> from datetime import date
47 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
48 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
49 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
50 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5, cost_per_unit=0.01)
51 >>> pf.position_delta_costs["cost"].to_list()
52 [0.0, 2.0, 3.0]
53 """
54 assets = [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()]
55 abs_position_changes = pl.sum_horizontal(pl.col(c).diff().abs().fill_null(0.0).fill_nan(0.0) for c in assets)
56 daily_cost = (abs_position_changes * self.cost_per_unit).alias("cost")
57 cols: list[str | pl.Expr] = []
58 if "date" in self.cashposition.columns:
59 cols.append("date")
60 cols.append(daily_cost)
61 return self.cashposition.select(cols)
63 @property
64 def net_cost_nav(self) -> pl.DataFrame:
65 """Net-of-cost cumulative additive NAV using the position-delta cost model.
67 Deducts :attr:`position_delta_costs` from daily portfolio profit and
68 computes the running cumulative sum offset by AUM. The result
69 represents the realised NAV path a strategy would achieve after paying
70 ``cost_per_unit`` on every unit of position change.
72 When ``cost_per_unit`` is zero the result equals :attr:`nav_accumulated`.
74 Returns:
75 pl.DataFrame: Frame with an optional ``'date'`` column,
76 ``'profit'``, ``'cost'``, and ``'NAV_accumulated_net'`` columns.
78 Examples:
79 >>> from jquantstats.portfolio import Portfolio
80 >>> import polars as pl
81 >>> from datetime import date
82 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
83 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
84 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
85 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5, cost_per_unit=0.0)
86 >>> net = pf.net_cost_nav
87 >>> list(net.columns)
88 ['date', 'profit', 'cost', 'NAV_accumulated_net']
89 """
90 profit_df = self.profit
91 cost_df = self.position_delta_costs
92 if "date" in profit_df.columns:
93 df = profit_df.join(cost_df, on="date", how="left")
94 else:
95 df = profit_df.hstack(cost_df.select(["cost"]))
96 return df.with_columns(((pl.col("profit") - pl.col("cost")).cum_sum() + self.aum).alias("NAV_accumulated_net"))
98 def cost_adjusted_returns(self, cost_bps: float | None = None) -> pl.DataFrame:
99 """Return daily portfolio returns net of estimated one-way trading costs.
101 Trading costs are modelled as a linear function of daily one-way
102 turnover: for every unit of AUM traded, the strategy incurs
103 ``cost_bps`` basis points (i.e. ``cost_bps / 10_000`` fractional
104 cost). The daily cost deduction is therefore::
106 daily_cost = turnover * (cost_bps / 10_000)
108 where ``turnover`` is the fraction-of-AUM one-way turnover already
109 computed by :attr:`turnover`. The deduction is applied to the
110 ``returns`` column of :attr:`returns`, leaving all other columns
111 (including ``date``) untouched.
113 Args:
114 cost_bps: One-way trading cost in basis points per unit of AUM
115 traded. Must be non-negative. Defaults to ``self.cost_bps``
116 set at construction time.
118 Returns:
119 pl.DataFrame: Same schema as :attr:`returns` but with the
120 ``returns`` column reduced by the per-period trading cost.
122 Raises:
123 ValueError: If ``cost_bps`` is negative.
125 Examples:
126 >>> from jquantstats.portfolio import Portfolio
127 >>> import polars as pl
128 >>> from datetime import date
129 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
130 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
131 >>> pos = pl.DataFrame({"date": _d, "A": [1000.0, 1200.0, 900.0]})
132 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
133 >>> adj = pf.cost_adjusted_returns(0.0)
134 >>> float(adj["returns"][1]) == float(pf.returns["returns"][1])
135 True
136 """
137 effective_bps = cost_bps if cost_bps is not None else self.cost_bps
138 if effective_bps < 0:
139 raise ValueError
140 base = self.returns
141 daily_cost = self.turnover["turnover"] * (effective_bps / 10_000.0)
142 return base.with_columns((pl.col("returns") - daily_cost).alias("returns"))
144 def trading_cost_impact(self, max_bps: int = 20) -> pl.DataFrame:
145 """Estimate the impact of trading costs on the Sharpe ratio.
147 Computes the annualised Sharpe ratio of cost-adjusted returns for
148 each integer cost level from 0 up to and including ``max_bps`` basis
149 points (1 bp = 0.01 %). The result lets you quickly assess at what
150 cost level the strategy's edge is eroded.
152 Args:
153 max_bps: Maximum one-way trading cost to evaluate, in basis
154 points. Defaults to 20 (i.e., evaluates 0, 1, 2, …, 20
155 bps). Must be a positive integer.
157 Returns:
158 pl.DataFrame: Frame with columns ``'cost_bps'`` (Int64) and
159 ``'sharpe'`` (Float64), one row per cost level from 0 to
160 ``max_bps`` inclusive.
162 Raises:
163 ValueError: If ``max_bps`` is not a positive integer.
165 Examples:
166 >>> from jquantstats.portfolio import Portfolio
167 >>> import polars as pl
168 >>> from datetime import date, timedelta
169 >>> import numpy as np
170 >>> start = date(2020, 1, 1)
171 >>> dates = pl.date_range(
172 ... start=start, end=start + timedelta(days=99), interval="1d", eager=True
173 ... )
174 >>> rng = np.random.default_rng(0)
175 >>> prices = pl.DataFrame({
176 ... "date": dates,
177 ... "A": pl.Series(np.cumprod(1 + rng.normal(0.001, 0.01, 100)) * 100),
178 ... })
179 >>> pos = pl.DataFrame({"date": dates, "A": pl.Series(np.ones(100) * 1000.0)})
180 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
181 >>> impact = pf.trading_cost_impact(max_bps=5)
182 >>> list(impact["cost_bps"])
183 [0, 1, 2, 3, 4, 5]
184 """
185 if not isinstance(max_bps, int) or max_bps < 1:
186 raise ValueError
187 import numpy as np
189 periods = self.data._periods_per_year # one Data object, outside the loop
190 _eps = np.finfo(np.float64).eps
191 sqrt_periods = float(np.sqrt(periods))
192 cost_levels = list(range(0, max_bps + 1))
194 # Extract base returns and turnover once — O(1) allocations regardless of max_bps
195 base_rets = self.returns["returns"]
196 turnover_s = self.turnover["turnover"]
198 # Build all cost-adjusted return columns in one vectorised DataFrame construction,
199 # then compute means and stds in a single aggregate pass (no per-iteration allocation).
200 sweep = pl.DataFrame({str(bps): base_rets - turnover_s * (bps / 10_000.0) for bps in cost_levels})
201 means_row = sweep.mean().row(0)
202 stds_row = sweep.std(ddof=1).row(0)
204 sharpe_values: list[float] = []
205 for mean_raw, std_raw in zip(means_row, stds_row, strict=False):
206 mean_val = 0.0 if mean_raw is None else float(mean_raw)
207 if std_raw is None or float(std_raw) <= _eps * max(abs(mean_val), _eps) * 10:
208 sharpe_values.append(float("nan"))
209 else:
210 sharpe_values.append(mean_val / float(std_raw) * sqrt_periods)
211 return pl.DataFrame({"cost_bps": pl.Series(cost_levels, dtype=pl.Int64), "sharpe": pl.Series(sharpe_values)})