Order Flow Toxicity

Order Flow Toxicity & Adverse Selection

Detects toxic (informed) order flow and quantifies adverse selection cost to protect passive market makers and limit-order-based strategies from predatory HFT activity. When loading this skill, you act as a quantitative risk analyst who measures the proportion of informed versus noise flow in real time and triggers protective actions when toxicity exceeds safe thresholds.

TL;DR Checklist

• Segment incoming trades into volume-synchronized buckets (not time buckets)
• Compute VPIN from intra-bucket imbalances over a rolling window of 50+ buckets
• Fit PIN via EM algorithm with convergence check on log-likelihood change < 1e-6
• Estimate adverse selection cost per fill as mid-price drift between submission and fill
• Activate kill switch when VPIN > 0.7 for 5+ consecutive buckets — reduce size by 50%, widen spreads 2x, flag for review
• Log full state dump (VPIN, PIN, adverse selection cost, order book snapshot, positions) on every kill switch event

When to Use

• You operate a passive quoting or market-making strategy and need early warning that informed traders are exploiting your quotes
• You suspect HFT predators (latency arbitrage, quote stuffing, spoofing) are degrading your execution quality
• You need to dynamically adjust position sizing, spread width, or order freshness based on measured flow toxicity
• You are designing a risk engine and need toxic-flow detection as a layer above standard stop-loss and drawdown controls
• You are post-trade analyzing execution quality by decomposing slippage into adverse selection vs. timing cost

When NOT to Use

• Purely directional / aggressive strategies that hunt liquidity — toxicity metrics apply to passive risk, not active taker flow (use ai-order-flow-analysis for directional signals instead)
• Low-frequency swing trading with minutes-to-hours holding periods — VPIN and PIN require high-frequency trade data to be meaningful; below ~10 trades per minute the signal degrades into noise
• When you have only OHLCV bars without individual trade timestamps, sizes, or buy/sell flags — bucket-based toxic flow measurement is impossible without tick-level data

Core Workflow

1. Segment Trades into Volume Buckets — Divide incoming trades into buckets of fixed size (default 50,000 shares). Each bucket represents roughly equal expected volume regardless of time elapsed. Track cumulative buy and sell volume within each bucket. Checkpoint: Verify bucket sizes are appropriate for the asset's typical trading frequency — crypto uses smaller buckets (~1 BTC) than equities; adjust via BUCKET_SIZE = 50_000 constant per instrument class.

2. Compute Intra-Bucket Imbalance — For each completed bucket, calculate imbalance = abs(buy_volume - sell_volume). Sum all intra-bucket imbalances over a rolling window of N buckets (default 50). Compute VPIN = sum_of_imbalances / total_bucket_volume. Checkpoint: VPIN ranges from 0 (perfectly balanced flow) to 1 (extremely toxic); values >0.6 trigger elevated risk flags, and values >0.7 require kill switch consideration.

3. Estimate PIN via EM Algorithm — Fit the Buy-Seller Orderflow Model (BSOM) using Expectation-Maximization: initialize α (informed probability) and λ (arrival rate), iterate E-step (estimate hidden informed states) and M-step (update parameters) for up to 200 iterations or until log-likelihood converges within 1e-6. Checkpoint: PIN is bounded [0,1]; values >0.5 indicate significant informed presence; values <0.1 suggest mostly noise trading.

4. Calculate Adverse Selection Cost — For each executed trade, estimate the cost of being on the wrong side: compare mid-price at order submission to mid-price at order fill. Positive cost means you were adverse-selected (informed flow hit your passive orders). Aggregate as a weighted average across recent fills. Checkpoint: If adverse selection cost exceeds 0.5x average spread for 10+ consecutive fills, halt passive quoting immediately and log the event.

5. Detect Spoofing & Layering — Track order book dynamics: if large orders appear and cancel repeatedly within <200ms without being hit, flag as potential spoofing. Count repeated appearances of same price level by same logical participant (inferred from order pattern clustering). **Checkpoint:** Spoofing detection requires ≥3 cancellations at same price level within the time window AND total canceled size >10x average depth at that level.

6. Activate Toxic Flow Kill Switch — When VPIN exceeds threshold (default 0.7) for M consecutive buckets (default 5), reduce position sizing by 50%, widen quote spreads by 2x, and flag all new signals for manual review. When toxicity normalizes below threshold for K buckets (default 10), gradually restore normal operations over a cooldown period. Checkpoint: Kill switch activation MUST log the full state dump: VPIN value, PIN estimate, adverse selection cost, order book snapshot, and active positions.

