Statistics and Math

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

op = HT_DCPERIOD(Input("FIELD:close", timeframe="1m", lookback=32))

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]

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

op = HT_DCPHASE(Input("FIELD:close", timeframe="1m", lookback=32))

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]

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

op = HT_TRENDLINE(Input("FIELD:close", timeframe="1m", lookback=32))

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]

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

op = HT_TRENDMODE(Input("FIELD:close", timeframe="1m", lookback=32))

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]

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

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.

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

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.

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

_{Source: apps/trading/operators/indicator/hilbert.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/minmax.py}

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
`high`	`'Input'`	Required
`low`	`'Input'`	Required
`period`	`int`	`14`

Usage

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.

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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.

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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.

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

_{Source: apps/trading/operators/indicator/minmax.py}

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

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.

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

_{Source: apps/trading/operators/indicator/price.py}

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

op = MEDPRICE(
    Input("FIELD:high", timeframe="1m", lookback=1),
    Input("FIELD:low", timeframe="1m", lookback=1),
)

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

_{Source: apps/trading/operators/indicator/price.py}

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

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.

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

_{Source: apps/trading/operators/indicator/price.py}

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

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.

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

_{Source: apps/trading/operators/indicator/price.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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.

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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.

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

_{Source: apps/trading/operators/indicator/stats.py}

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

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

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]

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

_{Source: apps/trading/operators/indicator/stats.py}

Operator Protocol

How operators implement the compute() interface

StatefulGraph

How operators compose into a DAG

​Overview

​Quick Reference

​HT_DCPERIOD

​Parameters

​Usage

​Source Code

​HT_DCPHASE

​Parameters

​Usage

​Source Code

​HT_TRENDLINE

​Parameters

​Usage

​Source Code

​HT_TRENDMODE

​Parameters

​Usage

​Source Code

​HT_SINE

​Parameters

​Usage

​Source Code

​HT_PHASOR

​Parameters

​Usage

​Source Code

​MAX

​Parameters

​Usage

​Source Code

​MAXINDEX

​Parameters

​Usage

​Source Code

​MIN

​Parameters

​Usage

​Source Code

​MININDEX

​Parameters

​Usage

​Source Code

​MIDPOINT

​Parameters

​Usage

​Source Code

​MIDPRICE

​Parameters

​Usage

​Source Code

​MINMAX

​Parameters

​Usage

​Source Code

​MINMAXINDEX

​Parameters

​Usage

​Source Code

​AVGPRICE

​Parameters

​Usage

​Source Code

​MEDPRICE

​Parameters

​Usage

​Source Code

​TYPPRICE

​Parameters

​Usage

​Source Code

​WCLPRICE

​Parameters

​Usage

​Source Code

​LINEARREG

​Parameters

​Usage

​Source Code

​LINEARREG_SLOPE

​Parameters

Overview

Quick Reference

HT_DCPERIOD

Parameters

Usage

Source Code

HT_DCPHASE

Parameters

Usage

Source Code

HT_TRENDLINE

Parameters

Usage

Source Code

HT_TRENDMODE

Parameters

Usage

Source Code

HT_SINE

Parameters

Usage

Source Code

HT_PHASOR

Parameters

Usage

Source Code

MAX

Parameters

Usage

Source Code

MAXINDEX

Parameters

Usage

Source Code

MIN

Parameters

Usage

Source Code

MININDEX

Parameters

Usage

Source Code

MIDPOINT

Parameters

Usage

Source Code

MIDPRICE

Parameters

Usage

Source Code

MINMAX

Parameters

Usage

Source Code

MINMAXINDEX

Parameters

Usage

Source Code

AVGPRICE

Parameters

Usage

Source Code

MEDPRICE

Parameters

Usage

Source Code

TYPPRICE

Parameters

Usage

Source Code

WCLPRICE

Parameters

Usage

Source Code

LINEARREG

Parameters

Usage

Source Code

LINEARREG_SLOPE

Parameters