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

# Portfolio Optimizers

> MeanVarianceOptimizer, RiskParityOptimizer, EqualWeight, ClipWeights, MaxPositions

## Overview

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

## Quick Reference

| Operator                  | Role      | Key Parameters                                   | Ephemeral |
| ------------------------- | --------- | ------------------------------------------------ | --------- |
| **MeanVarianceOptimizer** | `UNKNOWN` | `expected_returns`, `cov_matrix`, `symbol_order` | No        |
| **RiskParityOptimizer**   | `UNKNOWN` | `volatilities`, `symbol_order`, `min_vol=0.01`   | No        |
| **ClipWeights**           | `UNKNOWN` | `lower=-1.0`, `upper=1.0`, `renormalize=True`    | No        |
| **MaxPositions**          | `UNKNOWN` | `n=20`, `renormalize=True`                       | No        |
| **EqualWeight**           | `UNKNOWN` | `long_only=False`                                | No        |
| **TailSelector**          | `UNKNOWN` | `long_pct=0.2`, `short_pct=0.2`                  | No        |

***

## MeanVarianceOptimizer

Mean-Variance (Markowitz) portfolio optimization.

Computes optimal weights by maximizing expected return for a given
level of risk aversion. Solves the unconstrained quadratic program
and L1-normalizes the result.
Formula: w\_raw = Sigma^\{-1} \* mu / lambda; w = w\_raw / sum(|w\_raw|)

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

### Parameters

| Parameter          | Type                          | Default  | Description                                                           |
| ------------------ | ----------------------------- | -------- | --------------------------------------------------------------------- |
| `input`            | `'Input'`                     | Required | Input signal. timeframe specifies data frequency.                     |
| `expected_returns` | `Dict[str, float]`            | Required | Dict mapping symbol name to expected return.                          |
| `cov_matrix`       | `Dict[str, Dict[str, float]]` | Required | Nested dict \{sym\_i: \{sym\_j: cov}} defining the covariance matrix. |
| `symbol_order`     | `List[str]`                   | Required | Ordered list of symbol names matching the array index positions.      |
| `risk_aversion`    | `float`                       | `1.0`    | Risk aversion parameter lambda; higher means more conservative        |
| `regularize`       | `float`                       | `1e-06`  | Ridge regularization added to diagonal of covariance matrix           |

### Usage

```python theme={null}
optimizer = MeanVarianceOptimizer(
    Input("signals", timeframe="1m", lookback=1),
    expected_returns={"BTC": 0.1, "ETH": 0.15},
    cov_matrix={"BTC": {"BTC": 0.04, "ETH": 0.02}, "ETH": {"BTC": 0.02, "ETH": 0.09}},
    symbol_order=["BTC", "ETH"],
    risk_aversion=1.0,
)
graph.add_node("mvo_weights", optimizer)
```

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

        mu = np.array([
            self._expected_returns.get(sym, 0.0) for sym in valid_symbols
        ])

        Sigma = np.zeros((n_valid, n_valid))
        for i, sym_i in enumerate(valid_symbols):
            for j, sym_j in enumerate(valid_symbols):
                if sym_i in self._cov_matrix and sym_j in self._cov_matrix[sym_i]:
                    Sigma[i, j] = self._cov_matrix[sym_i][sym_j]
                elif i == j:
                    Sigma[i, j] = 0.04

        Sigma += self._regularize * np.eye(n_valid)

        try:
            Sigma_inv = np.linalg.inv(Sigma)
            raw_weights = (Sigma_inv @ mu) / self._risk_aversion

            total = np.sum(np.abs(raw_weights))
            if total > 1e-10:
                weights = raw_weights / total
            else:
                weights = np.zeros(n_valid)
        except np.linalg.LinAlgError:
            weights = np.zeros(n_valid)

        result[valid_indices] = weights

    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/optimizers.py`</sub>

***

## RiskParityOptimizer

Risk Parity portfolio weights proportional to inverse volatility.

Allocates weight to each asset inversely proportional to its
volatility, so that each asset contributes equally to portfolio
risk. Volatilities below min\_vol are floored to avoid extreme weights.
Formula: w\_i = (1 / vol\_i) / sum(1 / vol)

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

### Parameters

| Parameter      | Type               | Default  | Description                                                      |
| -------------- | ------------------ | -------- | ---------------------------------------------------------------- |
| `input`        | `'Input'`          | Required | Input signal. timeframe specifies data frequency.                |
| `volatilities` | `Dict[str, float]` | Required | Dict mapping symbol name to annualized volatility.               |
| `symbol_order` | `List[str]`        | Required | Ordered list of symbol names matching the array index positions. |
| `min_vol`      | `float`            | `0.01`   | Floor for volatility values to prevent extreme weights           |

### Usage

```python theme={null}
optimizer = RiskParityOptimizer(
    Input("signals", timeframe="1m", lookback=1),
    volatilities={"BTC": 0.5, "ETH": 0.6},
    symbol_order=["BTC", "ETH"],
)
graph.add_node("rp_weights", optimizer)
```

### 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_indices = np.where(compute_mask)[0]
        valid_symbols = [self._symbol_order[i] for i in valid_indices]

        vols = np.array([
            max(self._volatilities.get(sym, 0.20), self._min_vol)
            for sym in valid_symbols
        ])

        inv_vols = 1.0 / vols
        weights = inv_vols / np.sum(inv_vols)

        result[valid_indices] = weights

    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/optimizers.py`</sub>

***

## ClipWeights

Clip portfolio weights to bounds and optionally renormalize.

Constrains each weight to \[lower, upper] and optionally rescales
so that sum(|w\_i|) = 1 after clipping. This enforces per-asset
weight limits while maintaining full investment.
Formula: w\_i = clip(w\_i, lower, upper); if renormalize: w\_i = w\_i / sum(|w|)

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