Implementation Patterns

Pattern 1: VPIN (Volume-Synchronized Probability of Informed Trading) Calculator

VPIN segments trades into volume buckets rather than time buckets, making it robust across varying trading frequencies. The core insight is that informed flow creates persistent one-sidedness within each bucket — the greater the imbalance, the higher the toxicity estimate.

"""
VPIN Calculator — Volume-Synchronized Probability of Informed Trading.

File path convention: risk_engine/vpin_calculator.py

Uses volume-synchronized bucketing to detect toxic (informed) order flow.
Adapted from Easley, López de Prado, and O'Hara (2012).
"""

from __future__ import annotations

import math
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Optional


@dataclass
class Trade:
    """Single executed trade record."""
    timestamp_ns: int          # nanosecond precision for HFT contexts
    price: float
    size: float
    aggressor_side: str      # 'buy' (aggressor hit asks) or 'sell' (aggressor hit bids)


@dataclass
class VPINConfig:
    """Configuration for the VPIN calculator."""
    bucket_size: float = 50_000.0          # target volume per bucket
    rolling_window_buckets: int = 50       # number of buckets for rolling VPIN
    elevated_threshold: float = 0.6        # warn when exceeded
    critical_threshold: float = 0.7        # kill-switch trigger

    def __post_init__(self) -> None:
        if self.bucket_size <= 0:
            raise ValueError("bucket_size must be positive")
        if self.rolling_window_buckets < 5:
            raise ValueError("rolling_window_buckets must be >= 5 for stable VPIN")


@dataclass
class VolumeBucket:
    """Accumulates buy/sell volume until reaching target size."""
    buy_volume: float = 0.0
    sell_volume: float = 0.0
    completed: bool = False

    @property
    def total_volume(self) -> float:
        return self.buy_volume + self.sell_volume

    @property
    def imbalance(self) -> float:
        """Absolute intra-bucket volume imbalance."""
        return abs(self.buy_volume - self.sell_volume)


class VPINCalculator:
    """
    Real-time VPIN calculator using volume-synchronized bucketing.

    Usage:
        calc = VPINCalculator(VPINConfig())
        vpin = calc.process_trades(trade_stream)  # iterable of Trade
    """

    def __init__(self, config: Optional[VPINConfig] = None) -> None:
        self.config = config or VPINConfig()
        self._current_bucket: VolumeBucket = VolumeBucket()
        self._completed_buckets: Deque[VolumeBucket] = deque(maxlen=self.config.rolling_window_buckets * 2)
        self._imbalance_history: Deque[float] = deque(maxlen=self.config.rolling_window_buckets)
        self._total_volume_history: Deque[float] = deque(maxlen=self.config.rolling_window_buckets)

    def process_trade(self, trade: Trade) -> Optional[float]:
        """
        Process a single trade and return VPIN if a bucket just completed.

        Returns None while still accumulating the current bucket.
        Returns VPIN value when a bucket is completed and all rolling windows are filled.
        """
        # Accumulate into current bucket
        if trade.aggressor_side == 'buy':
            self._current_bucket.buy_volume += trade.size
        else:
            self._current_bucket.sell_volume += trade.size

        # Check if bucket is full
        if self._current_bucket.total_volume >= self.config.bucket_size:
            return self._complete_bucket()
        return None

    def _complete_bucket(self) -> Optional[float]:
        """Finalize the current bucket and compute VPIN if enough history exists."""
        completed = self._current_bucket
        self._completed_buckets.append(completed)
        self._imbalance_history.append(completed.imbalance)
        self._total_volume_history.append(completed.total_volume)

        # Need at least rolling_window_buckets for stable estimate
        if len(self._imbalance_history) < self.config.rolling_window_buckets:
            self._current_bucket = VolumeBucket()
            return None

        vpin = self._compute_vpin()
        self._current_bucket = VolumeBucket()
        return vpin

    def _compute_vpin(self) -> float:
        """
        Compute VPIN over the rolling window.

        VPIN = sum(abs(buy_i - sell_i)) / sum(total_volume_i) for i in rolling window

        Range: [0, 1] where 1 means perfect one-sided flow (maximally toxic).
        """
        n = len(self._imbalance_history)
        sum_imbalance = sum(self._imbalance_history)
        sum_total_volume = sum(self._total_volume_history)

        if sum_total_volume == 0:
            return 0.0

        vpin = sum_imbalance / sum_total_volume
        return min(max(vpin, 0.0), 1.0)  # clamp to [0, 1]

    def get_toxicity_state(self) -> dict:
        """Return current toxicity assessment and metrics."""
        if len(self._imbalance_history) < self.config.rolling_window_buckets:
            return {
                "vpin": None,
                "buckets_accumulated": len(self._completed_buckets),
                "toxicity_level": "insufficient_data",
                "action": "wait_for_more_data",
            }

        vpin = self._compute_vpin()

        if vpin >= self.config.critical_threshold:
            level = "critical"
            action = "activate_kill_switch"
        elif vpin >= self.config.elevated_threshold:
            level = "elevated"
            action = "reduce_sizing_widen_spreads"
        else:
            level = "normal"
            action = "continue_normal_operations"

        return {
            "vpin": round(vpin, 4),
            "buckets_accumulated": len(self._completed_buckets),
            "rolling_buckets": len(self._imbalance_history),
            "toxicity_level": level,
            "action": action,
        }

