Hawkes OFI + Deep LOB: two new strategies from advanced microstructure research

strategies/hawkes_ofi.py

@@ -0,0 +1,147 @@
+"""
+Hawkes Process Order Flow Imbalance Strategy.
+
+Models trade arrivals as self-exciting point processes.
+Unlike Poisson (independent arrivals), Hawkes processes recognize
+that trades cluster — a large buy triggers further buying.
+
+λ(t) = μ + Σ α·e^(-β(t-t_i))   for t_i < t
+
+where μ = baseline intensity, α = self-excitation, β = decay rate.
+
+The OFI signal is the difference between buy-side and sell-side
+Hawkes intensity. When OFI crosses a threshold, it predicts
+short-term price direction.
+
+Usage:
+    from strategies.hawkes_ofi import HawkesOFI
+    model = HawkesOFI(alpha=0.3, beta=0.5)
+    signal = model.update(trade_side, trade_size, trade_price)
+"""
+import math
+import time
+from collections import deque
+
+class HawkesOFI:
+    """Single-asset Hawkes process for OFI signal generation."""
+
+    def __init__(self, alpha: float = 0.3, beta: float = 0.5, mu: float = 0.05):
+        self.alpha = alpha    # Self-excitation strength
+        self.beta = beta      # Decay rate
+        self.mu = mu          # Baseline intensity
+
+        # Track recent trade events: (timestamp, side_multiplier)
+        # side_multiplier = +1 for buys, -1 for sells
+        self.events: deque = deque(maxlen=200)
+        self._last_update = time.time()
+
+        # Rolling statistics
+        self.buy_intensity = mu
+        self.sell_intensity = mu
+        self.ofi_history: deque = deque(maxlen=50)
+        self.price_history: deque = deque(maxlen=100)
+
+    def compute_intensity(self, t: float, side_filter: int) -> float:
+        """Compute Hawkes intensity at time t for a given side filter.
+        
+        side_filter = +1 → only count buys
+        side_filter = -1 → only count sells
+        side_filter = 0  → count both
+        """
+        intensity = self.mu
+        for ev_t, ev_side in self.events:
+            dt = t - ev_t
+            if dt > 10.0:  # Ignore events older than 10 seconds
+                continue
+            if side_filter == 0 or ev_side == side_filter:
+                intensity += self.alpha * math.exp(-self.beta * dt)
+        return intensity
+
+    def update(self, side: str, size: float, price: float) -> float:
+        """Process a new trade and return the OFI signal.
+        
+        Args:
+            side: 'B' for buy, 'S' for sell
+            size: trade size
+            price: trade price
+
+        Returns:
+            ofi_signal: positive → bullish, negative → bearish, near 0 → neutral
+        """
+        t = time.time()
+        side_mult = +1 if side in ('B', 'BUY', 'b') else -1
+
+        # Add event
+        self.events.append((t, side_mult))
+        self.price_history.append(price)
+
+        # Compute intensities
+        buy_intensity = self.compute_intensity(t, +1)
+        sell_intensity = self.compute_intensity(t, -1)
+
+        self.buy_intensity = buy_intensity
+        self.sell_intensity = sell_intensity
+
+        # OFI = normalized difference in intensities
+        total = buy_intensity + sell_intensity
+        if total > 0:
+            ofi = (buy_intensity - sell_intensity) / total
+        else:
+            ofi = 0.0
+
+        self.ofi_history.append(ofi)
+        self._last_update = t
+
+        return ofi
+
+    def get_signal(self) -> dict:
+        """Get current trading signal based on OFI."""
+        if len(self.ofi_history) < 10:
+            return {"signal": None, "strength": 0.0, "confidence": 0.0}
+
+        # Current OFI
+        current_ofi = self.ofi_history[-1]
+
+        # Trend in OFI (momentum of the imbalance)
+        if len(self.ofi_history) >= 5:
+            recent = list(self.ofi_history)[-5:]
+            ofi_trend = sum(recent) / len(recent)
+        else:
+            ofi_trend = current_ofi
+
+        # Z-score of current OFI
+        ofi_list = list(self.ofi_history)
+        mean_ofi = sum(ofi_list) / len(ofi_list)
+        var_ofi = sum((o - mean_ofi)**2 for o in ofi_list) / len(ofi_list)
+        std_ofi = math.sqrt(var_ofi) if var_ofi > 0 else 0.01
+
+        z_score = (current_ofi - mean_ofi) / std_ofi if std_ofi > 0 else 0.0
+
+        # Signal logic
+        threshold = 0.15
+        z_threshold = 1.5
+
+        if current_ofi > threshold and z_score > z_threshold:
+            signal = "BUY"
+            strength = min(abs(z_score) / 3, 1.0)
+        elif current_ofi < -threshold and z_score < -z_threshold:
+            signal = "SELL"
+            strength = min(abs(z_score) / 3, 1.0)
+        elif abs(ofi_trend) > threshold * 0.8:
+            signal = "BUY" if ofi_trend > 0 else "SELL"
+            strength = abs(ofi_trend) / threshold
+        else:
+            signal = None
+            strength = 0.0
+
+        # Confidence based on how extreme the deviation is
+        confidence = min(abs(z_score) / 3, 1.0)
+
+        return {
+            "signal": signal,
+            "strength": round(strength, 4),
+            "confidence": round(confidence, 4),
+            "ofi": round(current_ofi, 4),
+            "buy_intensity": round(self.buy_intensity, 4),
+            "sell_intensity": round(self.sell_intensity, 4),
+        }