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

1"""EWMA-based signal generation utilities. 

2 

3This module exposes helpers built around exponentially weighted moving averages 

4(EWMA) for use inside Polars expression pipelines. Functions operate column- 

5wise and are suitable for DataFrame.with_columns usage in notebooks and batch 

6pipelines. 

7""" 

8 

9import polars as pl 

10 

11 

12def ma_cross(prices: pl.Expr, fast: int, slow: int, min_samples: int = 1) -> pl.Expr: 

13 """Return the sign of the fast-vs-slow EWM moving-average cross per column. 

14 

15 Computes two exponentially weighted moving averages (EWM) of the input 

16 price series using windows ``fast`` and ``slow`` (interpreted as 

17 ``com=window-1``) and returns the sign of their difference. The output 

18 is -1, 0, or +1 after the warmup implied by ``min_samples``. 

19 

20 Args: 

21 prices: Polars expression containing the price series to transform. 

22 fast: Length for the fast EWM mean (``fast > 0``). Typically ``fast < slow``. 

23 slow: Length for the slow EWM mean (``slow > 0``). 

24 min_samples: Minimum number of observations required before EWM values 

25 are produced; earlier rows will be null. 

26 

27 Returns: 

28 pl.Expr: An expression yielding -1, 0, or +1 per row after warmup. 

29 

30 Example: 

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

32 >>> df = prices.with_columns( 

33 ... ma_cross(pl.col("A"), fast=2, slow=6, min_samples=3).alias("sig_A") 

34 ... ) 

35 """ 

36 return ( 

37 prices.ewm_mean(com=fast - 1, adjust=False, min_samples=min_samples) 

38 - prices.ewm_mean(com=slow - 1, adjust=False, min_samples=min_samples) 

39 ).sign()