Coverage for src/jquantstats/_portfolio_turnover.py: 100%
34 statements
« prev ^ index » next coverage.py v7.17.1, created at 2026-11-03 15:52 +0000
1"""Turnover analytics mixin for Portfolio."""
3from __future__ import annotations
5from typing import TYPE_CHECKING
7import polars as pl
9from ._cache import cached_in_slot
10from ._portfolio_base import _PortfolioMembers
13class PortfolioTurnoverMixin(_PortfolioMembers):
14 """Mixin providing turnover analytics for Portfolio."""
16 if TYPE_CHECKING:
17 _turnover_cache: pl.DataFrame | None
19 @property
20 @cached_in_slot("_turnover_cache")
21 def turnover(self) -> pl.DataFrame:
22 """Daily one-way portfolio turnover as a fraction of AUM.
24 Computes the traded notional summed across all assets for each period,
25 normalised by AUM — or, for a ``compounding`` portfolio, by the
26 previous period's NAV, the same capital `returns` uses, so
27 `cost_adjusted_returns` deducts costs on a consistent base. The trade
28 is measured in **units** and priced at the traded row —
29 ``|Δunits| * price`` — rather than as a difference of cash positions:
30 a cash position moves whenever the price moves, so ``|Δcashposition|``
31 would report turnover on a book that trades nothing.
33 The first row is always zero because there is no prior position to
34 form a difference against. A missing or zero price carries the units
35 held across it, so a gap in the data is not mistaken for a sale and a
36 rebuy.
38 The result is cached after first access so repeated calls are O(1).
40 Note:
41 Caching is not thread-safe. Concurrent access from multiple
42 threads may trigger redundant computation, but will never produce
43 incorrect results because each thread stores the same deterministic
44 value.
46 Returns:
47 pl.DataFrame: Frame with an optional ``'date'`` column and a
48 ``'turnover'`` column (dimensionless fraction of AUM).
50 Examples:
51 >>> from jquantstats.portfolio import Portfolio
52 >>> import polars as pl
53 >>> from datetime import date
54 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)]
55 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]})
56 >>> pos = pl.DataFrame({"date": prices["date"], "A": [1000.0, 1650.0, 1210.0]})
57 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5)
58 >>> pf.units["A"].to_list() # buy 5 units, then sell 5
59 [10.0, 15.0, 10.0]
60 >>> [round(t, 6) for t in pf.turnover["turnover"].to_list()] # 5 * 110 and 5 * 121, over aum
61 [0.0, 0.0055, 0.00605]
62 """
63 units = self.units
65 def _traded_notional(asset: str) -> pl.Series:
66 """``|Δunits| * price`` for *asset*, holding units across a missing price."""
67 held = units[asset].forward_fill().fill_null(0.0)
68 return (held.diff().abs() * self.prices[asset]).fill_null(0.0).fill_nan(0.0)
70 traded = pl.select(_traded_notional(asset) for asset in self.assets)
71 daily = traded.select((pl.sum_horizontal(pl.all()) / self._return_base).alias("turnover"))
72 if "date" in self.prices.columns:
73 return daily.insert_column(0, self.prices["date"])
74 return daily
76 @property
77 def turnover_weekly(self) -> pl.DataFrame:
78 """Weekly aggregated one-way portfolio turnover as a fraction of AUM.
80 When a ``'date'`` column is present, sums the daily turnover within
81 each calendar week (Monday-based ``group_by_dynamic``). Without a
82 date column, a rolling 5-period sum with ``min_samples=5`` is returned
83 (the first four rows will be ``null``).
85 Returns:
86 pl.DataFrame: Frame with an optional ``'date'`` column (week
87 start) and a ``'turnover'`` column (fraction of AUM, summed over
88 the week).
89 """
90 daily = self.turnover
91 if "date" not in daily.columns or not daily["date"].dtype.is_temporal():
92 return daily.with_columns(pl.col("turnover").rolling_sum(window_size=5, min_samples=5))
93 return daily.group_by_dynamic("date", every="1w").agg(pl.col("turnover").sum()).sort("date")
95 def turnover_summary(self) -> pl.DataFrame:
96 """Return a summary DataFrame of turnover statistics.
98 Computes three metrics from the daily turnover series:
100 - ``mean_daily_turnover``: mean of daily one-way turnover (fraction
101 of AUM).
102 - ``mean_weekly_turnover``: mean of weekly-aggregated turnover
103 (fraction of AUM).
104 - ``turnover_std``: standard deviation of daily turnover (fraction of
105 AUM); complements the mean to detect regime switches.
107 Returns:
108 pl.DataFrame: One row per metric with columns ``'metric'`` and
109 ``'value'``.
111 Examples:
112 >>> from jquantstats.portfolio import Portfolio
113 >>> import polars as pl
114 >>> from datetime import date, timedelta
115 >>> import numpy as np
116 >>> start = date(2020, 1, 1)
117 >>> dates = pl.date_range(start=start, end=start + timedelta(days=9), interval="1d", eager=True)
118 >>> prices = pl.DataFrame({"date": dates, "A": pl.Series(np.ones(10) * 100.0)})
119 >>> pos = pl.DataFrame({"date": dates, "A": pl.Series([float(i) * 100 for i in range(10)])})
120 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e4)
121 >>> summary = pf.turnover_summary()
122 >>> list(summary["metric"])
123 ['mean_daily_turnover', 'mean_weekly_turnover', 'turnover_std']
124 """
125 daily_col = self.turnover["turnover"]
126 _mean = daily_col.mean()
127 mean_daily = float(_mean) if isinstance(_mean, (int, float)) else 0.0
128 _std = daily_col.std()
129 std_daily = float(_std) if isinstance(_std, (int, float)) else 0.0
130 weekly_col = self.turnover_weekly["turnover"].drop_nulls()
131 _weekly_mean = weekly_col.mean()
132 mean_weekly = (
133 float(_weekly_mean) if weekly_col.len() > 0 and isinstance(_weekly_mean, (int, float)) else float("nan")
134 )
135 return pl.DataFrame(
136 {
137 "metric": ["mean_daily_turnover", "mean_weekly_turnover", "turnover_std"],
138 "value": [mean_daily, mean_weekly, std_daily],
139 }
140 )