Trend Indicators

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

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

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.

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),
    )

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

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG

​Overview

​Quick Reference

​ADX

​Parameters

​Usage

​Source Code

​ADXR

​Parameters

​Usage

​Source Code

​DX

​Parameters

​Usage

​Source Code

​AROON

​Parameters

​Usage

​Source Code

​AROONOSC

​Parameters

​Usage

​Source Code

​PLUS_DI

​Parameters

​Usage

​Source Code

​MINUS_DI

​Parameters

​Usage

​Source Code

​PLUS_DM

​Parameters

​Usage

​Source Code

​MINUS_DM

​Parameters

​Usage

​Source Code

​ICHIMOKU

​Parameters

​Usage

​Source Code

​SAR

​Parameters

​Usage

​Source Code

​SAREXT

​Parameters

​Usage

​Source Code

​Related Pages

Operator Protocol

StatefulGraph

Overview

Quick Reference

ADX

Parameters

Usage

Source Code

ADXR

Parameters

Usage

Source Code

DX

Parameters

Usage

Source Code

AROON

Parameters

Usage

Source Code

AROONOSC

Parameters

Usage

Source Code

PLUS_DI

Parameters

Usage

Source Code

MINUS_DI

Parameters

Usage

Source Code

PLUS_DM

Parameters

Usage

Source Code

MINUS_DM

Parameters

Usage

Source Code

ICHIMOKU

Parameters

Usage

Source Code

SAR

Parameters

Usage

Source Code

SAREXT

Parameters

Usage

Source Code

Related Pages