ftdt-quant-lab/strategies/queue_imbalance.py

"""
Queue Imbalance Model — Order Book Dynamics for Short-Term Prediction.

Based on the Stoikov & Sağlam and Cont et al. frameworks.
Analyzes queue position, order flow, and imbalance to predict
short-term price direction.

Key concepts:
  1. Queue position estimation — where we sit in the LOB queue
  2. Order flow imbalance at each level — net adds vs cancels
  3. Probability of mid-price move based on queue dynamics
  4. Adverse selection detection — when informed traders clear levels

Queue Imbalance (Q_i):
  Q_i = (BidSize_i - AskSize_i) / (BidSize_i + AskSize_i)

Weighted Queue Imbalance (WQI):
  WQI = Σ w_i · Q_i   where w_i decays with distance from mid

When WQI > 0: buying pressure → expected price increase
When WQI < 0: selling pressure → expected price decrease

Signal strength from queue dynamics is proportional to how
extreme the imbalance is relative to historical norms.

Usage:
    from strategies.queue_imbalance import QueueImbalance
    qi = QueueImbalance()
    signal = qi.analyze(bids, asks, historical_wqi)
"""
import math
from collections import deque

class QueueImbalance:
    """LOB queue dynamics model for short-term price prediction."""

    def __init__(self, depth_levels: int = 10):
        self.depth_levels = depth_levels
        self.wqi_history: deque = deque(maxlen=100)

    def compute_wqi(self, bids: list, asks: list) -> float:
        """Compute Weighted Queue Imbalance across LOB levels.

        Weights decay exponentially: w_i = e^(-i/3) for level i.
        This gives 3x more weight to top-of-book than 3 levels deep.
        """
        if not bids or not asks:
            return 0.0

        wqi = 0.0
        total_weight = 0.0

        for i in range(min(len(bids), len(asks), self.depth_levels)):
            bid_sz = bids[i][1]
            ask_sz = asks[i][1]
            total_sz = bid_sz + ask_sz

            if total_sz > 0:
                qi = (bid_sz - ask_sz) / total_sz
            else:
                qi = 0.0

            # Exponential decay weight
            weight = math.exp(-i / 3.0)
            wqi += weight * qi
            total_weight += weight

        return wqi / total_weight if total_weight > 0 else 0.0

    def compute_level_flow(self, bids: list, asks: list,
                           prev_bids: list, prev_asks: list) -> dict:
        """Compute net order flow at each level (adds minus cancels)."""
        flow = {"bid_flow": 0.0, "ask_flow": 0.0, "net_flow": 0.0}

        if not prev_bids or not prev_asks:
            return flow

        # Bid side: compare current level sizes with previous
        for i in range(min(len(bids), len(prev_bids))):
            flow["bid_flow"] += bids[i][1] - prev_bids[i][1]

        # Ask side
        for i in range(min(len(asks), len(prev_asks))):
            flow["ask_flow"] += asks[i][1] - prev_asks[i][1]

        flow["net_flow"] = flow["bid_flow"] - flow["ask_flow"]
        return flow

    def estimate_adverse_selection(self, bids: list, asks: list,
                                   mid_price: float,
                                   prev_mid: float) -> float:
        """Detect adverse selection: when price moves against the
        dominant side of the book (informed traders clearing levels).

        Returns 0-1 score where 1 = high adverse selection risk.
        """
        if prev_mid <= 0 or mid_price <= 0:
            return 0.0

        # Compute which side was dominant in the previous tick
        wqi = self.compute_wqi(bids, asks)
        price_move = (mid_price - prev_mid) / prev_mid

        # Adverse selection: price moves opposite to queue imbalance
        # e.g., bids dominant (WQI > 0) but price goes down
        if wqi > 0.1 and price_move < -0.0005:
            return min(abs(price_move) * 1000, 1.0)
        elif wqi < -0.1 and price_move > 0.0005:
            return min(abs(price_move) * 1000, 1.0)

        return 0.0

    def analyze(self, bids: list, asks: list, mid_price: float,
                prev_bids: list = None, prev_asks: list = None,
                prev_mid: float = 0) -> dict:
        """Full queue imbalance analysis.

        Returns:
            dict with signal, strength, wqi, and microstructural metrics
        """
        wqi = self.compute_wqi(bids, asks)
        self.wqi_history.append(wqi)

        # Compute WQI z-score
        if len(self.wqi_history) >= 20:
            history = list(self.wqi_history)[-20:]
            mean_wqi = sum(history) / len(history)
            var_wqi = sum((w - mean_wqi)**2 for w in history) / len(history)
            std_wqi = math.sqrt(var_wqi) if var_wqi > 0 else 0.01
            z_score = (wqi - mean_wqi) / std_wqi
        else:
            z_score = wqi * 3  # Rough scaling during warm-up

        # Order flow analysis (if previous state available)
        flow = {}
        if prev_bids and prev_asks:
            flow = self.compute_level_flow(bids, asks, prev_bids, prev_asks)

        # Adverse selection
        adverse = self.estimate_adverse_selection(
            bids, asks, mid_price, prev_mid) if prev_mid > 0 else 0

        # Signal generation
        signal = None
        strength = 0.0
        z_threshold = 1.2
        wqi_threshold = 0.15

        # Strong signal: extreme z-score AND high WQI
        if z_score > z_threshold and wqi > wqi_threshold:
            signal = "BUY"
            strength = min(abs(z_score) / 3.0, 1.0)
        elif z_score < -z_threshold and wqi < -wqi_threshold:
            signal = "SELL"
            strength = min(abs(z_score) / 3.0, 1.0)

        # Reduce confidence if adverse selection detected
        if adverse > 0.3:
            strength *= (1.0 - adverse)

        return {
            "signal": signal,
            "strength": round(strength, 4),
            "wqi": round(wqi, 4),
            "z_score": round(z_score, 3),
            "adverse_selection": round(adverse, 4),
            "bid_flow": round(flow.get("bid_flow", 0), 4),
            "ask_flow": round(flow.get("ask_flow", 0), 4),
            "net_flow": round(flow.get("net_flow", 0), 4),
        }