nautilus_trading/examples/strategies/hurst_vpin_directional/
strategy.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
16//! Hurst/VPIN directional strategy implementation.
17
18use std::{collections::VecDeque, fmt::Debug};
19
20use ahash::AHashSet;
21use nautilus_common::{
22    actor::DataActor,
23    config::{ConfigError, ConfigResult},
24};
25use nautilus_core::{DurationNanos, UnixNanos};
26use nautilus_model::{
27    data::{Bar, QuoteTick, TradeTick},
28    enums::{AggressorSide, OrderSide, PositionSide, TimeInForce},
29    events::{
30        OrderCanceled, OrderDenied, OrderExpired, OrderFilled, OrderRejected, PositionClosed,
31        PositionOpened,
32    },
33    identifiers::{ClientOrderId, PositionId},
34    orders::{Order, OrderCore},
35    types::Quantity,
36};
37
38use super::config::HurstVpinDirectionalConfig;
39use crate::{
40    nautilus_strategy,
41    strategy::{Strategy, StrategyCore},
42};
43
44/// Directional strategy combining a Hurst-exponent regime filter on dollar bars
45/// with a VPIN (Volume-synchronized Probability of Informed Trading) signal
46/// derived from trade aggressor flow, with entry timing gated by the live
47/// quote stream.
48///
49/// The strategy is sampled on information-driven (value) bars rather than
50/// clock time, following Lopez de Prado (*Advances in Financial Machine
51/// Learning*, Chapter 2). The Hurst exponent is estimated by rescaled range
52/// over the window of dollar bar log returns. VPIN is averaged over completed
53/// volume buckets, with a signed variant carrying the net informed direction.
54pub struct HurstVpinDirectional {
55    pub(super) core: StrategyCore,
56    pub(super) config: HurstVpinDirectionalConfig,
57    pub(super) returns: VecDeque<f64>,
58    pub(super) abs_imbalances: VecDeque<f64>,
59    pub(super) signed_imbalances: VecDeque<f64>,
60    pub(super) last_close: Option<f64>,
61    pub(super) bucket_buy_volume: f64,
62    pub(super) bucket_sell_volume: f64,
63    pub(super) hurst: Option<f64>,
64    pub(super) vpin: Option<f64>,
65    pub(super) signed_vpin: Option<f64>,
66    pub(super) position_opened_ns: Option<UnixNanos>,
67    pub(super) exit_cooldown: bool,
68    pub(super) entry_order_id: Option<ClientOrderId>,
69    pub(super) exit_order_ids: AHashSet<ClientOrderId>,
70}
71
72impl HurstVpinDirectional {
73    /// Creates a new [`HurstVpinDirectional`] instance from config with validation.
74    ///
75    /// # Errors
76    ///
77    /// Returns a [`ConfigError`] if either rolling window is outside `[1, 16_384]` or the base
78    /// strategy configuration is invalid.
79    pub fn new_checked(config: HurstVpinDirectionalConfig) -> ConfigResult<Self> {
80        config.validate()?;
81        let hurst_window = config.hurst_window;
82        let vpin_window = config.vpin_window;
83        let core = StrategyCore::new_checked(config.base.clone())
84            .map_err(|e| ConfigError::invalid_value("order_id_tag", e.to_string()))?;
85        Ok(Self {
86            core,
87            config,
88            returns: VecDeque::with_capacity(hurst_window),
89            abs_imbalances: VecDeque::with_capacity(vpin_window),
90            signed_imbalances: VecDeque::with_capacity(vpin_window),
91            last_close: None,
92            bucket_buy_volume: 0.0,
93            bucket_sell_volume: 0.0,
94            hurst: None,
95            vpin: None,
96            signed_vpin: None,
97            position_opened_ns: None,
98            exit_cooldown: false,
99            entry_order_id: None,
100            exit_order_ids: AHashSet::new(),
101        })
102    }
103
104    /// Creates a new [`HurstVpinDirectional`] instance from config.
105    ///
106    /// # Panics
107    ///
108    /// Panics if either rolling window is outside `[1, 16_384]` or the base strategy configuration
109    /// is invalid.
110    #[must_use]
111    pub fn new(config: HurstVpinDirectionalConfig) -> Self {
112        Self::new_checked(config).expect("invalid Hurst/VPIN config")
113    }
114
115    pub(super) fn signals_ready(&self) -> bool {
116        self.hurst.is_some()
117            && self.vpin.is_some()
118            && self.signed_vpin.is_some()
119            && self.returns.len() == self.config.hurst_window
120            && self.abs_imbalances.len() == self.config.vpin_window
121    }
122
123    pub(super) fn push_bounded(values: &mut VecDeque<f64>, capacity: usize, value: f64) {
124        if values.len() == capacity {
125            values.pop_front();
126        }
127        values.push_back(value);
128    }
129
130    pub(super) fn rolling_mean(values: &VecDeque<f64>) -> Option<f64> {
131        if values.is_empty() {
132            return None;
133        }
134        Some(values.iter().copied().sum::<f64>() / values.len() as f64)
135    }
136
137    #[allow(
138        clippy::cognitive_complexity,
139        reason = "R/S regression is inherently nested"
140    )]
141    pub(super) fn estimate_hurst(&self) -> Option<f64> {
142        if self.returns.len() < self.config.hurst_window {
143            return None;
144        }
145
146        let returns: Vec<f64> = self.returns.iter().copied().collect();
147        let mut log_lags: Vec<f64> = Vec::new();
148        let mut log_rs: Vec<f64> = Vec::new();
149
150        for &lag in &self.config.hurst_lags {
151            if lag < 2 || lag > returns.len() {
152                continue;
153            }
154
155            let mut rs_values: Vec<f64> = Vec::new();
156
157            for start in (0..=returns.len().saturating_sub(lag)).step_by(lag) {
158                let chunk = &returns[start..start + lag];
159                let mean = chunk.iter().sum::<f64>() / lag as f64;
160
161                let mut running = 0.0f64;
162                let mut cum_min = 0.0f64;
163                let mut cum_max = 0.0f64;
164                let mut var_sum = 0.0f64;
165
166                for value in chunk {
167                    let deviation = value - mean;
168                    running += deviation;
169                    if running < cum_min {
170                        cum_min = running;
171                    }
172
173                    if running > cum_max {
174                        cum_max = running;
175                    }
176                    var_sum += deviation * deviation;
177                }
178                let r_range = cum_max - cum_min;
179                let stdev = (var_sum / lag as f64).sqrt();
180                if r_range > 0.0 && stdev > 0.0 {
181                    rs_values.push(r_range / stdev);
182                }
183            }
184
185            if !rs_values.is_empty() {
186                let avg_rs = rs_values.iter().copied().sum::<f64>() / rs_values.len() as f64;
187                log_lags.push((lag as f64).ln());
188                log_rs.push(avg_rs.ln());
189            }
190        }
191
192        if log_lags.len() < 2 {
193            return None;
194        }
195
196        let n = log_lags.len() as f64;
197        let sx: f64 = log_lags.iter().sum();
198        let sy: f64 = log_rs.iter().sum();
199        let sxx: f64 = log_lags.iter().map(|x| x * x).sum();
200        let sxy: f64 = log_lags.iter().zip(log_rs.iter()).map(|(x, y)| x * y).sum();
201        let denom = n * sxx - sx * sx;
202        if denom == 0.0 {
203            return None;
204        }
205        Some((n * sxy - sx * sy) / denom)
206    }
207
208    pub(super) fn try_open_position(&mut self) -> anyhow::Result<()> {
209        let (hurst, vpin, signed_vpin) = match (self.hurst, self.vpin, self.signed_vpin) {
210            (Some(h), Some(v), Some(s)) => (h, v, s),
211            _ => return Ok(()),
212        };
213
214        if hurst < self.config.hurst_enter || vpin < self.config.vpin_threshold {
215            return Ok(());
216        }
217
218        if signed_vpin > 0.0 {
219            self.submit_entry(OrderSide::Buy)?;
220        } else if signed_vpin < 0.0 {
221            self.submit_entry(OrderSide::Sell)?;
222        }
223
224        Ok(())
225    }
226
227    pub(super) fn check_regime_exit(&mut self) -> anyhow::Result<()> {
228        if !self.exit_order_ids.is_empty() {
229            return Ok(());
230        }
231        let hurst = match self.hurst {
232            Some(h) => h,
233            None => return Ok(()),
234        };
235
236        if hurst >= self.config.hurst_exit {
237            return Ok(());
238        }
239
240        let has_open_position = self.has_open_position();
241        if !has_open_position {
242            return Ok(());
243        }
244
245        log::info!("Regime decay (Hurst={hurst:.3}); closing position");
246        self.submit_close()
247    }
248
249    pub(super) fn check_holding_timeout(&mut self, tick: &QuoteTick) -> anyhow::Result<()> {
250        if !self.exit_order_ids.is_empty() {
251            return Ok(());
252        }
253        let opened_ns = match self.position_opened_ns {
254            Some(ns) => ns,
255            None => return Ok(()),
256        };
257        let held = tick.ts_event.saturating_duration_since(opened_ns);
258        if held < DurationNanos::try_from_secs(self.config.max_holding_secs)? {
259            return Ok(());
260        }
261
262        log::info!("Holding timeout reached; closing position");
263        self.submit_close()
264    }
265
266    fn submit_entry(&mut self, side: OrderSide) -> anyhow::Result<()> {
267        let instrument_id = self.config.instrument_id;
268        let trade_size = self.config.trade_size;
269        let order = self.order().market(
270            instrument_id,
271            side,
272            trade_size,
273            Some(TimeInForce::Ioc),
274            None, // reduce_only
275            None, // quote_quantity
276            None, // exec_algorithm_id
277            None, // exec_algorithm_params
278            None, // tags
279            None, // client_order_id
280        );
281        self.entry_order_id = Some(order.client_order_id());
282        self.submit_order(order, None, None, None)
283    }
284
285    fn submit_close(&mut self) -> anyhow::Result<()> {
286        let instrument_id = self.config.instrument_id;
287        let strategy_id = self.strategy_id().expect("Strategy must be registered");
288
289        let positions: Vec<(PositionId, Quantity, PositionSide)> = self
290            .cache()
291            .positions_open(None, Some(&instrument_id), Some(&strategy_id), None, None)
292            .iter()
293            .map(|p| (p.id, p.quantity, p.side))
294            .collect();
295
296        if positions.is_empty() {
297            return Ok(());
298        }
299
300        self.exit_cooldown = true;
301
302        for (position_id, quantity, side) in positions {
303            let Some(closing_side) = OrderCore::closing_side(side) else {
304                continue;
305            };
306            let close_order = self.order().market(
307                instrument_id,
308                closing_side,
309                quantity,
310                Some(TimeInForce::Ioc),
311                Some(true), // reduce_only
312                None,
313                None,
314                None,
315                None,
316                None,
317            );
318            self.exit_order_ids.insert(close_order.client_order_id());
319            self.submit_order(close_order, Some(position_id), None, None)?;
320        }
321
322        Ok(())
323    }
324
325    fn has_open_position(&self) -> bool {
326        let instrument_id = self.config.instrument_id;
327        let strategy_id = self.strategy_id().expect("Strategy must be registered");
328        !self
329            .cache()
330            .positions_open(None, Some(&instrument_id), Some(&strategy_id), None, None)
331            .is_empty()
332    }
333
334    fn clear_latch_for(&mut self, client_order_id: ClientOrderId) {
335        if self.entry_order_id == Some(client_order_id) {
336            self.entry_order_id = None;
337        }
338        self.exit_order_ids.remove(&client_order_id);
339    }
340}
341
342nautilus_strategy!(HurstVpinDirectional, {
343    fn on_position_opened(&mut self, event: PositionOpened) {
344        if event.instrument_id == self.config.instrument_id {
345            self.position_opened_ns = Some(event.ts_event);
346        }
347    }
348
349    fn on_position_closed(&mut self, event: PositionClosed) {
350        if event.instrument_id == self.config.instrument_id {
351            self.position_opened_ns = None;
352        }
353    }
354
355    fn on_order_rejected(&mut self, event: OrderRejected) {
356        if event.instrument_id == self.config.instrument_id {
357            self.clear_latch_for(event.client_order_id);
358        }
359    }
360
361    fn on_order_expired(&mut self, event: OrderExpired) {
362        if event.instrument_id == self.config.instrument_id {
363            self.clear_latch_for(event.client_order_id);
364        }
365    }
366
367    fn on_order_denied(&mut self, event: OrderDenied) {
368        if event.instrument_id == self.config.instrument_id {
369            self.clear_latch_for(event.client_order_id);
370        }
371    }
372
373    fn on_order_filled(&mut self, event: &OrderFilled) {
374        if event.instrument_id != self.config.instrument_id {
375            return;
376        }
377
378        let closed = self
379            .cache()
380            .order(&event.client_order_id)
381            .is_some_and(|o| o.is_closed());
382
383        if closed {
384            self.clear_latch_for(event.client_order_id);
385        }
386    }
387
388    fn on_order_canceled(&mut self, event: &OrderCanceled) {
389        if event.instrument_id != self.config.instrument_id {
390            return;
391        }
392        self.clear_latch_for(event.client_order_id);
393    }
394});
395
396impl Debug for HurstVpinDirectional {
397    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
398        f.debug_struct(stringify!(HurstVpinDirectional))
399            .field("config", &self.config)
400            .field("hurst", &self.hurst)
401            .field("vpin", &self.vpin)
402            .field("signed_vpin", &self.signed_vpin)
403            .finish()
404    }
405}
406
407impl DataActor for HurstVpinDirectional {
408    fn on_start(&mut self) -> anyhow::Result<()> {
409        let instrument_id = self.config.instrument_id;
410        let bar_instrument_id = self.config.bar_type.instrument_id();
411        if bar_instrument_id != instrument_id {
412            anyhow::bail!(
413                "bar_type instrument {bar_instrument_id} does not match traded instrument {instrument_id}"
414            );
415        }
416        {
417            let cache = self.cache();
418            cache.try_instrument(&instrument_id)?;
419        }
420
421        self.subscribe_bars(self.config.bar_type, None, None);
422        self.subscribe_quotes(instrument_id, None, None);
423        self.subscribe_trades(instrument_id, None, None);
424        Ok(())
425    }
426
427    fn on_stop(&mut self) -> anyhow::Result<()> {
428        let instrument_id = self.config.instrument_id;
429        self.cancel_all_orders(instrument_id, None, None, true, None)?;
430        self.close_all_positions(instrument_id, None, None, None, None, None, None, None)?;
431        self.unsubscribe_bars(self.config.bar_type, None, None);
432        self.unsubscribe_quotes(instrument_id, None, None);
433        self.unsubscribe_trades(instrument_id, None, None);
434        Ok(())
435    }
436
437    fn on_trade(&mut self, tick: &TradeTick) -> anyhow::Result<()> {
438        let size = tick.size.as_f64();
439        match tick.aggressor_side {
440            AggressorSide::Buy => self.bucket_buy_volume += size,
441            AggressorSide::Sell => self.bucket_sell_volume += size,
442            _ => {}
443        }
444        Ok(())
445    }
446
447    fn on_bar(&mut self, bar: &Bar) -> anyhow::Result<()> {
448        let close = bar.close.as_f64();
449
450        if let Some(prev) = self.last_close
451            && prev > 0.0
452            && close > 0.0
453        {
454            let window = self.config.hurst_window;
455            Self::push_bounded(&mut self.returns, window, (close / prev).ln());
456        }
457        self.last_close = Some(close);
458
459        let total = self.bucket_buy_volume + self.bucket_sell_volume;
460        if total > 0.0 {
461            let imbalance = (self.bucket_buy_volume - self.bucket_sell_volume) / total;
462            let vpin_window = self.config.vpin_window;
463            Self::push_bounded(&mut self.abs_imbalances, vpin_window, imbalance.abs());
464            Self::push_bounded(&mut self.signed_imbalances, vpin_window, imbalance);
465        }
466        self.bucket_buy_volume = 0.0;
467        self.bucket_sell_volume = 0.0;
468
469        self.hurst = self.estimate_hurst();
470        self.vpin = Self::rolling_mean(&self.abs_imbalances);
471        self.signed_vpin = Self::rolling_mean(&self.signed_imbalances);
472
473        if let Some(h) = self.hurst {
474            log::info!(
475                "Hurst={h:.3} VPIN={:.3} signed={:+.3} bar_close={close:.2}",
476                self.vpin.unwrap_or(0.0),
477                self.signed_vpin.unwrap_or(0.0),
478            );
479        }
480
481        self.exit_cooldown = false;
482        self.check_regime_exit()
483    }
484
485    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
486        if !self.signals_ready() {
487            return Ok(());
488        }
489
490        if self.has_open_position() {
491            return self.check_holding_timeout(quote);
492        }
493
494        if self.exit_cooldown {
495            return Ok(());
496        }
497
498        if self.entry_order_id.is_some() || !self.exit_order_ids.is_empty() {
499            return Ok(());
500        }
501
502        let strategy_id = self.strategy_id().expect("Strategy must be registered");
503        let has_working = {
504            let cache = self.cache();
505            !cache
506                .orders_open(
507                    None,
508                    Some(&self.config.instrument_id),
509                    Some(&strategy_id),
510                    None,
511                    None,
512                )
513                .is_empty()
514                || !cache
515                    .orders_inflight(
516                        None,
517                        Some(&self.config.instrument_id),
518                        Some(&strategy_id),
519                        None,
520                        None,
521                    )
522                    .is_empty()
523        };
524
525        if has_working {
526            return Ok(());
527        }
528
529        self.try_open_position()
530    }
531
532    fn on_reset(&mut self) -> anyhow::Result<()> {
533        self.returns.clear();
534        self.abs_imbalances.clear();
535        self.signed_imbalances.clear();
536        self.last_close = None;
537        self.bucket_buy_volume = 0.0;
538        self.bucket_sell_volume = 0.0;
539        self.hurst = None;
540        self.vpin = None;
541        self.signed_vpin = None;
542        self.position_opened_ns = None;
543        self.exit_cooldown = false;
544        self.entry_order_id = None;
545        self.exit_order_ids.clear();
546        Ok(())
547    }
548}
nautilus_trading/examples/strategies/hurst_vpin_directional/strategy.rs

nautilus_trading/examples/strategies/hurst_vpin_directional/
strategy.rs
