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

1# Copyright (c) 2023 Thomas Schmelzer 

2# 

3# Permission is hereby granted, free of charge, to any person obtaining a copy 

4# of this software and associated documentation files (the "Software"), to deal 

5# in the Software without restriction, including without limitation the rights 

6# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

7# copies of the Software, and to permit persons to whom the Software is 

8# furnished to do so, subject to the following conditions: 

9# 

10# The above copyright notice and this permission notice shall be included in all 

11# copies or substantial portions of the Software. 

12"""Signal processing functions for trend-following CTA strategies. 

13 

14Provides oscillator computation and volatility-adjusted return calculations 

15used to generate trading signals from price data. 

16""" 

17 

18from __future__ import annotations 

19 

20import math 

21 

22import numpy as np 

23import polars as pl 

24 

25 

26def moving_absolute_deviation(x: pl.Expr, com: int = 32) -> pl.Expr: 

27 """Compute the rolling median absolute deviation (MAD) of log returns. 

28 

29 A robust alternative to moving standard deviation, less sensitive to outliers. 

30 Both the center and dispersion use rolling medians, making the estimate doubly 

31 robust. The result is scaled by 1/0.6745 to be a consistent estimator of std 

32 under normality. 

33 

34 Args: 

35 x: Polars expression representing the price series. 

36 com: Center of mass used to derive the rolling window as ``window = 2 * com - 1``. 

37 

38 Returns: 

39 Polars expression of scaled rolling MAD values consistent with std under normality. 

40 

41 Example: 

42 >>> import polars as pl 

43 >>> from tinycta.signal import moving_absolute_deviation 

44 >>> prices = pl.DataFrame({"A": [100.0, 101.5, 100.8, 103.2, 102.1, 105.0, 104.2, 107.5]}) 

45 >>> mad = prices.with_columns(moving_absolute_deviation(pl.col("A"), com=2).alias("mad")) 

46 

47 Two rolling medians of ``window = 2 * com - 1`` are chained over a log-return 

48 series that itself starts one row late, so the estimate needs 

49 ``2 * window - 1`` rows of returns before it emits a value: 

50 

51 >>> mad["mad"].null_count() 

52 5 

53 >>> float(mad["mad"][5]) > 0.0 

54 True 

55 

56 The estimate is a dispersion, so it never goes negative: 

57 

58 >>> all(v >= 0.0 for v in mad["mad"][5:]) 

59 True 

60 """ 

61 window = 2 * com - 1 

62 r = x.log(base=math.e).diff() 

63 rolling_median = r.rolling_median(window_size=window) 

64 return (r - rolling_median).abs().rolling_median(window_size=window) / 0.6745 

65 

66 

67def shrink2id(matrix: np.ndarray, lamb: float = 1.0) -> np.ndarray: 

68 """Shrink a square matrix towards the identity matrix by a weight factor. 

69 

70 Args: 

71 matrix: The input square matrix to be shrunk. 

72 lamb: Mixing ratio for shrinkage. A value of 1.0 retains the original 

73 matrix; 0.0 replaces it entirely with the identity matrix. Default is 1.0. 

74 

75 Returns: 

76 The resulting matrix after applying the shrinkage transformation. 

77 

78 Example: 

79 >>> import numpy as np 

80 >>> from tinycta.signal import shrink2id 

81 >>> corr = np.array([[1.0, 0.8], [0.8, 1.0]]) 

82 

83 ``lamb=1.0`` keeps the matrix as it is: 

84 

85 >>> shrink2id(corr, lamb=1.0) 

86 array([[1. , 0.8], 

87 [0.8, 1. ]]) 

88 

89 ``lamb=0.0`` replaces it entirely with the identity: 

90 

91 >>> shrink2id(corr, lamb=0.0) 

92 array([[1., 0.], 

93 [0., 1.]]) 

94 

95 In between, the unit diagonal is preserved and the off-diagonal 

96 correlation is pulled towards zero in proportion to ``1 - lamb``: 

97 

98 >>> shrink2id(corr, lamb=0.5) 

99 array([[1. , 0.4], 

100 [0.4, 1. ]]) 

101 """ 

102 return matrix * lamb + (1 - lamb) * np.eye(N=matrix.shape[0])