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

1"""Oscillator signal utilities built on Polars expressions. 

2 

3This module provides a helper to compute an oscillator from price series using 

4exponentially weighted moving averages (EWMA) and an analytical scaling factor. 

5The functions are designed to be used inside Polars pipelines 

6(e.g., with DataFrame.with_columns) and operate column-wise on numeric data. 

7""" 

8 

9import math 

10 

11import polars as pl 

12 

13 

14def _validate_windows(fast: int, slow: int) -> None: 

15 """Validate the fast/slow EWMA window parameters. 

16 

17 Args: 

18 fast: Fast EWMA length. Must be an integer greater than 1. 

19 slow: Slow EWMA length. Must be an integer greater than 1 and ``> fast``. 

20 

21 Raises: 

22 TypeError: If ``fast`` or ``slow`` are not integers. 

23 ValueError: If ``fast <= 1``, ``slow <= 1``, or ``fast >= slow``. 

24 """ 

25 for value, name in ((fast, "fast"), (slow, "slow")): 

26 if not isinstance(value, int): 

27 msg = f"{name} must be an integer" 

28 raise TypeError(msg) 

29 

30 value_checks = ( 

31 (fast <= 1, "fast must be greater than 1"), 

32 (slow <= 1, "slow must be greater than 1"), 

33 (fast >= slow, "fast must be less than slow"), 

34 ) 

35 for failed, msg in value_checks: 

36 if failed: 

37 raise ValueError(msg) 

38 

39 

40def osc(x: pl.Expr, fast: int, slow: int, min_samples: int = 1) -> pl.Expr: 

41 """Compute an analytically scaled EWMA-difference oscillator. 

42 

43 The oscillator is defined as (EMA_fast - EMA_slow) divided by the 

44 theoretical standard deviation of that difference under a unit-variance 

45 random walk: 

46 s = sqrt(1/(1-f²) - 2/(1-fg) + 1/(1-g²)) 

47 where f = 1 - 1/fast and g = 1 - 1/slow. 

48 

49 This gives consistent signal magnitudes regardless of the fast/slow 

50 parameter choice, without requiring a separate volatility lookback. 

51 

52 Args: 

53 x: Polars expression representing the price series to transform. 

54 fast: Fast EWMA length (interpreted via ``com=fast-1``). Must be > 1. 

55 slow: Slow EWMA length (interpreted via ``com=slow-1``). Must be > 1 and ``slow > fast``. 

56 min_samples: Minimum number of observations required before EWMA 

57 means are emitted; controls warmup period (earlier rows are 

58 null until this threshold is met). 

59 

60 Returns: 

61 pl.Expr: A Polars expression representing the oscillator values. 

62 

63 Raises: 

64 TypeError: If ``fast`` or ``slow`` are not integers. 

65 ValueError: If ``fast <= 1``, ``slow <= 1``, or ``fast >= slow``. 

66 

67 Example: 

68 >>> prices = pl.DataFrame({"A": [1,2,3,4,5,6,7,8,9,10]}) 

69 >>> df = prices.with_columns(osc(pl.col("A"), fast=2, slow=6).alias("osc_A")) 

70 

71 With ``min_samples=1`` the first row is 0.0 (both EWMAs equal the first 

72 observation), and the oscillator then rises as the fast mean pulls ahead 

73 of the slow one on a trending series: 

74 

75 >>> df["osc_A"].round(4).to_list() 

76 [0.0, 0.1117, 0.2836, 0.4985, 0.7389, 0.9898, 1.2398, 1.4809, 1.7086, 1.9202] 

77 

78 Note that the analytic scaling makes magnitudes comparable across 

79 ``fast``/``slow`` choices only once the slower EWMA has warmed up; over a 

80 span this short the slower pair is still in its transient. 

81 """ 

82 _validate_windows(fast, slow) 

83 

84 f, g = 1 - 1 / fast, 1 - 1 / slow 

85 s = math.sqrt(1.0 / (1 - f * f) - 2.0 / (1 - f * g) + 1.0 / (1 - g * g)) 

86 

87 diff = x.ewm_mean(com=fast - 1, adjust=True, min_samples=min_samples) - x.ewm_mean( 

88 com=slow - 1, adjust=True, min_samples=min_samples 

89 ) 

90 return diff / s