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 numpy as np 

21import pandas as pd 

22 

23 

24def osc(prices: pd.DataFrame, fast: int = 32, slow: int = 96, scaling: bool = True) -> pd.DataFrame: 

25 """Compute the oscillator for a given financial price data. 

26 

27 Use Exponential Weighted Moving Averages (EWM). The calculation involves 

28 the difference between fast and slow EWM means, optionally scaled by the 

29 standard deviation. 

30 

31 Args: 

32 prices: DataFrame containing the price data used for the oscillator computation. 

33 fast: The time period for the fast EWM calculation. Default is 32. 

34 slow: The time period for the slow EWM calculation. Default is 96. 

35 scaling: If True, the difference will be scaled using its standard deviation. 

36 If False, scaling is skipped. Default is True. 

37 

38 Returns: 

39 DataFrame containing the computed oscillator values. 

40 """ 

41 diff = prices.ewm(com=fast - 1).mean() - prices.ewm(com=slow - 1).mean() 

42 s = diff.std() if scaling else 1 

43 

44 return diff / s 

45 

46 

47def returns_adjust(price: pd.DataFrame, com: int = 32, min_periods: int = 300, clip: float = 4.2) -> pd.DataFrame: 

48 """Calculate and adjust log returns for a given price DataFrame. 

49 

50 Computes the logarithmic returns, normalizes them using exponentially 

51 weighted moving standard deviation, and clips the resulting values to a 

52 specified range. 

53 

54 Args: 

55 price: DataFrame containing price data for which log returns will be calculated. 

56 com: Center of mass for the exponentially weighted moving average. Default is 32. 

57 min_periods: Minimum number of periods required for the EWMA standard deviation 

58 to be valid. Default is 300. 

59 clip: Absolute value threshold to which the adjusted returns are clipped. 

60 Default is 4.2. 

61 

62 Returns: 

63 A DataFrame of normalized and clipped log returns for the input price data. 

64 """ 

65 r = price.apply(np.log).diff() 

66 return (r / r.ewm(com=com, min_periods=min_periods).std()).clip(-clip, +clip) 

67 

68 

69def moving_absolute_deviation(price: pd.DataFrame, com: int = 32) -> pd.DataFrame: 

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

71 

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

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

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

75 under normality. 

76 

77 Args: 

78 price: DataFrame containing price data. 

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

80 

81 Returns: 

82 DataFrame of scaled rolling MAD values consistent with std under normality. 

83 """ 

84 r = price.apply(np.log).diff() 

85 window = 2 * com - 1 

86 rolling_median = r.rolling(window=window).median() 

87 return (r - rolling_median).abs().rolling(window=window).median() / 0.6745 

88 

89 

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

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

92 

93 Args: 

94 matrix: The input square matrix to be shrunk. 

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

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

97 

98 Returns: 

99 The resulting matrix after applying the shrinkage transformation. 

100 """ 

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