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

# Momentum Indicators

> RSI, MACD, MOM, ROC, CMO, APO, PPO, STOCH, STOCHRSI, WILLR, ULTOSC, BOP, TRIX, CCI

## Overview

This page documents **17 operators** (role: `INDICATOR`).

## Quick Reference

| Operator     | Role        | Key Parameters                                        | Ephemeral |
| ------------ | ----------- | ----------------------------------------------------- | --------- |
| **APO**      | `INDICATOR` | `fast_period=12`, `slow_period=26`                    | No        |
| **BOP**      | `INDICATOR` | —                                                     | No        |
| **CMO**      | `INDICATOR` | `period=14`                                           | No        |
| **MACD**     | `INDICATOR` | `fast_period=12`, `slow_period=26`, `signal_period=9` | No        |
| **MOM**      | `INDICATOR` | `period=10`                                           | No        |
| **PPO**      | `INDICATOR` | `fast_period=12`, `slow_period=26`                    | No        |
| **ROC**      | `INDICATOR` | `period=10`                                           | No        |
| **ROCP**     | `INDICATOR` | `period=10`                                           | No        |
| **ROCR**     | `INDICATOR` | `period=10`                                           | No        |
| **ROCR100**  | `INDICATOR` | `period=10`                                           | No        |
| **RSI**      | `INDICATOR` | `period=14`                                           | No        |
| **STOCH**    | `INDICATOR` | `fastk_period=14`, `slowk_period=3`, `slowd_period=3` | No        |
| **STOCHF**   | `INDICATOR` | `fastk_period=14`, `fastd_period=3`, `output='fastk'` | No        |
| **STOCHRSI** | `INDICATOR` | `rsi_period=14`, `stoch_period=14`, `fastk_period=3`  | No        |
| **TRIX**     | `INDICATOR` | `period=30`                                           | No        |
| **ULTOSC**   | `INDICATOR` | `period1=7`, `period2=14`, `period3=28`               | No        |
| **WILLR**    | `INDICATOR` | `period=14`                                           | No        |

***

## APO

Absolute Price Oscillator indicator.

APO is the difference between a fast and slow exponential moving average.
APO = EMA(fast) - EMA(slow)

Positive values indicate upward momentum, negative values indicate downward.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter     | Type      | Default  | Description                                      |
| ------------- | --------- | -------- | ------------------------------------------------ |
| `input`       | `'Input'` | Required | Input signal (requires lookback >= slow\_period) |
| `fast_period` | `int`     | `12`     | Fast EMA period                                  |
| `slow_period` | `int`     | `26`     | Slow EMA period                                  |

### Usage

```python theme={null}
graph.add_node("apo", APO(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    fast_period=12,
    slow_period=26,
))
```

### 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:
                valid_prices = prices[valid_mask]
                fast_ema = self._calculate_ema(valid_prices, self._fast_alpha)
                slow_ema = self._calculate_ema(valid_prices, self._slow_alpha)
                result[i] = fast_ema[-1] - slow_ema[-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/apo.py`</sub>

***

## BOP

Balance of Power indicator.

BOP measures the strength of buyers vs sellers by assessing
the ability of each to push price to extreme levels.

BOP = (Close - Open) / (High - Low)

Values range from -1 to +1.
Positive = buyers in control, Negative = sellers in control.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `open_`   | `'Input'` | Required | Open price input  |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `close`   | `'Input'` | Required | Close price input |

### Usage

```python theme={null}
graph.add_node("bop", BOP(
    open_=Input("FIELD:gateio:spot:open", timeframe="1h", lookback=1),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=1),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=1),
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=1),
))
```

### 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) < 4:
        return self._empty_output()

    open_data, high_data, low_data, close_data = data[0], data[1], data[2], data[3]

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

    last_close = close_data[-1]
    n_symbols = len(last_close.exists)
    exists = last_close.exists
    valid = last_close.valid

    open_val = open_data[-1].value
    high_val = high_data[-1].value
    low_val = low_data[-1].value
    close_val = close_data[-1].value

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            o, h, l, c = open_val[i], high_val[i], low_val[i], close_val[i]
            if not any(np.isnan([o, h, l, c])):
                hl_range = h - l
                if hl_range > 0:
                    result[i] = (c - o) / hl_range

    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/bop.py`</sub>

***

## CMO

Chande Momentum Oscillator indicator.

CMO is a modified RSI that measures momentum.
CMO = ((Sum of gains - Sum of losses) / (Sum of gains + Sum of losses)) \* 100

