Scalers

ZScore, Rank, MinMaxScale, L1Norm, L2Norm, Softmax, Clip, Winsorize

Overview

This page documents 11 operators (role: various).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
ZScore	UNKNOWN	min_std=1e-08	No
MinMaxScale	UNKNOWN	feature_range=...	No
Rank	UNKNOWN	—	No
Percentile	UNKNOWN	—	No
L1Norm	UNKNOWN	long_only=False	No
L2Norm	UNKNOWN	—	No
Softmax	UNKNOWN	temperature=1.0	No
Clip	UNKNOWN	lower=None, upper=None	No
Winsorize	UNKNOWN	std_mult=3.0	No
Demean	UNKNOWN	—	No
RollingStd	UNKNOWN	window=20	No

---

ZScore

Z-Score normalization across symbols.

Standardizes values by removing the cross-sectional mean and scaling by the cross-sectional standard deviation. Formula: z_i = (x_i - mean(x)) / max(std(x), min_std)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
min_std	float	1e-08	Minimum standard deviation to avoid division by zero

Usage

zscore = ZScore(Input("momentum", timeframe="1m", lookback=1))
graph.add_node("zscore", zscore)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        valid_values = values[compute_mask]
        mean = np.mean(valid_values)
        std = max(np.std(valid_values), self._min_std)
        result[compute_mask] = (values[compute_mask] - mean) / std

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

MinMaxScale

Min-Max scaling to a specified feature range.

Linearly rescales values so the minimum maps to the lower bound and the maximum maps to the upper bound of the target range. When all valid values are identical, returns the midpoint of the range. Formula: scaled_i = (x_i - min(x)) / (max(x) - min(x)) * (range_max - range_min) + range_min

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
feature_range	tuple	...	Target (min, max) range for scaled values

Usage

scaled = MinMaxScale(
    Input("momentum", timeframe="1m", lookback=1),
    feature_range=(0, 1),
)
graph.add_node("scaled", scaled)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        valid_values = values[compute_mask]
        v_min, v_max = np.min(valid_values), np.max(valid_values)

        if v_max - v_min < 1e-10:
            result[compute_mask] = (self._min + self._max) / 2
        else:
            scaled = (values[compute_mask] - v_min) / (v_max - v_min)
            result[compute_mask] = scaled * (self._max - self._min) + self._min

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Rank

Cross-sectional rank transform to [0, 1].

Assigns each symbol a rank based on its value relative to other symbols, then normalizes ranks to the [0, 1] interval. Uses ordinal ranking via double argsort. Formula: rank_i = argsort(argsort(x))[i] / (N - 1), where N is the number of valid symbols.

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.

Usage

ranked = Rank(Input("alpha", timeframe="1m", lookback=1))
graph.add_node("ranked", ranked)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    n_valid = np.sum(compute_mask)
    if n_valid > 0:
        valid_indices = np.where(compute_mask)[0]
        valid_values = values[compute_mask]

        order = np.argsort(np.argsort(valid_values))
        ranks = order / (n_valid - 1) if n_valid > 1 else np.zeros_like(order, dtype=float)

        result[valid_indices] = ranks

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Percentile

Cross-sectional percentile rank scaled to 0-100.

Computes the percentage of valid symbols whose value is less than or equal to each symbol’s value. Produces a score in the range [0, 100]. Formula: percentile_i = (count(x <= x_i) / N) * 100, where N is the number of valid symbols.

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.

Usage

pctl = Percentile(Input("alpha", timeframe="1m", lookback=1))
graph.add_node("percentile", pctl)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    n_valid = np.sum(compute_mask)
    if n_valid > 0:
        valid_indices = np.where(compute_mask)[0]
        valid_values = values[compute_mask]

        for i, idx in enumerate(valid_indices):
            val = valid_values[i]
            result[idx] = np.sum(valid_values <= val) / n_valid * 100

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

L1Norm

L1 normalization so absolute values sum to one.

Divides each value by the sum of absolute values across all valid symbols, producing weights that satisfy sum(|w_i|) = 1. Optionally clamps negative values to zero before normalizing (long-only mode). Formula: w_i = x_i / sum(|x|)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
long_only	bool	False	If True, clamp negative values to zero before normalizing

Usage

weights = L1Norm(
    Input("alpha", timeframe="1m", lookback=1),
    long_only=False,
)
graph.add_node("weights", weights)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        work_values = values.copy()
        if self._long_only:
            work_values = np.maximum(work_values, 0)

        total = np.sum(np.abs(work_values[compute_mask]))
        if total < 1e-10:
            result[compute_mask] = 0.0
        else:
            result[compute_mask] = work_values[compute_mask] / total

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

L2Norm

L2 normalization to unit Euclidean norm.