Bucket sizing guidance per asset class:

Pattern 2: PIN (Probability of Informed Trading) via Bayesian EM Estimation

The Buy-Seller Orderflow Model (BSOM) estimates the probability that incoming trades are informed. The EM algorithm handles the hidden state (informed vs. uninformed arrivals) by alternating between estimating latent variables and updating parameters. This pattern fits the model on a rolling window of trade counts.

"""
PIN Estimator — Probability of Informed Trading via Bayesian EM.

File path convention: risk_engine/pin_estimator.py

Implements the BSOM (Buy-Seller Orderflow Model) with Expectation-Maximization
to estimate alpha (probability of information event) and lambda (arrival rate).

Adapted from Easley, Kiefer, O'Hara, and Paperman (1996).
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import List


@dataclass
class PINResult:
    """Results of PIN estimation."""
    alpha: float            # probability of information event (informed trading)
    lambda_buy: float       # arrival rate of buy orders (uninformed + informed buys)
    lambda_sell: float      # arrival rate of sell orders (uninformed + informed sells)
    lambda_total: float     # total arrival rate
    log_likelihood: float   # final log-likelihood value
    iterations: int         # number of EM iterations performed
    converged: bool         # whether convergence was achieved

    @property
    def is_toxic(self) -> bool:
        """Return True if PIN estimate indicates significant informed trading."""
        return self.alpha > 0.5


class PinEstimator:
    """
    Bayesian EM estimator for the Probability of Informed Trading (PIN).

    The BSOM assumes each time period can have one of four states:
      - No information event (probability 1 - alpha): uninformed buys/sells only
      - Information event with buy orders (probability alpha/2)
      - Information event with sell orders (probability alpha/2)
      - Information event with both (probability alpha*(1-exp(-mu))/2, where mu = informed arrival)

    The EM algorithm iteratively estimates the hidden state given parameters,
    then updates parameters given the estimated state.

    Args:
        buy_counts: List of buy order counts per time period (e.g., per minute)
        sell_counts: List of sell order counts per time period
        max_iterations: Maximum EM iterations (default 200)
        convergence_tol: Log-likelihood convergence tolerance (default 1e-6)
    """

    def __init__(
        self,
        buy_counts: List[int],
        sell_counts: List[int],
        max_iterations: int = 200,
        convergence_tol: float = 1e-6,
    ) -> None:
        if len(buy_counts) != len(sell_counts):
            raise ValueError("buy_counts and sell_counts must have equal length")
        if len(buy_counts) < 30:
            raise ValueError(f"Need at least 30 observations for PIN estimation; got {len(buy_counts)}")

        self.buy_counts = buy_counts
        self.sell_counts = sell_counts
        self.max_iterations = max_iterations
        self.convergence_tol = convergence_tol
        self.n_periods = len(buy_counts)

    def estimate(self) -> PINResult:
        """
        Run the EM algorithm to estimate PIN parameters.

        Returns PINResult with estimated alpha, lambda values, and convergence info.
        """
        # Initialize parameters
        # alpha: probability of information event (bounded in (0, 1))
        alpha = 0.3
        # lambda_p: uninformed buy arrival rate
        # lambda_m: uninformed sell arrival rate
        # mu: informed order arrival rate (conditional on information event)
        avg_buy = sum(self.buy_counts) / self.n_periods
        avg_sell = sum(self.sell_counts) / self.n_periods
        lambda_p = max(avg_buy * 0.5, 1.0)
        lambda_m = max(avg_sell * 0.5, 1.0)
        mu = max(min(avg_buy, avg_sell) * 0.3, 1.0)

        prev_ll = -float('inf')
        iterations = 0

        for iteration in range(self.max_iterations):
            # ===== E-STEP: Compute expected hidden state probabilities =====
            # For each period t, compute posterior probability of each state given observed (b_t, s_t)

            # Log-likelihood accumulator
            log_lik = 0.0

            for t in range(self.n_periods):
                b = self.buy_counts[t]
                s = self.sell_counts[t]

                # State 1: No info event — uninformed buys only (lambda_p), sells only (lambda_m)
                # P(b,s | no_info) = Poisson(lambda_p, b) * Poisson(lambda_m, s)
                ll_no_info = self._poisson_log(b, lambda_p) + self._poisson_log(s, lambda_m)

                # State 2: Buy-side info event — extra informed buys (lambda_p + mu), sells only
                ll_buy_info = self._poisson_log(b, lambda_p + mu) + self._poisson_log(s, lambda_m)

                # State 3: Sell-side info event — buys only, extra informed sells
                ll_sell_info = self._poisson_log(b, lambda_p) + self._poisson_log(s, lambda_m + mu)

                # State 4: Both-sided info event — both extra
                ll_both_info = self._poisson_log(b, lambda_p + mu) + self._poisson_log(s, lambda_m + mu)

                # Log-sum-exp for numerical stability
                log_probs = [
                    math.log(1 - alpha) + ll_no_info,
                    math.log(alpha / 2) + ll_buy_info,
                    math.log(alpha / 2) + ll_sell_info,
                    math.log(alpha * (1 - math.exp(-mu)) / 2 + alpha * math.exp(-mu) if mu > 0 else 0) + ll_both_info,
                ]
                max_log = max(log_probs)
                log_sum = max_log + math.log(sum(math.log(p) if False else math.exp(lp - max_log) for lp in log_probs))

                # Posterior probabilities (weights for M-step)
                posteriors = [math.exp(lp - log_sum) for lp in log_probs]

                # Accumulate log-likelihood
                log_lik += log_sum

                # Store posteriors for M-step accumulation
                if not hasattr(self, '_posterior_sums'):
                    self._posterior_sums = [0.0, 0.0, 0.0, 0.0]
                    self._weighted_buy = 0.0
                    self._weighted_sell = 0.0
                    self._weighted_both = 0.0

                self._posterior_sums[0] += posteriors[0]
                self._posterior_sums[1] += posteriors[1]
                self._posterior_sums[2] += posteriors[2]
                self._posterior_sums[3] += posteriors[3]

            # ===== M-STEP: Update parameters given expected hidden states =====
            total_posterior = sum(self._posterior_sums)
            if total_posterior < 1e-12:
                break

            # Update alpha (probability of information event)
            informed_weight = self._posterior_sums[1] + self._posterior_sums[2] + self._posterior_sums[3]
            new_alpha = min(max(informed_weight / total_posterior, 0.01), 0.99)

            # Update mu (informed arrival rate) — weighted average of observed excess counts
            numerator_mu = sum(
                max(self.buy_counts[t] - lambda_p, 0) * (self._posterior_sums[1] + self._posterior_sums[3]) / total_posterior
                + max(self.sell_counts[t] - lambda_m, 0) * (self._posterior_sums[2] + self._posterior_sums[3]) / total_posterior
                for t in range(self.n_periods)
            )
            new_mu = min(max(numerator_mu / self.n_periods, 0.1), max(avg_buy, avg_sell))

            # Update lambda_p and lambda_m (uninformed arrival rates)
            new_lambda_p = sum(
                (self.buy_counts[t] - mu * posteriors_t[1] - mu * posteriors_t[3]) / total_posterior
                for t, posteriors_t in enumerate(self._get_period_posteriors())
            ) / self.n_periods
            new_lambda_m = sum(
                (self.sell_counts[t] - mu * posteriors_t[2] - mu * posteriors_t[3]) / total_posterior
                for t, posteriors_t in enumerate(self._get_period_posteriors())
            ) / self.n_periods

            # Clamp parameters to valid ranges
            new_lambda_p = max(new_lambda_p, 0.1)
            new_lambda_m = max(new_lambda_m, 0.1)
            new_mu = max(new_mu, 0.1)

            alpha = new_alpha
            lambda_p = new_lambda_p
            lambda_m = new_lambda_m
            mu = new_mu
            iterations = iteration + 1

            # Check convergence
            ll_change = abs(log_lik - prev_ll)
            if ll_change < self.convergence_tol:
                prev_ll = log_lik
                break
            prev_ll = log_lik

        lambda_total = lambda_p + lambda_m + mu

        return PINResult(
            alpha=round(alpha, 6),
            lambda_buy=round(lambda_p + mu * alpha / 2, 4),
            lambda_sell=round(lambda_m + mu * alpha / 2, 4),
            lambda_total=round(lambda_total, 4),
            log_likelihood=round(prev_ll, 4),
            iterations=iterations,
            converged=(prev_ll - log_lik if 'log_lik' in dir() else 0) < self.convergence_tol,
        )

    @staticmethod
    def _poisson_log(k: int, lam: float) -> float:
        """Log of Poisson probability mass function."""
        if lam <= 0:
            return -float('inf') if k > 0 else 0.0
        return k * math.log(lam) - lam - PinEstimator._log_factorial(k)

    @staticmethod
    def _log_factorial(n: int) -> float:
        """Compute log(n!) using Stirling's approximation for large n, exact for small."""
        if n <= 1:
            return 0.0
        if n < 170:
            result = 0.0
            for i in range(2, n + 1):
                result += math.log(i)
            return result
        # Stirling's approximation
        return n * math.log(n) - n + 0.5 * math.log(2 * math.pi / n)

    def _get_period_posteriors(self) -> List[List[float]]:
        """Recompute posteriors for M-step (called during M-step)."""
        # Simplified placeholder — full implementation would recompute
        return [[1.0, 0.0, 0.0, 0.0]] * self.n_periods


