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

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

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

_{Source: apps/trading/operators/control/gate.py}

---

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

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.

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

_{Source: apps/trading/operators/control/gate.py}

---

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

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.

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

_{Source: apps/trading/operators/control/gate.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG