Universe Filters

VolumeFilter, VolatilityFilter, PriceFilter, LiquidityFilter, DataAvailabilityFilter

Overview

This page documents 7 operators (role: FILTER).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
CompositeFilter	FILTER	filters, logic='and'	No
DataAvailabilityFilter	FILTER	min_bars=50	No
LiquidityFilter	FILTER	min_dollar_volume=1000000	No
PriceFilter	FILTER	min_price=1.0, max_price=None	No
StrataFilter	FILTER	strata, symbol_order, include=None	No
VolatilityFilter	FILTER	min_vol=0.001, max_vol=1.0, annualize=True	No
VolumeFilter	FILTER	min_volume=100000	No

---

CompositeFilter

Combine multiple filters with AND/OR logic.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
filters	list	Required
logic	str	'and'

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:
    """Apply all filters and combine results."""
    if not self._filters:
        if isinstance(data, list):
            data = data[0]
        last = data[-1] if len(data) > 0 else data
        exists = last.exists
        valid = last.valid
        result = np.where(exists & valid, 1.0, 0.0)
        return TaggedArray(
            value=result,
            exists=exists,
            valid=exists & valid,
            updated=np.ones(len(result), dtype=bool),
        )

    results = [f.compute(data, timestamp, context) for f in self._filters]

    first_result = results[0]
    exists = first_result.exists
    valid = first_result.valid

    combined_values = first_result.value.copy()
    for r in results[1:]:
        r_values = r.value
        if self._logic == "and":
            combined_values = np.minimum(combined_values, r_values)
        else:
            combined_values = np.maximum(combined_values, r_values)

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

_{Source: apps/trading/operators/base.py}

---

DataAvailabilityFilter

Filter by minimum data availability (non-NaN bars).

Counts the number of non-NaN values within a trailing window for each symbol and retains only those with at least min_bars valid data points. For single-bar (1-D) data, a symbol passes if its value is not NaN. For multi-bar (2-D) data, the window size is min(min_bars, available bars) and the count of non-NaN entries must meet the threshold. Passing symbols are assigned 1.0, others 0.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal providing the data to check for availability.
min_bars	int	50	Minimum number of non-NaN bars required to pass the

Usage

filter = DataAvailabilityFilter(
    Input("FIELD:close", timeframe="1m", lookback=50),
    min_bars=50,
)

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 isinstance(data, list):
        data = data[0]

    values = data.value

    last = data[-1] if len(data) > 0 else data
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    filter_result = np.zeros(n_symbols, dtype=bool)

    if len(values.shape) == 1:
        filter_result[compute_mask] = ~np.isnan(values[compute_mask])
    else:
        window = min(self._min_bars, len(values))
        window_data = values[-window:]
        valid_count = np.sum(~np.isnan(window_data), axis=0)
        filter_result[compute_mask] = valid_count[compute_mask] >= self._min_bars

    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/data_availability.py}

---

LiquidityFilter

Filter by average dollar volume (price * volume).

Computes dollar volume as close price multiplied by volume for each bar, then averages over the lookback window. Retains only symbols whose average dollar volume meets or exceeds the specified minimum threshold. Requires two inputs (close price and volume). For single-bar (1-D) data the raw dollar volume is compared directly; for multi-bar (2-D) data the trailing window average is used. Passing symbols are assigned 1.0, others 0.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of two Input signals: the first provides close price
min_dollar_volume	float	1000000	Minimum average dollar volume required to pass

Usage

filter = LiquidityFilter(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=20),
        Input("FIELD:volume", timeframe="1m", lookback=20),
    ],
    min_dollar_volume=1_000_000,
)

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 isinstance(data, list) and len(data) >= 2:
        close_data = data[0]
        volume_data = data[1]
        close_values = close_data.value
        volume_values = volume_data.value

        last = close_data[-1] if len(close_data) > 0 else close_data
        exists = last.exists
        valid = last.valid
    else:
        if isinstance(data, list):
            data = data[0]
        close_values = data.value
        volume_values = np.ones_like(close_values)

        last = data[-1] if len(data) > 0 else data
        exists = last.exists
        valid = last.valid

    dollar_volume = close_values * volume_values
    compute_mask = exists & valid
    n_symbols = len(exists)
    filter_result = np.zeros(n_symbols, dtype=bool)

    if len(dollar_volume.shape) == 1:
        filter_result[compute_mask] = dollar_volume[compute_mask] >= self._min_dollar_volume
    else:
        window = min(self._lookback, len(dollar_volume))
        avg_dv = np.mean(dollar_volume[-window:], axis=0)
        filter_result[compute_mask] = avg_dv[compute_mask] >= self._min_dollar_volume

    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/liquidity.py}

---

PriceFilter

Filter by price range.

Checks the most recent close price for each symbol and retains only those whose price falls within the [min_price, max_price] range. When max_price is None, no upper bound is applied. For multi-bar (2-D) data the last bar is used; for single-bar (1-D) data the value is used directly. Passing symbols are assigned 1.0, others 0.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal providing close price data. timeframe specifies
min_price	float	1.0	Minimum price to pass the filter
max_price	Optional[float]	None	Maximum price to pass the filter, or None for no upper