class RollingPinEstimator:
    """
    Wraps PIN estimation to run on a rolling window of observations.

    Re-estimates PIN every `reestimate_every` new periods using the latest
    `window_size` observations.
    """

    def __init__(
        self,
        window_size: int = 120,
        reestimate_every: int = 10,
    ) -> None:
        self.window_size = window_size
        self.reestimate_every = reestimate_every
        self._buy_buffer: List[int] = []
        self._sell_buffer: List[int] = []
        self._last_result: Optional[PINResult] = None
        self._periods_since_reestimate = 0

    def feed_period(self, buy_count: int, sell_count: int) -> Optional[PINResult]:
        """
        Feed one period (e.g., one minute) of trade counts.
        Returns updated PINResult when re-estimation occurs, else None.
        """
        self._buy_buffer.append(buy_count)
        self._sell_buffer.append(sell_count)
        self._periods_since_reestimate += 1

        # Trim to window size
        if len(self._buy_buffer) > self.window_size:
            self._buy_buffer = self._buy_buffer[-self.window_size:]
            self._sell_buffer = self._sell_buffer[-self.window_size:]

        if self._periods_since_reestimate >= self.reestimate_every and len(self._buy_buffer) >= 30:
            estimator = PinEstimator(
                buy_counts=self._buy_buffer,
                sell_counts=self._sell_buffer,
            )
            self._last_result = estimator.estimate()
            self._periods_since_reestimate = 0
            return self._last_result

        return None

    def get_current_pin(self) -> Optional[float]:
        """Return the most recent PIN (alpha) estimate."""
        if self._last_result:
            return self._last_result.alpha
        return None

PIN interpretation guide:

Pattern 3: Adverse Selection Cost Estimator + Toxic Flow Kill Switch

This pattern combines two related concerns: measuring how much cost you incur from being on the wrong side of informed flow, and automatically triggering protective actions when toxicity metrics cross safety thresholds. The kill switch is a circuit breaker that degrades strategy aggressiveness rather than shutting down entirely — it buys time for manual review.

"""
Adverse Selection & Toxic Flow Kill Switch.

File path convention: risk_engine/adverse_selection.py
             also: risk_engine/toxic_flow_kill_switch.py

Measures per-fill adverse selection cost and implements a multi-state kill switch
that degrades strategy aggressiveness when toxic flow is detected.

Design principle: The kill switch is an INDEPENDENT safety layer — it does not
depend on stop-losses, drawdown controls, or any other risk module. It fires on
real-time flow toxicity signals only.
"""

from __future__ import annotations

import time
from collections import deque
from dataclasses import dataclass, field
from enum import Enum
from typing import Deque, List, Optional


class KillSwitchState(Enum):
    """States of the toxic flow kill switch."""
    NORMAL = "normal"                        # All operations at full capacity
    REDUCED = "reduced_risk"                 # Size cut 50%, spreads widened 2x
    QUIET = "quiet_mode"                    # Passive only; no new positions
    HALTED = "halted"                       # All quoting stopped; manual review required


@dataclass
class FillRecord:
    """Record of a single trade fill for adverse selection measurement."""
    timestamp_ns: int
    mid_price_at_submission: float          # Mid-price when order was placed
    filled_price: float                     # Actual execution price
    side: str                               # 'buy' or 'sell'
    size: float
    symbol: str


@dataclass
class KillSwitchConfig:
    """Configuration for the toxic flow kill switch."""
    # VPIN thresholds
    vpin_elevated: float = 0.6
    vpin_critical: float = 0.7
    vpin_buckets_to_escalate: int = 5       # consecutive buckets above threshold

    # Recovery
    recovery_buckets: int = 10              # buckets below critical to begin recovery
    cooldown_buckets: int = 5               # additional normal buckets before full restore

    # Kill switch parameters
    size_reduction_factor: float = 0.5      # reduce sizing by this factor
    spread_widening_factor: float = 2.0     # widen spreads by this multiplier
    max_position_reduction_pct: float = 0.5 # cap position reduction at 50%

    # Adverse selection thresholds
    adverse_selection_spread_ratio: float = 0.5  # AS cost > 50% of avg spread → alert
    consecutive_adverse_fill_limit: int = 10     # consecutive fills with high AS → halt


@dataclass
class KillSwitchStateDump:
    """Full state dump for logging when kill switch activates."""
    timestamp_ns: int
    vpin_value: Optional[float]
    pin_alpha: Optional[float]
    avg_adverse_selection_cost: float
    avg_spread_bps: float
    as_to_spread_ratio: float
    current_state: KillSwitchState
    positions_total_notional: float
    recent_fill_count: int
    recovery_progress_buckets: int


class AdverseSelectionEstimator:
    """
    Estimates adverse selection cost per fill and over a rolling window.

    Adverse selection cost = mid-price drift after your fill, measured
    in the direction opposite to your trade side. If you bought and the price
    immediately drops, that's adverse selection — informed sellers hit your quote.
    """

    def __init__(
        self,
        window_size: int = 50,
    ) -> None:
        self.window_size = window_size
        self._fills: Deque[FillRecord] = deque(maxlen=window_size)
        self._recent_mid_prices: Deque[tuple[int, float]] = deque(maxlen=window_size * 2)

    def record_fill(self, fill: FillRecord) -> Optional[float]:
        """
        Record a fill and compute its adverse selection cost.

        The adverse selection cost is the price movement against you after
        your fill, measured as the absolute change in mid-price from your
        submission to the next observed mid-price.

        Returns:
            Adverse selection cost in dollar terms, or None if insufficient data.
        """
        self._fills.append(fill)

        # In a real system, you'd query the latest mid-price here
        # For now, we track fills for aggregate measurement
        return self.get_average_cost()

    def record_mid_price(self, timestamp_ns: int, mid_price: float) -> None:
        """Record the current mid-price for adverse selection computation."""
        self._recent_mid_prices.append((timestamp_ns, mid_price))

    def get_adverse_selection_cost_for_fill(self, fill: FillRecord) -> float:
        """
        Compute adverse selection cost for a specific fill.

        For a buy fill: AS cost = max(0, mid_at_submission - latest_mid)
            (price dropped after you bought — you were hit by informed sellers)
        For a sell fill: AS cost = max(0, latest_mid - mid_at_submission)
            (price rose after you sold — you were hit by informed buyers)

        Returns cost in same units as price (e.g., dollars per share).
        """
        if not self._recent_mid_prices:
            return 0.0

        latest_mid = self._recent_mid_prices[-1][1]

        if fill.side == 'buy':
            # You bought; adverse selection if price goes down after
            as_cost = max(0.0, fill.mid_price_at_submission - latest_mid)
        else:
            # You sold; adverse selection if price goes up after
            as_cost = max(0.0, latest_mid - fill.mid_price_at_submission)

        return as_cost

    def get_average_cost(self) -> Optional[float]:
        """Return the volume-weighted average adverse selection cost over the window."""
        if not self._fills or not self._recent_mid_prices:
            return None

        total_as_cost = 0.0
        total_volume = 0.0

        for fill in self._fills:
            as_cost = self.get_adverse_selection_cost_for_fill(fill)
            total_as_cost += as_cost * fill.size
            total_volume += fill.size

        if total_volume == 0:
            return None

        return total_as_cost / total_volume

    def get_consecutive_adverse_ratio(self, spread_bps: float) -> tuple[int, float]:
        """
        Count consecutive fills where adverse selection cost > threshold.

        Returns:
            (consecutive_count, as_to_spread_ratio)
        """
        if not self._fills or not self._recent_mid_prices:
            return 0, 0.0

        avg_as = self.get_average_cost() or 0.0
        avg_spread = spread_bps  # in dollar terms
        threshold = avg_spread * 0.5  # 50% of spread

        consecutive = 0
        for fill in reversed(list(self._fills)):
            as_cost = self.get_adverse_selection_cost_for_fill(fill)
            if as_cost > threshold:
                consecutive += 1
            else:
                break

        ratio = avg_as / avg_spread if avg_spread > 0 else 0.0
        return consecutive, ratio


class ToxicFlowKillSwitch:
    """
    Multi-state kill switch that degrades strategy aggressiveness based on
    real-time flow toxicity measurements.

    States (progressive escalation):
      NORMAL → REDUCED → QUIET → HALTED

    Recovery is gradual — the system does not jump back to NORMAL immediately.
    It must pass through a cooldown period with sustained low toxicity.

    Usage:
        switch = ToxicFlowKillSwitch(config)
        state = switch.evaluate(vpin_value, pin_alpha, as_cost, spread_bps)
    """

    def __init__(self, config: Optional[KillSwitchConfig] = None) -> None:
        self.config = config or KillSwitchConfig()
        self.state = KillSwitchState.NORMAL
        self._vpin_history: Deque[float] = deque(maxlen=100)
        self._recovery_counter: int = 0
        self._escalation_counter: int = 0
        self._activation_log: List[KillSwitchStateDump] = []

    def evaluate(
        self,
        vpin: Optional[float],
        pin_alpha: Optional[float],
        avg_as_cost: Optional[float],
        spread_bps: float,
    ) -> KillSwitchState:
        """
        Evaluate current toxicity signals and update kill switch state.

        Args:
            vpin: Current VPIN value (0-1), or None if insufficient data
            pin_alpha: Current PIN estimate (0-1), or None
            avg_as_cost: Average adverse selection cost, or None
            spread_bps: Average spread in basis points for reference

        Returns:
            Current kill switch state after evaluation
        """
        # ===== Check for escalation =====
        if vpin is not None:
            self._vpin_history.append(vpin)

            if vpin >= self.config.vpin_critical:
                self._escalation_counter += 1
                if (self._escalation_counter >= self.config.vpin_buckets_to_escalate and
                        self.state != KillSwitchState.HALTED):
                    self.state = self._escalate_state()
                    self._log_activation(vpin, pin_alpha, avg_as_cost, spread_bps)
            elif vpin >= self.config.vpin_elevated:
                self._escalation_counter += 1
                if (self._escalation_counter >= 3 and
                        self.state == KillSwitchState.NORMAL):
                    self.state = KillSwitchState.REDUCED
            else:
                # Below elevated threshold — start recovery
                self._escalation_counter = 0
                self._recovery_counter += 1

        # ===== Check adverse selection independent halt =====
        if avg_as_cost is not None and spread_bps > 0:
            as_ratio = avg_as_cost / (spread_bps * 0.0001)  # convert bps to dollar terms
            if as_ratio > self.config.adverse_selection_spread_ratio:
                # Check consecutive fills
                if self._escalation_counter >= self.config.consecutive_adverse_fill_limit:
                    self.state = KillSwitchState.HALTED
                    self._log_activation(vpin, pin_alpha, avg_as_cost, spread_bps)

        return self.state

    def _escalate_state(self) -> KillSwitchState:
        """Move to the next higher severity state."""
        transitions = {
            KillSwitchState.NORMAL: KillSwitchState.REDUCED,
            KillSwitchState.REDUCED: KillSwitchState.QUIET,
            KillSwitchState.QUIET: KillSwitchState.HALTED,
            KillSwitchState.HALTED: KillSwitchState.HALTED,  # stays halted
        }
        return transitions[self.state]

    def get_execution_params(self) -> dict:
        """
        Return execution parameters adjusted for current kill switch state.

        This is called by the order router to determine position sizing,
        spread width, and order freshness.
        """
        if self.state == KillSwitchState.NORMAL:
            return {
                "size_multiplier": 1.0,
                "spread_widening": 1.0,
                "order_freshness_ms": 500,
                "allow_new_positions": True,
                "requires_manual_review": False,
            }

        if self.state == KillSwitchState.REDUCED:
            return {
                "size_multiplier": self.config.size_reduction_factor,
                "spread_widening": self.config.spread_widening_factor,
                "order_freshness_ms": 200,
                "allow_new_positions": True,
                "requires_manual_review": False,
            }

        if self.state == KillSwitchState.QUIET:
            return {
                "size_multiplier": self.config.size_reduction_factor * 0.5,
                "spread_widening": self.config.spread_widening_factor * 2.0,
                "order_freshness_ms": 100,
                "allow_new_positions": False,
                "requires_manual_review": True,
            }

        # HALTED
        return {
            "size_multiplier": 0.0,
            "spread_widening": float('inf'),
            "order_freshness_ms": 0,
            "allow_new_positions": False,
            "requires_manual_review": True,
        }

    def _log_activation(self, vpin, pin_alpha, as_cost, spread_bps) -> None:
        """Log full state dump when kill switch changes state."""
        dump = KillSwitchStateDump(
            timestamp_ns=int(time.time() * 1e9),
            vpin_value=vpin,
            pin_alpha=pin_alpha,
            avg_adverse_selection_cost=as_cost or 0.0,
            avg_spread_bps=spread_bps,
            as_to_spread_ratio=(as_cost / (spread_bps * 1e-4)) if (as_cost and spread_bps) else 0.0,
            current_state=self.state,
            positions_total_notional=0.0,  # populated by caller
            recent_fill_count=len(self._vpin_history),
            recovery_progress_buckets=self._recovery_counter,
        )
        self._activation_log.append(dump)

    def get_state_dump(self, positions_notional: float = 0.0) -> KillSwitchStateDump:
        """Generate current state dump for logging/alerting."""
        return KillSwitchStateDump(
            timestamp_ns=int(time.time() * 1e9),
            vpin_value=self._vpin_history[-1] if self._vpin_history else None,
            pin_alpha=None,  # populated by caller
            avg_adverse_selection_cost=0.0,  # populated by caller
            avg_spread_bps=0.0,  # populated by caller
            as_to_spread_ratio=0.0,  # populated by caller
            current_state=self.state,
            positions_total_notional=positions_notional,
            recent_fill_count=len(self._vpin_history),
            recovery_progress_buckets=self._recovery_counter,
        )

    def reset(self) -> None:
        """Reset kill switch to normal state (called after manual review)."""
        self.state = KillSwitchState.NORMAL
        self._escalation_counter = 0
        self._recovery_counter = 0

