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

nautilus_trading/examples/strategies/hurst_vpin_directional/
strategy.rs
