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

# Trend Indicators

> ADX, DI, AROON, SAR, ICHIMOKU

## Overview

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

## Quick Reference

| Operator      | Role        | Key Parameters                                                     | Ephemeral |
| ------------- | ----------- | ------------------------------------------------------------------ | --------- |
| **ADX**       | `INDICATOR` | `period=14`                                                        | No        |
| **ADXR**      | `INDICATOR` | `period=14`                                                        | No        |
| **DX**        | `INDICATOR` | `period=14`                                                        | No        |
| **AROON**     | `INDICATOR` | `period=14`, `output='up'`                                         | No        |
| **AROONOSC**  | `INDICATOR` | `period=14`                                                        | No        |
| **PLUS\_DI**  | `INDICATOR` | `period=14`                                                        | No        |
| **MINUS\_DI** | `INDICATOR` | `period=14`                                                        | No        |
| **PLUS\_DM**  | `INDICATOR` | `period=14`                                                        | No        |
| **MINUS\_DM** | `INDICATOR` | `period=14`                                                        | No        |
| **ICHIMOKU**  | `INDICATOR` | `tenkan_period=9`, `kijun_period=26`, `senkou_b_period=52`         | No        |
| **SAR**       | `INDICATOR` | `acceleration=0.02`, `maximum=0.2`                                 | No        |
| **SAREXT**    | `INDICATOR` | `start_value=0.0`, `offset_on_reverse=0.0`, `accel_init_long=0.02` | No        |

***

## ADX

Average Directional Movement Index indicator.

ADX measures the strength of a trend, regardless of direction.
Values above 25 indicate a strong trend.

**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("adx", ADX(
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=30),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=30),
    low=Input("FIELD:gateio:spot:low", 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 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._period * 2:
                result[i] = self._calculate_adx(
                    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/adx.py`</sub>

***

## ADXR

Average Directional Movement Index Rating indicator.

ADXR is the average of current ADX and ADX from period days ago.
ADXR = (ADX\_today + ADX\_period\_days\_ago) / 2

**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}
op = ADXR(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:close", timeframe="1m", lookback=14),
    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._period * 3:
                result[i] = self._calculate_adxr(
                    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/adx.py`</sub>

***

## DX

Directional Movement Index indicator.

DX = (|+DI - -DI| / (+DI + -DI)) \* 100

**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}
op = DX(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:close", timeframe="1m", lookback=14),
    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._period + 1:
                result[i] = self._calculate_dx(
                    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/adx.py`</sub>

***

## AROON

Aroon indicator.

Aroon measures the time since the highest high and lowest low.
Aroon Up = ((period - periods since highest high) / period) \* 100
Aroon Down = ((period - periods since lowest low) / period) \* 100

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

### Parameters

| Parameter | Type      | Default  | Description      |
| --------- | --------- | -------- | ---------------- |
| `high`    | `'Input'` | Required | High price input |
| `low`     | `'Input'` | Required | Low price input  |
| `period`  | `int`     | `14`     | Lookback period  |
| `output`  | `str`     | `'up'`   | "up" or "down"   |

### Usage

```python theme={null}
graph.add_node("aroon_up", AROON(
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=20),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=20),
    period=14,
    output="up",
))
```

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

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

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

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

    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]:
            highs = high_values[-period:, i]
            lows = low_values[-period:, i]

            if not np.any(np.isnan(highs)) and not np.any(np.isnan(lows)):
                # Find periods since highest high and lowest low
                high_idx = np.argmax(highs)
                low_idx = np.argmin(lows)

                periods_since_high = period - 1 - high_idx
                periods_since_low = period - 1 - low_idx

                aroon_up = ((period - periods_since_high) / period) * 100
                aroon_down = ((period - periods_since_low) / period) * 100

                result[i] = aroon_down if self._output == "down" else aroon_up

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

***

## AROONOSC

Aroon Oscillator indicator.

Aroon Oscillator = Aroon Up - Aroon Down

Values range from -100 to +100.
Positive values indicate uptrend, negative values indicate downtrend.

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

### Parameters

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

### Usage

```python theme={null}
op = AROONOSC(
    Input("FIELD:high", timeframe="1m", lookback=25),
    Input("FIELD:low", timeframe="1m", lookback=25),
    period=25,
)
```

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

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

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

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

    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]:
            highs = high_values[-period:, i]
            lows = low_values[-period:, i]

            if not np.any(np.isnan(highs)) and not np.any(np.isnan(lows)):
                high_idx = np.argmax(highs)
                low_idx = np.argmin(lows)

                periods_since_high = period - 1 - high_idx
                periods_since_low = period - 1 - low_idx

                aroon_up = ((period - periods_since_high) / period) * 100
                aroon_down = ((period - periods_since_low) / period) * 100

                result[i] = aroon_up - aroon_down

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

***

## PLUS\_DI

Plus Directional Indicator (+DI).

+DI measures upward price movement strength.
+DI = (Smoothed +DM / ATR) \* 100

**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}
op = PLUS_DI(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:close", timeframe="1m", lookback=14),
    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._period + 1:
                result[i] = self._calculate_plus_di(
                    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/di.py`</sub>

***

## MINUS\_DI

Minus Directional Indicator (-DI).

-DI measures downward price movement strength.
-DI = (Smoothed -DM / ATR) \* 100

**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}
op = MINUS_DI(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:close", timeframe="1m", lookback=14),
    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._period + 1:
                result[i] = self._calculate_minus_di(
                    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/di.py`</sub>

***

## PLUS\_DM

Plus Directional Movement (+DM).

+DM = High - Previous High (if above 0 and > -DM, else 0)

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

### Parameters

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

### Usage

```python theme={null}
op = PLUS_DM(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    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) < 2:
        return self._empty_output()

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

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

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

    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]:
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= self._period + 1:
                result[i] = self._calculate_plus_dm(
                    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/di.py`</sub>

***

## MINUS\_DM

Minus Directional Movement (-DM).

-DM = Previous Low - Low (if above 0 and > +DM, else 0)

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

### Parameters

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

### Usage

```python theme={null}
op = MINUS_DM(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    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) < 2:
        return self._empty_output()

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

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

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

    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]:
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= self._period + 1:
                result[i] = self._calculate_minus_dm(
                    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/di.py`</sub>

***

## ICHIMOKU

Ichimoku Kinko Hyo (Ichimoku Cloud) indicator.

Components:

* Tenkan-sen (Conversion Line): (9-period high + 9-period low) / 2
* Kijun-sen (Base Line): (26-period high + 26-period low) / 2
* Senkou Span A (Leading Span A): (Tenkan-sen + Kijun-sen) / 2, plotted 26 periods ahead
* Senkou Span B (Leading Span B): (52-period high + 52-period low) / 2, plotted 26 periods ahead
* Chikou Span (Lagging Span): Close plotted 26 periods back

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

### Parameters

| Parameter         | Type      | Default    | Description                                                                 |
| ----------------- | --------- | ---------- | --------------------------------------------------------------------------- |
| `high`            | `'Input'` | Required   | High price input                                                            |
| `low`             | `'Input'` | Required   | Low price input                                                             |
| `close`           | `'Input'` | Required   | Close price input                                                           |
| `tenkan_period`   | `int`     | `9`        | Tenkan-sen period                                                           |
| `kijun_period`    | `int`     | `26`       | Kijun-sen period                                                            |
| `senkou_b_period` | `int`     | `52`       | Senkou Span B period                                                        |
| `output`          | `str`     | `'tenkan'` | Which line to output: "tenkan", "kijun", "senkou\_a", "senkou\_b", "chikou" |

### Usage

```python theme={null}
graph.add_node("ichimoku_tenkan", ICHIMOKU(
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=60),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=60),
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=60),
    output="tenkan",
))
```

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

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

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

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

    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)])
    close_values = np.array([close_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]:
            highs = high_values[:, i]
            lows = low_values[:, i]
            closes = close_values[:, i]

            valid_mask = ~(np.isnan(highs) | np.isnan(lows) | np.isnan(closes))
            if np.sum(valid_mask) >= self._senkou_b_period:
                result[i] = self._calculate_ichimoku(
                    highs[valid_mask],
                    lows[valid_mask],
                    closes[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/ichimoku.py`</sub>

***

## SAR

Parabolic SAR (Stop and Reverse) indicator.

SAR provides potential entry and exit points.
When price is above SAR, it's a bullish signal.
When price is below SAR, it's a bearish signal.

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

### Parameters

| Parameter      | Type      | Default  | Description                 |
| -------------- | --------- | -------- | --------------------------- |
| `high`         | `'Input'` | Required | High price input            |
| `low`          | `'Input'` | Required | Low price input             |
| `acceleration` | `float`   | `0.02`   | Initial acceleration factor |
| `maximum`      | `float`   | `0.2`    | Maximum acceleration factor |

### Usage

```python theme={null}
graph.add_node("sar", SAR(
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=30),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=30),
    acceleration=0.02,
    maximum=0.2,
))
```

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

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

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

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

    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]:
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= 5:
                result[i] = self._calculate_sar(
                    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/sar.py`</sub>

***

## SAREXT

Parabolic SAR - Extended indicator.

Extended version of SAR with more configurable parameters.

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

### Parameters

| Parameter           | Type      | Default  | Description            |
| ------------------- | --------- | -------- | ---------------------- |
| `high`              | `'Input'` | Required | High price input       |
| `low`               | `'Input'` | Required | Low price input        |
| `start_value`       | `float`   | `0.0`    | Initial SAR value      |
| `offset_on_reverse` | `float`   | `0.0`    | SAR offset on reversal |
| `accel_init_long`   | `float`   | `0.02`   | Initial AF for long    |
| `accel_long`        | `float`   | `0.02`   | AF increment for long  |
| `accel_max_long`    | `float`   | `0.2`    | Max AF for long        |
| `accel_init_short`  | `float`   | `0.02`   | Initial AF for short   |
| `accel_short`       | `float`   | `0.02`   | AF increment for short |
| `accel_max_short`   | `float`   | `0.2`    | Max AF for short       |

### Usage

```python theme={null}
op = SAREXT(
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
    accel_init_long=0.02,
    accel_max_long=0.2,
    accel_init_short=0.02,
    accel_max_short=0.2,
)
```

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

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

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

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

    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]:
            highs = high_values[:, i]
            lows = low_values[:, i]

            valid_mask = ~(np.isnan(highs) | np.isnan(lows))
            if np.sum(valid_mask) >= 5:
                result[i] = self._calculate_sarext(
                    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/sar.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>