Utility Operators

Identity, Resample, FieldMerge, SymbolSelect, SymbolDrop, Constant, IntervalGate

Overview

This page documents 4 operators (role: various).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
IntervalGate	UNKNOWN	trigger_input, interval='5m', output_timeframe=None	No
Resample	UNKNOWN	rule='1h', method='last', min_valid_ratio=0.5	No
Identity	UNKNOWN	input_spec=None, window=1, axes=None	No
Constant	UNKNOWN	value, axes=None	No

---

IntervalGate

Time-based gate that opens on timeframe boundaries.

Outputs 1.0 when the current timestamp falls on a specified interval boundary, and 0.0 otherwise. Used to control execution frequency of operators that lack internal data aggregation (e.g., WebSearch, LLM). The gate checks whether the timestamp is aligned to the interval grid. Formula: output = 1.0 if timestamp % interval == 0, else 0.0

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
trigger_input	Input	Required	Any input to trigger computation and infer n_symbols.
interval	str	'5m'	Target interval string (e.g., “5m”, “1h”, “1d”)
output_timeframe	Optional[str]	None	Output timeframe. Defaults to the trigger input’s

Usage

# Gate that opens every 5 minutes
graph.add_node("every_5m", IntervalGate(
    trigger_input=Input("rsi", timeframe="1m", lookback=1),
    interval="5m",
))

# Use with GateAnd to combine with conditions
graph.add_node("search_gate", GateAnd([
    Input("every_5m", timeframe="1m", lookback=1),
    Input("rsi_extreme_gate", timeframe="1m", lookback=1),
]))

