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 

24# compute the oscillator 

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

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

27 

28 Use Exponential Weighted Moving Averages (EWM). 

29 The calculation involves the difference between fast and 

30 slow EWM means, optionally scaled by the standard deviation. 

31 

32 Parameters: 

33 prices (pd.DataFrame) 

34 DataFrame containing the price data used for the oscillator computation. 

35 fast (int, optional) 

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

37 slow (int, optional) 

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

39 scaling (bool, optional) 

40 If True, the difference will be scaled using its standard deviation. If 

41 False, scaling is skipped. Default is True. 

42 

43 Returns: 

44 pd.DataFrame 

45 DataFrame containing the computed oscillator values. 

46 """ 

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

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

49 

50 return diff / s 

51 

52 

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

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

55 

56 This function computes the logarithmic returns of a given price DataFrame, 

57 normalizes them using exponentially weighted moving standard deviation with 

58 specified parameters, and clips the resulting values to a specified range. 

59 

60 Parameters: 

61 price : pd.DataFrame 

62 The DataFrame containing price data for which log returns will be calculated. 

63 com : int, default=32 

64 Specifies the center of mass for the exponentially weighted moving average 

65 calculation. 

66 min_periods : int, default=300 

67 Minimum number of periods required for the calculation of the exponentially 

68 weighted moving standard deviation to be valid. 

69 clip : float, default=4.2 

70 The absolute value threshold to which the adjusted returns are clipped. 

71 

72 Returns: 

73 pd.DataFrame 

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

75 """ 

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

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

78 

79 

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

81 """Performs shrinkage of a given square matrix towards the identity matrix by a weight factor. 

82 

83 This function modifies the input matrix by shrinking it towards the identity matrix. The 

84 shrinking is controlled by the `lamb` parameter, which determines the weighting between the 

85 original matrix and the identity matrix. 

86 

87 Parameters: 

88 matrix (np.ndarray): The input square matrix to be shrunk. 

89 lamb (float): The mixing ratio for shrinkage. Defaults to 1.0. A value of 1.0 retains the 

90 original matrix, while 0.0 completely replaces it with the identity matrix. 

91 

92 Returns: 

93 np.ndarray: The resulting matrix after applying the shrinkage transformation. 

94 """ 

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