Alpha Signals

21 alpha signals across momentum, mean reversion, volatility, volume, and liquidity categories

Overview

This page documents 21 operators (role: ALPHA).

Quick Reference

Operator	Role	Key Parameters	Ephemeral
AmihudAlpha	ALPHA	—	No
EffectiveSpreadAlpha	ALPHA	—	No
VolatilityOfVolatilityAlpha	ALPHA	vol_window=5, outer_lookback=20	No
BollingerAlpha	ALPHA	num_std=2.0	No
ZScoreAlpha	ALPHA	clip_value=3.0	No
PercentileAlpha	ALPHA	—	No
MomentumAlpha	ALPHA	—	No
RSIAlpha	ALPHA	—	No
TrendStrengthAlpha	ALPHA	annualization=252	No
VolumeStabilityAlpha	ALPHA	—	No
PriceImpactAlpha	ALPHA	—	No
MarketDepthProxyAlpha	ALPHA	—	No
DollarVolumeSizeAlpha	ALPHA	—	No
RealizedSpreadAlpha	ALPHA	—	No
PriceEfficiencyAlpha	ALPHA	—	No
ImpliedBasisAlpha	ALPHA	—	No
VolatilityAlpha	ALPHA	annualize=True	No
VolatilityRegimeAlpha	ALPHA	short_window=5	No
VolumeAlpha	ALPHA	—	No
DollarVolumeAlpha	ALPHA	—	No
VolumeRatioAlpha	ALPHA	short_window=5	No

---

AmihudAlpha

Amihud illiquidity alpha signal favoring higher liquidity.

Computes the Amihud (2002) illiquidity ratio as the average of absolute returns divided by dollar volume, then negates it so that more liquid assets receive higher scores. Formula: illiquidity = mean(|r_t| / (close_t * volume_t)) alpha = -illiquidity

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 2 Inputs: [close, volume]. timeframe must match

Usage

alpha = AmihudAlpha(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=21),
        Input("FIELD:volume", timeframe="1m", lookback=21),
    ],
)

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 not isinstance(data, list) or len(data) < 2:
        raise ValueError("AmihudAlpha requires [close, volume] inputs")

    close_data = data[0]
    volume_data = data[1]

    close = close_data.value
    volume = volume_data.value

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

    vol_last = volume_data[-1] if len(volume_data) > 0 else volume_data
    exists = exists & vol_last.exists
    valid = valid & vol_last.valid

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

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

        abs_returns = np.abs(returns)
        dollar_volume = close[1:] * volume[1:]

        with np.errstate(divide='ignore', invalid='ignore'):
            illiquidity = abs_returns / dollar_volume

        window = min(self._lookback - 1, len(illiquidity))
        avg_illiquidity = np.nanmean(illiquidity[-window:], axis=0)

        score = -avg_illiquidity
        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/liquidity.py}

---

EffectiveSpreadAlpha

Effective spread alpha signal favoring tighter spreads.

Estimates the effective bid-ask spread using the high-low range normalized by the close price, averaged over the lookback window. The result is negated so that tighter spreads produce higher scores. Formula: spread_t = (high_t - low_t) / close_t alpha = -mean(spread[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 3 Inputs: [high, low, close]. timeframe must match

Usage

alpha = EffectiveSpreadAlpha(
    inputs=[
        Input("FIELD:high", timeframe="1m", lookback=20),
        Input("FIELD:low", timeframe="1m", lookback=20),
        Input("FIELD:close", timeframe="1m", lookback=20),
    ],
)

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 not isinstance(data, list) or len(data) < 3:
        raise ValueError("EffectiveSpreadAlpha requires [high, low, close] inputs")

    high_data = data[0]
    low_data = data[1]
    close_data = data[2]

    high = high_data.value
    low = low_data.value
    close = close_data.value

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

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

    if len(high.shape) > 1:
        window = min(self._lookback, len(high))

        range_val = high[-window:] - low[-window:]
        with np.errstate(divide='ignore', invalid='ignore'):
            spread = range_val / close[-window:]

        avg_spread = np.nanmean(spread, axis=0)

        score = -avg_spread
        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/liquidity.py}

---

VolatilityOfVolatilityAlpha

Volatility-of-volatility alpha signal favoring stable volatility regimes.

Computes rolling realized volatility of returns over a short window, then measures the standard deviation of those rolling volatility estimates over a longer outer window. The result is negated so that assets with more stable volatility receive higher scores. Formula: rolling_vol_i = std(returns[i:i+vol_window]) alpha = -std(rolling_vol[-outer_lookback:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input for close price data. timeframe specifies data frequency.
vol_window	int	5	Number of periods for each rolling volatility estimate
outer_lookback	int	20	Number of rolling-vol observations used to compute

Usage

alpha = VolatilityOfVolatilityAlpha(
    Input("FIELD:close", timeframe="1m", lookback=26),
    vol_window=5,
    outer_lookback=20,
)

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)

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

        n_periods = len(returns) - self._vol_window + 1
        if n_periods >= 2:
            rolling_vol = np.zeros((n_periods, returns.shape[1]))
            for i in range(n_periods):
                rolling_vol[i] = np.std(
                    returns[i:i + self._vol_window], axis=0, ddof=1
                )

            window = min(self._outer_lookback, len(rolling_vol))
            vol_of_vol = np.std(rolling_vol[-window:], axis=0, ddof=1)

            score = -vol_of_vol
            compute_mask = exists & valid
            result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/liquidity.py}

---

BollingerAlpha

Bollinger mean reversion alpha.

Measures deviation from the Bollinger Band middle line (SMA), normalized by bandwidth. Positive score when price is below mean (buy signal). Formula: score = clip((SMA - price) / (std * num_std), -1, 1)

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback defines the SMA window.
num_std	float	2.0	Number of standard deviations for bandwidth

Usage

alpha = BollingerAlpha(
    Input("FIELD:close", timeframe="1m", lookback=20),
    num_std=2.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

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

    if len(values.shape) > 1:
        window = min(self._lookback, len(values))
        if window >= 2:
            recent = values[-window:]
            middle = np.mean(recent, axis=0)
            std = np.std(recent, axis=0, ddof=1)
            current = values[-1]

            with np.errstate(divide='ignore', invalid='ignore'):
                bandwidth = std * self._num_std
                deviation = middle - current
                score = deviation / bandwidth

            score = np.clip(score, -1.0, 1.0)
            compute_mask = exists & valid
            result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/mean_reversion.py}

---

ZScoreAlpha

Z-Score mean reversion alpha.

Computes z-score of current price vs rolling window, then negates it for mean reversion (below-mean = positive signal). Formula: score = clip(-(price - mean) / std, -clip_value, clip_value)

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback defines the rolling window.
clip_value	float	3.0	Maximum absolute z-score to clip at

Usage

alpha = ZScoreAlpha(
    Input("FIELD:close", timeframe="1m", lookback=20),
    clip_value=3.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

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

    if len(values.shape) > 1:
        window = min(self._lookback, len(values))
        if window >= 2:
            recent = values[-window:]
            mean = np.mean(recent, axis=0)
            std = np.std(recent, axis=0, ddof=1)
            current = values[-1]

            with np.errstate(divide='ignore', invalid='ignore'):
                zscore = (current - mean) / std

            score = -zscore
            score = np.clip(score, -self._clip_value, self._clip_value)
            compute_mask = exists & valid
            result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/mean_reversion.py}

---

PercentileAlpha

Percentile mean reversion alpha.

Computes the percentile rank of current price within the lookback window, then inverts for mean reversion. Price at bottom of range yields positive signal. Formula: score = -(percentile_rank - 0.5) * 2

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback defines the ranking window.

Usage

alpha = PercentileAlpha(
    Input("FIELD:close", timeframe="1m", lookback=20),
)

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)
    compute_mask = exists & valid

    if len(values.shape) > 1:
        window = min(self._lookback, len(values))
        if window >= 2:
            recent = values[-window:]
            current = values[-1]

            percentiles = np.zeros(n_symbols)
            for i in range(n_symbols):
                if compute_mask[i]:
                    percentiles[i] = np.mean(recent[:, i] <= current[i])

            centered = percentiles - 0.5
            score = -centered * 2
            result[compute_mask] = score[compute_mask]

    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/signal/alpha/mean_reversion.py}

---

MomentumAlpha

Simple momentum alpha (return over lookback period).

Computes the price return over the lookback window as a momentum signal. Formula: momentum = (price_t - price_{t-n}) / price_{t-n}

Positive values indicate upward momentum, negative values indicate downward momentum. The lookback is derived from Input.lookback.

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). timeframe specifies the data

Usage

momentum = MomentumAlpha(Input("FIELD:close", timeframe="1m", lookback=20))
graph.add_node("momentum", momentum)

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)

    if len(values.shape) > 1 and len(values) >= 2:
        window = min(self._lookback, len(values) - 1)
        if window >= 1:
            current = values[-1]
            past = values[-(window + 1)]

            with np.errstate(divide='ignore', invalid='ignore'):
                momentum = (current - past) / past

            compute_mask = exists & valid
            result[compute_mask] = momentum[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/momentum.py}

---

RSIAlpha

RSI alpha normalized to [-1, 1].

Computes RSI and normalizes to [-1, 1] range for direct use as alpha signal. Formula: RSI = 100 - 100 / (1 + avg_gain / avg_loss) Normalized: (RSI - 50) / 50

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback >= period + 1

Usage

rsi = RSIAlpha(Input("FIELD:close", timeframe="1m", lookback=15))
graph.add_node("rsi", rsi)

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)

    if len(values.shape) > 1 and len(values) >= 2:
        returns = np.diff(values, axis=0)
        window = min(self._lookback - 1, len(returns))

        if window >= 1:
            recent_returns = returns[-window:]
            gains = np.where(recent_returns > 0, recent_returns, 0)
            losses = np.where(recent_returns < 0, -recent_returns, 0)

            avg_gain = np.mean(gains, axis=0)
            avg_loss = np.mean(losses, axis=0)

            with np.errstate(divide='ignore', invalid='ignore'):
                rs = avg_gain / avg_loss
                rsi = 100 - (100 / (1 + rs))

            normalized = (rsi - 50) / 50
            compute_mask = exists & valid
            result[compute_mask] = normalized[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/momentum.py}

---

TrendStrengthAlpha

Trend strength alpha (annualized slope of log prices).

Fits a linear regression on log prices over the lookback window and annualizes the slope. Higher slope indicates stronger upward trend. Formula: slope = cov(t, log(price)) / var(t), annualized by * 252

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback >= 2 for regression.
annualization	int	252	Annualization factor

Usage

trend = TrendStrengthAlpha(Input("FIELD:close", timeframe="1m", lookback=20))
graph.add_node("trend", trend)

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)

    if len(values.shape) > 1 and len(values) >= 2:
        window = min(self._lookback, len(values))
        if window >= 2:
            with np.errstate(divide='ignore', invalid='ignore'):
                log_prices = np.log(values[-window:])

            t = np.arange(window).reshape(-1, 1)
            t_mean = t.mean()
            t_var = np.var(t, ddof=0)

            if t_var > 0:
                log_mean = np.mean(log_prices, axis=0)
                cov = np.sum((t - t_mean) * (log_prices - log_mean), axis=0) / window
                slope = cov / t_var
                annualized = slope * self._annualization

                compute_mask = exists & valid
                result[compute_mask] = annualized[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/momentum.py}

---

VolumeStabilityAlpha

Volume stability alpha signal favoring consistent trading volume.

Measures volume stability as the inverse of the coefficient of variation (CV) of volume over the lookback window. Assets with more stable volume patterns receive higher scores. Formula: CV = std(volume[-window:]) / mean(volume[-window:]) alpha = 1 / CV

Role: ALPHA | Ephemeral: No

Parameters

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

Usage

alpha = VolumeStabilityAlpha(
    Input("FIELD:volume", timeframe="1m", lookback=20),
)

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)

    if len(values.shape) > 1:
        window = min(self._lookback, len(values))
        if window >= 2:
            recent = values[-window:]
            mean_vol = np.mean(recent, axis=0)
            std_vol = np.std(recent, axis=0, ddof=1)

            with np.errstate(divide='ignore', invalid='ignore'):
                cv = std_vol / mean_vol
                stability = 1.0 / cv

            compute_mask = exists & valid
            result[compute_mask] = stability[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/quality.py}

