> ## Documentation Index
> Fetch the complete documentation index at: https://docs.clypt.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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

```python theme={null}
graph.add_node("dema", DEMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/dema.py`</sub>

***

## 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

```python theme={null}
# 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.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/ema.py`</sub>

***

## 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

```python theme={null}
graph.add_node("kama", KAMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=10,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/kama.py`</sub>

***

## 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

```python theme={null}
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.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/ma.py`</sub>

***

## 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

```python theme={null}
op = MACDEXT(Input("FIELD:close", timeframe="1m", lookback=35))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/ma.py`</sub>

***

## 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

```python theme={null}
op = MACDFIX(Input("FIELD:close", timeframe="1m", lookback=35))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/ma.py`</sub>

***

## 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

```python theme={null}
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.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/mama.py`</sub>

***

## 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

```python theme={null}
graph.add_node("sma", SMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    period=20,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/sma.py`</sub>

***

## T3

Triple Exponential Moving Average (T3) indicator.

T3 is a smoother version of TEMA using a volume factor.
T3 = c1*e6 + c2*e5 + c3*e4 + c4*e3
where:
e1 = EMA(price), e2 = EMA(e1), ..., e6 = EMA(e5)
c1 = -a^3, c2 = 3*a^2 + 3*a^3, c3 = -6*a^2 - 3*a - 3*a^3, c4 = 1 + 3*a + 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

```python theme={null}
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.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/t3.py`</sub>

***

## 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

```python theme={null}
graph.add_node("tema", TEMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/tema.py`</sub>

***

## 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

```python theme={null}
graph.add_node("trima", TRIMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    period=14,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/trima.py`</sub>

***

## WMA

Weighted Moving Average indicator.

WMA assigns linearly increasing weights to more recent data points.
WMA = (n\*P\_n + (n-1)*P\_\{n-1} + ... + 1*P\_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

```python theme={null}
graph.add_node("wma", WMA(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    period=14,
))
```

### Source Code

Full `compute()` implementation — no hidden logic.

```python theme={null}
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),
    )
```

<sub>Source: `apps/trading/operators/indicator/wma.py`</sub>

## Related Pages

<CardGroup cols={2}>
  <Card title="Operator Protocol" icon="gear" href="/engine/operator-protocol">
    How operators implement the compute() interface
  </Card>

  <Card title="StatefulGraph" icon="diagram-project" href="/engine/stateful-graph">
    How operators compose into a DAG
  </Card>
</CardGroup>