### Parameters

| Parameter     | Type      | Default  | Description                                       |
| ------------- | --------- | -------- | ------------------------------------------------- |
| `input`       | `'Input'` | Required | Input signal. timeframe specifies data frequency. |
| `lower`       | `float`   | `-1.0`   | Lower bound for each weight                       |
| `upper`       | `float`   | `1.0`    | Upper bound for each weight                       |
| `renormalize` | `bool`    | `True`   | If True, L1-normalize weights after clipping      |

### Usage

```python theme={null}
transform = ClipWeights(
    Input("weights", timeframe="1m", lookback=1),
    lower=-0.5,
    upper=0.5,
    renormalize=True,
)
graph.add_node("clipped_weights", transform)
```

### 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 = values.copy()
    result[compute_mask] = np.clip(values[compute_mask], self._lower, self._upper)

    if self._renormalize and np.any(compute_mask):
        total = np.sum(np.abs(result[compute_mask]))
        if total > 1e-10:
            result[compute_mask] = result[compute_mask] / total

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

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

***

## MaxPositions

Limit the number of positions to the top K by absolute weight.

Keeps only the top-N symbols ranked by absolute weight value and
zeros out the rest. Optionally renormalizes the surviving weights
so that sum(|w\_i|) = 1.
Algorithm: sort symbols by |w\_i| descending, keep top N, zero the rest.

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

### Parameters

| Parameter     | Type      | Default  | Description                                       |
| ------------- | --------- | -------- | ------------------------------------------------- |
| `input`       | `'Input'` | Required | Input signal. timeframe specifies data frequency. |
| `n`           | `int`     | `20`     | Maximum number of positions to retain             |
| `renormalize` | `bool`    | `True`   | If True, L1-normalize weights after filtering     |

### Usage

```python theme={null}
transform = MaxPositions(
    Input("weights", timeframe="1m", lookback=1),
    n=20,
    renormalize=True,
)
graph.add_node("top_positions", transform)
```

### 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_indices = np.where(compute_mask)[0]
        valid_values = values[compute_mask]

        abs_values = np.abs(valid_values)
        n_keep = min(self._n, len(valid_values))

        top_k_local = np.argsort(abs_values)[-n_keep:]

        sparse_values = np.zeros_like(valid_values)
        sparse_values[top_k_local] = valid_values[top_k_local]

        if self._renormalize:
            total = np.sum(np.abs(sparse_values))
            if total > 1e-10:
                sparse_values = sparse_values / total

        result[valid_indices] = sparse_values

    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/optimizers.py`</sub>

***

## EqualWeight

Equal weight allocation across symbols based on signal sign.

Assigns equal absolute weight to every valid symbol, using the sign
of the input signal to determine direction (long or short). In
long-only mode, only symbols with positive signal values receive
weight.
Formula: w\_i = sign(x\_i) / N, where N is the number of active positions.

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

### Parameters

| Parameter   | Type      | Default  | Description                                                   |
| ----------- | --------- | -------- | ------------------------------------------------------------- |
| `input`     | `'Input'` | Required | Input signal. timeframe specifies data frequency.             |
| `long_only` | `bool`    | `False`  | If True, only allocate to symbols with positive signal values |

### Usage

```python theme={null}
transform = EqualWeight(
    Input("signals", timeframe="1m", lookback=1),
    long_only=False,
)
graph.add_node("equal_weights", transform)
```

### 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):
        if self._long_only:
            work_mask = compute_mask & (values > 0)
        else:
            work_mask = compute_mask

        n_positions = np.sum(work_mask)
        if n_positions == 0:
            result[compute_mask] = 0.0
        else:
            result[work_mask] = np.sign(values[work_mask]) / n_positions
            result[compute_mask & ~work_mask] = 0.0

    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/optimizers.py`</sub>

***

## TailSelector

Quintile-spread portfolio: long top percentile, short bottom percentile.

Ranks the input signal cross-sectionally, assigns +1 to symbols in the
top quantile and -1 to symbols in the bottom quantile, and zeros out
the middle. This produces a dollar-neutral long-short portfolio
concentrated on the strongest and weakest signals.

With 30 symbols and long\_pct=short\_pct=0.2, the top 6 symbols get +1
(long) and the bottom 6 get -1 (short). The remaining 18 are zeroed out.
Feed the output through L1Norm for proper weight normalization.

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

### Parameters

| Parameter   | Type      | Default  | Description                                                                  |
| ----------- | --------- | -------- | ---------------------------------------------------------------------------- |
| `input`     | `'Input'` | Required | Input signal (e.g. neutralized alpha). timeframe specifies data frequency.   |
| `long_pct`  | `float`   | `0.2`    | Fraction of symbols to go long (top quantile). Default 0.2 (top 20%).        |
| `short_pct` | `float`   | `0.2`    | Fraction of symbols to go short (bottom quantile). Default 0.2 (bottom 20%). |

### Usage

```python theme={null}
transform = TailSelector(
    Input("neutralized", timeframe="1h", lookback=1),
    long_pct=0.2, short_pct=0.2,
)
graph.add_node("tails", transform)
```

### 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.zeros(n_symbols)

    valid_indices = np.where(compute_mask)[0]
    n_valid = len(valid_indices)

    if n_valid > 0:
        valid_values = values[valid_indices]
        ranked = np.argsort(np.argsort(valid_values))
        n_long = max(1, int(n_valid * self._long_pct))
        n_short = max(1, int(n_valid * self._short_pct))
        result[valid_indices[ranked >= n_valid - n_long]] = 1.0
        result[valid_indices[ranked < n_short]] = -1.0

    result_valid = compute_mask.copy()

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

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