---

PriceImpactAlpha

Price impact alpha signal favoring lower market impact.

Estimates price impact as the average ratio of absolute returns to log-volume over the lookback window. The result is negated so that assets with lower price impact (better execution quality) score higher. Formula: impact_t = |r_t| / log(volume_t + 1) alpha = -mean(impact[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 2 Inputs: [close, volume]. timeframe must match

Usage

alpha = PriceImpactAlpha(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=21),
        Input("FIELD:volume", timeframe="1m", lookback=21),
    ],
)

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 not isinstance(data, list) or len(data) < 2:
        raise ValueError("PriceImpactAlpha requires [close, volume] inputs")

    close_data = data[0]
    volume_data = data[1]

    close = close_data.value
    volume = volume_data.value

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

    vol_last = volume_data[-1] if len(volume_data) > 0 else volume_data
    exists = exists & vol_last.exists
    valid = valid & vol_last.valid

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

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

        abs_returns = np.abs(returns)
        log_volume = np.log(volume[1:] + 1.0)

        with np.errstate(divide='ignore', invalid='ignore'):
            impact = abs_returns / log_volume

        window = min(self._lookback - 1, len(impact))
        avg_impact = np.nanmean(impact[-window:], axis=0)

        score = -avg_impact
        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/quality.py}

---

MarketDepthProxyAlpha

Market depth proxy alpha signal favoring deeper, more liquid markets.

Approximates market depth as the ratio of average volume to return volatility over the lookback window. Higher values indicate that the asset can absorb larger trades with less price movement. Formula: volatility = std(returns[-window:]) alpha = mean(volume[-window:]) / volatility

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 2 Inputs: [close, volume]. timeframe must match

Usage

alpha = MarketDepthProxyAlpha(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=21),
        Input("FIELD:volume", timeframe="1m", lookback=21),
    ],
)

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 not isinstance(data, list) or len(data) < 2:
        raise ValueError("MarketDepthProxyAlpha requires [close, volume] inputs")

    close_data = data[0]
    volume_data = data[1]

    close = close_data.value
    volume = volume_data.value

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

    vol_last = volume_data[-1] if len(volume_data) > 0 else volume_data
    exists = exists & vol_last.exists
    valid = valid & vol_last.valid

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

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

        window = min(self._lookback - 1, len(returns))
        volatility = np.std(returns[-window:], axis=0, ddof=1)
        avg_volume = np.mean(volume[-window:], axis=0)

        with np.errstate(divide='ignore', invalid='ignore'):
            depth_proxy = avg_volume / volatility

        compute_mask = exists & valid
        result[compute_mask] = depth_proxy[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/quality.py}

---

DollarVolumeSizeAlpha

Dollar volume size alpha signal based on log-scaled average dollar volume.

Computes the natural logarithm of the average dollar volume (close multiplied by volume) over the lookback window. The log transform compresses the scale, making the signal suitable for cross-sectional comparison of assets with vastly different trading activity. Formula: alpha = log(mean(close[-window:] * volume[-window:]) + 1)

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 2 Inputs: [close, volume]. timeframe must match

Usage

alpha = DollarVolumeSizeAlpha(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=20),
        Input("FIELD:volume", timeframe="1m", lookback=20),
    ],
)

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 not isinstance(data, list) or len(data) < 2:
        raise ValueError("DollarVolumeSizeAlpha requires [close, volume] inputs")

    close_data = data[0]
    volume_data = data[1]

    close = close_data.value
    volume = volume_data.value

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

    vol_last = volume_data[-1] if len(volume_data) > 0 else volume_data
    exists = exists & vol_last.exists
    valid = valid & vol_last.valid

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

    if len(close.shape) > 1:
        dollar_volume = close * volume
        window = min(self._lookback, len(dollar_volume))

        avg_dv = np.mean(dollar_volume[-window:], axis=0)
        score = np.log(avg_dv + 1.0)

        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/size.py}

---

RealizedSpreadAlpha

Realized spread alpha signal favoring tighter realized spreads.

Computes the realized bid-ask spread proxy using the high-low range normalized by the close price, averaged over the lookback window. The result is negated so that assets with tighter spreads (better value) receive higher scores. Formula: spread_t = (high_t - low_t) / close_t alpha = -mean(spread[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 3 Inputs: [high, low, close]. timeframe must match

Usage

alpha = RealizedSpreadAlpha(
    inputs=[
        Input("FIELD:high", timeframe="1m", lookback=20),
        Input("FIELD:low", timeframe="1m", lookback=20),
        Input("FIELD:close", timeframe="1m", lookback=20),
    ],
)

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 not isinstance(data, list) or len(data) < 3:
        raise ValueError("RealizedSpreadAlpha requires [high, low, close] inputs")

    high_data = data[0]
    low_data = data[1]
    close_data = data[2]

    high = high_data.value
    low = low_data.value
    close = close_data.value

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

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

    if len(high.shape) > 1:
        window = min(self._lookback, len(high))

        range_val = high[-window:] - low[-window:]
        with np.errstate(divide='ignore', invalid='ignore'):
            spread = range_val / close[-window:]

        avg_spread = np.nanmean(spread, axis=0)

        score = -avg_spread
        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/value.py}

---

PriceEfficiencyAlpha

Price efficiency alpha signal favoring prices that close near the midpoint.

Measures how far the close price deviates from the mid-price (average of high and low), normalized by the price range. The result is negated so that assets whose close consistently lands near the midpoint (indicating efficient price discovery) receive higher scores. Formula: mid_t = (high_t + low_t) / 2 deviation_t = |close_t - mid_t| / (high_t - low_t) alpha = -mean(deviation[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 3 Inputs: [high, low, close]. timeframe must match

Usage

alpha = PriceEfficiencyAlpha(
    inputs=[
        Input("FIELD:high", timeframe="1m", lookback=20),
        Input("FIELD:low", timeframe="1m", lookback=20),
        Input("FIELD:close", timeframe="1m", lookback=20),
    ],
)

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 not isinstance(data, list) or len(data) < 3:
        raise ValueError("PriceEfficiencyAlpha requires [high, low, close] inputs")

    high_data = data[0]
    low_data = data[1]
    close_data = data[2]

    high = high_data.value
    low = low_data.value
    close = close_data.value

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

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

    if len(high.shape) > 1:
        window = min(self._lookback, len(high))

        mid_price = (high[-window:] + low[-window:]) / 2.0
        price_range = high[-window:] - low[-window:]
        deviation = np.abs(close[-window:] - mid_price)

        with np.errstate(divide='ignore', invalid='ignore'):
            relative_deviation = deviation / price_range

        avg_deviation = np.nanmean(relative_deviation, axis=0)

        score = -avg_deviation
        compute_mask = exists & valid
        result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/value.py}

---

ImpliedBasisAlpha

Implied basis alpha signal as a Sharpe-ratio-style momentum proxy.

Computes the mean return divided by the standard deviation of returns over the lookback window, producing a risk-adjusted momentum measure analogous to the Sharpe ratio. Formula: alpha = mean(returns[-window:]) / std(returns[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input for close price data. timeframe specifies data frequency.

Usage

alpha = ImpliedBasisAlpha(
    Input("FIELD:close", timeframe="1m", lookback=21),
)

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)

    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))
        if window >= 2:
            recent = returns[-window:]
            mean_return = np.mean(recent, axis=0)
            std_return = np.std(recent, axis=0, ddof=1)

            with np.errstate(divide='ignore', invalid='ignore'):
                basis_proxy = mean_return / std_return

            compute_mask = exists & valid
            result[compute_mask] = basis_proxy[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/value.py}

---

VolatilityAlpha

Inverse volatility alpha (lower volatility = higher score).

Computes annualized volatility of returns and negates it so that lower-volatility assets receive higher scores (defensive signal). Formula: score = -(std(returns) * sqrt(252))

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback >= 2 for return calculation.
annualize	bool	True	Whether to annualize volatility

Usage

alpha = VolatilityAlpha(
    Input("FIELD:close", timeframe="1m", lookback=21),
    annualize=True,
)

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)

    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))
        if window >= 2:
            vol = np.std(returns[-window:], axis=0, ddof=1)

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

            inv_vol = -vol
            compute_mask = exists & valid
            result[compute_mask] = inv_vol[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/volatility.py}

