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

# Neutralizers

> Demean, Neutralize, GroupNeutralize, BetaNeutralize, FactorNeutralize, BarraNeutralizer

## Overview

This page documents **7 operators** (role: various).

## Quick Reference

| Operator             | Role      | Key Parameters                                  | Ephemeral |
| -------------------- | --------- | ----------------------------------------------- | --------- |
| **Demean**           | `UNKNOWN` | —                                               | No        |
| **Neutralize**       | `UNKNOWN` | `scale=True`                                    | No        |
| **GroupNeutralize**  | `UNKNOWN` | `group_mapping`, `symbol_order`                 | No        |
| **BetaNeutralize**   | `UNKNOWN` | `betas`, `symbol_order`                         | No        |
| **FactorNeutralize** | `UNKNOWN` | `factor_exposures`, `symbol_order`              | No        |
| **Normalize**        | `UNKNOWN` | —                                               | No        |
| **BarraNeutralizer** | `UNKNOWN` | `alpha_input`, `returns_input`, `factor_inputs` | No        |

***

## Demean

Remove cross-sectional mean from signal values.

Subtracts the mean of all valid symbol values, centering the
cross-section around zero. This is the simplest form of
neutralization, removing the common level across symbols.
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

```python theme={null}
demean = Demean(Input("zscore", timeframe="1m", lookback=1))
graph.add_node("weights", demean)
```

### 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:
    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):
        mean = np.mean(values[compute_mask])
        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),
    )
```

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## Neutralize

Neutralize signal to zero mean with optional unit-variance scaling.

Removes the cross-sectional mean and optionally divides by the
cross-sectional standard deviation to produce a zero-mean, unit-
variance signal. When scale is False, only demeaning is applied.
Formula: result\_i = (x\_i - mean(x)) / std(x)  \[when scale=True]

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                                                |
| --------- | --------- | -------- | ---------------------------------------------------------- |
| `input`   | `'Input'` | Required | Input signal. timeframe specifies data frequency.          |
| `scale`   | `bool`    | `True`   | If True, also divide by standard deviation after demeaning |

### Usage

```python theme={null}
neutral = Neutralize(
    Input("alpha", timeframe="1m", lookback=1),
    scale=True,
)
graph.add_node("neutral", neutral)
```

### 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:
    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)
        demeaned = valid_values - mean

        if self._scale:
            std = np.std(valid_values)
            if std > 1e-10:
                demeaned = demeaned / std

        result[compute_mask] = demeaned

    result_valid = compute_mask & ~np.isnan(result)

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

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## GroupNeutralize

Neutralize signal within groups such as sectors or categories.

Demeans each symbol within its assigned group so that every group
has zero mean exposure. Symbols not found in the mapping are
assigned to the "Unknown" group.
Formula: result\_i = x\_i - mean(x\_g), where g is the group of symbol i.

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

| Parameter       | Type             | Default  | Description                                                        |
| --------------- | ---------------- | -------- | ------------------------------------------------------------------ |
| `input`         | `'Input'`        | Required | Input signal. timeframe specifies data frequency.                  |
| `group_mapping` | `Dict[str, str]` | Required | Dict mapping symbol name to group name (e.g., \{"BTC": "layer1"}). |
| `symbol_order`  | `List[str]`      | Required | Ordered list of symbol names matching the array index positions.   |

### Usage

```python theme={null}
sector_neutral = GroupNeutralize(
    Input("alpha", timeframe="1m", lookback=1),
    group_mapping={"BTC": "layer1", "ETH": "layer1", "SOL": "layer1", "DOGE": "meme"},
    symbol_order=["BTC", "ETH", "SOL", "DOGE"],
)
graph.add_node("sector_neutral", sector_neutral)
```

### 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:
    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):
        groups: Dict[str, List[int]] = {}
        for i, symbol in enumerate(self._symbol_order):
            if compute_mask[i]:
                group = self._group_mapping.get(symbol, "Unknown")
                if group not in groups:
                    groups[group] = []
                groups[group].append(i)

        for group, indices in groups.items():
            if len(indices) == 0:
                continue
            group_values = values[indices]
            group_mean = np.mean(group_values)
            result[indices] = group_values - group_mean

    result_valid = compute_mask & ~np.isnan(result)

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

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## BetaNeutralize

Neutralize market beta exposure from a signal.

Removes the beta-weighted market component from each symbol's value.
Computes the beta-weighted average signal and subtracts each symbol's
proportional beta contribution.
Formula: adjustment = sum(x\_i \* beta\_i) / sum(beta\_i);
result\_i = x\_i - adjustment \* beta\_i

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

| Parameter      | Type               | Default  | Description                                                                 |
| -------------- | ------------------ | -------- | --------------------------------------------------------------------------- |
| `input`        | `'Input'`          | Required | Input signal. timeframe specifies data frequency.                           |
| `betas`        | `Dict[str, float]` | Required | Dict mapping symbol name to its market beta (default beta: 1.0 if missing). |
| `symbol_order` | `List[str]`        | Required | Ordered list of symbol names matching the array index positions.            |

### Usage

```python theme={null}
beta_neutral = BetaNeutralize(
    Input("alpha", timeframe="1m", lookback=1),
    betas={"BTC": 1.0, "ETH": 1.2, "SOL": 1.5, "DOGE": 1.8},
    symbol_order=["BTC", "ETH", "SOL", "DOGE"],
)
graph.add_node("beta_neutral", beta_neutral)
```

### 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:
    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):
        beta_arr = np.array([
            self._betas.get(sym, 1.0) for sym in self._symbol_order
        ])

        valid_values = values[compute_mask]
        valid_betas = beta_arr[compute_mask]

        weighted_sum = np.sum(valid_values * valid_betas)
        total_beta = np.sum(valid_betas)

        if abs(total_beta) < 1e-10:
            result[compute_mask] = valid_values
        else:
            adjustment = weighted_sum / total_beta
            beta_component = adjustment * valid_betas
            result[compute_mask] = valid_values - beta_component

    result_valid = compute_mask & ~np.isnan(result)

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

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## FactorNeutralize

Remove multiple factor exposures via OLS cross-sectional regression.

Regresses the signal on a matrix of factor exposures (with an intercept)
and returns the residuals. This removes any linear dependence on the
specified factors. Falls back to simple demeaning if the factor matrix
is singular.
Formula: alpha = X \* beta + residual, where X = \[1, factors];
result = alpha - X \* (X'X)^\{-1} \* X' \* alpha

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

| Parameter          | Type                          | Default  | Description                                                                      |
| ------------------ | ----------------------------- | -------- | -------------------------------------------------------------------------------- |
| `input`            | `'Input'`                     | Required | Input signal. timeframe specifies data frequency.                                |
| `factor_exposures` | `Dict[str, Dict[str, float]]` | Required | Dict of \{factor\_name: \{symbol: exposure}} defining the factor loading matrix. |
| `symbol_order`     | `List[str]`                   | Required | Ordered list of symbol names matching the array index positions.                 |

### Usage

```python theme={null}
factor_neutral = FactorNeutralize(
    Input("alpha", timeframe="1m", lookback=1),
    factor_exposures={
        "momentum": {"BTC": 0.5, "ETH": 0.8, "SOL": 1.2},
        "size": {"BTC": 2.0, "ETH": 1.5, "SOL": 0.5},
    },
    symbol_order=["BTC", "ETH", "SOL"],
)
graph.add_node("factor_neutral", factor_neutral)
```

### 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:
    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_symbols = [self._symbol_order[i] for i in valid_indices]
        y = values[compute_mask]

        factor_names = list(self._factor_exposures.keys())
        n_factors = len(factor_names)

        if n_factors == 0:
            result[compute_mask] = y
        else:
            X = np.zeros((n_valid, n_factors))
            for j, factor_name in enumerate(factor_names):
                factor_dict = self._factor_exposures[factor_name]
                for i, sym in enumerate(valid_symbols):
                    X[i, j] = factor_dict.get(sym, 0.0)

            X = np.column_stack([np.ones(n_valid), X])

            try:
                XtX_inv = np.linalg.inv(X.T @ X)
                beta = XtX_inv @ X.T @ y
                prediction = X @ beta
                residuals = y - prediction
            except np.linalg.LinAlgError:
                residuals = y - np.mean(y)

            result[compute_mask] = residuals

    result_valid = compute_mask & ~np.isnan(result)

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

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## Normalize

Combined demeaning and L1 normalization to produce dollar-neutral weights.

First removes the cross-sectional mean, then scales the result so that
the sum of absolute values equals one. This produces a zero-net-exposure,
unit-leverage weight vector suitable for long-short portfolios.
Formula: demeaned\_i = x\_i - mean(x); result\_i = demeaned\_i / sum(|demeaned|)

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

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

### Usage

```python theme={null}
weights = Normalize(Input("alpha", timeframe="1m", lookback=1))
graph.add_node("weights", weights)
```

### 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:
    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]
        demeaned = valid_values - np.mean(valid_values)

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

    result_valid = compute_mask & ~np.isnan(result)

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

<sub>Source: `apps/trading/operators/transform/neutralizers.py`</sub>

***

## BarraNeutralizer

BARRA-style two-stage factor neutralization with rolling estimation.

Performs a full BARRA neutralization process in two stages:
Stage 1 -- Estimate factor loading matrix B via time-series regression:
R\_i,t = sum(beta\_i,k \* F\_k,t) + epsilon\_i,t
Stage 2 -- Neutralize alpha via cross-sectional regression:
alpha\_raw = B \* gamma + alpha\_pure
Formula: alpha\_pure = alpha\_raw - B \* (B'B)^\{-1} \* B' \* alpha\_raw

Maintains rolling history buffers for loading estimation. Falls back
to simple demeaning when insufficient history is available.

**Role**: `UNKNOWN` | **Ephemeral**: No

### Parameters

| Parameter           | Type                  | Default  | Description                                                         |
| ------------------- | --------------------- | -------- | ------------------------------------------------------------------- |
| `alpha_input`       | `'Input'`             | Required | Raw alpha signal to neutralize. timeframe specifies data frequency. |
| `returns_input`     | `'Input'`             | Required | Asset returns for loading estimation.                               |
| `factor_inputs`     | `List['Input']`       | Required | List of factor characteristic inputs (e.g., volatility, size).      |
| `factor_names`      | `Optional[List[str]]` | `None`   | Human-readable names for factors, used for debugging                |
| `estimation_window` | `int`                 | `60`     | Rolling window length for loading estimation                        |
| `min_obs`           | `int`                 | `30`     | Minimum number of observations required for regression              |
| `include_market`    | `bool`                | `True`   | If True, include an equal-weight market factor                      |
| `output_timeframe`  | `Optional[str]`       | `None`   | Output timeframe override                                           |

### Usage

```python theme={null}
graph.add_node("neutral_alpha", BarraNeutralizer(
    alpha_input=Input("momentum", timeframe="1d", lookback=1),
    returns_input=Input("returns", timeframe="1d", lookback=60),
    factor_inputs=[
        Input("volatility", timeframe="1d", lookback=60),
        Input("size", timeframe="1d", lookback=60),
    ],
    factor_names=["VOL", "SIZE"],
    estimation_window=60,
))
```

### 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:
    if not isinstance(data, list) or len(data) < 2 + self._n_factors:
        return self._passthrough(data, 0)

    # Extract inputs
    alpha = self._extract_values(data, 0)
    returns = self._extract_values(data, 1)
    factors = [self._extract_values(data, 2 + i) for i in range(self._n_factors)]

    n_symbols = len(alpha)
    if n_symbols == 0:
        return self._empty_output()

    # Update rolling history
    self._update_history(returns, factors)

    # Estimate loading matrix if enough history
    if len(self._returns_history) >= self._min_obs:
        self._estimate_loading_matrix()

    # Neutralize alpha
    if self._loading_matrix is not None:
        neutralized = self._neutralize_alpha(alpha)
    else:
        # Not enough history - return demeaned alpha
        neutralized = alpha - np.nanmean(alpha)

    return TaggedArray(
        value=neutralized,
        exists=np.ones(n_symbols, dtype=bool),
        valid=~np.isnan(neutralized),
        updated=np.ones(n_symbols, dtype=bool),
    )
```

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