Constraints

MUST DO

• Use volume-synchronized bucketing (NOT time-based) for VPIN — time buckets distort signal during low and high volatility regimes equally
• Clamp all VPIN values to [0, 1] and PIN alpha to [0, 1] — out-of-range values indicate bugs in the estimator
• Run PIN EM convergence check: stop only when abs(current_log_likelihood - previous_log_likelihood) < 1e-6 OR max iterations reached (200)
• Log full kill switch state dump on every activation and deactivation event including VPIN, PIN, adverse selection cost, order book snapshot, and positions
• Implement the kill switch as an INDEPENDENT safety layer — it must function even if stop-loss modules or drawdown controls are disabled
• Validate bucket sizes per instrument class — using equity bucket sizes for crypto (or vice versa) produces meaningless VPIN values
• Track consecutive buckets above threshold before escalating — a single spike should NOT trigger the kill switch; use the configured vpin_buckets_to_escalate (default 5)
• Widen spreads by at least 2x when entering REDUCED state to compensate for elevated adverse selection risk

MUST NOT DO

• Use raw trade count imbalance instead of volume-weighted imbalance — a single large informed trade is more toxic than many small uninformed trades
• Reset the VPIN rolling window on each new bucket — this destroys the time-decay weighting and produces an inaccurate toxicity estimate
• Bypass the kill switch even for "testing" or "demo mode" — safety layers must be present in all environments, including paper trading
• Use a fixed VPIN threshold across all instruments — highly liquid ETFs may have normal VPIN of 0.2 while illiquid small-caps may naturally run at 0.4; calibrate per instrument
• Feed fewer than 30 observations into the PIN EM algorithm — the estimator will not converge reliably and produces noise
• Allow the kill switch to auto-recover directly from HALTED to NORMAL — must pass through cooldown buckets with sustained low toxicity first

Output Template

When implementing toxic flow detection or a kill switch, produce:

1. VPIN Configuration — Bucket size (per asset class), rolling window length, threshold values for elevated and critical states
2. *PIN Estimation Setup

…(truncated)