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 

12 

13class PortfolioAttributionMixin(_PortfolioMembers): 

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

15 

16 if TYPE_CHECKING: 

17 _tilt_cache: Self | None 

18 

19 @classmethod 

20 def from_cash_position( 

21 cls, 

22 prices: pl.DataFrame, 

23 cash_position: pl.DataFrame, 

24 aum: float, 

25 cost_per_unit: float = 0.0, 

26 cost_bps: float = 0.0, 

27 ) -> Self: 

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

29 ... 

30 

31 @property 

32 @cached_in_slot("_tilt_cache") 

33 def tilt(self) -> Self: 

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

35 

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

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

38 applied across time. Prices and AUM are preserved. 

39 

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

41 

42 Note: 

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

44 threads may trigger redundant computation, but will never produce 

45 incorrect results because each thread stores the same deterministic 

46 value. 

47 """ 

48 const_position = self.cashposition.with_columns( 

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

50 ) 

51 return type(self).from_cash_position( 

52 self.prices, 

53 const_position, 

54 aum=self.aum, 

55 cost_per_unit=self.cost_per_unit, 

56 cost_bps=self.cost_bps, 

57 ) 

58 

59 @property 

60 def timing(self) -> Self: 

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

62 

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

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

65 effects. Prices and AUM are preserved. 

66 """ 

67 const_position = self.tilt.cashposition 

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

69 return type(self).from_cash_position( 

70 self.prices, 

71 position, 

72 aum=self.aum, 

73 cost_per_unit=self.cost_per_unit, 

74 cost_bps=self.cost_bps, 

75 ) 

76 

77 @property 

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

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

80 

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

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

83 """ 

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

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

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

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

88 

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

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

91 ) 

92 else: 

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

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

95 {"NAV_accumulated": "NAV_accumulated_tilt"} 

96 ) 

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

98 {"NAV_accumulated": "NAV_accumulated_timing"} 

99 ) 

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

101 

102 merged_df = merged_df.rename( 

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

104 ) 

105 return merged_df