Coverage for src / jquantstats / _stats / _core.py: 100%

1"""Module helpers and method decorators for statistical computations. 

2 

3Provides: 

4 

5- :func:`_drawdown_series` — drawdown series from a returns series. 

6- :func:`_to_float` — safe Polars aggregation result → Python float. 

7- :func:`columnwise_stat` — decorator: apply a metric to every asset column. 

8- :func:`to_frame` — decorator: build a per-column Polars DataFrame result. 

9 

10These building blocks are shared across the stats mixin modules 

11(:mod:`~jquantstats._stats._basic`, :mod:`~jquantstats._stats._performance`, 

12:mod:`~jquantstats._stats._reporting`, :mod:`~jquantstats._stats._rolling`). 

13""" 

14 

15from __future__ import annotations 

16 

17from collections.abc import Callable 

18from datetime import timedelta 

19from functools import wraps 

20from typing import Any, cast 

21 

22import polars as pl 

23 

24# ── Module helpers ──────────────────────────────────────────────────────────── 

25 

26 

27def _drawdown_series(series: pl.Series) -> pl.Series: 

28 """Compute the drawdown percentage series from a returns series. 

29 

30 Treats ``series`` as additive daily returns and builds a normalised NAV 

31 starting at 1.0. The high-water mark is the running maximum of that NAV; 

32 drawdown is expressed as the fraction below the high-water mark. 

33 

34 Args: 

35 series: A Polars Series of additive returns (profit / AUM). 

36 

37 Returns: 

38 A Polars Float64 Series whose values are in [0, 1]. A value of 0 

39 means the NAV is at its all-time high; a value of 0.2 means the NAV 

40 is 20 % below its previous peak. 

41 

42 Examples: 

43 >>> import polars as pl 

44 >>> s = pl.Series([0.0, -0.1, 0.2]) 

45 >>> [round(x, 10) for x in _drawdown_series(s).to_list()] 

46 [0.0, 0.1, 0.0] 

47 """ 

48 nav = 1.0 + series.cast(pl.Float64).cum_sum() 

49 hwm = nav.cum_max() 

50 hwm_safe = hwm.clip(lower_bound=1e-10) 

51 return ((hwm - nav) / hwm_safe).clip(lower_bound=0.0) 

52 

53 

54def _to_float(value: object) -> float: 

55 """Safely convert a Polars aggregation result to float. 

56 

57 Examples: 

58 >>> _to_float(2.0) 

59 2.0 

60 >>> _to_float(None) 

61 0.0 

62 """ 

63 if value is None: 

64 return 0.0 

65 if isinstance(value, timedelta): 

66 return value.total_seconds() 

67 return float(cast(float, value)) 

68 

69 

70# ── Module-level decorators ────────────────────────────────────────────────── 

71 

72 

73def columnwise_stat(func: Callable[..., Any]) -> Callable[..., dict[str, float]]: 

74 """Apply a column-wise statistical function to all numeric columns. 

75 

76 Args: 

77 func (Callable): The function to decorate. 

78 

79 Returns: 

80 Callable: The decorated function. 

81 

82 """ 

83 

84 @wraps(func) 

85 def wrapper(self: Any, *args: Any, **kwargs: Any) -> dict[str, float]: 

86 """Apply *func* to every column and return a ``{column: value}`` mapping.""" 

87 return {col: func(self, series, *args, **kwargs) for col, series in self.data.items()} 

88 

89 return wrapper 

90 

91 

92def to_frame(func: Callable[..., Any]) -> Callable[..., pl.DataFrame]: 

93 """Apply per-column expressions and evaluates with .with_columns(...). 

94 

95 Args: 

96 func (Callable): The function to decorate. 

97 

98 Returns: 

99 Callable: The decorated function. 

100 

101 """ 

102 

103 @wraps(func) 

104 def wrapper(self: Any, *args: Any, **kwargs: Any) -> pl.DataFrame: 

105 """Apply *func* per column and return the result as a Polars DataFrame.""" 

106 return cast(pl.DataFrame, self.all).select( 

107 [pl.col(name) for name in self.data.date_col] 

108 + [func(self, series, *args, **kwargs).alias(col) for col, series in self.data.items()] 

109 ) 

110 

111 return wrapper