Coverage for src / jquantstats / _portfolio_turnover.py: 100%

1"""Turnover analytics mixin for Portfolio.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6from typing import TYPE_CHECKING 

7 

8import polars as pl 

9 

10 

11class PortfolioTurnoverMixin: 

12 """Mixin providing turnover analytics for Portfolio.""" 

13 

14 if TYPE_CHECKING: 

15 cashposition: pl.DataFrame 

16 aum: float 

17 _turnover_cache: pl.DataFrame | None 

18 

19 @property 

20 def turnover(self) -> pl.DataFrame: 

21 """Daily one-way portfolio turnover as a fraction of AUM. 

22 

23 Computes the sum of absolute position changes across all assets for 

24 each period, normalised by AUM. The first row is always zero because 

25 there is no prior position to form a difference against. 

26 

27 The result is cached after first access so repeated calls are O(1). 

28 

29 Note: 

30 Caching is not thread-safe. Concurrent access from multiple 

31 threads may trigger redundant computation, but will never produce 

32 incorrect results because each thread stores the same deterministic 

33 value. 

34 

35 Returns: 

36 pl.DataFrame: Frame with an optional ``'date'`` column and a 

37 ``'turnover'`` column (dimensionless fraction of AUM). 

38 

39 Examples: 

40 >>> from jquantstats.portfolio import Portfolio 

41 >>> import polars as pl 

42 >>> from datetime import date 

43 >>> _d = [date(2020, 1, 1), date(2020, 1, 2), date(2020, 1, 3)] 

44 >>> prices = pl.DataFrame({"date": _d, "A": [100.0, 110.0, 121.0]}) 

45 >>> pos = pl.DataFrame({"date": prices["date"], "A": [1000.0, 1200.0, 900.0]}) 

46 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e5) 

47 >>> pf.turnover["turnover"].to_list() 

48 [0.0, 0.002, 0.003] 

49 """ 

50 cache = getattr(self, "_turnover_cache", None) 

51 if cache is not None: 

52 return cache 

53 

54 assets = [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()] 

55 daily_abs_chg = ( 

56 pl.sum_horizontal(pl.col(c).diff().abs().fill_null(0.0).fill_nan(0.0) for c in assets) / self.aum 

57 ).alias("turnover") 

58 cols: list[str | pl.Expr] = [] 

59 if "date" in self.cashposition.columns: 

60 cols.append("date") 

61 cols.append(daily_abs_chg) 

62 result = self.cashposition.select(cols) 

63 

64 with contextlib.suppress(AttributeError, TypeError): 

65 object.__setattr__(self, "_turnover_cache", result) 

66 return result 

67 

68 @property 

69 def turnover_weekly(self) -> pl.DataFrame: 

70 """Weekly aggregated one-way portfolio turnover as a fraction of AUM. 

71 

72 When a ``'date'`` column is present, sums the daily turnover within 

73 each calendar week (Monday-based ``group_by_dynamic``). Without a 

74 date column, a rolling 5-period sum with ``min_samples=5`` is returned 

75 (the first four rows will be ``null``). 

76 

77 Returns: 

78 pl.DataFrame: Frame with an optional ``'date'`` column (week 

79 start) and a ``'turnover'`` column (fraction of AUM, summed over 

80 the week). 

81 """ 

82 daily = self.turnover 

83 if "date" not in daily.columns or not daily["date"].dtype.is_temporal(): 

84 return daily.with_columns(pl.col("turnover").rolling_sum(window_size=5, min_samples=5)) 

85 return daily.group_by_dynamic("date", every="1w").agg(pl.col("turnover").sum()).sort("date") 

86 

87 def turnover_summary(self) -> pl.DataFrame: 

88 """Return a summary DataFrame of turnover statistics. 

89 

90 Computes three metrics from the daily turnover series: 

91 

92 - ``mean_daily_turnover``: mean of daily one-way turnover (fraction 

93 of AUM). 

94 - ``mean_weekly_turnover``: mean of weekly-aggregated turnover 

95 (fraction of AUM). 

96 - ``turnover_std``: standard deviation of daily turnover (fraction of 

97 AUM); complements the mean to detect regime switches. 

98 

99 Returns: 

100 pl.DataFrame: One row per metric with columns ``'metric'`` and 

101 ``'value'``. 

102 

103 Examples: 

104 >>> from jquantstats.portfolio import Portfolio 

105 >>> import polars as pl 

106 >>> from datetime import date, timedelta 

107 >>> import numpy as np 

108 >>> start = date(2020, 1, 1) 

109 >>> dates = pl.date_range(start=start, end=start + timedelta(days=9), interval="1d", eager=True) 

110 >>> prices = pl.DataFrame({"date": dates, "A": pl.Series(np.ones(10) * 100.0)}) 

111 >>> pos = pl.DataFrame({"date": dates, "A": pl.Series([float(i) * 100 for i in range(10)])}) 

112 >>> pf = Portfolio(prices=prices, cashposition=pos, aum=1e4) 

113 >>> summary = pf.turnover_summary() 

114 >>> list(summary["metric"]) 

115 ['mean_daily_turnover', 'mean_weekly_turnover', 'turnover_std'] 

116 """ 

117 daily_col = self.turnover["turnover"] 

118 _mean = daily_col.mean() 

119 mean_daily = float(_mean) if isinstance(_mean, (int, float)) else 0.0 

120 _std = daily_col.std() 

121 std_daily = float(_std) if isinstance(_std, (int, float)) else 0.0 

122 weekly_col = self.turnover_weekly["turnover"].drop_nulls() 

123 _weekly_mean = weekly_col.mean() 

124 mean_weekly = ( 

125 float(_weekly_mean) if weekly_col.len() > 0 and isinstance(_weekly_mean, (int, float)) else float("nan") 

126 ) 

127 return pl.DataFrame( 

128 { 

129 "metric": ["mean_daily_turnover", "mean_weekly_turnover", "turnover_std"], 

130 "value": [mean_daily, mean_weekly, std_daily], 

131 } 

132 )