Coverage for src/tinycta/util.py: 100%

1"""Volatility adjustment and price normalization helpers (Polars expressions). 

2 

3This module provides expression-level building blocks used to standardize 

4log returns by an exponentially weighted volatility estimate and to integrate 

5those standardized returns into adjusted log-price series. These are designed 

6for use within Polars pipelines (e.g., DataFrame.with_columns) and operate 

7column-wise. 

8 

9Functions: 

10- vol_adj: Standardize log returns using EWMA volatility and clip extremes. 

11- adj_log_prices: Cumulative sum (integration) of standardized, clipped returns. 

12""" 

13 

14import polars as pl 

15 

16 

17def vol_adj(x: pl.Expr, vola: int, clip: float, min_samples: int = 1) -> pl.Expr: 

18 """Compute clipped, volatility-adjusted log returns per column. 

19 

20 Args: 

21 x: Price series to transform. 

22 vola: EWMA lookback (span-equivalent) for std. 

23 clip: Symmetric clipping threshold applied after standardization. 

24 min_samples: Minimum samples required by EWM to yield non-null values. 

25 Note that ``ewm_std`` is undefined for a single observation, so the 

26 first log return is null regardless of this value — the output 

27 therefore starts at the *second* log return. 

28 

29 Returns: 

30 pl.Expr: Standardized and clipped log returns. 

31 

32 Example: 

33 >>> import polars as pl 

34 >>> from tinycta.util import vol_adj 

35 >>> prices = pl.DataFrame({"A": [100.0, 102.0, 101.0, 104.0, 103.0, 106.0]}) 

36 >>> out = prices.with_columns(vol_adj(pl.col("A"), vola=3, clip=4.2).alias("adj")) 

37 

38 The first row has no log return and the second has no ``ewm_std`` (it is 

39 undefined for a single observation), so the series starts on the third row: 

40 

41 >>> out["adj"].null_count() 

42 2 

43 

44 Standardised returns keep the sign of the underlying move: 

45 

46 >>> [v > 0 for v in out["adj"][2:]] 

47 [False, True, False, True] 

48 

49 ``clip`` bounds the output symmetrically, which is what keeps a single 

50 volatility spike from dominating a downstream signal: 

51 

52 >>> tight = prices.with_columns(vol_adj(pl.col("A"), vola=3, clip=1.0).alias("adj")) 

53 >>> all(-1.0 <= v <= 1.0 for v in tight["adj"][2:]) 

54 True 

55 """ 

56 log_returns = x.log().diff() 

57 vol = log_returns.ewm_std(com=vola - 1, adjust=True, min_samples=min_samples) 

58 return (log_returns / vol).clip(-clip, clip) 

59 

60 

61def adj_log_prices(x: pl.Expr, vola: int, clip: float, min_samples: int = 1) -> pl.Expr: 

62 """Integrate clipped, volatility-adjusted log returns to adjusted log prices. 

63 

64 Uses ``vol_adj`` to standardize/clamp log returns and then integrates them 

65 via cumulative sum. The resulting series behaves like a standardized price- 

66 like process with roughly unit volatility. 

67 

68 Args: 

69 x: Polars expression of the price series to transform. 

70 vola: EWMA lookback (span-equivalent) used to estimate volatility. 

71 clip: Symmetric clipping threshold applied after standardization. 

72 min_samples: Minimum samples required by EWM to emit non-null values. 

73 

74 Returns: 

75 pl.Expr: Adjusted-log-price series obtained by cumulative sum of 

76 standardized returns. 

77 

78 Example: 

79 >>> import polars as pl 

80 >>> from tinycta.util import adj_log_prices, vol_adj 

81 >>> prices = pl.DataFrame({"A": [100.0, 102.0, 101.0, 104.0, 103.0, 106.0]}) 

82 >>> out = prices.with_columns( 

83 ... vol_adj(pl.col("A"), vola=3, clip=4.2).alias("adj"), 

84 ... adj_log_prices(pl.col("A"), vola=3, clip=4.2).alias("level"), 

85 ... ) 

86 

87 The result is the running total of the standardised returns, so each level 

88 is the previous one plus the current adjusted return: 

89 

90 >>> float(out["level"][2]) == float(out["adj"][2]) 

91 True 

92 >>> round(float(out["level"][3]) - float(out["level"][2]), 12) == round(float(out["adj"][3]), 12) 

93 True 

94 

95 ``cum_sum`` carries the leading nulls through, so the level series starts 

96 where the adjusted returns do: 

97 

98 >>> out["level"].null_count() 

99 2 

100 """ 

101 return vol_adj(x, vola=vola, clip=clip, min_samples=min_samples).cum_sum()