Moving Averages

SMA, EMA, DEMA, TEMA, WMA, TRIMA, T3, KAMA, MAMA, MA

Overview

This page documents 12 operators (role: INDICATOR).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
DEMA	INDICATOR	period=14	No
EMA	UNKNOWN	self_input, span=None, alpha=None	No
KAMA	INDICATOR	period=10, fast_period=2, slow_period=30	No
MA	INDICATOR	period=30, ma_type=0	No
MACDEXT	INDICATOR	fast_period=12, fast_ma_type=1, slow_period=26	No
MACDFIX	INDICATOR	signal_period=9, output='macd'	No
MAMA	INDICATOR	fast_limit=0.5, slow_limit=0.05	No
SMA	INDICATOR	period=14	No
T3	INDICATOR	period=5, volume_factor=0.7	No
TEMA	INDICATOR	period=14	No
TRIMA	INDICATOR	period=14	No
WMA	INDICATOR	period=14	No

---

DEMA

Double Exponential Moving Average indicator.

DEMA reduces the lag compared to a traditional EMA by applying the formula: DEMA = 2 * EMA(price) - EMA(EMA(price))

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period)
period	int	14	Number of periods for EMA calculation

Usage

graph.add_node("dema", DEMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period:
                # Calculate first EMA
                ema1 = self._calculate_ema(prices[valid_mask])
                # Calculate EMA of EMA
                ema2 = self._calculate_ema(ema1)
                # DEMA = 2 * EMA - EMA(EMA)
                result[i] = 2 * ema1[-1] - ema2[-1]

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/dema.py}

---

EMA

Exponential Moving Average for signal smoothing.

Applies EMA smoothing across time to reduce signal noise and turnover. Uses the formula: EMA_t = alpha * x_t + (1 - alpha) * EMA_{t-1}

This operator uses self-reference: it receives its own previous output via the Graph’s lookback buffer, making it stateless.

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal to smooth (lookback=1 for current value)
self_input	'Input'	Required	Self-reference input for previous EMA (lookback=1)
span	Optional[int]	None	EMA span (number of periods). alpha = 2 / (span + 1)
alpha	Optional[float]	None	Direct alpha value (overrides span if provided)

Usage

# In Graph setup:
graph.add_node("alpha_ema", EMA(
    Input("raw_alpha", timeframe="1h", lookback=1),
    Input("alpha_ema", timeframe="1h", lookback=1),  # Self-reference
    span=6,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if not isinstance(data, list) or len(data) < 2:
        # Fallback: just return input
        if isinstance(data, list):
            data = data[0]
        last = data[-1] if len(data) > 0 else data
        return last

    input_data, self_data = data[0], data[1]

    # Current input value
    last_input = input_data[-1] if len(input_data) > 0 else input_data
    values = last_input.value
    exists = last_input.exists
    valid = last_input.valid

    # Previous EMA value (self-reference)
    last_self = self_data[-1] if len(self_data) > 0 else self_data
    prev_ema = last_self.value if hasattr(last_self, 'value') else np.full_like(values, np.nan)

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if not compute_mask[i] or np.isnan(values[i]):
            # No valid input: keep previous EMA
            result[i] = prev_ema[i] if not np.isnan(prev_ema[i]) else np.nan
        elif np.isnan(prev_ema[i]):
            # First valid value: initialize EMA
            result[i] = values[i]
        else:
            # EMA update: alpha * new + (1 - alpha) * old
            result[i] = self._alpha * values[i] + (1 - self._alpha) * prev_ema[i]

    result_valid = compute_mask & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/ema.py}

---

KAMA

Kaufman Adaptive Moving Average indicator.

KAMA adapts to market volatility by adjusting its smoothing constant. When the market is trending, KAMA responds quickly; when it’s ranging, it slows down.

Efficiency Ratio (ER) = Change / Volatility Smoothing Constant (SC) = [ER * (fast_sc - slow_sc) + slow_sc]^2 KAMA_t = KAMA_{t-1} + SC * (Price_t - KAMA_{t-1})

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period)
period	int	10	Efficiency ratio period
fast_period	int	2	Fast EMA period
slow_period	int	30	Slow EMA period

Usage

graph.add_node("kama", KAMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=10,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period + 1:
                valid_prices = prices[valid_mask]
                result[i] = self._calculate_kama(valid_prices)

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/kama.py}

---

MA

Generic Moving Average indicator.

Supports multiple MA types via ma_type parameter.

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal
period	int	30	MA period
ma_type	int	0	Type of MA: 0=SMA, 1=EMA, 2=WMA, 3=DEMA, 4=TEMA

Usage

graph.add_node("ma", MA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=35),
    period=30,
    ma_type=0,  # SMA
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period:
                valid_prices = prices[valid_mask]
                result[i] = self._calculate_ma(valid_prices)

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/ma.py}

---

MACDEXT

MACD with controllable MA type.

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal
fast_period	int	12	Fast MA period
fast_ma_type	int	1	Fast MA type
slow_period	int	26	Slow MA period
slow_ma_type	int	1	Slow MA type
signal_period	int	9	Signal MA period
signal_ma_type	int	1	Signal MA type
output	str	'macd'	”macd”, “signal”, or “histogram”

Usage

op = MACDEXT(Input("FIELD:close", timeframe="1m", lookback=35))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._slow_period + self._signal_period:
                valid_prices = prices[valid_mask]
                macd, signal, hist = self._calculate_macdext(valid_prices)
                if self._output == "signal":
                    result[i] = signal
                elif self._output == "histogram":
                    result[i] = hist
                else:
                    result[i] = macd

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/ma.py}

