nautilus_indicators/volatility/
kc.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, typical_price},
25    volatility::atr::AverageTrueRange,
26};
27
28/// Keltner channel.
29#[repr(C)]
30#[derive(Debug)]
31#[cfg_attr(
32    feature = "python",
33    pyo3::pyclass(module = "nautilus_trader.indicators", unsendable)
34)]
35#[cfg_attr(
36    feature = "python",
37    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.indicators")
38)]
39pub struct KeltnerChannel {
40    pub period: usize,
41    pub atr_period: usize,
42    pub k_multiplier: f64,
43    pub ma_type: MovingAverageType,
44    pub ma_type_atr: MovingAverageType,
45    pub use_previous: bool,
46    pub atr_floor: f64,
47    pub upper: f64,
48    pub middle: f64,
49    pub lower: f64,
50    pub initialized: bool,
51    has_inputs: bool,
52    ma: Box<dyn MovingAverage + Send + 'static>,
53    atr: AverageTrueRange,
54}
55
56impl Display for KeltnerChannel {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        write!(f, "{}({})", self.name(), self.period)
59    }
60}
61
62impl Indicator for KeltnerChannel {
63    fn name(&self) -> String {
64        stringify!(KeltnerChannel).to_string()
65    }
66
67    fn has_inputs(&self) -> bool {
68        self.has_inputs
69    }
70
71    fn initialized(&self) -> bool {
72        self.initialized
73    }
74
75    fn handle_bar(&mut self, bar: &Bar) {
76        self.update_raw((&bar.high).into(), (&bar.low).into(), (&bar.close).into());
77    }
78
79    fn reset(&mut self) {
80        self.ma.reset();
81        self.atr.reset();
82        self.upper = 0.0;
83        self.middle = 0.0;
84        self.lower = 0.0;
85        self.has_inputs = false;
86        self.initialized = false;
87    }
88}
89
90impl KeltnerChannel {
91    /// Creates a new [`KeltnerChannel`] instance.
92    ///
93    /// # Panics
94    ///
95    /// Panics if either period is outside `1..=MAX_PERIOD`, the multiplier is not positive and finite,
96    /// or the ATR floor is negative or non-finite.
97    ///
98    /// The defaults are the standard Keltner convention: an EMA centerline
99    /// over the typical price and a Wilder smoothed ATR. `atr_period`
100    /// defaults to `period` when not given.
101    #[must_use]
102    pub fn new(
103        period: usize,
104        k_multiplier: f64,
105        atr_period: Option<usize>,
106        ma_type: Option<MovingAverageType>,
107        ma_type_atr: Option<MovingAverageType>,
108        use_previous: Option<bool>,
109        atr_floor: Option<f64>,
110    ) -> Self {
111        Self::new_checked(
112            period,
113            k_multiplier,
114            atr_period,
115            ma_type,
116            ma_type_atr,
117            use_previous,
118            atr_floor,
119        )
120        .expect(FAILED)
121    }
122
123    pub(crate) fn new_checked(
124        period: usize,
125        k_multiplier: f64,
126        atr_period: Option<usize>,
127        ma_type: Option<MovingAverageType>,
128        ma_type_atr: Option<MovingAverageType>,
129        use_previous: Option<bool>,
130        atr_floor: Option<f64>,
131    ) -> anyhow::Result<Self> {
132        anyhow::ensure!(period <= MAX_PERIOD, "period cannot exceed {MAX_PERIOD}");
133        let atr_period = atr_period.unwrap_or(period);
134        anyhow::ensure!(period > 0, "period must be positive");
135        anyhow::ensure!(
136            (1..=MAX_PERIOD).contains(&atr_period),
137            "atr_period must be in 1..={MAX_PERIOD}"
138        );
139        anyhow::ensure!(
140            k_multiplier.is_finite() && k_multiplier > 0.0,
141            "k_multiplier must be finite and positive"
142        );
143        let atr = AverageTrueRange::new_checked(atr_period, ma_type_atr, use_previous, atr_floor)?;
144        Ok(Self {
145            period,
146            atr_period,
147            k_multiplier,
148            ma_type: ma_type.unwrap_or(MovingAverageType::Exponential),
149            ma_type_atr: ma_type_atr.unwrap_or(MovingAverageType::Wilder),
150            use_previous: use_previous.unwrap_or(true),
151            atr_floor: atr_floor.unwrap_or(0.0),
152            upper: 0.0,
153            middle: 0.0,
154            lower: 0.0,
155            has_inputs: false,
156            initialized: false,
157            ma: MovingAverageFactory::create(
158                ma_type.unwrap_or(MovingAverageType::Exponential),
159                period,
160            ),
161            atr,
162        })
163    }
164
165    pub fn update_raw(&mut self, high: f64, low: f64, close: f64) {
166        let count = self.atr.count;
167        self.atr.update_raw(high, low, close);
168        if self.atr.count == count {
169            return;
170        }
171        self.ma.update_raw(typical_price(high, low, close));
172        self.has_inputs = true;
173
174        if !self.ma.initialized() || !self.atr.initialized {
175            return;
176        }
177
178        self.upper = self.atr.value.mul_add(self.k_multiplier, self.ma.value());
179        self.middle = self.ma.value();
180        self.lower = self.atr.value.mul_add(-self.k_multiplier, self.ma.value());
181
182        // Bands are only meaningful once both the centerline MA and the ATR are warm
183        if !self.initialized {
184            self.has_inputs = true;
185
186            if self.ma.initialized() && self.atr.initialized {
187                self.initialized = true;
188            }
189        }
190    }
191}
192
193#[cfg(test)]
194mod tests {
195    use rstest::rstest;
196
197    use super::*;
198    use crate::{stubs::kc_10, testing::assert_approx_equal};
199
200    #[rstest]
201    fn test_name_returns_expected_string(kc_10: KeltnerChannel) {
202        assert_eq!(kc_10.name(), "KeltnerChannel");
203    }
204
205    #[rstest]
206    fn test_str_repr_returns_expected_string(kc_10: KeltnerChannel) {
207        assert_eq!(format!("{kc_10}"), "KeltnerChannel(10)");
208    }
209
210    #[rstest]
211    fn test_period_returns_expected_value(kc_10: KeltnerChannel) {
212        assert_eq!(kc_10.period, 10);
213        assert_eq!(kc_10.k_multiplier, 2.0);
214    }
215
216    #[rstest]
217    fn test_initialized_without_inputs_returns_false(kc_10: KeltnerChannel) {
218        assert!(!kc_10.initialized());
219    }
220
221    #[rstest]
222    fn test_value_with_all_higher_inputs_returns_expected_value(mut kc_10: KeltnerChannel) {
223        let high_values = [
224            1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0,
225        ];
226        let low_values = [
227            0.9, 1.9, 2.9, 3.9, 4.9, 5.9, 6.9, 7.9, 8.9, 9.9, 10.1, 10.2, 10.3, 11.1, 11.4,
228        ];
229
230        let close_values = [
231            0.95, 1.95, 2.95, 3.95, 4.95, 5.95, 6.95, 7.95, 8.95, 9.95, 10.5, 11.0, 11.5, 12.5,
232            13.0,
233        ];
234
235        for i in 0..15 {
236            kc_10.update_raw(high_values[i], low_values[i], close_values[i]);
237        }
238
239        assert!(kc_10.initialized());
240        let middle = (5..15)
241            .map(|i| (high_values[i] + low_values[i] + close_values[i]) / 3.0)
242            .sum::<f64>()
243            / 10.0;
244        let atr = (5..15)
245            .map(|i| {
246                f64::max(high_values[i], close_values[i - 1])
247                    - f64::min(low_values[i], close_values[i - 1])
248            })
249            .sum::<f64>()
250            / 10.0;
251        assert_approx_equal(kc_10.upper, middle + 2.0 * atr);
252        assert_approx_equal(kc_10.middle, middle);
253        assert_approx_equal(kc_10.lower, middle - 2.0 * atr);
254    }
255
256    #[rstest]
257    fn test_reset_successfully_returns_indicator_to_fresh_state(mut kc_10: KeltnerChannel) {
258        for _ in 0..10 {
259            kc_10.update_raw(11.0, 9.0, 10.0);
260        }
261        // Typical price 10 and true range 2 give bands at 10 -/+ 2 * 2
262        assert!(kc_10.initialized());
263        assert_eq!(kc_10.upper, 14.0);
264        assert_eq!(kc_10.middle, 10.0);
265        assert_eq!(kc_10.lower, 6.0);
266
267        kc_10.reset();
268
269        assert!(!kc_10.initialized());
270        assert!(!kc_10.has_inputs);
271        assert_eq!(kc_10.upper, 0.0);
272        assert_eq!(kc_10.middle, 0.0);
273        assert_eq!(kc_10.lower, 0.0);
274    }
275}
nautilus_indicators/volatility/kc.rs

nautilus_indicators/volatility/
kc.rs
