nautilus_indicators/momentum/
kvo.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::fmt::{Debug, Display};
17
18use nautilus_core::correctness::FAILED;
19use nautilus_model::data::Bar;
20
21use crate::{
22    average::{MovingAverageFactory, MovingAverageType},
23    indicator::{Indicator, MovingAverage},
24    support::{MAX_PERIOD, is_valid_hlc},
25};
26
27/// Stephen J. Klinger's Volume Oscillator: a fast/slow moving-average
28/// difference over the per-bar "volume force".
29///
30/// ```text
31/// dm_t   = high_t - low_t                           (the daily measurement)
32/// trend  = sign(hlc_t - hlc_{t-1}), carried over when equal, with hlc = high + low + close
33/// cm_t   = cm_{t-1} + dm_t        while the trend holds
34/// cm_t   = dm_{t-1} + dm_t        when the trend flips
35/// vf_t   = volume_t * |2 * (dm_t / cm_t - 1)| * trend * 100
36/// KVO_t  = MA(vf, fast) - MA(vf, slow)
37/// ```
38///
39/// Klinger's textbook configuration is `fast = 34, slow = 55` with exponential
40/// averages, which is the default `ma_type`.
41#[repr(C)]
42#[derive(Debug)]
43#[cfg_attr(
44    feature = "python",
45    pyo3::pyclass(module = "nautilus_trader.indicators", unsendable)
46)]
47#[cfg_attr(
48    feature = "python",
49    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.indicators")
50)]
51pub struct KlingerVolumeOscillator {
52    pub fast_period: usize,
53    pub slow_period: usize,
54    pub ma_type: MovingAverageType,
55    pub value: f64,
56    pub initialized: bool,
57    has_inputs: bool,
58    fast_ma: Box<dyn MovingAverage + Send + 'static>,
59    slow_ma: Box<dyn MovingAverage + Send + 'static>,
60    previous_hlc: Option<f64>,
61    previous_dm: f64,
62    trend: i8,
63    cm: f64,
64}
65
66impl Display for KlingerVolumeOscillator {
67    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68        write!(
69            f,
70            "{}({},{},{})",
71            self.name(),
72            self.fast_period,
73            self.slow_period,
74            self.ma_type,
75        )
76    }
77}
78
79impl Indicator for KlingerVolumeOscillator {
80    fn name(&self) -> String {
81        stringify!(KlingerVolumeOscillator).to_string()
82    }
83
84    fn has_inputs(&self) -> bool {
85        self.has_inputs
86    }
87
88    fn initialized(&self) -> bool {
89        self.initialized
90    }
91
92    fn handle_bar(&mut self, bar: &Bar) {
93        self.update_raw(
94            (&bar.high).into(),
95            (&bar.low).into(),
96            (&bar.close).into(),
97            (&bar.volume).into(),
98        );
99    }
100
101    fn reset(&mut self) {
102        self.fast_ma.reset();
103        self.slow_ma.reset();
104        self.previous_hlc = None;
105        self.previous_dm = 0.0;
106        self.trend = 0;
107        self.cm = 0.0;
108        self.value = 0.0;
109        self.has_inputs = false;
110        self.initialized = false;
111    }
112}
113
114impl KlingerVolumeOscillator {
115    /// Creates a new [`KlingerVolumeOscillator`] instance.
116    ///
117    /// # Panics
118    ///
119    /// Panics if `fast_period` is zero or not less than `slow_period`.
120    #[must_use]
121    pub fn new(fast_period: usize, slow_period: usize, ma_type: Option<MovingAverageType>) -> Self {
122        Self::new_checked(fast_period, slow_period, ma_type).expect(FAILED)
123    }
124
125    pub(crate) fn new_checked(
126        fast_period: usize,
127        slow_period: usize,
128        ma_type: Option<MovingAverageType>,
129    ) -> anyhow::Result<Self> {
130        anyhow::ensure!(
131            fast_period <= MAX_PERIOD,
132            "fast_period must not exceed {MAX_PERIOD}"
133        );
134        anyhow::ensure!(
135            slow_period <= MAX_PERIOD,
136            "slow_period must not exceed {MAX_PERIOD}"
137        );
138
139        anyhow::ensure!(
140            fast_period > 0 && fast_period < slow_period,
141            "KlingerVolumeOscillator: fast_period must be > 0 and < slow_period (received fast {fast_period}, slow {slow_period})"
142        );
143        let ma_type = ma_type.unwrap_or(MovingAverageType::Exponential);
144
145        Ok(Self {
146            fast_period,
147            slow_period,
148            ma_type,
149            value: 0.0,
150            fast_ma: MovingAverageFactory::create(ma_type, fast_period),
151            slow_ma: MovingAverageFactory::create(ma_type, slow_period),
152            previous_hlc: None,
153            previous_dm: 0.0,
154            trend: 0,
155            cm: 0.0,
156            has_inputs: false,
157            initialized: false,
158        })
159    }
160
161    pub fn update_raw(&mut self, high: f64, low: f64, close: f64, volume: f64) {
162        if !is_valid_hlc(high, low, close) || !volume.is_finite() || volume < 0.0 {
163            return;
164        }
165
166        self.has_inputs = true;
167        let hlc = high + low + close;
168        let dm = high - low;
169
170        let Some(previous_hlc) = self.previous_hlc else {
171            // The first bar only establishes the previous trend sum and daily measurement
172            self.previous_hlc = Some(hlc);
173            self.previous_dm = dm;
174            return;
175        };
176
177        let new_trend: i8 = if hlc > previous_hlc {
178            1
179        } else if hlc < previous_hlc {
180            -1
181        } else {
182            self.trend
183        };
184
185        // The cumulative measurement resets to (previous_dm + dm) whenever the
186        // trend flips, and seeds the same way on the first sign read
187        if new_trend == self.trend && self.trend != 0 {
188            self.cm += dm;
189        } else {
190            self.cm = self.previous_dm + dm;
191        }
192        self.trend = new_trend;
193
194        let volume_force = if self.cm == 0.0 {
195            // Zero-range bars since the last trend flip: no force to register
196            0.0
197        } else {
198            volume * (2.0 * (dm / self.cm - 1.0)).abs() * f64::from(new_trend) * 100.0
199        };
200
201        self.previous_hlc = Some(hlc);
202        self.previous_dm = dm;
203
204        self.fast_ma.update_raw(volume_force);
205        self.slow_ma.update_raw(volume_force);
206        if !self.initialized && self.fast_ma.initialized() && self.slow_ma.initialized() {
207            self.initialized = true;
208        }
209
210        if self.initialized {
211            self.value = self.fast_ma.value() - self.slow_ma.value();
212        }
213    }
214}
215
216#[cfg(test)]
217mod tests {
218    use rstest::{fixture, rstest};
219
220    use super::*;
221
222    #[fixture]
223    fn kvo_34() -> KlingerVolumeOscillator {
224        KlingerVolumeOscillator::new(3, 4, Some(MovingAverageType::Simple))
225    }
226
227    #[rstest]
228    fn test_name_returns_expected_string(kvo_34: KlingerVolumeOscillator) {
229        assert_eq!(kvo_34.name(), "KlingerVolumeOscillator");
230    }
231
232    #[rstest]
233    fn test_str_repr_returns_expected_string(kvo_34: KlingerVolumeOscillator) {
234        assert_eq!(format!("{kvo_34}"), "KlingerVolumeOscillator(3,4,SIMPLE)");
235    }
236
237    #[rstest]
238    fn test_period_returns_expected_value(kvo_34: KlingerVolumeOscillator) {
239        assert_eq!(kvo_34.fast_period, 3);
240        assert_eq!(kvo_34.slow_period, 4);
241    }
242
243    #[rstest]
244    #[should_panic(expected = "fast_period must be > 0 and < slow_period")]
245    fn test_fast_not_less_than_slow_panics() {
246        let _ = KlingerVolumeOscillator::new(4, 4, None);
247    }
248
249    #[rstest]
250    fn test_initialized_without_inputs_returns_false(kvo_34: KlingerVolumeOscillator) {
251        assert!(!kvo_34.initialized());
252    }
253
254    #[rstest]
255    fn test_volume_force_matches_reference(mut kvo_34: KlingerVolumeOscillator) {
256        // (high, low, close) with hlc = high + low + close and dm = high - low:
257        // hlc = [5, 10, 11, 18, 13], dm = [1, 2, 1, 3, 2], volume 10.
258        // Bar 2: trend +1, cm = 1 + 2 = 3, vf = 10 * |2 * (2/3 - 1)| * 100 = 2000/3
259        // Bar 3: trend +1, cm = 3 + 1 = 4, vf = 10 * |2 * (1/4 - 1)| * 100 = 1500
260        // Bar 4: trend +1, cm = 4 + 3 = 7, vf = 10 * |2 * (3/7 - 1)| * 100 = 8000/7
261        // Bar 5: trend -1, cm = 3 + 2 = 5, vf = -10 * |2 * (2/5 - 1)| * 100 = -1200
262        let bars = [
263            (2.0, 1.0, 2.0),
264            (4.0, 2.0, 4.0),
265            (4.0, 3.0, 4.0),
266            (7.0, 4.0, 7.0),
267            (5.0, 3.0, 5.0),
268        ];
269
270        for (high, low, close) in bars {
271            kvo_34.update_raw(high, low, close, 10.0);
272        }
273        assert!(kvo_34.initialized());
274        let expected = (1500.0 + 8000.0 / 7.0 - 1200.0) / 3.0
275            - (2000.0 / 3.0 + 1500.0 + 8000.0 / 7.0 - 1200.0) / 4.0;
276        assert!((kvo_34.value - expected).abs() < 1e-9);
277        assert_eq!(kvo_34.trend, -1);
278        assert_eq!(kvo_34.cm, 5.0);
279    }
280
281    #[rstest]
282    fn test_flat_dm_carries_trend_forward(mut kvo_34: KlingerVolumeOscillator) {
283        kvo_34.update_raw(1.0, 1.0, 1.0, 10.0);
284        kvo_34.update_raw(2.0, 2.0, 2.0, 10.0);
285        let trend_after_up = kvo_34.trend;
286        kvo_34.update_raw(2.0, 2.0, 2.0, 10.0);
287        assert_eq!(kvo_34.trend, trend_after_up);
288    }
289
290    #[rstest]
291    fn test_reset_successfully_returns_indicator_to_fresh_state(
292        mut kvo_34: KlingerVolumeOscillator,
293    ) {
294        for (high, low, close, volume) in [
295            (12.0, 10.0, 11.0, 100.0),
296            (13.0, 11.0, 12.0, 200.0),
297            (15.0, 12.0, 14.0, 300.0),
298            (16.0, 13.0, 14.0, 400.0),
299            (17.0, 14.0, 16.0, 500.0),
300        ] {
301            kvo_34.update_raw(high, low, close, volume);
302        }
303        // An uptrend from bar 2 accumulates cm = 4, 7, 10, 13 against dm = 2, 3, 3, 3
304        let forces = [20_000.0, 240_000.0 / 7.0, 56_000.0, 1_000_000.0 / 13.0];
305        let fast = forces[1..].iter().sum::<f64>() / 3.0;
306        let slow = forces.iter().sum::<f64>() / 4.0;
307        assert!(kvo_34.initialized());
308        assert!((kvo_34.value - (fast - slow)).abs() < 1e-9);
309        assert_eq!(kvo_34.trend, 1);
310        assert_eq!(kvo_34.cm, 13.0);
311        assert_eq!(kvo_34.previous_hlc, Some(47.0));
312        assert_eq!(kvo_34.previous_dm, 3.0);
313
314        kvo_34.reset();
315
316        assert!(!kvo_34.initialized());
317        assert_eq!(kvo_34.value, 0.0);
318        assert_eq!(kvo_34.trend, 0);
319        assert_eq!(kvo_34.cm, 0.0);
320        assert!(kvo_34.previous_hlc.is_none());
321        assert_eq!(kvo_34.previous_dm, 0.0);
322    }
323}
nautilus_indicators/momentum/kvo.rs

nautilus_indicators/momentum/
kvo.rs