# Then use as gate for WebSearch
graph.add_node("news", WebSearchOperator(
    query_template="BTC news",
    gate_input=Input("search_gate", timeframe="1m", 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:
    """Compute gate value based on timestamp boundary.

    Returns 1.0 if timestamp is on interval boundary, 0.0 otherwise.
    """
    # Extract input to get n_symbols
    if isinstance(data, list):
        input_data = data[0]
    else:
        input_data = data

    if len(input_data) > 0:
        last_tick = input_data[-1]
        n_symbols = len(last_tick.value)
        exists = last_tick.exists
    else:
        n_symbols = 1
        exists = np.array([True])

    # Check if on boundary
    is_boundary = False

    if timestamp is not None:
        # Method 1: Use context boundary info (from graph)
        if context and "is_boundary" in context:
            is_boundary = context["is_boundary"].get(self.interval, False)
        else:
            # Method 2: Check timestamp directly
            is_boundary = self._check_boundary(timestamp)

    # Create output
    gate_value = 1.0 if is_boundary else 0.0

    # Log gate status
    ts_str = timestamp.strftime('%H:%M:%S') if timestamp else 'N/A'
    gate_status = "🔓 OPEN" if is_boundary else "🔒 CLOSED"
    logger.info(f"IntervalGate[{self.interval}] @ {ts_str}: {gate_status}")

    return TaggedArray(
        value=np.full(n_symbols, gate_value),
        exists=exists,
        valid=np.ones(n_symbols, dtype=bool),
        updated=np.ones(n_symbols, dtype=bool),
    )

_{Source: operator/interval_gate.py}

---

Resample

Time-axis compression (1m -> 1h, 1d, etc.).

Converts higher-frequency data to lower-frequency by aggregating ticks at time boundaries. The Graph detects boundaries using timestamp.floor(rule) and passes is_boundary via context.

Why timeframe in Input matters: Graph uses Input.timeframe to:

1. Validate timeframe compatibility between connected nodes
2. Compute required buffer sizes (lookback * timeframe_seconds)
3. Register resample rules for boundary detection

Even if your data source is 1m, you can build indicators on any higher timeframe by chaining Resample.

Multi-timeframe pattern:

Raw 1m data -> 1h candles -> indicators on 1h

graph.add_node(“close_1h”, Resample( input=Input(“FIELD:close”, timeframe=“1m”), rule=“1h”, method=“last”, )) graph.add_node(“sma_1h”, SMA( Input(“close_1h”, timeframe=“1h”, lookback=20), period=20, ))

4h from the same 1m source

graph.add_node(“close_4h”, Resample( input=Input(“FIELD:close”, timeframe=“1m”), rule=“4h”, ))

Volume bars (sum aggregation)

graph.add_node(“vol_1h”, Resample( input=Input(“FIELD:volume”, timeframe=“1m”), rule=“1h”, method=“sum”, ))

Available aggregation methods:

• “last”: Last value in window (default, for close prices)
• “first”: First value (for open prices)
• “mean”: Average (for indicators)
• “sum”: Sum (for volume)
• “min”: Minimum (for low prices)
• “max”: Maximum (for high prices)
• “ohlc”: Full OHLC (returns close as main value)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Source Input. timeframe must match the raw data frequency.
rule	str	'1h'	Target timeframe string (e.g., “1h”, “5m”, “1d”, “4h”)
method	Literal['last', 'first', 'mean', 'sum', 'min', 'max', 'ohlc']	'last'	Aggregation method
min_valid_ratio	float	0.5	Minimum ratio of valid ticks in window

Usage

# 1m close -> 1h close
resample = Resample(
    input=Input("FIELD:close", timeframe="1m"),
    rule="1h",
)
graph.add_node("close_1h", resample)

# Then use the resampled output as input to other operators
momentum = MomentumAlpha(Input("close_1h", timeframe="1h", lookback=20))
graph.add_node("momentum_1h", momentum)

Source Code

Full compute() implementation — no hidden logic.

def compute(
    self,
    data: TaggedArray,
    timestamp: Optional[pd.Timestamp] = None,
    context: Optional[Dict[str, Any]] = None,
) -> TaggedArray:
    """Resample data at time boundaries."""
    if len(data) == 0:
        return TaggedArray(
            value=np.array([]),
            exists=np.array([], dtype=bool),
            valid=np.array([], dtype=bool),
            updated=np.array([], dtype=bool),
        )

    is_boundary = False
    if context and "is_boundary" in context:
        is_boundary = context["is_boundary"].get(self.rule, False)

    if not is_boundary:
        current = data[-1] if len(data) > 0 else data

        if self._last_value is not None:
            result_value = self._last_value
            result_exists = self._last_exists
            result_valid = self._last_valid
        else:
            result_value = current.value
            result_exists = current.exists
            result_valid = current.valid

        return TaggedArray(
            value=result_value,
            exists=result_exists,
            valid=result_valid,
            updated=np.zeros_like(result_exists, dtype=bool),
        )

    # Check which ticks have updates (any symbol updated in that tick)
    tick_has_update = np.any(data.updated, axis=-1) if data.updated.ndim > 1 else data.updated

    if not np.any(tick_has_update):
        current = data[-1]
        if self._last_value is not None:
            result_value = self._last_value
            result_exists = self._last_exists
            result_valid = self._last_valid
        else:
            result_value = current.value
            result_exists = current.exists
            result_valid = current.valid

        return TaggedArray(
            value=result_value,
            exists=result_exists,
            valid=result_valid,
            updated=np.zeros_like(result_exists, dtype=bool),
        )

    # Filter to updated ticks only
    updated_values = data.value[tick_has_update]
    updated_exists = data.exists[tick_has_update]
    updated_valid = data.valid[tick_has_update]

    num_ticks = len(updated_values)
    valid_count = np.sum(np.any(updated_valid, axis=-1) if updated_valid.ndim > 1 else updated_valid)
    valid_ratio = valid_count / num_ticks if num_ticks > 0 else 0.0

    if valid_ratio < self.min_valid_ratio:
        current = data[-1]
        if self._last_value is not None:
            result_value = self._last_value
            result_exists = self._last_exists
            result_valid = self._last_valid
        else:
            result_value = current.value
            result_exists = current.exists
            result_valid = current.valid

        return TaggedArray(
            value=result_value,
            exists=result_exists,
            valid=result_valid,
            updated=np.zeros_like(result_exists, dtype=bool),
        )

    result_values, result_exists, result_valid = self._aggregate(
        updated_values, updated_exists, updated_valid
    )

    self._last_value = result_values
    self._last_exists = result_exists
    self._last_valid = result_valid

    result_updated = np.ones_like(result_exists, dtype=bool)

    return TaggedArray(
        value=result_values,
        exists=result_exists,
        valid=result_valid,
        updated=result_updated,
    )

_{Source: operator/resample.py}

---

Identity

Pass through the most recent tick unchanged.

Returns the latest data slice without any transformation. Commonly used to alias FIELD inputs under shorter node names for cleaner graph definitions, or to adapt input specifications. Formula: result = x[t] (identity function, no transformation)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input_spec		None	Optional Input specification for FIELD aliasing
window	int	1	Lookback window size
axes	Optional[List[str]]	None	Output axes override

Usage

graph.add_node("close", Identity(
    Input("FIELD:binance:futures:close", timeframe="1d"),
))

# Then use in other operators
graph.add_node("mom", MomentumAlpha(
    Input("close", timeframe="1d", 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:
    """Return most recent tick."""
    if len(data) > 0:
        return data[-1]
    return data

_{Source: operator/utility.py}

---

Constant

Emit a fixed constant value at every time step.

Produces a TaggedArray filled with the specified constant value, shaped to match the input dimensions. Useful for thresholds, scaling factors, or any fixed parameter in the computation graph. Formula: result = c (constant for all elements)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
value	Union[float, int, list, np.ndarray]	Required	The constant value to emit. Can be a scalar, list, or
axes	Optional[List[str]]	None	Output axes override

Usage

graph.add_node("threshold", Constant(
    Input("signal", timeframe="1m", lookback=1),
    value=0.5,
))

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:
    """Return constant value."""
    current = data[-1] if len(data) > 0 else data

    if isinstance(self.value, (int, float)):
        if self._axes:
            const_values = np.full_like(current.value, self.value)
        else:
            const_values = np.array([self.value])
    else:
        const_values = np.asarray(self.value)

    if self._axes:
        result_exists = np.ones_like(current.exists, dtype=bool)
        result_valid = np.ones_like(current.valid, dtype=bool)
        result_updated = current.updated
    else:
        result_exists = np.array([True])
        result_valid = np.array([True])
        result_updated = np.array([np.any(current.updated)])

    return TaggedArray(
        value=const_values,
        exists=result_exists,
        valid=result_valid,
        updated=result_updated,
    )

_{Source: operator/utility.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG