Skip to content

Oscillator

Analytically scaled EWMA-difference oscillator for trend-following signals.

tinycta.osc

Oscillator signal utilities built on Polars expressions.

This module provides a helper to compute an oscillator from price series using exponentially weighted moving averages (EWMA) and an analytical scaling factor. The functions are designed to be used inside Polars pipelines (e.g., with DataFrame.with_columns) and operate column-wise on numeric data.

osc(x, fast, slow, min_samples=1)

Compute an analytically scaled EWMA-difference oscillator.

The oscillator is defined as (EMA_fast - EMA_slow) divided by the theoretical standard deviation of that difference under a unit-variance random walk: s = sqrt(1/(1-f²) - 2/(1-fg) + 1/(1-g²)) where f = 1 - 1/fast and g = 1 - 1/slow.

This gives consistent signal magnitudes regardless of the fast/slow parameter choice, without requiring a separate volatility lookback.

Parameters:

Name Type Description Default
x Expr

Polars expression representing the price series to transform.

 required
fast int

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

 required
slow int

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

 required
min_samples int

Minimum number of observations required before EWMA means are emitted; controls warmup period (earlier rows are null until this threshold is met).

 1

Returns:

Type Description
Expr

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

Raises:

Type Description
TypeError

If fast or slow are not integers.
ValueError

If fast <= 1, slow <= 1, or fast >= slow.
Example

prices = pl.DataFrame({"A": [1,2,3,4,5,6,7,8,9,10]}) df = prices.with_columns(osc(pl.col("A"), fast=2, slow=6).alias("osc_A"))

Source code in src/tinycta/osc.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def osc(x: pl.Expr, fast: int, slow: int, min_samples: int = 1) -> pl.Expr:
    """Compute an analytically scaled EWMA-difference oscillator.

    The oscillator is defined as (EMA_fast - EMA_slow) divided by the
    theoretical standard deviation of that difference under a unit-variance
    random walk:
        s = sqrt(1/(1-f²) - 2/(1-fg) + 1/(1-g²))
    where f = 1 - 1/fast and g = 1 - 1/slow.

    This gives consistent signal magnitudes regardless of the fast/slow
    parameter choice, without requiring a separate volatility lookback.

    Args:
        x: Polars expression representing the price series to transform.
        fast: Fast EWMA length (interpreted via ``com=fast-1``). Must be > 1.
        slow: Slow EWMA length (interpreted via ``com=slow-1``). Must be > 1 and ``slow > fast``.
        min_samples: Minimum number of observations required before EWMA
            means are emitted; controls warmup period (earlier rows are
            null until this threshold is met).

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

    Raises:
        TypeError: If ``fast`` or ``slow`` are not integers.
        ValueError: If ``fast <= 1``, ``slow <= 1``, or ``fast >= slow``.

    Example:
        >>> prices = pl.DataFrame({"A": [1,2,3,4,5,6,7,8,9,10]})
        >>> df = prices.with_columns(osc(pl.col("A"), fast=2, slow=6).alias("osc_A"))
    """
    if not isinstance(fast, int):
        msg = "fast must be an integer"
        raise TypeError(msg)
    if not isinstance(slow, int):
        msg = "slow must be an integer"
        raise TypeError(msg)
    if fast <= 1:
        msg = "fast must be greater than 1"
        raise ValueError(msg)
    if slow <= 1:
        msg = "slow must be greater than 1"
        raise ValueError(msg)
    if fast >= slow:
        msg = "fast must be less than slow"
        raise ValueError(msg)

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

    diff = x.ewm_mean(com=fast - 1, adjust=True, min_samples=min_samples) - x.ewm_mean(
        com=slow - 1, adjust=True, min_samples=min_samples
    )
    return diff / s