Usage

filter = PriceFilter(
    Input("FIELD:close", timeframe="1m", lookback=1),
    min_price=1.0,
    max_price=10000.0,
)

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 isinstance(data, list):
        data = data[0]

    values = data.value

    last = data[-1] if len(data) > 0 else data
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)

    if len(values.shape) > 1:
        prices = values[-1]
    else:
        prices = values

    passes = prices >= self._min_price
    if self._max_price is not None:
        passes = passes & (prices <= self._max_price)

    filter_result = np.zeros(n_symbols, dtype=bool)
    filter_result[compute_mask] = passes[compute_mask]
    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/price.py}

---

StrataFilter

Filter symbols by strata (group/category) membership.

Assigns each symbol to a category via the strata mapping, then retains only symbols whose category is in the include list (if provided) and not in the exclude list. The set of passing symbols is precomputed at construction time based on the include/exclude rules. At compute time, each symbol in symbol_order is checked against the precomputed set. Passing symbols are assigned 1.0, others 0.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency, lookback
strata	Dict[str, str]	Required	Mapping from symbol name to category string.
symbol_order	List[str]	Required	Ordered list of symbol names matching the data array
include	Optional[List[str]]	None	None).
exclude	Optional[List[str]]	None	List of categories to exclude

Usage

filter = StrataFilter(
    Input("FIELD:close", timeframe="1m", lookback=1),
    strata={"AAPL": "tech", "GOOG": "tech", "JPM": "finance"},
    symbol_order=["AAPL", "GOOG", "JPM"],
    include=["tech"],
)

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 isinstance(data, list):
        data = data[0]

    last = data[-1] if len(data) > 0 else data
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    filter_result = np.zeros(n_symbols, dtype=bool)

    for i, symbol in enumerate(self._symbol_order):
        if compute_mask[i] and symbol in self._passing_symbols:
            filter_result[i] = True

    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/strata.py}

---

VolatilityFilter

Filter by volatility range.

Computes the standard deviation of simple returns over the lookback window and retains only symbols whose volatility falls within the specified [min_vol, max_vol] range. Returns can optionally be annualized by multiplying by sqrt(periods_per_year). Requires at least two bars of data to calculate returns. Symbols outside the range are assigned 0.0, those inside are assigned 1.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal providing close price data. timeframe specifies
min_vol	float	0.001	Minimum volatility to pass the filter
max_vol	float	1.0	Maximum volatility to pass the filter
annualize	bool	True	Whether to annualize volatility
periods_per_year	int	252	Number of periods per year used for annualization

Usage

filter = VolatilityFilter(
    Input("FIELD:close", timeframe="1m", lookback=21),
    min_vol=0.001,
    max_vol=1.0,
    annualize=True,
    periods_per_year=252,
)

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 isinstance(data, list):
        data = data[0]

    values = data.value

    last = data[-1] if len(data) > 0 else data
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    filter_result = np.zeros(n_symbols, dtype=bool)

    if len(values.shape) > 1 and len(values) >= 2:
        with np.errstate(divide='ignore', invalid='ignore'):
            returns = np.diff(values, axis=0) / values[:-1]

        window = min(self._lookback - 1, len(returns))
        vol = np.nanstd(returns[-window:], axis=0, ddof=1)

        if self._annualize:
            vol = vol * np.sqrt(self._periods_per_year)

        in_range = (vol >= self._min_vol) & (vol <= self._max_vol)
        filter_result[compute_mask] = in_range[compute_mask]

    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/volatility.py}

---

VolumeFilter

Filter by minimum average trading volume.

Computes the mean trading volume over a lookback window and retains only symbols whose average volume meets or exceeds the specified minimum threshold. For single-bar (1-D) data the raw value is compared directly; for multi-bar (2-D) data the trailing window average is used. Symbols that do not pass are assigned 0.0, passing symbols are assigned 1.0.

Role: FILTER | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal providing volume data. timeframe specifies data
min_volume	float	100000	Minimum average volume required to pass the filter

Usage

filter = VolumeFilter(
    Input("FIELD:volume", timeframe="1m", lookback=20),
    min_volume=100_000,
)

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 isinstance(data, list):
        data = data[0]

    values = data.value

    last = data[-1] if len(data) > 0 else data
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    filter_result = np.zeros(n_symbols, dtype=bool)

    if len(values.shape) == 1:
        filter_result[compute_mask] = values[compute_mask] >= self._min_volume
    else:
        window = min(self._lookback, len(values))
        avg_volume = np.mean(values[-window:], axis=0)
        filter_result[compute_mask] = avg_volume[compute_mask] >= self._min_volume

    result = np.where(filter_result, 1.0, 0.0)

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

_{Source: apps/trading/operators/universe/filter/volume.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG