""" Hurst/VPIN NautilusTrader strategy. Hurst exponent regime detection combined with VPIN (Volume-synchronized Probability of INformed trading) for directional flow imbalance. Hurst > 0.55 → trending regime High VPIN → informed flow present Entry: trending + high VPIN + directional alignment Exit: Hurst drops below 0.45 (mean-reverting regime) or VPIN normalizes Based on the existing HurstVPINLive signal generator used in the production node. Backtest shows 96% win rate on synthetic data — this port enables testing on real Hyperliquid candles. """ from __future__ import annotations import logging import math from collections import deque import numpy as np from nautilus_trader.model.data import Bar from nautilus_trader.model.enums import OrderSide from framework.base_strategy import BaseHlStrategy from framework.config import StrategyConfig logger = logging.getLogger(__name__) def _hurst_rs(returns: list[float]) -> float: n = len(returns) if n < 32: return 0.50 max_lag = min(n // 2, 64) lags = [] rs = [] for lag in range(4, max_lag): segs = n // lag if segs < 2: continue vals = [] for s in range(segs): seg = returns[s * lag:(s + 1) * lag] mean = np.mean(seg) dev = np.cumsum(seg - mean) r = float(np.max(dev) - np.min(dev)) sd = float(np.std(seg, ddof=1)) if sd > 1e-12: vals.append(r / sd) if vals: lags.append(np.log(lag)) rs.append(np.log(np.mean(vals))) if len(lags) < 4: return 0.50 slope = float(np.polyfit(lags, rs, 1)[0]) return max(0.20, min(0.80, slope)) class HurstVPINNT(BaseHlStrategy): """Hurst exponent + VPIN directional signal on real candle data.""" def __init__(self, config: StrategyConfig): super().__init__(config) # Hurst self._hurst_window = config.params.get("hurst_window", 64) self._hurst_entry = config.params.get("hurst_entry", 0.55) self._hurst_exit = config.params.get("hurst_exit", 0.45) # VPIN self._vpin_window = config.params.get("vpin_window", 50) self._vpin_threshold = config.params.get("vpin_threshold", 0.25) self._dollar_threshold = config.params.get("dollar_threshold", 100000.0) # State self._close_history: deque[float] = deque(maxlen=self._hurst_window) self._vpin_values: deque[float] = deque(maxlen=self._vpin_window) self._vpin_directions: deque[float] = deque(maxlen=self._vpin_window) # Dollar bar accumulator self._bar_volume = 0.0 self._bar_buy_vol = 0.0 self._bar_sell_vol = 0.0 self._bar_close = 0.0 self._bar_open = 0.0 # Rolling state self._returns: deque[float] = deque(maxlen=self._hurst_window) self._last_emit_close = 0.0 self._in_trade = False self._trade_direction: str | None = None def on_bar(self, bar: Bar): price = float(bar.close) self._close_history.append(price) # Accumulate notional for dollar bars notional = price * float(bar.volume) if hasattr(bar, 'volume') else price * 100 is_buy = float(bar.close) > float(bar.open) self._bar_volume += notional if is_buy: self._bar_buy_vol += notional else: self._bar_sell_vol += notional self._bar_close = price if self._bar_open == 0: self._bar_open = float(bar.open) if self._bar_volume < self._dollar_threshold: return # Dollar bar complete — emit self._emit_dollar_bar() signal = self._compute_hurst_vpin_signal() if signal: self._last_signal = signal self.handle_signal(signal) def _emit_dollar_bar(self): total = self._bar_buy_vol + self._bar_sell_vol vpin = abs(self._bar_buy_vol - self._bar_sell_vol) / total if total > 1 else 0.0 direction = (self._bar_buy_vol - self._bar_sell_vol) / total if total > 1 else 0.0 self._vpin_values.append(vpin) self._vpin_directions.append(direction) if self._last_emit_close > 0 and self._bar_close > 0: self._returns.append(math.log(self._bar_close / self._last_emit_close)) self._last_emit_close = self._bar_close # Reset accumulator self._bar_volume = 0.0 self._bar_buy_vol = 0.0 self._bar_sell_vol = 0.0 self._bar_open = self._bar_close def _compute_hurst_vpin_signal(self) -> dict | None: if len(self._returns) < 32 or len(self._vpin_values) < 10: return None hurst = _hurst_rs(list(self._returns)) vpin = float(np.mean(self._vpin_values)) direction = float(np.mean(self._vpin_directions)) trending = hurst >= self._hurst_entry high_vpin = vpin >= self._vpin_threshold # Exit logic if self._in_trade: if hurst < self._hurst_exit: self._in_trade = False self._trade_direction = None return { "signal": "SELL" if self._trade_direction == "long" else "BUY", "strength": 1.0, "hurst": round(hurst, 3), "vpin": round(vpin, 3), "reason": "exit_hurst_fade", } # Exit on direction flip with high certainty if self._trade_direction == "long" and direction < -0.5 and high_vpin: self._in_trade = False self._trade_direction = None return { "signal": "SELL", "strength": abs(direction), "hurst": round(hurst, 3), "vpin": round(vpin, 3), "reason": "exit_direction_flip", } elif self._trade_direction == "short" and direction > 0.5 and high_vpin: self._in_trade = False self._trade_direction = None return { "signal": "BUY", "strength": abs(direction), "hurst": round(hurst, 3), "vpin": round(vpin, 3), "reason": "exit_direction_flip", } return None # Entry: trending + informed flow + directional alignment if trending and high_vpin: if direction > 0.05: self._in_trade = True self._trade_direction = "long" return { "signal": "BUY", "strength": max(0.15, direction), "hurst": round(hurst, 3), "vpin": round(vpin, 3), "direction": round(direction, 3), "reason": "entry_trending_vpin", } elif direction < -0.05: self._in_trade = True self._trade_direction = "short" return { "signal": "SELL", "strength": max(0.15, abs(direction)), "hurst": round(hurst, 3), "vpin": round(vpin, 3), "direction": round(direction, 3), "reason": "entry_trending_vpin", } return None def compute_signal(self, price: float | None = None) -> dict | None: """External signal compute for paper trader / deploy orchestrator.""" if price is None: return None self._close_history.append(price) # Simplified: just use price-based dollar bar if len(self._close_history) < 2: return None last = self._close_history[-2] cur = self._close_history[-1] notional = cur * abs(cur - last) * 100 is_buy = cur > last self._bar_volume += notional if is_buy: self._bar_buy_vol += notional else: self._bar_sell_vol += notional self._bar_close = cur if self._bar_volume < self._dollar_threshold: return None self._emit_dollar_bar() return self._compute_hurst_vpin_signal() def handle_signal(self, signal: dict): side_str = signal["signal"] if "BUY" in side_str: self._submit_order(OrderSide.BUY, size=self._cfg.order_size) elif "SELL" in side_str: self._submit_order(OrderSide.SELL, size=self._cfg.order_size)