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

# Statistics and Math

> LINEARREG, BETA, CORREL, VAR, TSF, MINMAX, Hilbert Transform

## Overview

This page documents **26 operators** (role: `INDICATOR`).

## Quick Reference

| Operator                 | Role        | Key Parameters                  | Ephemeral |
| ------------------------ | ----------- | ------------------------------- | --------- |
| **HT\_DCPERIOD**         | `INDICATOR` | —                               | No        |
| **HT\_DCPHASE**          | `INDICATOR` | —                               | No        |
| **HT\_TRENDLINE**        | `INDICATOR` | —                               | No        |
| **HT\_TRENDMODE**        | `INDICATOR` | —                               | No        |
| **HT\_SINE**             | `INDICATOR` | `output='sine'`                 | No        |
| **HT\_PHASOR**           | `INDICATOR` | `output='inphase'`              | No        |
| **MAX**                  | `INDICATOR` | `period=30`                     | No        |
| **MAXINDEX**             | `INDICATOR` | `period=30`                     | No        |
| **MIN**                  | `INDICATOR` | `period=30`                     | No        |
| **MININDEX**             | `INDICATOR` | `period=30`                     | No        |
| **MIDPOINT**             | `INDICATOR` | `period=14`                     | No        |
| **MIDPRICE**             | `INDICATOR` | `period=14`                     | No        |
| **MINMAX**               | `INDICATOR` | `period=30`, `output='min'`     | No        |
| **MINMAXINDEX**          | `INDICATOR` | `period=30`, `output='minidx'`  | No        |
| **AVGPRICE**             | `INDICATOR` | —                               | No        |
| **MEDPRICE**             | `INDICATOR` | —                               | No        |
| **TYPPRICE**             | `INDICATOR` | —                               | No        |
| **WCLPRICE**             | `INDICATOR` | —                               | No        |
| **LINEARREG**            | `INDICATOR` | `period=14`                     | No        |
| **LINEARREG\_SLOPE**     | `INDICATOR` | `period=14`                     | No        |
| **LINEARREG\_INTERCEPT** | `INDICATOR` | `period=14`                     | No        |
| **LINEARREG\_ANGLE**     | `INDICATOR` | `period=14`                     | No        |
| **TSF**                  | `INDICATOR` | `period=14`                     | No        |
| **CORREL**               | `INDICATOR` | `input1`, `input2`, `period=30` | No        |
| **BETA**                 | `INDICATOR` | `input1`, `input2`, `period=5`  | No        |
| **SUM**                  | `INDICATOR` | `period=30`                     | No        |

***

## HT\_DCPERIOD

Hilbert Transform - Dominant Cycle Period indicator.

Returns the dominant cycle period of the price data.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                          |
| --------- | --------- | -------- | ------------------------------------ |
| `input`   | `'Input'` | Required | Input signal (typically close price) |

### Usage

```python theme={null}
op = HT_DCPERIOD(Input("FIELD:close", timeframe="1m", lookback=32))
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 32:
                result[i] = self._calculate_dc_period(prices[valid_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/indicator/hilbert.py`</sub>

***

## HT\_DCPHASE

Hilbert Transform - Dominant Cycle Phase indicator.

Returns the phase of the dominant cycle.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                          |
| --------- | --------- | -------- | ------------------------------------ |
| `input`   | `'Input'` | Required | Input signal (typically close price) |

### Usage

```python theme={null}
op = HT_DCPHASE(Input("FIELD:close", timeframe="1m", lookback=32))
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 32:
                result[i] = self._calculate_dc_phase(prices[valid_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/indicator/hilbert.py`</sub>

***

## HT\_TRENDLINE

Hilbert Transform - Instantaneous Trendline indicator.

Returns the trendline component of the price.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                          |
| --------- | --------- | -------- | ------------------------------------ |
| `input`   | `'Input'` | Required | Input signal (typically close price) |

### Usage

```python theme={null}
op = HT_TRENDLINE(Input("FIELD:close", timeframe="1m", lookback=32))
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 12:
                result[i] = self._calculate_trendline(prices[valid_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/indicator/hilbert.py`</sub>

***

## HT\_TRENDMODE

Hilbert Transform - Trend vs Cycle Mode indicator.

Returns 1 for trend mode, 0 for cycle mode.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                          |
| --------- | --------- | -------- | ------------------------------------ |
| `input`   | `'Input'` | Required | Input signal (typically close price) |

### Usage

```python theme={null}
op = HT_TRENDMODE(Input("FIELD:close", timeframe="1m", lookback=32))
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 20:
                result[i] = self._calculate_trend_mode(prices[valid_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/indicator/hilbert.py`</sub>

***

## HT\_SINE

Hilbert Transform - SineWave indicator.

Returns the sine or lead sine of the dominant cycle.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description                          |
| --------- | --------- | -------- | ------------------------------------ |
| `input`   | `'Input'` | Required | Input signal (typically close price) |
| `output`  | `str`     | `'sine'` | "sine" or "leadsine"                 |

### Usage

```python theme={null}
sine = HT_SINE(Input("FIELD:close", timeframe="1m", lookback=32), output="sine")
leadsine = HT_SINE(Input("FIELD:close", timeframe="1m", lookback=32), output="leadsine")
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 32:
                sine, leadsine = self._calculate_sine(prices[valid_mask])
                result[i] = leadsine if self._output == "leadsine" else sine

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

***

## HT\_PHASOR

Hilbert Transform - Phasor Components indicator.

Returns the in-phase or quadrature component.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default     | Description                          |
| --------- | --------- | ----------- | ------------------------------------ |
| `input`   | `'Input'` | Required    | Input signal (typically close price) |
| `output`  | `str`     | `'inphase'` | "inphase" or "quadrature"            |

### Usage

```python theme={null}
inphase = HT_PHASOR(Input("FIELD:close", timeframe="1m", lookback=32), output="inphase")
quadrature = HT_PHASOR(Input("FIELD:close", timeframe="1m", lookback=32), output="quadrature")
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            prices = all_values[:, i]
            valid_mask = ~np.isnan(prices)
            if np.sum(valid_mask) >= 32:
                inphase, quadrature = self._calculate_phasor(prices[valid_mask])
                result[i] = quadrature if self._output == "quadrature" else inphase

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

***

## MAX

Highest value over a specified period.

Example:
op = MAX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `30`     |             |

### Usage

```python theme={null}
op = MAX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            valid_vals = window[~np.isnan(window)]
            if len(valid_vals) > 0:
                result[i] = np.max(valid_vals)

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

***

## MAXINDEX

Index of highest value over a specified period.

Example:
op = MAXINDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `30`     |             |

### Usage

```python theme={null}
op = MAXINDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)):
                result[i] = float(np.argmax(window))

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

***

## MIN

Lowest value over a specified period.

Example:
op = MIN(Input("FIELD:close", timeframe="1m", lookback=30), period=30)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `30`     |             |

### Usage

```python theme={null}
op = MIN(Input("FIELD:close", timeframe="1m", lookback=30), period=30)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            valid_vals = window[~np.isnan(window)]
            if len(valid_vals) > 0:
                result[i] = np.min(valid_vals)

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

***

## MININDEX

Index of lowest value over a specified period.

Example:
op = MININDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `30`     |             |

### Usage

```python theme={null}
op = MININDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)):
                result[i] = float(np.argmin(window))

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

***

## MIDPOINT

MidPoint over period: (highest + lowest) / 2.

Example:
op = MIDPOINT(Input("FIELD:close", timeframe="1m", lookback=30), period=30)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `14`     |             |

### Usage

```python theme={null}
op = MIDPOINT(Input("FIELD:close", timeframe="1m", lookback=30), period=30)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            valid_vals = window[~np.isnan(window)]
            if len(valid_vals) > 0:
                result[i] = (np.max(valid_vals) + np.min(valid_vals)) / 2

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

***

## MIDPRICE

Midpoint Price over period: (highest high + lowest low) / 2.

Example:
op = MIDPRICE(
Input("FIELD:high", timeframe="1m", lookback=14),
Input("FIELD:low", timeframe="1m", lookback=14),
period=14,
)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `high`    | `'Input'` | Required |             |
| `low`     | `'Input'` | Required |             |
| `period`  | `int`     | `14`     |             |

### Usage

```python theme={null}
op = MIDPRICE(
    Input("FIELD:high", timeframe="1m", lookback=14),
    Input("FIELD:low", timeframe="1m", lookback=14),
    period=14,
)
```

### 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:
        return self._empty_output()

    high_data, low_data = data[0], data[1]

    n_time = len(high_data)
    if n_time == 0:
        return self._empty_output()

    last = high_data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    high_values = np.array([high_data[t].value for t in range(n_time)])
    low_values = np.array([low_data[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            highs = high_values[-period:, i]
            lows = low_values[-period:, i]
            if not np.any(np.isnan(highs)) and not np.any(np.isnan(lows)):
                result[i] = (np.max(highs) + np.min(lows)) / 2

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

***

## MINMAX

Lowest and highest values over a specified period.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description     |
| --------- | --------- | -------- | --------------- |
| `input`   | `'Input'` | Required | Input signal    |
| `period`  | `int`     | `30`     | Lookback period |
| `output`  | `str`     | `'min'`  | "min" or "max"  |

### Usage

```python theme={null}
lowest = MINMAX(Input("FIELD:close", timeframe="1m", lookback=30), period=30, output="min")
highest = MINMAX(Input("FIELD:close", timeframe="1m", lookback=30), period=30, output="max")
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            valid_vals = window[~np.isnan(window)]
            if len(valid_vals) > 0:
                if self._output == "max":
                    result[i] = np.max(valid_vals)
                else:
                    result[i] = np.min(valid_vals)

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

***

## MINMAXINDEX

Indices of lowest and highest values over a specified period.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default    | Description          |
| --------- | --------- | ---------- | -------------------- |
| `input`   | `'Input'` | Required   | Input signal         |
| `period`  | `int`     | `30`       | Lookback period      |
| `output`  | `str`     | `'minidx'` | "minidx" or "maxidx" |

### Usage

```python theme={null}
min_idx = MINMAXINDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30, output="minidx")
max_idx = MINMAXINDEX(Input("FIELD:close", timeframe="1m", lookback=30), period=30, output="maxidx")
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)):
                if self._output == "maxidx":
                    result[i] = float(np.argmax(window))
                else:
                    result[i] = float(np.argmin(window))

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

***

## AVGPRICE

Average Price indicator.

AVGPRICE = (Open + High + Low + Close) / 4

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `open_`   | `'Input'` | Required | Open price input  |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `close`   | `'Input'` | Required | Close price input |

### Usage

```python theme={null}
op = AVGPRICE(
    Input("FIELD:open", timeframe="1m", lookback=1),
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
    Input("FIELD:close", timeframe="1m", lookback=1),
)
```

### 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) < 4:
        return self._empty_output()

    open_data, high_data, low_data, close_data = data[0], data[1], data[2], data[3]

    n_time = len(close_data)
    if n_time == 0:
        return self._empty_output()

    last = close_data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    o = open_data[-1].value
    h = high_data[-1].value
    l = low_data[-1].value
    c = close_data[-1].value

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            if not any(np.isnan([o[i], h[i], l[i], c[i]])):
                result[i] = (o[i] + h[i] + l[i] + c[i]) / 4

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

***

## MEDPRICE

Median Price indicator.

MEDPRICE = (High + Low) / 2

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description      |
| --------- | --------- | -------- | ---------------- |
| `high`    | `'Input'` | Required | High price input |
| `low`     | `'Input'` | Required | Low price input  |

### Usage

```python theme={null}
op = MEDPRICE(
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
)
```

### 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:
        return self._empty_output()

    high_data, low_data = data[0], data[1]

    n_time = len(high_data)
    if n_time == 0:
        return self._empty_output()

    last = high_data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    h = high_data[-1].value
    l = low_data[-1].value

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            if not any(np.isnan([h[i], l[i]])):
                result[i] = (h[i] + l[i]) / 2

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

***

## TYPPRICE

Typical Price indicator.

TYPPRICE = (High + Low + Close) / 3

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `close`   | `'Input'` | Required | Close price input |

### Usage

```python theme={null}
op = TYPPRICE(
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
    Input("FIELD:close", timeframe="1m", lookback=1),
)
```

### 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) < 3:
        return self._empty_output()

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

    n_time = len(close_data)
    if n_time == 0:
        return self._empty_output()

    last = close_data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    h = high_data[-1].value
    l = low_data[-1].value
    c = close_data[-1].value

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            if not any(np.isnan([h[i], l[i], c[i]])):
                result[i] = (h[i] + l[i] + c[i]) / 3

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

***

## WCLPRICE

Weighted Close Price indicator.

WCLPRICE = (High + Low + Close \* 2) / 4

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description       |
| --------- | --------- | -------- | ----------------- |
| `high`    | `'Input'` | Required | High price input  |
| `low`     | `'Input'` | Required | Low price input   |
| `close`   | `'Input'` | Required | Close price input |

### Usage

```python theme={null}
op = WCLPRICE(
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
    Input("FIELD:close", timeframe="1m", lookback=1),
)
```

### 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) < 3:
        return self._empty_output()

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

    n_time = len(close_data)
    if n_time == 0:
        return self._empty_output()

    last = close_data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    h = high_data[-1].value
    l = low_data[-1].value
    c = close_data[-1].value

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

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            if not any(np.isnan([h[i], l[i], c[i]])):
                result[i] = (h[i] + l[i] + c[i] * 2) / 4

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

***

## LINEARREG

Linear Regression indicator.

Returns the linear regression value at the end of the period.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description     |
| --------- | --------- | -------- | --------------- |
| `input`   | `'Input'` | Required | Input signal    |
| `period`  | `int`     | `14`     | Lookback period |

### Usage

```python theme={null}
op = LINEARREG(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])

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

    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)) and len(window) == period:
                x = np.arange(period)
                try:
                    slope, intercept = np.polyfit(x, window, 1)
                    result[i] = intercept + slope * (period - 1)
                except np.linalg.LinAlgError:
                    pass

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

***

## LINEARREG\_SLOPE

Linear Regression Slope indicator.

Example:
op = LINEARREG\_SLOPE(Input("FIELD:close", timeframe="1m", lookback=14), period=14)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `14`     |             |

### Usage

```python theme={null}
op = LINEARREG_SLOPE(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)) and len(window) == period:
                x = np.arange(period)
                try:
                    slope, _ = np.polyfit(x, window, 1)
                    result[i] = slope
                except np.linalg.LinAlgError:
                    pass

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

***

## LINEARREG\_INTERCEPT

Linear Regression Intercept indicator.

Example:
op = LINEARREG\_INTERCEPT(Input("FIELD:close", timeframe="1m", lookback=14), period=14)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `14`     |             |

### Usage

```python theme={null}
op = LINEARREG_INTERCEPT(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)) and len(window) == period:
                x = np.arange(period)
                try:
                    _, intercept = np.polyfit(x, window, 1)
                    result[i] = intercept
                except np.linalg.LinAlgError:
                    pass

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

***

## LINEARREG\_ANGLE

Linear Regression Angle indicator (in degrees).

Example:
op = LINEARREG\_ANGLE(Input("FIELD:close", timeframe="1m", lookback=14), period=14)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `14`     |             |

### Usage

```python theme={null}
op = LINEARREG_ANGLE(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)) and len(window) == period:
                x = np.arange(period)
                try:
                    slope, _ = np.polyfit(x, window, 1)
                    result[i] = np.degrees(np.arctan(slope))
                except np.linalg.LinAlgError:
                    pass

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

***

## TSF

Time Series Forecast indicator.

TSF = LINEARREG + LINEARREG\_SLOPE

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description     |
| --------- | --------- | -------- | --------------- |
| `input`   | `'Input'` | Required | Input signal    |
| `period`  | `int`     | `14`     | Lookback period |

### Usage

```python theme={null}
op = TSF(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)) and len(window) == period:
                x = np.arange(period)
                try:
                    slope, intercept = np.polyfit(x, window, 1)
                    result[i] = intercept + slope * period
                except np.linalg.LinAlgError:
                    pass

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

***

## CORREL

Pearson's Correlation Coefficient indicator.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description         |
| --------- | --------- | -------- | ------------------- |
| `input1`  | `'Input'` | Required | First input signal  |
| `input2`  | `'Input'` | Required | Second input signal |
| `period`  | `int`     | `30`     | Lookback period     |

### Usage

```python theme={null}
op = CORREL(
    Input("FIELD:close", timeframe="1m", lookback=30),
    Input("FIELD:volume", timeframe="1m", lookback=30),
    period=30,
)
```

### 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:
        return self._empty_output()

    data1, data2 = data[0], data[1]

    n_time = len(data1)
    if n_time == 0:
        return self._empty_output()

    last = data1[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    values1 = np.array([data1[t].value for t in range(n_time)])
    values2 = np.array([data2[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            w1 = values1[-period:, i]
            w2 = values2[-period:, i]
            if not np.any(np.isnan(w1)) and not np.any(np.isnan(w2)):
                result[i] = np.corrcoef(w1, w2)[0, 1]

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

***

## BETA

Beta indicator - measures volatility relative to market.

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description     |
| --------- | --------- | -------- | --------------- |
| `input1`  | `'Input'` | Required | Asset returns   |
| `input2`  | `'Input'` | Required | Market returns  |
| `period`  | `int`     | `5`      | Lookback period |

### Usage

```python theme={null}
op = BETA(
    Input("FIELD:close", timeframe="1m", lookback=30),
    Input("FIELD:volume", timeframe="1m", lookback=30),
    period=30,
)
```

### 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:
        return self._empty_output()

    data1, data2 = data[0], data[1]

    n_time = len(data1)
    if n_time == 0:
        return self._empty_output()

    last = data1[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    values1 = np.array([data1[t].value for t in range(n_time)])
    values2 = np.array([data2[t].value for t in range(n_time)])

    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            w1 = values1[-period:, i]
            w2 = values2[-period:, i]
            if not np.any(np.isnan(w1)) and not np.any(np.isnan(w2)):
                var_market = np.var(w2)
                if var_market > 0:
                    cov = np.cov(w1, w2)[0, 1]
                    result[i] = cov / var_market

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

***

## SUM

Summation over a period.

Example:
op = SUM(Input("FIELD:close", timeframe="1m", lookback=14), period=14)

**Role**: `INDICATOR` | **Ephemeral**: No

### Parameters

| Parameter | Type      | Default  | Description |
| --------- | --------- | -------- | ----------- |
| `input`   | `'Input'` | Required |             |
| `period`  | `int`     | `30`     |             |

### Usage

```python theme={null}
op = SUM(Input("FIELD:close", timeframe="1m", lookback=14), period=14)
```

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

    n_time = len(data)
    if n_time == 0:
        return self._empty_output()

    last = data[-1]
    n_symbols = len(last.exists)
    exists = last.exists
    valid = last.valid

    all_values = np.array([data[t].value for t in range(n_time)])
    result = np.full(n_symbols, np.nan)
    period = min(self._period, n_time)

    for i in range(n_symbols):
        if exists[i] and valid[i]:
            window = all_values[-period:, i]
            if not np.any(np.isnan(window)):
                result[i] = np.sum(window)

    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/indicator/stats.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>