pysatl_tsp.implementations.processor.rma_handler

Module Contents

Classes

RMAHandler

Wilder’s Moving Average (RMA) Handler.

API

class pysatl_tsp.implementations.processor.rma_handler.RMAHandler(length: int = 10, source: pysatl_tsp.core.Handler[Any, float | None] | None = None)

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

Wilder’s Moving Average (RMA) Handler.

The Wilder’s Moving Average is an Exponential Moving Average (EMA) with a modified alpha = 1 / length. It was introduced by J. Welles Wilder and is also known as the Smoothed Moving Average.

RMA gives greater weight to recent data and less weight to older data, but does so more gradually than a standard EMA, resulting in a smoother line. It’s commonly used in technical analysis for calculating indicators like the Relative Strength Index (RSI).

Parameters:

• length – The number of periods for the moving average calculation, defaults to 10

• source – Source handler providing the input data, 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 RMA handler with length of 5
rma_handler = RMAHandler(length=5)
rma_handler.set_source(data_source)

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

# First values will be None until we have 'length' values
# Then RMA values will be calculated with alpha = 1/5

Initialization

Initialize a Wilder’s Moving Average handler.

Parameters:

• length – The number of periods for the moving average calculation, defaults to 10

• source – Source handler providing the input data, defaults to None

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

Calculate the RMA value based on current state.

Returns the current RMA value if there’s a non-zero denominator and we have seen at least ‘length’ non-None values. Otherwise returns None.

Parameters:

state – Current state dictionary

Returns:

Current RMA value or None if conditions aren’t met

_initialize_state() → dict[str, float | int]

Initialize state for RMA calculation.

Creates an initial state dictionary with counters and accumulators set to zero.

Returns:

Dictionary containing initial state variables

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

Update state with a new value.

Updates the running totals using the RMA formula with alpha = 1/length. Handles None values by treating them as zeros in the calculation but tracking how many non-None values have been seen.

Parameters:

• state – Current state dictionary

• value – New value to incorporate into the RMA calculation

Returns:

Updated state dictionary