---

VolatilityRegimeAlpha

Volatility regime alpha (current vol / average vol).

Compares short-term volatility to long-term volatility to detect regime changes. Low ratio means calm market (positive signal). Formula: score = -(short_vol / long_vol)

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Price Input (e.g., FIELD:close). lookback defines the long window.
short_window	int	5	Short-term volatility window

Usage

alpha = VolatilityRegimeAlpha(
    Input("FIELD:close", timeframe="1m", lookback=21),
    short_window=5,
)

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)

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

        if len(returns) >= self._long_window:
            short_vol = np.std(returns[-self._short_window:], axis=0, ddof=1)
            long_vol = np.std(returns[-self._long_window:], axis=0, ddof=1)

            with np.errstate(divide='ignore', invalid='ignore'):
                ratio = short_vol / long_vol

            score = -ratio
            compute_mask = exists & valid
            result[compute_mask] = score[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/volatility.py}

---

VolumeAlpha

Volume alpha signal based on relative volume.

Computes the ratio of the most recent volume to the average volume over a lookback window. Values greater than 1 indicate above-average activity. Formula: alpha = volume_current / mean(volume[-lookback:])

Role: ALPHA | Ephemeral: No

Parameters

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

Usage

alpha = VolumeAlpha(
    Input("FIELD:volume", timeframe="1m", lookback=20),
)

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)

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

            with np.errstate(divide='ignore', invalid='ignore'):
                ratio = current / avg_volume

            compute_mask = exists & valid
            result[compute_mask] = ratio[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/volume.py}

---

DollarVolumeAlpha

Dollar volume alpha signal based on average dollar volume.

Computes the mean dollar volume (price multiplied by volume) over a lookback window. Higher dollar volume indicates greater market activity and liquidity. Formula: alpha = mean(close[-window:] * volume[-window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
inputs	List['Input']	Required	List of 2 Inputs: [close, volume]. timeframe must match

Usage

alpha = DollarVolumeAlpha(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=20),
        Input("FIELD:volume", timeframe="1m", lookback=20),
    ],
)

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 not isinstance(data, list) or len(data) < 2:
        raise ValueError("DollarVolumeAlpha requires [close, volume] inputs")

    close_data = data[0]
    volume_data = data[1]

    close = close_data.value
    volume = volume_data.value

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

    vol_last = volume_data[-1] if len(volume_data) > 0 else volume_data
    exists = exists & vol_last.exists
    valid = valid & vol_last.valid

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

    if len(close.shape) > 1:
        dollar_volume = close * volume
        window = min(self._lookback, len(dollar_volume))

        avg_dv = np.mean(dollar_volume[-window:], axis=0)
        compute_mask = exists & valid
        result[compute_mask] = avg_dv[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/volume.py}

---

VolumeRatioAlpha

Volume ratio alpha signal comparing short-term to long-term average volume.

Computes the ratio of a short-window moving average of volume to a long-window moving average. Values greater than 1 indicate recent volume acceleration relative to the longer-term trend. Formula: alpha = mean(volume[-short_window:]) / mean(volume[-long_window:])

Role: ALPHA | Ephemeral: No

Parameters

Parameter	Type	Default	Description
input	'Input'	Required	Input for volume data. timeframe specifies data frequency.
short_window	int	5	Number of periods for the short-term average

Usage

alpha = VolumeRatioAlpha(
    Input("FIELD:volume", timeframe="1m", lookback=20),
    short_window=5,
)

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)

    if len(values.shape) > 1 and len(values) >= self._long_window:
        short_avg = np.mean(values[-self._short_window:], axis=0)
        long_avg = np.mean(values[-self._long_window:], axis=0)

        with np.errstate(divide='ignore', invalid='ignore'):
            ratio = short_avg / long_avg

        compute_mask = exists & valid
        result[compute_mask] = ratio[compute_mask]

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

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

_{Source: apps/trading/operators/signal/alpha/volume.py}

Related Pages

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG