Detrend

DetrendOscillator

Bases: IndicatorBase

Detrend Oscillator indicator.

Measures the spread between two smoothed price estimates (short and long horizon moving averages): Detrend(t) = SMA_short(t) - SMA_long(t)

This construction removes the slow-moving trend component and leaves a zero-centered oscillation.

The rolling windows are maintained independently for each symbol. Until both windows are fully populated, the indicator yields numpy.nan.

name property

Canonical indicator name.

Returns:

Type	Description
str	Stable identifier for the Detrend Oscillator.

Source code in src/onesecondtrader/indicators/oscillators/detrend.py

def name(self) -> str:
    """
    Canonical indicator name.

    Returns:
        Stable identifier for the Detrend Oscillator.
    """
    return f"DETREND({self.short_period},{self.long_period})"

__init__(short_period=3, long_period=7, max_history=100, bar_field=models.BarField.CLOSE, plot_at=99, plot_as=models.PlotStyle.LINE, plot_color=models.PlotColor.BLACK)

Parameters:

Name	Type	Description	Default
short_period	int	Window size for the short-horizon moving average.	3
long_period	int	Window size for the long-horizon moving average.	7
max_history	int	Maximum number of computed indicator values retained per symbol.	100
bar_field	BarField	Bar field used as the input series.	CLOSE
plot_at	int	Opaque plotting identifier forwarded to the charting backend.	99
plot_as	PlotStyle	Visual style used to render the indicator.	LINE
plot_color	PlotColor	Color used to render the indicator.	BLACK

Source code in src/onesecondtrader/indicators/oscillators/detrend.py

def __init__(
    self,
    short_period: int = 3,
    long_period: int = 7,
    max_history: int = 100,
    bar_field: models.BarField = models.BarField.CLOSE,
    plot_at: int = 99,
    plot_as: models.PlotStyle = models.PlotStyle.LINE,
    plot_color: models.PlotColor = models.PlotColor.BLACK,
) -> None:
    """
    Parameters:
        short_period:
            Window size for the short-horizon moving average.
        long_period:
            Window size for the long-horizon moving average.
        max_history:
            Maximum number of computed indicator values retained per symbol.
        bar_field:
            Bar field used as the input series.
        plot_at:
            Opaque plotting identifier forwarded to the charting backend.
        plot_as:
            Visual style used to render the indicator.
        plot_color:
            Color used to render the indicator.
    """
    super().__init__(
        max_history=max_history,
        plot_at=plot_at,
        plot_as=plot_as,
        plot_color=plot_color,
    )

    self.short_period: int = max(1, int(short_period))
    self.long_period: int = max(1, int(long_period))
    self.bar_field: models.BarField = bar_field
    self._short_window: dict[str, collections.deque[float]] = {}
    self._long_window: dict[str, collections.deque[float]] = {}

_compute_indicator(incoming_bar)

Compute the detrend oscillator for a single received bar.

Parameters:

Name	Type	Description	Default
incoming_bar	BarReceived	Market bar used as input for the computation.	required

Returns:

Type	Description
float	Detrend oscillator value, or numpy.nan if the rolling windows are not yet fully populated.

Source code in src/onesecondtrader/indicators/oscillators/detrend.py

def _compute_indicator(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Compute the detrend oscillator for a single received bar.

    Parameters:
        incoming_bar:
            Market bar used as input for the computation.

    Returns:
        Detrend oscillator value, or `numpy.nan` if the rolling windows are not yet fully populated.
    """
    symbol = incoming_bar.symbol
    if symbol not in self._short_window:
        self._short_window[symbol] = collections.deque(maxlen=self.short_period)
    if symbol not in self._long_window:
        self._long_window[symbol] = collections.deque(maxlen=self.long_period)

    value = self._extract_field(incoming_bar)
    self._short_window[symbol].append(value)
    self._long_window[symbol].append(value)

    if len(self._short_window[symbol]) < self.short_period:
        return np.nan
    if len(self._long_window[symbol]) < self.long_period:
        return np.nan

    sma_short = sum(self._short_window[symbol]) / self.short_period
    sma_long = sum(self._long_window[symbol]) / self.long_period

    return sma_short - sma_long

_extract_field(incoming_bar)

Extract the configured bar field from an incoming bar.

Parameters:

Name	Type	Description	Default
incoming_bar	BarReceived	Market bar providing the input data.	required

Returns:

Type	Description
float	Extracted field value, or numpy.nan if unavailable.

Source code in src/onesecondtrader/indicators/oscillators/detrend.py

def _extract_field(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Extract the configured bar field from an incoming bar.

    Parameters:
        incoming_bar:
            Market bar providing the input data.

    Returns:
        Extracted field value, or `numpy.nan` if unavailable.
    """
    match self.bar_field:
        case models.BarField.OPEN:
            return incoming_bar.open
        case models.BarField.HIGH:
            return incoming_bar.high
        case models.BarField.LOW:
            return incoming_bar.low
        case models.BarField.CLOSE:
            return incoming_bar.close
        case models.BarField.VOLUME:
            return (
                float(incoming_bar.volume)
                if incoming_bar.volume is not None
                else np.nan
            )
        case _:
            return incoming_bar.close