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

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

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7import polars as pl 

8 

9from ._cache import cached_in_slot 

10from ._portfolio_base import _PortfolioMembers 

11 

12 

13class PortfolioTurnoverMixin(_PortfolioMembers): 

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

15 

16 if TYPE_CHECKING: 

17 _turnover_cache: pl.DataFrame | None 

18 

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. 

23 

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

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

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

27 

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

29 

30 Note: 

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

32 threads may trigger redundant computation, but will never produce 

33 incorrect results because each thread stores the same deterministic 

34 value. 

35 

36 Returns: 

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

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

39 

40 Examples: 

41 >>> from jquantstats.portfolio import Portfolio 

42 >>> import polars as pl 

43 >>> from datetime import date 

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

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

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

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

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

49 [0.0, 0.002, 0.003] 

50 """ 

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

52 daily_abs_chg = ( 

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

54 ).alias("turnover") 

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

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

57 cols.append("date") 

58 cols.append(daily_abs_chg) 

59 return self.cashposition.select(cols) 

60 

61 @property 

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

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

64 

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

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

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

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

69 

70 Returns: 

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

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

73 the week). 

74 """ 

75 daily = self.turnover 

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

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

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

79 

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

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

82 

83 Computes three metrics from the daily turnover series: 

84 

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

86 of AUM). 

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

88 (fraction of AUM). 

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

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

91 

92 Returns: 

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

94 ``'value'``. 

95 

96 Examples: 

97 >>> from jquantstats.portfolio import Portfolio 

98 >>> import polars as pl 

99 >>> from datetime import date, timedelta 

100 >>> import numpy as np 

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

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

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

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

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

106 >>> summary = pf.turnover_summary() 

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

108 ['mean_daily_turnover', 'mean_weekly_turnover', 'turnover_std'] 

109 """ 

110 daily_col = self.turnover["turnover"] 

111 _mean = daily_col.mean() 

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

113 _std = daily_col.std() 

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

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

116 _weekly_mean = weekly_col.mean() 

117 mean_weekly = ( 

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

119 ) 

120 return pl.DataFrame( 

121 { 

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

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

124 } 

125 )