Divides each value by the L2 (Euclidean) norm of the valid values, producing a vector with unit length. Formula: w_i = x_i / sqrt(sum(x^2))

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.

Usage

normed = L2Norm(Input("alpha", timeframe="1m", lookback=1))
graph.add_node("normed", normed)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        norm = np.sqrt(np.sum(values[compute_mask] ** 2))
        if norm < 1e-10:
            result[compute_mask] = 0.0
        else:
            result[compute_mask] = values[compute_mask] / norm

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Softmax

Softmax normalization with configurable temperature.

Applies the softmax function to produce a probability distribution over symbols. A lower temperature sharpens the distribution (more weight on the largest values); a higher temperature flattens it. Uses the numerically stable form with max subtraction. Formula: w_i = exp((x_i - max(x)) / T) / sum(exp((x - max(x)) / T))

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
temperature	float	1.0	Temperature parameter controlling distribution sharpness

Usage

weights = Softmax(
    Input("alpha", timeframe="1m", lookback=1),
    temperature=0.5,
)
graph.add_node("softmax_weights", weights)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        valid_values = values[compute_mask] / self._temperature
        exp_values = np.exp(valid_values - np.max(valid_values))
        softmax_values = exp_values / np.sum(exp_values)
        result[compute_mask] = softmax_values

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Clip

Clip values to a specified lower and upper bound.

Constrains each value to lie within [lower, upper]. Values below the lower bound are set to the lower bound; values above the upper bound are set to the upper bound. Either bound may be None to leave that side unconstrained. Formula: result_i = max(lower, min(x_i, upper))

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
lower	float	None	Lower clipping bound, or None for no lower bound
upper	float	None	Upper clipping bound, or None for no upper bound

Usage

clipped = Clip(
    Input("alpha", timeframe="1m", lookback=1),
    lower=-0.05,
    upper=0.05,
)
graph.add_node("clipped", clipped)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = values.copy()

    if self._lower is not None:
        result = np.maximum(result, self._lower)
    if self._upper is not None:
        result = np.minimum(result, self._upper)

    result[~compute_mask] = np.nan
    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Winsorize

Winsorize outliers by clamping to standard-deviation bounds.

Computes the cross-sectional mean and standard deviation, then clips values to [mean - kstd, mean + kstd]. This limits the influence of extreme outliers while preserving the relative ordering of non-outlier values. Formula: result_i = clip(x_i, mean(x) - kstd(x), mean(x) + kstd(x))

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
std_mult	float	3.0	Number of standard deviations for the clipping bounds

Usage

cleaned = Winsorize(
    Input("alpha", timeframe="1m", lookback=1),
    std_mult=3.0,
)
graph.add_node("winsorized", cleaned)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = values.copy()

    if np.any(compute_mask):
        valid_values = values[compute_mask]
        mean = np.mean(valid_values)
        std = np.std(valid_values)

        lower = mean - self._std_mult * std
        upper = mean + self._std_mult * std

        result[compute_mask] = np.clip(values[compute_mask], lower, upper)

    result[~compute_mask] = np.nan
    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

Demean

Cross-sectional demeaning by subtracting the mean across symbols.

Centers the signal around zero by subtracting the cross-sectional mean of all valid values. This removes the common level so that only relative differences between symbols remain. Formula: result_i = x_i - mean(x)

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.

Usage

demean = Demean(Input("alpha", timeframe="1m", lookback=1))
graph.add_node("demeaned", demean)

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
    values = last.value
    exists = last.exists
    valid = last.valid

    compute_mask = exists & valid
    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    if np.any(compute_mask):
        valid_values = values[compute_mask]
        mean = np.nanmean(valid_values)
        result[compute_mask] = values[compute_mask] - mean

    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

---

RollingStd

Rolling standard deviation per symbol over a lookback window.

Computes the sample standard deviation (ddof=1) of the last window observations for each symbol along the time axis. Requires at least 2 data points in the time dimension. Formula: std_i = sqrt(sum((x_t,i - mean_i)^2) / (T - 1)) for t in [now - window, now]

Role: UNKNOWN | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input signal. timeframe specifies data frequency.
window	int	20	Number of trailing observations to include

Usage

volatility = RollingStd(
    Input("returns", timeframe="1h", lookback=20),
    window=20,
)
graph.add_node("volatility", volatility)

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

    n_symbols = len(exists)
    result = np.full(n_symbols, np.nan)

    # Need at least 2 data points for std
    if len(values.shape) > 1 and len(values) >= 2:
        window = min(self._window, len(values))
        window_data = values[-window:]

        with np.errstate(invalid='ignore'):
            result = np.nanstd(window_data, axis=0, ddof=1)

    compute_mask = exists & valid
    result_valid = compute_mask & ~np.isnan(result)

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

_{Source: apps/trading/operators/transform/scalers.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG