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

# 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

```python theme={null}
# 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.

```python theme={null}
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),
    )
```

<sub>Source: `operator/interval_gate.py`</sub>

***

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

```python theme={null}
# 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.

```python theme={null}
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,
    )
```

<sub>Source: `operator/resample.py`</sub>

***

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

```python theme={null}
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.

```python theme={null}
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
```

<sub>Source: `operator/utility.py`</sub>

***

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

```python theme={null}
graph.add_node("threshold", Constant(
    Input("signal", timeframe="1m", lookback=1),
    value=0.5,
))
```

### 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:
    """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,
    )
```

<sub>Source: `operator/utility.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>