pysatl_tsp.implementations.processor.ema_handler

Module Contents

Classes

EMAHandler

Exponential Moving Average (EMA) handler.

API

class pysatl_tsp.implementations.processor.ema_handler.EMAHandler(length: int = 10, adjust: bool = False, sma: bool = True, alpha: float | None = None, source: pysatl_tsp.core.Handler[Any, float | None] | None = None)

Bases: pysatl_tsp.core.processor.InductiveHandler[float | None, float | None]

Exponential Moving Average (EMA) handler.

Calculates EMA values for a sequence of input values, matching the functionality of pandas_ta.EMA implementation.

Parameters:

• length – The period for EMA calculation, defaults to 10

• adjust – Whether to use adjusted weights in calculation, defaults to False

• sma – Whether to use SMA for initial value, defaults to True

• alpha – Custom smoothing factor, defaults to 2/(length+1) if None

• source – Input data source, defaults to None

Example:

# Create a data source with numeric values
data_source = SimpleDataProvider([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0])

# Create an EMA handler with length of 5
ema_handler = EMAHandler(length=5)
ema_handler.set_source(data_source)

# Process the data
for value in ema_handler:
    print(value)

# The first 4 values will be None since we're using SMA initialization
# The 5th value will be the SMA of the first 5 values
# Subsequent values will be EMA values based on the formula

Initialization

Initialize EMA handler with specified parameters.

Parameters:

• length – The period for EMA calculation, defaults to 10

• adjust – Whether to use adjusted weights in calculation, defaults to False

• sma – Whether to use SMA for initial value, defaults to True

• alpha – Custom smoothing factor, defaults to 2/(length+1) if None

• source – Input data source, defaults to None

_compute_result(state: dict[str, Any]) → float | None

Return the current EMA value or None if not yet initialized.

Calculates the final EMA value by dividing the numerator by the denominator if the denominator exists (indicating that EMA is initialized).

Parameters:

state – Current state of the handler

Returns:

Current EMA value or None if not yet calculated

_initialize_state() → dict[str, Any]

Initialize state for EMA calculation.

Creates the initial state dictionary with a window to collect values, variables to track the EMA calculation, and a position counter.

Returns:

Dictionary containing initial state variables

_update_state(state: dict[str, Any], value: float | None) → dict[str, Any]

Update state with a new value.

Implements the same logic as the original pandas_ta.EMA function: - If sma=True, initialize EMA with the SMA of first ‘length’ values - If sma=False, initialize EMA with the first value - Then apply the standard EMA formula for subsequent values

When adjust=True, uses an adjusted weighting method that gives more weight to recent observations.

Parameters:

• state – Current state dictionary

• value – New value to incorporate into the EMA calculation

Returns:

Updated state dictionary