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

11 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-07-30 04:16 +0000

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 window = 2 * com - 1 

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

43 rolling_median = r.rolling_median(window_size=window) 

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

45 

46 

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

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

49 

50 Args: 

51 matrix: The input square matrix to be shrunk. 

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

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

54 

55 Returns: 

56 The resulting matrix after applying the shrinkage transformation. 

57 """ 

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