Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,16 @@ tests = [
"pytest",
"pytest-cov",
]

docs = [
"scipy",
"pdoc3",
"matplotlib",
"plotly",
"seaborn",
"nbformat==5.4.0",
"ipython",
"notebook",
"ipywidgets",
"yfinance"
]
663 changes: 663 additions & 0 deletions tests/test_american_tf.py

Large diffs are not rendered by default.

138 changes: 89 additions & 49 deletions tests/test_filters.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,27 @@
"""
Test suite for filter functions in trendfollower.core.filters.

This module tests the exponentially weighted moving average (EWMA) filters,
including standard EWMA, variance-preserving EWMA, and long-short variance-preserving EWMA.
"""

from itertools import combinations, product

import numpy as np
import polars as pl
import pytest

from trendfollower.filters import ewma, variance_preserving_ewma, long_short_variance_preserving_ewma
from trendfollower.core.filters import (
ewma,
variance_preserving_ewma,
long_short_variance_preserving_ewma
)


# Test Configuration Constants
RNG = np.random.default_rng(seed=42)
SMOOTHING_PARS = 0.1 + 0.1*np.arange(8)
VARS = [0.1, 1.0, 2.0]
SMOOTHING_PARS = 0.1 + 0.1 * np.arange(8) # [0.1, 0.2, ..., 0.8]
VARS = [0.1, 1.0, 2.0] # Different variance levels for testing


def sample_long_series_with_input_variance(var: float) -> pl.Series:
Expand All @@ -24,61 +36,89 @@ def sample_long_series_with_input_variance(var: float) -> pl.Series:
-------
pl.Series
A sample long series with the specified variance.

"""
samples = RNG.normal(loc=0, scale=np.sqrt(var), size=100000)
return pl.Series(samples)


@pytest.fixture(scope="module", autouse=True)
def sample_series_per_variance() -> dict[float, pl.Series]:
"""Generate sample series for each variance level."""
return {var: sample_long_series_with_input_variance(var) for var in VARS}


@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_ewma_filter_mean(sample_series_per_variance, input_var, alpha):
series = sample_series_per_variance[input_var]
filtered = ewma(series, alpha=alpha)
expected = series.mean()
calculated = filtered.mean()
msg = f"Under ewma transformation, mean remains unchanged, expected {expected}, got {calculated}"
assert calculated == pytest.approx(expected, rel=1e-1), msg


@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_ewma_filter_variance(sample_series_per_variance, input_var, alpha):
series = sample_series_per_variance[input_var]
filtered = ewma(series, alpha=alpha)
expected = input_var * ((1 - alpha) / (1 + alpha))
calculated = filtered.var()
msg = f"Variance of ewma of a series with input_variance var must be (1 - alpha) / (1 + alpha) * var, expected {expected}, got {calculated}"
assert calculated == pytest.approx(expected, rel=1e-1), msg


@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_variance_preserving_ewma(sample_series_per_variance, input_var, alpha):
series = sample_series_per_variance[input_var]
filtered = variance_preserving_ewma(series, alpha=alpha)
expected = input_var
calculated = filtered.var()
msg = f"Variance of variance_preserving_ewma of a series with input_variance var must be var, expected {expected}, got {calculated}"
assert calculated == pytest.approx(expected, rel=1e-1), msg


@pytest.mark.parametrize("input_var, alphas", product(VARS, combinations(SMOOTHING_PARS, 2)))
def test_variance_long_short_variance_preserving_ewma(sample_series_per_variance, input_var, alphas):
series = sample_series_per_variance[input_var]
filtered = long_short_variance_preserving_ewma(series, alpha1=alphas[0], alpha2=alphas[1])
expected = input_var
calculated = filtered.var()
msg = f"Variance of long_short_variance_preserving_ewma of a series with input_variance var must be var, expected {expected}, got {calculated}"
assert calculated == pytest.approx(expected, rel=1e-1), msg


@pytest.mark.parametrize("alpha", SMOOTHING_PARS)
def test_trivial_long_short_variance_preserving_ewma(sample_series_per_variance, alpha):
series = sample_series_per_variance[0.1]
with pytest.raises(ValueError, match="alpha1 and alpha2 must be different. When they are equal, the long-short filter is ill-defined."):
long_short_variance_preserving_ewma(series, alpha1=alpha, alpha2=alpha)
class TestEWMAFilter:
"""Test cases for the standard EWMA filter."""

@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_mean_preservation(self, sample_series_per_variance, input_var, alpha):
"""Test that EWMA preserves the mean of the input series."""
series = sample_series_per_variance[input_var]
filtered = ewma(series, nu=alpha)
expected = series.mean()
calculated = filtered.mean()

msg = f"EWMA should preserve mean: expected {expected}, got {calculated}"
assert calculated == pytest.approx(expected, rel=1e-1), msg

@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_variance_reduction(self, sample_series_per_variance, input_var, alpha):
"""Test that EWMA reduces variance according to the theoretical formula."""
series = sample_series_per_variance[input_var]
filtered = ewma(series, nu=alpha)

# Theoretical variance for EWMA: var * (1 - alpha) / (1 + alpha)
expected = input_var * ((1 - alpha) / (1 + alpha))
calculated = filtered.var()

msg = (f"EWMA variance should be (1 - alpha) / (1 + alpha) * var: "
f"expected {expected}, got {calculated}")
assert calculated == pytest.approx(expected, rel=1e-1), msg


class TestVariancePreservingEWMA:
"""Test cases for the variance-preserving EWMA filter."""

@pytest.mark.parametrize("input_var, alpha", product(VARS, SMOOTHING_PARS))
def test_variance_preservation(self, sample_series_per_variance, input_var, alpha):
"""Test that variance-preserving EWMA maintains input variance."""
series = sample_series_per_variance[input_var]
filtered = variance_preserving_ewma(series, nu=alpha)

expected = input_var
calculated = filtered.var()

msg = (f"Variance-preserving EWMA should maintain input variance: "
f"expected {expected}, got {calculated}")
assert calculated == pytest.approx(expected, rel=1e-1), msg


class TestLongShortVariancePreservingEWMA:
"""Test cases for the long-short variance-preserving EWMA filter."""

@pytest.mark.parametrize("input_var, alphas", product(VARS, combinations(SMOOTHING_PARS, 2)))
def test_variance_preservation(self, sample_series_per_variance, input_var, alphas):
"""Test that long-short variance-preserving EWMA maintains input variance."""
series = sample_series_per_variance[input_var]
filtered = long_short_variance_preserving_ewma(
series, nu1=alphas[0], nu2=alphas[1]
)

expected = input_var
calculated = filtered.var()

msg = (f"Long-short variance-preserving EWMA should maintain input variance: "
f"expected {expected}, got {calculated}")
assert calculated == pytest.approx(expected, rel=1e-1), msg

@pytest.mark.parametrize("alpha", SMOOTHING_PARS)
def test_equal_alphas_error(self, sample_series_per_variance, alpha):
"""Test that equal alpha values raise a ValueError."""
series = sample_series_per_variance[0.1]

error_msg = ("nu1 and nu2 must be different. When they are equal, the long-short filter is ill-defined.")

with pytest.raises(ValueError, match=error_msg):
long_short_variance_preserving_ewma(series, nu1=alpha, nu2=alpha)


Loading
Loading