Skip to content

Parabolic Sar

_SARState dataclass

Per-symbol mutable state for the Parabolic SAR computation.

Source code in src/onesecondtrader/indicators/wilders/parabolic_sar.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@dataclasses.dataclass
class _SARState:
    """Per-symbol mutable state for the Parabolic SAR computation."""

    count: int = 0
    is_long: bool = True
    sar: float = 0.0
    ep: float = 0.0
    af: float = 0.0
    high_buffer: collections.deque[float] = dataclasses.field(
        default_factory=lambda: collections.deque(maxlen=2)
    )
    low_buffer: collections.deque[float] = dataclasses.field(
        default_factory=lambda: collections.deque(maxlen=2)
    )

ParabolicSAR

Bases: IndicatorBase

Parabolic Stop and Reverse (SAR) indicator.

Computes the Parabolic SAR as described by J. Welles Wilder Jr. One scalar value is produced per incoming bar and stored per symbol.

The rolling state is maintained independently for each symbol. Until enough bars are received to initialise, 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/parabolic_sar.py 
66
67
68
69
70
71
72
73
def name(self) -> str:
    """
    Canonical indicator name.

    Returns:
        Stable identifier encoding all configuration parameters.
    """
    return f"PSAR_{self.af_start}_{self.af_step}_{self.af_max}"

__init__(af_start=0.02, af_step=0.02, af_max=0.2, max_history=100)

Parameters:

Name Type Description Default
af_start float

Initial acceleration factor.

 0.02
af_step float

Acceleration factor increment when a new extreme point is reached.

 0.02
af_max float

Maximum acceleration factor.

 0.2
max_history int

Maximum number of computed indicator values retained per symbol.

 100
Source code in src/onesecondtrader/indicators/wilders/parabolic_sar.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    af_start: float = 0.02,
    af_step: float = 0.02,
    af_max: float = 0.20,
    max_history: int = 100,
) -> None:
    """
    Parameters:
        af_start:
            Initial acceleration factor.
        af_step:
            Acceleration factor increment when a new extreme point is reached.
        af_max:
            Maximum acceleration factor.
        max_history:
            Maximum number of computed indicator values retained per symbol.
    """
    super().__init__(max_history=max_history)

    self.af_start: float = af_start
    self.af_step: float = af_step
    self.af_max: float = af_max

    self._state: dict[str, _SARState] = {}

_compute_indicator(incoming_bar)

Compute the Parabolic SAR 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

SAR value, or numpy.nan if not enough data is available.
Source code in src/onesecondtrader/indicators/wilders/parabolic_sar.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def _compute_indicator(self, incoming_bar: events.market.BarReceived) -> float:
    """
    Compute the Parabolic SAR value for a single received bar.

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

    Returns:
        SAR value, or `numpy.nan` if not enough data is available.
    """
    symbol = incoming_bar.symbol
    high = incoming_bar.high
    low = incoming_bar.low

    if symbol not in self._state:
        self._state[symbol] = _SARState()

    s = self._state[symbol]
    s.count += 1

    if s.count == 1:
        s.high_buffer.append(high)
        s.low_buffer.append(low)
        return np.nan

    if s.count == 2:
        prev_high = s.high_buffer[-1]
        prev_low = s.low_buffer[-1]
        s.high_buffer.append(high)
        s.low_buffer.append(low)

        if high >= prev_high:
            s.is_long = True
            s.sar = prev_low
            s.ep = high
        else:
            s.is_long = False
            s.sar = prev_high
            s.ep = low

        s.af = self.af_start
        return s.sar

    # Advance SAR
    sar = s.sar + s.af * (s.ep - s.sar)

    # Clamp SAR against prior two bars' extremes
    if s.is_long:
        sar = min(sar, min(s.low_buffer))
        if low < sar:
            # Reversal to short
            s.is_long = False
            sar = s.ep
            s.ep = low
            s.af = self.af_start
        else:
            if high > s.ep:
                s.ep = high
                s.af = min(s.af + self.af_step, self.af_max)
    else:
        sar = max(sar, max(s.high_buffer))
        if high > sar:
            # Reversal to long
            s.is_long = True
            sar = s.ep
            s.ep = high
            s.af = self.af_start
        else:
            if low < s.ep:
                s.ep = low
                s.af = min(s.af + self.af_step, self.af_max)

    s.sar = sar
    s.high_buffer.append(high)
    s.low_buffer.append(low)

    return sar