Values range from -100 to +100.
Above 50 indicates overbought, below -50 indicates oversold.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
graph.add_node("cmo", CMO(
    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

    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_cmo(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/cmo.py`</sub>

***

## MACD

Moving Average Convergence/Divergence indicator.

MACD shows the relationship between two EMAs of a security's price.
MACD Line = EMA(fast) - EMA(slow)
Signal Line = EMA(MACD Line, signal\_period)
Histogram = MACD Line - Signal Line

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter       | Type      | Default  | Description                                                       |
| --------------- | --------- | -------- | ----------------------------------------------------------------- |
| `input`         | `'Input'` | Required | Input signal (requires lookback >= slow\_period + signal\_period) |
| `fast_period`   | `int`     | `12`     | Fast EMA period                                                   |
| `slow_period`   | `int`     | `26`     | Slow EMA period                                                   |
| `signal_period` | `int`     | `9`      | Signal line EMA period                                            |
| `output`        | `str`     | `'macd'` | Which output to return: "macd", "signal", or "histogram"          |

### Usage

```python theme={null}
graph.add_node("macd", MACD(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=40),
    fast_period=12,
    slow_period=26,
    signal_period=9,
))
```

### 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._slow_period + self._signal_period:
                valid_prices = prices[valid_mask]
                macd, signal, histogram = self._calculate_macd(valid_prices)
                if self._output == "signal":
                    result[i] = signal
                elif self._output == "histogram":
                    result[i] = histogram
                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/macd.py`</sub>

***

## MOM

Momentum indicator.

Momentum measures the rate of change of a security's price.
MOM = Price\_today - Price\_n\_periods\_ago

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
graph.add_node("mom", MOM(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=15),
    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)

    if n_time > self._period:
        for i in range(n_symbols):
            if exists[i] and valid[i]:
                current = all_values[-1, i]
                past = all_values[-(self._period + 1), i]
                if not np.isnan(current) and not np.isnan(past):
                    result[i] = current - past

    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/mom.py`</sub>

***

## PPO

Percentage Price Oscillator indicator.

PPO is similar to MACD but expressed as a percentage.
PPO = ((EMA(fast) - EMA(slow)) / EMA(slow)) \* 100

This makes it easier to compare across different price levels.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter     | Type      | Default  | Description                                      |
| ------------- | --------- | -------- | ------------------------------------------------ |
| `input`       | `'Input'` | Required | Input signal (requires lookback >= slow\_period) |
| `fast_period` | `int`     | `12`     | Fast EMA period                                  |
| `slow_period` | `int`     | `26`     | Slow EMA period                                  |

### Usage

```python theme={null}
graph.add_node("ppo", PPO(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    fast_period=12,
    slow_period=26,
))
```

### 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:
                valid_prices = prices[valid_mask]
                fast_ema = self._calculate_ema(valid_prices, self._fast_alpha)
                slow_ema = self._calculate_ema(valid_prices, self._slow_alpha)
                if slow_ema[-1] != 0:
                    result[i] = ((fast_ema[-1] - slow_ema[-1]) / slow_ema[-1]) * 100

    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/ppo.py`</sub>

***

## ROC

Rate of Change indicator.

ROC = ((Price\_today - Price\_n\_periods\_ago) / Price\_n\_periods\_ago) \* 100

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
graph.add_node("roc", ROC(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=15),
    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

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

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

    if n_time > self._period:
        for i in range(n_symbols):
            if exists[i] and valid[i]:
                current = all_values[-1, i]
                past = all_values[-(self._period + 1), i]
                if not np.isnan(current) and not np.isnan(past) and past != 0:
                    result[i] = ((current - past) / past) * 100

    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/roc.py`</sub>

***

## ROCP

Rate of Change Percentage indicator.

ROCP = (Price\_today - Price\_n\_periods\_ago) / Price\_n\_periods\_ago

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
op = ROCP(Input("FIELD:close", timeframe="1m", lookback=10), 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

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

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

    if n_time > self._period:
        for i in range(n_symbols):
            if exists[i] and valid[i]:
                current = all_values[-1, i]
                past = all_values[-(self._period + 1), i]
                if not np.isnan(current) and not np.isnan(past) and past != 0:
                    result[i] = (current - past) / past

    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/roc.py`</sub>

***

## ROCR

Rate of Change Ratio indicator.

ROCR = Price\_today / Price\_n\_periods\_ago

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
op = ROCR(Input("FIELD:close", timeframe="1m", lookback=10), 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

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

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

    if n_time > self._period:
        for i in range(n_symbols):
            if exists[i] and valid[i]:
                current = all_values[-1, i]
                past = all_values[-(self._period + 1), i]
                if not np.isnan(current) and not np.isnan(past) and past != 0:
                    result[i] = current / past

    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/roc.py`</sub>

***

## ROCR100

Rate of Change Ratio 100 scale indicator.

ROCR100 = (Price\_today / Price\_n\_periods\_ago) \* 100

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
op = ROCR100(Input("FIELD:close", timeframe="1m", lookback=10), 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

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

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

    if n_time > self._period:
        for i in range(n_symbols):
            if exists[i] and valid[i]:
                current = all_values[-1, i]
                past = all_values[-(self._period + 1), i]
                if not np.isnan(current) and not np.isnan(past) and past != 0:
                    result[i] = (current / past) * 100

    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/roc.py`</sub>

***

## RSI

Relative Strength Index indicator.

RSI measures the magnitude of recent price changes to evaluate overbought
or oversold conditions.

RSI = 100 - (100 / (1 + RS))
where RS = Average Gain / Average Loss over the period

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
graph.add_node("rsi", RSI(
    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)

    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_rsi(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/rsi.py`</sub>

***

## STOCH

Stochastic Oscillator indicator.

Stochastic measures the close position relative to the high-low range.
%K = ((Close - Lowest Low) / (Highest High - Lowest Low)) \* 100
%D = SMA(%K, slowd\_period)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter      | Type      | Default   | Description         |
| -------------- | --------- | --------- | ------------------- |
| `close`        | `'Input'` | Required  | Close price input   |
| `high`         | `'Input'` | Required  | High price input    |
| `low`          | `'Input'` | Required  | Low price input     |
| `fastk_period` | `int`     | `14`      | %K period           |
| `slowk_period` | `int`     | `3`       | %K smoothing period |
| `slowd_period` | `int`     | `3`       | %D period           |
| `output`       | `str`     | `'slowk'` | "slowk" or "slowd"  |

### Usage

```python theme={null}
graph.add_node("stoch", STOCH(
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=20),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=20),
    fastk_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 not isinstance(data, list) or len(data) < 3:
        return self._empty_output()

    close_data, high_data, low_data = data[0], data[1], data[2]

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

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

    close_values = np.array([close_data[t].value for t in range(n_time)])
    high_values = np.array([high_data[t].value for t in range(n_time)])
    low_values = np.array([low_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]:
            closes = close_values[:, i]
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= self._fastk_period + self._slowk_period:
                slowk, slowd = self._calculate_stoch(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask]
                )
                result[i] = slowd if self._output == "slowd" else slowk

    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/stoch.py`</sub>

***

## STOCHF

Stochastic Fast indicator.

Fast Stochastic returns raw %K without smoothing.
%K = ((Close - Lowest Low) / (Highest High - Lowest Low)) \* 100
%D = SMA(%K, fastd\_period)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter      | Type      | Default   | Description        |
| -------------- | --------- | --------- | ------------------ |
| `close`        | `'Input'` | Required  | Close price input  |
| `high`         | `'Input'` | Required  | High price input   |
| `low`          | `'Input'` | Required  | Low price input    |
| `fastk_period` | `int`     | `14`      | %K period          |
| `fastd_period` | `int`     | `3`       | %D period          |
| `output`       | `str`     | `'fastk'` | "fastk" or "fastd" |

### Usage

```python theme={null}
op = STOCHF(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:close", timeframe="1m", lookback=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 not isinstance(data, list) or len(data) < 3:
        return self._empty_output()

    close_data, high_data, low_data = data[0], data[1], data[2]

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

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

    close_values = np.array([close_data[t].value for t in range(n_time)])
    high_values = np.array([high_data[t].value for t in range(n_time)])
    low_values = np.array([low_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]:
            closes = close_values[:, i]
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= self._fastk_period:
                fastk, fastd = self._calculate_stochf(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask]
                )
                result[i] = fastd if self._output == "fastd" else fastk

    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/stoch.py`</sub>

***

## STOCHRSI

Stochastic RSI indicator.

StochRSI applies the Stochastic formula to RSI values instead of prices.
StochRSI = (RSI - Lowest RSI) / (Highest RSI - Lowest RSI)

Values range from 0 to 1 (or 0 to 100 when multiplied).

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter      | Type      | Default   | Description                                                     |
| -------------- | --------- | --------- | --------------------------------------------------------------- |
| `input`        | `'Input'` | Required  | Input signal (requires lookback >= rsi\_period + stoch\_period) |
| `rsi_period`   | `int`     | `14`      | RSI period                                                      |
| `stoch_period` | `int`     | `14`      | Stochastic period                                               |
| `fastk_period` | `int`     | `3`       | Fast %K smoothing                                               |
| `fastd_period` | `int`     | `3`       | Fast %D smoothing                                               |
| `output`       | `str`     | `'fastk'` | "fastk" or "fastd"                                              |

### Usage

```python theme={null}
graph.add_node("stochrsi", STOCHRSI(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=35),
    rsi_period=14,
    stoch_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

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

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

    min_periods = self._rsi_period + self._stoch_period + 1

    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) >= min_periods:
                valid_prices = prices[valid_mask]
                fastk, fastd = self._calculate_stochrsi(valid_prices)
                result[i] = fastd if self._output == "fastd" else fastk

    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/stochrsi.py`</sub>

***

## TRIX

TRIX - 1-day Rate-Of-Change (ROC) of a Triple Smooth EMA.

TRIX = 100 \* (EMA3\_today - EMA3\_yesterday) / EMA3\_yesterday
where EMA3 = EMA(EMA(EMA(price)))

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description  |
| --------- | --------- | -------- | ------------ |
| `input`   | `'Input'` | Required | Input signal |
| `period`  | `int`     | `30`     | EMA period   |

### Usage

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

### 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 * 2:
                valid_prices = prices[valid_mask]
                result[i] = self._calculate_trix(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/trix.py`</sub>

***

## ULTOSC

Ultimate Oscillator indicator.

The Ultimate Oscillator uses weighted averages of three different periods
to reduce volatility and false trading signals.

UO = 100 \* \[(4 \* Avg7) + (2 \* Avg14) + Avg28] / (4 + 2 + 1)
where Avg = Sum(BP) / Sum(TR)
BP (Buying Pressure) = Close - Min(Low, PrevClose)
TR (True Range) = Max(High, PrevClose) - Min(Low, PrevClose)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `close`   | `'Input'` | Required | Close price input |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `period1` | `int`     | `7`      | First period      |
| `period2` | `int`     | `14`     | Second period     |
| `period3` | `int`     | `28`     | Third period      |

### Usage

```python theme={null}
graph.add_node("ultosc", ULTOSC(
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=35),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=35),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", 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 not isinstance(data, list) or len(data) < 3:
        return self._empty_output()

    close_data, high_data, low_data = data[0], data[1], data[2]

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

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

    close_values = np.array([close_data[t].value for t in range(n_time)])
    high_values = np.array([high_data[t].value for t in range(n_time)])
    low_values = np.array([low_data[t].value for t in range(n_time)])

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

    min_periods = self._period3 + 1

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            closes = close_values[:, i]
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= min_periods:
                result[i] = self._calculate_ultosc(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask]
                )

    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/ultosc.py`</sub>

***

## WILLR

Williams %R indicator.

Williams %R is a momentum indicator that measures overbought/oversold levels.
%R = ((Highest High - Close) / (Highest High - Lowest Low)) \* -100

Values range from -100 to 0.
-20 to 0 = overbought, -100 to -80 = oversold.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `close`   | `'Input'` | Required | Close price input |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `period`  | `int`     | `14`     | Lookback period   |

### Usage

```python theme={null}
graph.add_node("willr", WILLR(
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=20),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=20),
    low=Input("FIELD:gateio:spot:low", 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 not isinstance(data, list) or len(data) < 3:
        return self._empty_output()

    close_data, high_data, low_data = data[0], data[1], data[2]

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

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

    close_values = np.array([close_data[t].value for t in range(n_time)])
    high_values = np.array([high_data[t].value for t in range(n_time)])
    low_values = np.array([low_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]:
            close = close_values[-1, i]
            highs = high_values[-period:, i]
            lows = low_values[-period:, i]

            if not np.any(np.isnan(highs)) and not np.any(np.isnan(lows)) and not np.isnan(close):
                highest_high = np.max(highs)
                lowest_low = np.min(lows)
                hl_range = highest_high - lowest_low

                if hl_range > 0:
                    result[i] = ((highest_high - close) / hl_range) * -100

    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/willr.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>