Momentum Indicators

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

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.

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

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

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

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.

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

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

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

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

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

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.

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

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

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

graph.add_node("mom", MOM(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=15),
    period=10,
))

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

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

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

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.

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

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

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

graph.add_node("roc", ROC(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=15),
    period=10,
))

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

op = ROCP(Input("FIELD:close", timeframe="1m", lookback=10), period=10)

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

op = ROCR(Input("FIELD:close", timeframe="1m", lookback=10), period=10)

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

op = ROCR100(Input("FIELD:close", timeframe="1m", lookback=10), period=10)

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

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

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

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

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

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

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.

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

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

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

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.

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

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

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

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.

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

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

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

graph.add_node("trix", TRIX(
    Input("FIELD:gateio:spot:close", timeframe="1h", lookback=50),
    period=30,
))

Source Code

Full compute() implementation — no hidden logic.

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

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

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

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

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= self._period * 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),
    )

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

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

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.

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

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

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

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.

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

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

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG

​Overview

​Quick Reference

​APO

​Parameters

​Usage

​Source Code

​BOP

​Parameters

​Usage

​Source Code

​CMO

​Parameters

​Usage

​Source Code

​MACD

​Parameters

​Usage

​Source Code

​MOM

​Parameters

​Usage

​Source Code

​PPO

​Parameters

​Usage

​Source Code

​ROC

​Parameters

​Usage

​Source Code

​ROCP

​Parameters

​Usage

​Source Code

​ROCR

​Parameters

​Usage

​Source Code

​ROCR100

​Parameters

​Usage

​Source Code

​RSI

​Parameters

​Usage

​Source Code

​STOCH

​Parameters

​Usage

​Source Code

​STOCHF

​Parameters

​Usage

​Source Code

​STOCHRSI

​Parameters

​Usage

​Source Code

​TRIX

​Parameters

​Usage

​Source Code

​ULTOSC

​Parameters

​Usage

​Source Code

​WILLR

​Parameters

​Usage

​Source Code

​Related Pages

Operator Protocol

StatefulGraph

Overview

Quick Reference

APO

Parameters

Usage

Source Code

BOP

Parameters

Usage

Source Code

CMO

Parameters

Usage

Source Code

MACD

Parameters

Usage

Source Code

MOM

Parameters

Usage

Source Code

PPO

Parameters

Usage

Source Code

ROC

Parameters

Usage

Source Code

ROCP

Parameters

Usage

Source Code

ROCR

Parameters

Usage

Source Code

ROCR100

Parameters

Usage

Source Code

RSI

Parameters

Usage

Source Code

STOCH

Parameters

Usage

Source Code

STOCHF

Parameters

Usage

Source Code

STOCHRSI

Parameters

Usage

Source Code

TRIX

Parameters

Usage

Source Code

ULTOSC

Parameters

Usage

Source Code

WILLR

Parameters

Usage

Source Code

Related Pages