ftdt-quant-lab/microstructure/signals.py

"""
Composite signal construction from microstructure features.

Combines book imbalance, trade flow, toxicity, and funding signals
into a single directional signal with confidence score.
"""

from __future__ import annotations

from typing import Optional

import numpy as np


# ── Composite signal ────────────────────────────────────────

def composite_signal(
    obi: float,                         # [-1, 1] order book imbalance
    trade_imbalance: float = 0.0,       # [-1, 1] recent trade aggressor skew
    vpin: float = 0.0,                  # [0, 1] flow toxicity (higher = toxic)
    funding_regime: str = "neutral",    # "high_positive", "negative", etc.
    spread_bps: float = 1.0,            # current spread
    weight_obi: float = 0.35,
    weight_trade: float = 0.25,
    weight_vpin: float = -0.20,         # negative: high VPIN → reduce confidence
    weight_funding: float = 0.20,
) -> dict:
    """Combine microstructure features into a single directional signal.

    Returns:
        signal:  "buy", "sell", or "neutral"
        score:   [-1, 1] raw composite (positive = buy pressure)
        confidence: [0, 1] confidence in the signal
        breakdown: per-component contributions
    """
    obi_score = np.clip(obi, -1.0, 1.0)
    trade_score = np.clip(trade_imbalance, -1.0, 1.0)
    vpin_score = np.clip(vpin, 0.0, 1.0)

    # Funding: positive funding → short gets paid → sell bias; negative → buy bias
    funding_score_map = {
        "high_positive": -0.8,
        "positive": -0.4,
        "neutral": 0.0,
        "negative": 0.4,
        "high_negative": 0.8,
    }
    funding_score = funding_score_map.get(funding_regime, 0.0)

    raw = (
        weight_obi * obi_score +
        weight_trade * trade_score +
        weight_vpin * vpin_score +
        weight_funding * funding_score
    )

    # Confidence: base from signal magnitude, reduced by VPIN and spread
    signal_magnitude = abs(raw)
    vpin_penalty = np.clip(vpin * 0.5, 0.0, 0.3) if raw != 0 else 0
    spread_penalty = min(spread_bps / 50.0, 0.3)  # wide spread → lower confidence

    confidence = max(0.0, min(1.0, signal_magnitude * 1.5 - vpin_penalty - spread_penalty))

    if raw > 0.1:
        signal = "buy"
    elif raw < -0.1:
        signal = "sell"
    else:
        signal = "neutral"

    return {
        "signal": signal,
        "score": round(raw, 4),
        "confidence": round(confidence, 4),
        "breakdown": {
            "obi": round(obi_score * weight_obi, 4),
            "trade": round(trade_score * weight_trade, 4),
            "vpin": round(vpin_score * weight_vpin, 4),
            "funding": round(funding_score * weight_funding, 4),
        },
    }


# ── Signal pipeline ─────────────────────────────────────────

class SignalPipeline:
    """Stateful pipeline that accumulates microstructure data and emits signals.

    Usage:
        pipeline = SignalPipeline()
        pipeline.update_book(bids, asks)
        pipeline.update_trade(px, sz, mid)
        signal = pipeline.emit()
    """

    def __init__(
        self,
        obi_window: int = 100,
        trade_window: int = 500,
        vpin_volume_size: float = 100.0,
        vpin_buckets: int = 50,
    ):
        self._obi_window = obi_window
        self._trade_window = trade_window
        self._vpin_volume_size = vpin_volume_size
        self._vpin_buckets = vpin_buckets

        self._buy_vol: list[float] = []
        self._sell_vol: list[float] = []
        self._obuys: int = 0
        self._osells: int = 0
        self._obuy: int = 1

    def update_book(self, bids: dict[float, float], asks: dict[float, float]):
        from microstructure.book import order_book_imbalance
        self._obuys = order_book_imbalance(bids, asks)

    def update_trade(self, px: float, sz: float, mid: float):
        if px >= mid:
            self._buy_vol.append(sz)
        else:
            self._sell_vol.append(sz)

        if len(self._buy_vol) > self._trade_window:
            self._buy_vol = self._buy_vol[-self._trade_window:]
        if len(self._sell_vol) > self._trade_window:
            self._sell_vol = self._sell_vol[-self._trade_window:]

    def recent_trade_imbalance(self) -> float:
        bv = sum(self._buy_vol[-100:])
        sv = sum(self._sell_vol[-100:])
        total = bv + sv
        return (bv - sv) / total if total > 0 else 0.0

    def current_vpin(self) -> float:
        from microstructure.toxicity import compute_vpin
        result = compute_vpin(
            self._buy_vol, self._sell_vol,
            volume_bucket_size=self._vpin_volume_size,
            n_buckets=self._vpin_buckets,
        )
        return result.get("vpin_value", 0.0)

    def emit(self) -> dict:
        return composite_signal(
            obi=self._obuys,
            trade_imbalance=self.recent_trade_imbalance(),
            vpin=self.current_vpin(),
        )


# ── Regime detection ─────────────────────────────────────────

def detect_hft_regime(
    obi_std: float,
    spread_mean_bps: float,
    trade_rate_per_sec: float,
    vpin: float,
) -> str:
    """Classify current market regime for HFT strategy selection.

    Returns one of:
      - "trending" — directional, high OBI variance, low VPIN
      - "ranging" — low OBI variance, tight spread, active
      - "toxic" — high VPIN, wide spread → don't quote
      - "quiet" — low activity, avoid
    """
    if vpin > 0.4:
        return "toxic"
    if trade_rate_per_sec < 0.1:
        return "quiet"
    if obi_std > 0.3 and spread_mean_bps < 5:
        return "trending"
    if obi_std < 0.15 and spread_mean_bps < 3:
        return "ranging"
    if spread_mean_bps > 10:
        return "toxic"
    return "quiet"