ftdt-quant-lab/strategies/hurst_vpin_live.py

"""
Hurst/VPIN integration module — provides compact signal generators
for live trading, paper trading, and backtesting.

Live: feeds price tick stream into Hurst dollar bars.
Paper/Backtest: feeds real trade data.
"""

import math, time, numpy as np
from collections import deque


# ═══════════════════════════════════════════════════════════
# 1. Hurst Exponent — R/S on log returns
# ═══════════════════════════════════════════════════════════
def _hurst_rs(returns: list) -> float:
    """R/S estimate from log returns. Returns 0.20–0.80."""
    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))


# ═══════════════════════════════════════════════════════════
# 2. Dollar Bar Builder (notional-based)
# ═══════════════════════════════════════════════════════════
class DollarBar:
    def __init__(self, threshold: float = 10000.0):
        self.threshold = threshold
        self.vol = 0.0
        self.buy_vol = 0.0
        self.sell_vol = 0.0
        self.close = 0.0

    def add(self, price: float, notional: float, is_buy: bool):
        self.vol += notional
        if is_buy:
            self.buy_vol += notional
        else:
            self.sell_vol += notional
        self.close = price

    @property
    def ready(self) -> bool:
        return self.vol >= self.threshold

    def emit(self) -> dict:
        total = self.buy_vol + self.sell_vol
        data = {
            "close": self.close,
            "vpin": abs(self.buy_vol - self.sell_vol) / total if total > 1 else 0.0,
            "direction": (self.buy_vol - self.sell_vol) / total if total > 1 else 0.0,
        }
        self.vol = 0.0; self.buy_vol = 0.0; self.sell_vol = 0.0
        return data


# ═══════════════════════════════════════════════════════════
# 3. Hurst/VPIN Signal (price-tick mode for live trading)
# ═══════════════════════════════════════════════════════════
class HurstVPINLive:
    """Lightweight Hurst/VPIN for live price tick stream.
    
    Uses notional bars ($10K) from mid-price changes.
    Each tick adds notional ≈ price * |Δprice| * 100 as volume proxy.
    """
    def __init__(self, threshold: float = 10000.0,
                 hurst_window: int = 128,
                 vpin_window: int = 50,
                 hurst_entry: float = 0.55,
                 vpin_threshold: float = 0.25):
        self.threshold = threshold
        self.vpin_window = vpin_window
        self.hurst_entry = hurst_entry
        self.vpin_threshold = vpin_threshold

        self.bar = DollarBar(threshold)
        self.vpin_buf = deque(maxlen=vpin_window)
        self.vpin_dir_buf = deque(maxlen=vpin_window)
        self.returns = deque(maxlen=hurst_window)
        self.last_close = 0.0
        self.last_price = 0.0

    def feed_price(self, price: float):
        """Feed a mid-price tick. Returns signal dict or None."""
        if self.last_price <= 0:
            self.last_price = price
            return None

        delta = price - self.last_price
        is_buy = delta > 0
        notional = price * abs(delta) * 100  # volume proxy
        self.last_price = price

        self.bar.add(price, notional, is_buy)
        if not self.bar.ready:
            return None

        bar_data = self.bar.emit()

        # VPIN
        self.vpin_buf.append(bar_data["vpin"])
        self.vpin_dir_buf.append(bar_data["direction"])
        vpin = float(np.mean(self.vpin_buf)) if len(self.vpin_buf) >= self.vpin_window else 0.0
        direction = float(np.mean(self.vpin_dir_buf)) if len(self.vpin_dir_buf) >= self.vpin_window else 0.0

        # Hurst
        if self.last_close > 0:
            self.returns.append(math.log(bar_data["close"] / self.last_close))
        self.last_close = bar_data["close"]

        hurst = _hurst_rs(list(self.returns)) if len(self.returns) >= 64 else 0.50

        # Signal
        trending = hurst >= self.hurst_entry
        high_vpin = vpin >= self.vpin_threshold

        if trending and high_vpin:
            if direction > 0.02:
                return {"signal": "BUY", "hurst": round(hurst, 3), "vpin": round(vpin, 3), "direction": round(direction, 3)}
            elif direction < -0.02:
                return {"signal": "SELL", "hurst": round(hurst, 3), "vpin": round(vpin, 3), "direction": round(direction, 3)}

        return None


# ═══════════════════════════════════════════════════════════
# 4. Hurst/VPIN for backtest (full trade data)
# ═══════════════════════════════════════════════════════════
from strategies.hurst_vpin import run_hurst_vpin, HurstVPINSignal

# Expose for easy import
def hurst_vpin_backtest(trades, capital=100.0, size=0.00024):
    return run_hurst_vpin(trades, starting_capital=capital, size=size)