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

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.

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

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

---

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

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.

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

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

---

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

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.

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

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

---

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

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.

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

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

---

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

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.

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

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

---

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

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.

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

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

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG