Order Flow Toxicity: How AI Agents Detect and Avoid Informed Trading

import asyncio
import numpy as np
import aiohttp
from collections import deque
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from enum import Enum
import json

class ToxicityLevel(Enum):
    LOW = "low"          # VPIN < 0.40
    MODERATE = "moderate"  # 0.40 - 0.60
    HIGH = "high"         # 0.60 - 0.75
    CRITICAL = "critical"  # VPIN > 0.75

@dataclass
class VolumeBucket:
    """Equal-size volume bucket for VPIN calculation."""
    target_volume: float
    filled_volume: float = 0.0
    buy_volume: float = 0.0
    sell_volume: float = 0.0
    open_price: Optional[float] = None
    close_price: Optional[float] = None
    timestamp_start: Optional[float] = None
    timestamp_end: Optional[float] = None

    @property
    def is_complete(self) -> bool:
        return self.filled_volume >= self.target_volume

    @property
    def imbalance(self) -> float:
        return abs(self.buy_volume - self.sell_volume)

@dataclass
class VPINResult:
    vpin: float
    toxicity: ToxicityLevel
    buy_volume_pct: float
    buckets_in_window: int
    recommendation: str
    position_size_mult: float  # 0.0 = halt, 1.0 = full size

class VPINCalculator:
    """
    Real-time VPIN calculator using Bulk Volume Classification.
    Connects to Purple Flea Trading API for live trade data.

    Parameters:
        bucket_size: Volume per bucket (e.g., 50 BTC)
        window_size: Rolling window in buckets (typically 50)
        symbol: Trading pair (e.g., 'BTC-USD')
    """

    def __init__(
        self,
        bucket_size: float,
        window_size: int = 50,
        symbol: str = "BTC-USD",
    ):
        self.bucket_size = bucket_size
        self.window_size = window_size
        self.symbol = symbol
        self.completed_buckets: deque = deque(maxlen=window_size)
        self.current_bucket = VolumeBucket(target_volume=bucket_size)
        self.callbacks: List[Callable] = []
        self._last_price: Optional[float] = None

    def on_vpin_update(self, callback: Callable):
        """Register callback for VPIN updates."""
        self.callbacks.append(callback)

    def _classify_trade(self, price: float, volume: float) -> tuple[float, float]:
        """
        Bulk Volume Classification: classify trade as buy or sell.
        Uses price-change direction (simplified BVC).
        Returns (buy_volume, sell_volume).
        """
        if self._last_price is None:
            self._last_price = price
            return volume / 2, volume / 2

        # Using price change direction as BVC signal
        # Positive price change = buy pressure; negative = sell pressure
        price_change = price - self._last_price
        self._last_price = price

        if price_change > 0:
            # Full volume classified as buy-initiated
            return volume, 0.0
        elif price_change < 0:
            # Full volume classified as sell-initiated
            return 0.0, volume
        else:
            # No price change: split 50/50
            return volume / 2, volume / 2

    def process_trade(self, price: float, volume: float, timestamp: float):
        """Process a single trade and update VPIN if bucket completes."""
        buy_vol, sell_vol = self._classify_trade(price, volume)

        if self.current_bucket.open_price is None:
            self.current_bucket.open_price = price
            self.current_bucket.timestamp_start = timestamp

        remaining = self.current_bucket.target_volume - self.current_bucket.filled_volume

        if volume <= remaining:
            self.current_bucket.buy_volume += buy_vol
            self.current_bucket.sell_volume += sell_vol
            self.current_bucket.filled_volume += volume
        else:
            # Fill the current bucket partially
            frac = remaining / volume
            self.current_bucket.buy_volume += buy_vol * frac
            self.current_bucket.sell_volume += sell_vol * frac
            self.current_bucket.filled_volume = self.current_bucket.target_volume

        if self.current_bucket.is_complete:
            self.current_bucket.close_price = price
            self.current_bucket.timestamp_end = timestamp
            self.completed_buckets.append(self.current_bucket)

            # Start new bucket with overflow volume
            overflow = volume - remaining
            self.current_bucket = VolumeBucket(target_volume=self.bucket_size)
            if overflow > 0:
                self.process_trade(price, overflow, timestamp)

            # Calculate and broadcast VPIN
            result = self.calculate_vpin()
            if result:
                for cb in self.callbacks:
                    cb(result)

    def calculate_vpin(self) -> Optional[VPINResult]:
        if len(self.completed_buckets) < 5:
            return None  # Need minimum buckets for stable estimate

        buckets = list(self.completed_buckets)
        total_imbalance = sum(b.imbalance for b in buckets)
        total_volume = sum(b.filled_volume for b in buckets)
        total_buy = sum(b.buy_volume for b in buckets)

        vpin = total_imbalance / total_volume
        buy_pct = total_buy / total_volume

        # Determine toxicity level and trading recommendation
        if vpin < 0.40:
            toxicity = ToxicityLevel.LOW
            recommendation = "Normal trading operations. Full position sizes permitted."
            size_mult = 1.0
        elif vpin < 0.60:
            toxicity = ToxicityLevel.MODERATE
            recommendation = "Elevated toxicity. Reduce position sizes by 50%. Widen quotes."
            size_mult = 0.5
        elif vpin < 0.75:
            toxicity = ToxicityLevel.HIGH
            recommendation = "High toxicity. Stop market making. Only directional trades."
            size_mult = 0.2
        else:
            toxicity = ToxicityLevel.CRITICAL
            recommendation = "CRITICAL: Halt all new positions. Close open exposure."
            size_mult = 0.0

        return VPINResult(
            vpin=round(vpin, 4),
            toxicity=toxicity,
            buy_volume_pct=round(buy_pct, 4),
            buckets_in_window=len(buckets),
            recommendation=recommendation,
            position_size_mult=size_mult,
        )

# Live integration with Purple Flea Trading API
async def run_vpin_monitor(api_key: str, symbol: str):
    calculator = VPINCalculator(
        bucket_size=50.0,   # 50 BTC per bucket
        window_size=50,
        symbol=symbol
    )

    def on_vpin(result: VPINResult):
        color = {
            ToxicityLevel.LOW: "GREEN",
            ToxicityLevel.MODERATE: "YELLOW",
            ToxicityLevel.HIGH: "RED",
            ToxicityLevel.CRITICAL: "CRITICAL",
        }[result.toxicity]
        print(f"[{color}] VPIN={result.vpin:.4f} | Buy%={result.buy_volume_pct:.1%}")
        print(f"  Size mult: {result.position_size_mult}x | {result.recommendation}")

    calculator.on_vpin_update(on_vpin)

    ws_url = f"wss://purpleflea.com/trading-api/ws/trades/{symbol}"
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(ws_url, headers={"X-API-Key": api_key}) as ws:
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    trade = json.loads(msg.data)
                    calculator.process_trade(
                        price=float(trade['price']),
                        volume=float(trade['size']),
                        timestamp=float(trade['timestamp']),
                    )

asyncio.run(run_vpin_monitor("your-api-key", "BTC-USD"))