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

# Control Operators

> Conditional gates for controlling execution flow

## Overview

This page documents **3 operators** (role: `CONTROL`).

***

## ConditionalGate

Gate operator that outputs 1 or 0 based on condition.

Used to conditionally enable/disable downstream operators
(especially semantic operators that make API calls).

Output:

* 1.0: Condition is met (gate open)
* 0.0: Condition is not met (gate closed)

Downstream operators can use this as gate\_input to skip expensive
operations when the gate is closed.

Example:

# Open gate when ATR above 2%

graph.add\_node("high\_vol\_gate", ConditionalGate(
condition\_input=Input("atr", timeframe="1h", lookback=1),
threshold=0.02,
comparison="gt",
))

# Open gate when RSI below 30 OR RSI above 70 (using two gates)

graph.add\_node("oversold\_gate", ConditionalGate(
condition\_input=Input("rsi", timeframe="1h", lookback=1),
threshold=30,
comparison="lt",
))

graph.add\_node("overbought\_gate", ConditionalGate(
condition\_input=Input("rsi", timeframe="1h", lookback=1),
threshold=70,
comparison="gt",
))

# Combine gates with OR logic (max of two gate values)

graph.add\_node("extreme\_rsi\_gate", GateOr(\[
Input("oversold\_gate", timeframe="1h", lookback=1),
Input("overbought\_gate", timeframe="1h", lookback=1),
]))

**Role**: `CONTROL` | **Ephemeral**: No

### Parameters

| Parameter          | Type             | Default  | Description |
| ------------------ | ---------------- | -------- | ----------- |
| `condition_input`  | `Input`          | Required |             |
| `threshold`        | `float`          | `0.5`    |             |
| `comparison`       | `ComparisonType` | `'gt'`   |             |
| `per_symbol`       | `bool`           | `True`   |             |
| `any_symbol`       | `bool`           | `False`  |             |
| `output_timeframe` | `Optional[str]`  | `None`   |             |

### Usage

```python theme={null}
# Open gate when ATR > 2%
graph.add_node("high_vol_gate", ConditionalGate(
    condition_input=Input("atr", timeframe="1h", lookback=1),
    threshold=0.02,
    comparison="gt",
))

# Open gate when RSI < 30 OR RSI > 70 (using two gates)
graph.add_node("oversold_gate", ConditionalGate(
    condition_input=Input("rsi", timeframe="1h", lookback=1),
    threshold=30,
    comparison="lt",
))

graph.add_node("overbought_gate", ConditionalGate(
    condition_input=Input("rsi", timeframe="1h", lookback=1),
    threshold=70,
    comparison="gt",
))

# Combine gates with OR logic (max of two gate values)
graph.add_node("extreme_rsi_gate", GateOr([
    Input("oversold_gate", timeframe="1h", lookback=1),
    Input("overbought_gate", timeframe="1h", 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:
    """Evaluate gate condition."""
    # Extract values
    if isinstance(data, list):
        input_data = data[0]
    else:
        input_data = data

    if len(input_data) > 0:
        last_tick = input_data[-1]
        values = last_tick.value
        exists = last_tick.exists
        valid = last_tick.valid
    else:
        # No data - return closed gate
        n = 1
        return TaggedArray(
            value=[0.0],
            exists=[True],
            valid=[False],
            updated=np.array([True]),
        )

    n = len(values)

    # Evaluate condition
    condition_met = self._evaluate_condition(values)

    if self._per_symbol:
        # Per-symbol gate values
        gate_values = np.where(condition_met & valid, 1.0, 0.0)
    else:
        # Aggregate across symbols
        if self._any_symbol:
            # Gate opens if ANY symbol meets condition
            any_met = np.any(condition_met & valid)
            gate_values = np.full(n, 1.0 if any_met else 0.0)
        else:
            # Gate opens only if ALL valid symbols meet condition
            valid_symbols = valid.sum()
            if valid_symbols > 0:
                all_met = np.all(condition_met[valid])
                gate_values = np.full(n, 1.0 if all_met else 0.0)
            else:
                gate_values = np.zeros(n)

    # Log gate status per symbol
    symbols = self._get_symbols(context)
    open_symbols = [symbols[i] for i in range(n) if gate_values[i] > 0.5]
    closed_symbols = [symbols[i] for i in range(n) if gate_values[i] <= 0.5]
    comp_str = f"{self._comparison} {self._threshold}"
    logger.debug(
        f"ConditionalGate({comp_str}): "
        f"🔓 {open_symbols if open_symbols else 'none'} | "
        f"🔒 {closed_symbols if closed_symbols else 'none'}"
    )

    return TaggedArray(
        value=gate_values,
        exists=exists,
        valid=valid,
        updated=np.ones(n, dtype=bool),
    )
```

<sub>Source: `apps/trading/operators/control/gate.py`</sub>

***

## GateOr

Combine multiple gates with OR logic (max of gate values).

**Role**: `CONTROL` | **Ephemeral**: No

### Parameters

| Parameter          | Type            | Default  | Description |
| ------------------ | --------------- | -------- | ----------- |
| `gate_inputs`      | `List[Input]`   | Required |             |
| `output_timeframe` | `Optional[str]` | `None`   |             |

### Usage

```python theme={null}
gate = GateOr(
    gate_inputs=[
        Input("oversold_gate", timeframe="1h", lookback=1),
        Input("overbought_gate", timeframe="1h", 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:
    """Combine gates with OR (max)."""
    if not isinstance(data, list):
        data = [data]

    combined = None
    exists = None
    valid = None

    for gate_data in data:
        if len(gate_data) > 0:
            last_tick = gate_data[-1]
            values = last_tick.value

            if combined is None:
                combined = values.copy()
                exists = last_tick.exists
                valid = last_tick.valid
            else:
                combined = np.maximum(combined, values)
                valid = valid | last_tick.valid

    if combined is None:
        logger.debug("GateOr: No data, returning closed")
        return TaggedArray(
            value=[0.0],
            exists=[True],
            valid=[False],
            updated=np.array([True]),
        )

    # Log combined gate status
    n_open = np.sum(combined > 0.5)
    n_total = len(combined)
    logger.debug(f"GateOr: {n_open}/{n_total} symbols open")

    return TaggedArray(
        value=combined,
        exists=exists,
        valid=valid,
        updated=np.ones(len(combined), dtype=bool),
    )
```

<sub>Source: `apps/trading/operators/control/gate.py`</sub>

***

## GateAnd

Combine multiple gates with AND logic (min of gate values).

**Role**: `CONTROL` | **Ephemeral**: No

### Parameters

| Parameter          | Type            | Default  | Description |
| ------------------ | --------------- | -------- | ----------- |
| `gate_inputs`      | `List[Input]`   | Required |             |
| `output_timeframe` | `Optional[str]` | `None`   |             |

### Usage

```python theme={null}
gate = GateAnd(
    gate_inputs=[
        Input("high_vol_gate", timeframe="1h", lookback=1),
        Input("trend_gate", timeframe="1h", 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:
    """Combine gates with AND (min)."""
    if not isinstance(data, list):
        data = [data]

    combined = None
    exists = None
    valid = None

    for gate_data in data:
        if len(gate_data) > 0:
            last_tick = gate_data[-1]
            values = last_tick.value

            if combined is None:
                combined = values.copy()
                exists = last_tick.exists
                valid = last_tick.valid
            else:
                combined = np.minimum(combined, values)
                valid = valid & last_tick.valid

    if combined is None:
        logger.debug("GateAnd: No data, returning closed")
        return TaggedArray(
            value=[0.0],
            exists=[True],
            valid=[False],
            updated=np.array([True]),
        )

    # Log combined gate status
    n_open = np.sum(combined > 0.5)
    n_total = len(combined)
    logger.debug(f"GateAnd: {n_open}/{n_total} symbols open")

    return TaggedArray(
        value=combined,
        exists=exists,
        valid=valid,
        updated=np.ones(len(combined), dtype=bool),
    )
```

<sub>Source: `apps/trading/operators/control/gate.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>