---

MACDFIX

MACD Fix 12/26 - MACD with fixed periods.

Uses fixed 12/26/9 periods like traditional MACD.

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal
signal_period	int	9	Signal period
output	str	'macd'	”macd”, “signal”, or “histogram”

Usage

op = MACDFIX(Input("FIELD:close", timeframe="1m", lookback=35))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 35:  # 26 + 9
                valid_prices = prices[valid_mask]
                macd, signal, hist = self._calculate_macdfix(valid_prices)
                if self._output == "signal":
                    result[i] = signal
                elif self._output == "histogram":
                    result[i] = hist
                else:
                    result[i] = macd

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/ma.py}

---

MAMA

MESA Adaptive Moving Average indicator.

MAMA adapts to price movement based on the rate of change of phase (from Hilbert Transform). It provides both MAMA and FAMA (Following Adaptive Moving Average) values.

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= 32)
fast_limit	float	0.5	Fast limit for alpha
slow_limit	float	0.05	Slow limit for alpha

Usage

graph.add_node("mama", MAMA(
        Input("FIELD:gateio:spot:close", timeframe="1h", lookback=50),
        fast_limit=0.5,
        slow_limit=0.05,
    ))

Note:
    This implementation returns MAMA value. For FAMA, use a separate FAMA operator
    or access via context.

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 32:
                valid_prices = prices[valid_mask]
                mama, _ = self._calculate_mama(valid_prices)
                result[i] = mama

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/mama.py}

---

SMA

Simple Moving Average indicator.

Calculates the arithmetic mean of a given set of values over a specified period. SMA_t = (x_t + x_{t-1} + … + x_{t-n+1}) / n

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period)
period	int	14	Number of periods to average

Usage

graph.add_node("sma", SMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    period=20,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            valid_values = window[~np.isnan(window)]
            if len(valid_values) >= period:
                result[i] = np.mean(valid_values)

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/sma.py}

---

T3

Triple Exponential Moving Average (T3) indicator.

T3 is a smoother version of TEMA using a volume factor. T3 = c1e6 + c2e5 + c3e4 + c4e3 where: e1 = EMA(price), e2 = EMA(e1), …, e6 = EMA(e5) c1 = -a^3, c2 = 3a^2 + 3a^3, c3 = -6a^2 - 3a - 3a^3, c4 = 1 + 3a + a^3 + 3*a^2 a = volume_factor (default 0.7)

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period * 3)
period	int	5	Number of periods for EMA calculation
volume_factor	float	0.7	Smoothing factor

Usage

graph.add_node("t3", T3(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=5,
    volume_factor=0.7,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period * 2:
                valid_prices = prices[valid_mask]
                # Calculate 6 EMAs
                e1 = self._calculate_ema(valid_prices)
                e2 = self._calculate_ema(e1)
                e3 = self._calculate_ema(e2)
                e4 = self._calculate_ema(e3)
                e5 = self._calculate_ema(e4)
                e6 = self._calculate_ema(e5)
                # T3 formula
                result[i] = (self._c1 * e6[-1] + self._c2 * e5[-1] +
                             self._c3 * e4[-1] + self._c4 * e3[-1])

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/t3.py}

---

TEMA

Triple Exponential Moving Average indicator.

TEMA further reduces lag by applying the formula: TEMA = 3 * EMA - 3 * EMA(EMA) + EMA(EMA(EMA))

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period * 2)
period	int	14	Number of periods for EMA calculation

Usage

graph.add_node("tema", TEMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period:
                # Calculate EMAs
                ema1 = self._calculate_ema(prices[valid_mask])
                ema2 = self._calculate_ema(ema1)
                ema3 = self._calculate_ema(ema2)
                # TEMA = 3 * EMA - 3 * EMA(EMA) + EMA(EMA(EMA))
                result[i] = 3 * ema1[-1] - 3 * ema2[-1] + ema3[-1]

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/tema.py}

---

TRIMA

Triangular Moving Average indicator.

TRIMA is a double-smoothed SMA that gives more weight to middle values. For odd periods: TRIMA = SMA(SMA(price, (n+1)/2), (n+1)/2) For even periods: TRIMA = SMA(SMA(price, n/2+1), n/2)

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period)
period	int	14	Number of periods

Usage

graph.add_node("trima", TRIMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            valid_prices = prices[valid_mask]
            if len(valid_prices) >= self._period:
                result[i] = self._calculate_trima(valid_prices)

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/trima.py}

---

WMA

Weighted Moving Average indicator.

WMA assigns linearly increasing weights to more recent data points. WMA = (n*P_n + (n-1)P_{n-1} + … + 1P_1) / (n + (n-1) + … + 1)

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal (requires lookback >= period)
period	int	14	Number of periods to average

Usage

graph.add_node("wma", WMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    period=14,
))

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: Union[TaggedArray, List[TaggedArray]],
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    if isinstance(data, list):
        data = data[0]

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    # Extract values for all time periods
    all_values = np.array([data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)):
                weights = self._weights[-period:]
                weight_sum = weights.sum()
                result[i] = np.sum(weights * window) / weight_sum

    result_valid = exists & valid & ~np.isnan(result)

    return TaggedArray(
        value=result,
        exists=exists,
        valid=result_valid,
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: apps/trading/operators/indicator/wma.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG