Skip to content

Plus Di

PlusDI

Bases: IndicatorBase

Positive Directional Indicator (+DI).

Computes +DI using Wilder's smoothing method. One scalar value in the range 0-100 is produced per incoming bar and stored per symbol.

The rolling state is maintained independently for each symbol. Until enough bars are received to compute the initial smoothed values, the indicator yields numpy.nan.

name property

Canonical indicator name.

Returns:

Type Description
str

Stable identifier encoding all configuration parameters.

Source code in src/onesecondtrader/indicators/wilders/plus_di.py
def name(self) -> str:
    """
    Canonical indicator name.

    Returns:
        Stable identifier encoding all configuration parameters.
    """
    return f"PLUS_DI_{self.period}"

__init__(period=14, max_history=100, plot_at=1, plot_as=models.PlotStyle.LINE, plot_color=models.PlotColor.GREEN)

Parameters:

Name Type Description Default
period int

Lookback period for the +DI calculation.

14
max_history int

Maximum number of computed indicator values retained per symbol.

100
plot_at int

Opaque plotting identifier forwarded to the charting backend.

1
plot_as PlotStyle

Visual style used to render the indicator.

LINE
plot_color PlotColor

Color used to render the indicator.

GREEN
Source code in src/onesecondtrader/indicators/wilders/plus_di.py
def __init__(
    self,
    period: int = 14,
    max_history: int = 100,
    plot_at: int = 1,
    plot_as: models.PlotStyle = models.PlotStyle.LINE,
    plot_color: models.PlotColor = models.PlotColor.GREEN,
) -> None:
    """
    Parameters:
        period:
            Lookback period for the +DI calculation.
        max_history:
            Maximum number of computed indicator values retained per symbol.
        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.period: int = max(1, int(period))

    self._prev_high: dict[str, float] = {}
    self._prev_low: dict[str, float] = {}
    self._prev_close: dict[str, float] = {}
    self._count: dict[str, int] = {}
    self._plus_dm_sum: dict[str, float] = {}
    self._tr_sum: dict[str, float] = {}
    self._smoothed_plus_dm: dict[str, float] = {}
    self._smoothed_tr: dict[str, float] = {}

_compute_indicator(incoming_bar)

Compute the +DI value 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

+DI value (0-100), or numpy.nan if not enough data is available.

Source code in src/onesecondtrader/indicators/wilders/plus_di.py
def _compute_indicator(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Compute the +DI value for a single received bar.

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

    Returns:
        +DI value (0-100), or `numpy.nan` if not enough data is available.
    """
    symbol = incoming_bar.symbol
    high = incoming_bar.high
    low = incoming_bar.low
    close = incoming_bar.close

    if symbol not in self._prev_close:
        self._prev_high[symbol] = high
        self._prev_low[symbol] = low
        self._prev_close[symbol] = close
        self._count[symbol] = 0
        self._plus_dm_sum[symbol] = 0.0
        self._tr_sum[symbol] = 0.0
        self._smoothed_plus_dm[symbol] = 0.0
        self._smoothed_tr[symbol] = 0.0
        return np.nan

    tr = _true_range(high, low, self._prev_close[symbol])
    plus_dm, _ = _directional_movement(
        high, low, self._prev_high[symbol], self._prev_low[symbol]
    )

    self._prev_high[symbol] = high
    self._prev_low[symbol] = low
    self._prev_close[symbol] = close

    self._count[symbol] += 1
    count = self._count[symbol]

    if count < self.period:
        self._plus_dm_sum[symbol] += plus_dm
        self._tr_sum[symbol] += tr
        return np.nan

    if count == self.period:
        self._smoothed_plus_dm[symbol] = self._plus_dm_sum[symbol] + plus_dm
        self._smoothed_tr[symbol] = self._tr_sum[symbol] + tr
    else:
        self._smoothed_plus_dm[symbol] = (
            self._smoothed_plus_dm[symbol]
            - self._smoothed_plus_dm[symbol] / self.period
            + plus_dm
        )
        self._smoothed_tr[symbol] = (
            self._smoothed_tr[symbol] - self._smoothed_tr[symbol] / self.period + tr
        )

    smoothed_tr = self._smoothed_tr[symbol]
    if smoothed_tr <= 1e-12:
        return 0.0

    return 100.0 * self._smoothed_plus_dm[symbol] / smoothed_tr

_true_range(high, low, prev_close)

Source code in src/onesecondtrader/indicators/wilders/plus_di.py
def _true_range(high: float, low: float, prev_close: float) -> float:
    return max(high - low, abs(high - prev_close), abs(low - prev_close))

_directional_movement(high, low, prev_high, prev_low)

Source code in src/onesecondtrader/indicators/wilders/plus_di.py
def _directional_movement(
    high: float, low: float, prev_high: float, prev_low: float
) -> tuple[float, float]:
    up_move = high - prev_high
    down_move = prev_low - low
    plus_dm = up_move if up_move > down_move and up_move > 0.0 else 0.0
    minus_dm = down_move if down_move > up_move and down_move > 0.0 else 0.0
    return plus_dm, minus_dm