Volume Indicators

OBV, AD, ADOSC, MFI, VWAP, CCI

Overview

This page documents 5 operators (role: INDICATOR).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
AD	INDICATOR	—	No
ADOSC	INDICATOR	fast_period=3, slow_period=10	No
OBV	INDICATOR	—	No
MFI	INDICATOR	period=14	No
CCI	INDICATOR	period=20	No

---

AD

Chaikin A/D Line (Accumulation/Distribution) indicator.

AD = Previous AD + Current Money Flow Volume Money Flow Multiplier = [(Close - Low) - (High - Close)] / (High - Low) Money Flow Volume = Multiplier * Volume

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
volume	'Input'	Required	Volume input

Usage

graph.add_node("ad", AD(
    close=Input("FIELD:gateio:spot:close", timeframe="1h", lookback=1),
    high=Input("FIELD:gateio:spot:high", timeframe="1h", lookback=1),
    low=Input("FIELD:gateio:spot:low", timeframe="1h", lookback=1),
    volume=Input("FIELD:gateio:spot:volume", 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()

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

    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

    # Get all values
    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)])
    vol_values = np.array([volume_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]
            vols = vol_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows) | np.isnan(vols))
            if np.sum(valid_mask) > 0:
                result[i] = self._calculate_ad(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask],
                    vols[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/volume.py}

---

ADOSC

Chaikin A/D Oscillator indicator.

ADOSC = EMA(AD, fast) - EMA(AD, slow)

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
volume	'Input'	Required	Volume input
fast_period	int	3	Fast EMA period
slow_period	int	10	Slow EMA period

Usage

op = ADOSC(
    Input("FIELD:close", timeframe="1m", lookback=10),
    Input("FIELD:high", timeframe="1m", lookback=10),
    Input("FIELD:low", timeframe="1m", lookback=10),
    Input("FIELD:volume", timeframe="1m", lookback=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 not isinstance(data, list) or len(data) < 4:
        return self._empty_output()

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

    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)])
    vol_values = np.array([volume_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]
            vols = vol_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows) | np.isnan(vols))
            if np.sum(valid_mask) >= self._slow_period:
                result[i] = self._calculate_adosc(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask],
                    vols[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/volume.py}

---

OBV

On Balance Volume indicator.

OBV adds volume on up days and subtracts volume on down days. If Close > PrevClose: OBV = PrevOBV + Volume If Close < PrevClose: OBV = PrevOBV - Volume If Close = PrevClose: OBV = PrevOBV

Role: INDICATOR | Ephemeral: No

Parameters

Parameter	Type	Default	Description
close	'Input'	Required	Close price input
volume	'Input'	Required	Volume input

Usage

op = OBV(
    Input("FIELD:close", timeframe="1m", lookback=10),
    Input("FIELD:volume", timeframe="1m", lookback=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 not isinstance(data, list) or len(data) < 2:
        return self._empty_output()

    close_data, volume_data = data[0], data[1]

    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)])
    vol_values = np.array([volume_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]
            vols = vol_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(vols))
            if np.sum(valid_mask) >= 2:
                result[i] = self._calculate_obv(
                    closes[valid_mask],
                    vols[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/volume.py}

---

MFI

Money Flow Index indicator.

MFI is like RSI but incorporates volume. MFI = 100 - (100 / (1 + Money Flow Ratio))

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
volume	'Input'	Required	Volume input
period	int	14	Lookback period

Usage

op = MFI(
    Input("FIELD:close", timeframe="1m", lookback=14),
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    Input("FIELD:volume", 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) < 4:
        return self._empty_output()

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

    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)])
    vol_values = np.array([volume_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]
            vols = vol_values[:, i]

            valid_mask = ~(np.isnan(closes) | np.isnan(highs) | np.isnan(lows) | np.isnan(vols))
            if np.sum(valid_mask) >= self._period + 1:
                result[i] = self._calculate_mfi(
                    closes[valid_mask],
                    highs[valid_mask],
                    lows[valid_mask],
                    vols[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/volume.py}

---

CCI

Commodity Channel Index indicator.

CCI = (Typical Price - SMA(Typical Price)) / (0.015 * Mean Deviation)

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	20	Lookback period

Usage

op = CCI(
    Input("FIELD:close", timeframe="1m", lookback=20),
    Input("FIELD:high", timeframe="1m", lookback=20),
    Input("FIELD:low", timeframe="1m", lookback=20),
    period=20,
)

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:
                result[i] = self._calculate_cci(
                    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/volume.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG