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

1"""Tilt/timing attribution mixin for Portfolio.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Self 

6 

7import polars as pl 

8 

9from ._cache import cached_in_slot 

10from ._portfolio_base import _PortfolioMembers 

11 

12if TYPE_CHECKING: 

13 from ._cost_model import CostModel 

14 

15 

16class PortfolioAttributionMixin(_PortfolioMembers): 

17 """Mixin providing tilt/timing attribution properties for Portfolio.""" 

18 

19 if TYPE_CHECKING: 

20 _tilt_cache: Self | None 

21 

22 @classmethod 

23 def from_cash_position( 

24 cls, 

25 prices: pl.DataFrame, 

26 cash_position: pl.DataFrame, 

27 aum: float, 

28 cost_per_unit: float = 0.0, 

29 cost_bps: float = 0.0, 

30 cost_model: CostModel | None = None, 

31 annual_fee: float = 0.0, 

32 compounding: bool = False, 

33 realized_costs: pl.DataFrame | None = None, 

34 ) -> Self: 

35 """Create a Portfolio directly from cash positions aligned with prices.""" 

36 ... 

37 

38 @property 

39 @cached_in_slot("_tilt_cache") 

40 def tilt(self) -> Self: 

41 """Return the 'tilt' portfolio with constant average weights. 

42 

43 Computes the time-average of each asset's cash position (ignoring 

44 nulls/NaNs) and builds a new Portfolio with those constant weights 

45 applied across time. Prices and AUM are preserved. 

46 

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

48 

49 Note: 

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

51 threads may trigger redundant computation, but will never produce 

52 incorrect results because each thread stores the same deterministic 

53 value. 

54 """ 

55 const_position = self.cashposition.with_columns( 

56 pl.col(col).drop_nulls().drop_nans().mean().alias(col) for col in self.assets 

57 ) 

58 return type(self).from_cash_position( 

59 self.prices, 

60 const_position, 

61 aum=self.aum, 

62 cost_per_unit=self.cost_per_unit, 

63 cost_bps=self.cost_bps, 

64 annual_fee=self.annual_fee, 

65 compounding=self.compounding, 

66 ) 

67 

68 @property 

69 def timing(self) -> Self: 

70 """Return the 'timing' portfolio capturing deviations from the tilt. 

71 

72 Constructs weights as original cash positions minus the tilt's 

73 constant positions, per asset. This isolates timing (alloc-demeaned) 

74 effects. Prices and AUM are preserved. 

75 """ 

76 const_position = self.tilt.cashposition 

77 position = self.cashposition.with_columns((pl.col(col) - const_position[col]).alias(col) for col in self.assets) 

78 return type(self).from_cash_position( 

79 self.prices, 

80 position, 

81 aum=self.aum, 

82 cost_per_unit=self.cost_per_unit, 

83 cost_bps=self.cost_bps, 

84 annual_fee=self.annual_fee, 

85 compounding=self.compounding, 

86 ) 

87 

88 @property 

89 def tilt_timing_decomp(self) -> pl.DataFrame: 

90 """Return the portfolio's tilt/timing NAV decomposition. 

91 

92 When a ``'date'`` column is present the three NAV series are joined on 

93 it. When data is integer-indexed the frames are stacked horizontally. 

94 """ 

95 if "date" in self.nav_accumulated.columns: 

96 nav_portfolio = self.nav_accumulated.select(["date", "NAV_accumulated"]) 

97 nav_tilt = self.tilt.nav_accumulated.select(["date", "NAV_accumulated"]) 

98 nav_timing = self.timing.nav_accumulated.select(["date", "NAV_accumulated"]) 

99 

100 merged_df = nav_portfolio.join(nav_tilt, on="date", how="inner", suffix="_tilt").join( 

101 nav_timing, on="date", how="inner", suffix="_timing" 

102 ) 

103 else: 

104 nav_portfolio = self.nav_accumulated.select(["NAV_accumulated"]) 

105 nav_tilt = self.tilt.nav_accumulated.select(["NAV_accumulated"]).rename( 

106 {"NAV_accumulated": "NAV_accumulated_tilt"} 

107 ) 

108 nav_timing = self.timing.nav_accumulated.select(["NAV_accumulated"]).rename( 

109 {"NAV_accumulated": "NAV_accumulated_timing"} 

110 ) 

111 merged_df = nav_portfolio.hstack(nav_tilt).hstack(nav_timing) 

112 

113 merged_df = merged_df.rename( 

114 {"NAV_accumulated_tilt": "tilt", "NAV_accumulated_timing": "timing", "NAV_accumulated": "portfolio"} 

115 ) 

116 return merged_df