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

# Cross-Sectional Factors

> 8 risk factors for portfolio construction

## Overview

This page documents **8 operators** (role: `FACTOR`).

## Quick Reference

| Operator                    | Role     | Key Parameters    | Ephemeral |
| --------------------------- | -------- | ----------------- | --------- |
| **LiquidityFactor**         | `FACTOR` | —                 | No        |
| **MeanReversionFactor**     | `FACTOR` | —                 | No        |
| **MomentumFactor**          | `FACTOR` | —                 | No        |
| **ShortTermReversalFactor** | `FACTOR` | —                 | No        |
| **QualityFactor**           | `FACTOR` | —                 | No        |
| **SizeFactor**              | `FACTOR` | —                 | No        |
| **VolatilityFactor**        | `FACTOR` | `annualize=False` | No        |
| **VolumeTrendFactor**       | `FACTOR` | —                 | No        |

***

## LiquidityFactor

Liquidity Factor (Inverse of Amihud Illiquidity).

Requires two inputs: close and volume.

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type            | Default  | Description |
| --------- | --------------- | -------- | ----------- |
| `inputs`  | `List['Input']` | Required |             |

### Usage

```python theme={null}
factor = LiquidityFactor(
    inputs=[
        Input("FIELD:close", timeframe="1m", lookback=21),
        Input("FIELD:volume", timeframe="1m", lookback=21),
    ],
)
```

### 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:
        raise ValueError("LiquidityFactor 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)

        with np.errstate(divide='ignore', invalid='ignore'):
            illiquidity = abs_returns / volume[1:]

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

        # Liquidity = negative illiquidity
        liquidity = -avg_illiquidity
        compute_mask = exists & valid
        result[compute_mask] = liquidity[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/liquidity.py`</sub>

***

## MeanReversionFactor

Mean Reversion Factor (Z-score of price deviation).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = MeanReversionFactor(
    Input("FIELD:close", timeframe="1m", lookback=20),
)
```

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

    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

            compute_mask = exists & valid
            result[compute_mask] = zscore[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/mean_reversion.py`</sub>

***

## MomentumFactor

Momentum Factor (rate of price change).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = MomentumFactor(
    Input("FIELD:close", timeframe="1m", lookback=20),
)
```

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

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

<sub>Source: `apps/trading/operators/signal/factor/momentum.py`</sub>

***

## ShortTermReversalFactor

Short-Term Reversal Factor (5-day return reversal).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = ShortTermReversalFactor(
    Input("FIELD:close", timeframe="1m", lookback=5),
)
```

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

    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

            # Reversal: negate momentum
            reversal = -momentum
            compute_mask = exists & valid
            result[compute_mask] = reversal[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/momentum.py`</sub>

***

## QualityFactor

Quality Factor (Volume Stability Proxy - lower CV = higher quality).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = QualityFactor(
    Input("FIELD:volume", timeframe="1m", lookback=20),
)
```

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

    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

            # Negate: lower CV = higher quality
            quality = -cv
            compute_mask = exists & valid
            result[compute_mask] = quality[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/quality.py`</sub>

***

## SizeFactor

Size Factor (volume-based size proxy).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = SizeFactor(
    Input("FIELD:volume", timeframe="1m", lookback=20),
)
```

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

    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))
        avg_volume = np.mean(values[-window:], axis=0)

        compute_mask = exists & valid
        result[compute_mask] = avg_volume[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/size.py`</sub>

***

## VolatilityFactor

Volatility Factor (std of returns).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter   | Type      | Default  | Description |
| ----------- | --------- | -------- | ----------- |
| `input`     | `'Input'` | Required |             |
| `annualize` | `bool`    | `False`  |             |

### Usage

```python theme={null}
factor = VolatilityFactor(
    Input("FIELD:close", timeframe="1m", lookback=21),
    annualize=False,
)
```

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

    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)

            compute_mask = exists & valid
            result[compute_mask] = 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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/volatility.py`</sub>

***

## VolumeTrendFactor

Volume Trend Factor (rate of change in volume).

**Role**: `FACTOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |

### Usage

```python theme={null}
factor = VolumeTrendFactor(
    Input("FIELD:volume", timeframe="1m", lookback=10),
)
```

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

    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'):
                trend = (current - past) / past

            compute_mask = exists & valid
            result[compute_mask] = trend[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),
    )
```

<sub>Source: `apps/trading/operators/signal/factor/volume.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>