59. Liquidity Depletion Early Warning System¶

services/liquidity-warning-center/
├── src/
│   ├── main.py
│   ├── fetcher/
│   │   ├── depth_fetcher.py
│   │   ├── order_book_collector.py
│   │   └── trade_flow_monitor.py
│   ├── calculator/
│   │   ├── liquidity_calculator.py
│   │   ├── depth_analyzer.py
│   │   └── imbalance_calculator.py
│   ├── detector/
│   │   ├── liquidity_change_detector.py
│   │   ├── crisis_detector.py
│   │   └── stress_indicator.py
│   ├── controller/
│   │   ├── warning_controller.py
│   │   ├── alert_manager.py
│   │   └── risk_responder.py
│   ├── adjuster/
│   │   ├── strategy_throttler.py
│   │   ├── position_manager.py
│   │   └── frequency_controller.py
│   ├── api/
│   │   ├── liquidity_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class DepthFetcher:
    def __init__(self, order_book_collector, trade_flow_monitor):
        self.order_book_collector = order_book_collector
        self.trade_flow_monitor = trade_flow_monitor
        self.depth_cache = {}
        self.collection_stats = {
            "total_symbols": 0,
            "active_venues": 0,
            "last_update": None,
            "update_frequency_ms": 100
        }

    async def start_depth_collection(self):
        """Start real-time order book collection"""

        # Initialize venue connections
        await self.initialize_venues()

        # Start order book monitoring
        await self.start_order_book_monitoring()

        # Start trade flow monitoring
        await self.start_trade_flow_monitoring()

    async def initialize_venues(self):
        """Initialize connections to all trading venues"""

        venues = self.get_configured_venues()

        for venue in venues:
            try:
                collector = await self.order_book_collector.connect(venue)
                await collector.initialize()
                self.collection_stats["active_venues"] += 1
            except Exception as e:
                print(f"Failed to connect to {venue}: {e}")

    async def start_order_book_monitoring(self):
        """Start monitoring order books for all symbols"""

        symbols = self.get_monitored_symbols()
        self.collection_stats["total_symbols"] = len(symbols)

        for symbol in symbols:
            await self.start_symbol_monitoring(symbol)

    async def start_symbol_monitoring(self, symbol: str):
        """Start monitoring order book for a specific symbol"""

        venues = self.get_venues_for_symbol(symbol)

        for venue in venues:
            try:
                # Subscribe to order book updates
                await self.subscribe_to_order_book(symbol, venue)

                # Start order book collection task
                asyncio.create_task(self.collect_order_book(symbol, venue))

            except Exception as e:
                print(f"Failed to monitor {symbol} on {venue}: {e}")

    async def collect_order_book(self, symbol: str, venue: str):
        """Collect order book data for a symbol from a venue"""

        while True:
            try:
                # Get current order book
                order_book = await self.get_current_order_book(symbol, venue)

                if order_book:
                    # Update depth cache
                    await self.update_depth_cache(symbol, venue, order_book)

                    # Trigger liquidity analysis
                    await self.trigger_liquidity_analysis(symbol, venue)

                # Wait for next update
                await asyncio.sleep(self.collection_stats["update_frequency_ms"] / 1000)

            except Exception as e:
                print(f"Error collecting order book for {symbol} on {venue}: {e}")
                await asyncio.sleep(1)

    async def update_depth_cache(self, symbol: str, venue: str, order_book: Dict):
        """Update depth cache with new order book data"""

        cache_key = f"{symbol}_{venue}"

        self.depth_cache[cache_key] = {
            "symbol": symbol,
            "venue": venue,
            "bids": order_book.get("bids", []),
            "asks": order_book.get("asks", []),
            "timestamp": datetime.now().isoformat(),
            "sequence": order_book.get("sequence", 0),
            "last_update_id": order_book.get("last_update_id", 0)
        }

        # Update collection statistics
        self.collection_stats["last_update"] = datetime.now().isoformat()

    async def get_current_order_book(self, symbol: str, venue: str) -> Optional[Dict]:
        """Get current order book from venue"""

        try:
            collector = self.order_book_collector.get_collector(venue)
            order_book = await collector.get_order_book(symbol)
            return order_book
        except Exception as e:
            print(f"Error getting order book for {symbol} on {venue}: {e}")
            return None

    async def get_order_book_for_symbol(self, symbol: str) -> List[Dict]:
        """Get order books for a symbol across all venues"""

        order_books = []
        venues = self.get_venues_for_symbol(symbol)

        for venue in venues:
            cache_key = f"{symbol}_{venue}"
            if cache_key in self.depth_cache:
                order_books.append(self.depth_cache[cache_key])

        return order_books

    async def get_all_order_books(self) -> Dict:
        """Get all current order books from cache"""

        return {
            "order_books": self.depth_cache,
            "statistics": self.collection_stats,
            "timestamp": datetime.now().isoformat()
        }

    def get_configured_venues(self) -> List[str]:
        """Get list of configured trading venues"""

        return [
            "binance", "okx", "bybit", "coinbase", "kraken",
            "nyse", "nasdaq", "lse", "tse", "hkex"
        ]

    def get_monitored_symbols(self) -> List[str]:
        """Get list of monitored symbols"""

        return [
            "BTC/USDT", "ETH/USDT", "AAPL", "MSFT", "NVDA",
            "GOOGL", "AMZN", "TSLA", "SPY", "QQQ"
        ]

    def get_venues_for_symbol(self, symbol: str) -> List[str]:
        """Get venues that support a symbol"""

        # Crypto symbols
        if "/" in symbol:
            return ["binance", "okx", "bybit", "coinbase", "kraken"]

        # Stock symbols
        else:
            return ["nyse", "nasdaq", "lse", "tse", "hkex"]

class LiquidityCalculator:
    def __init__(self, depth_analyzer, imbalance_calculator):
        self.depth_analyzer = depth_analyzer
        self.imbalance_calculator = imbalance_calculator
        self.liquidity_history = {}
        self.calculation_levels = [1, 5, 10, 20, 50]  # Price levels to analyze

    async def calculate_liquidity_metrics(self, order_book: Dict) -> Dict:
        """Calculate comprehensive liquidity metrics"""

        bids = order_book.get("bids", [])
        asks = order_book.get("asks", [])

        # Calculate basic liquidity metrics
        basic_metrics = self.calculate_basic_liquidity(bids, asks)

        # Calculate depth analysis
        depth_analysis = self.calculate_depth_analysis(bids, asks)

        # Calculate imbalance metrics
        imbalance_metrics = self.calculate_imbalance_metrics(bids, asks)

        # Calculate spread metrics
        spread_metrics = self.calculate_spread_metrics(bids, asks)

        # Calculate concentration metrics
        concentration_metrics = self.calculate_concentration_metrics(bids, asks)

        return {
            "symbol": order_book.get("symbol"),
            "venue": order_book.get("venue"),
            "timestamp": order_book.get("timestamp"),
            "basic_metrics": basic_metrics,
            "depth_analysis": depth_analysis,
            "imbalance_metrics": imbalance_metrics,
            "spread_metrics": spread_metrics,
            "concentration_metrics": concentration_metrics
        }

    def calculate_basic_liquidity(self, bids: List, asks: List) -> Dict:
        """Calculate basic liquidity metrics"""

        # Calculate total bid and ask liquidity
        total_bid_liquidity = sum(bid["quantity"] for bid in bids[:10])
        total_ask_liquidity = sum(ask["quantity"] for ask in asks[:10])

        # Calculate average order size
        avg_bid_size = total_bid_liquidity / len(bids[:10]) if bids else 0
        avg_ask_size = total_ask_liquidity / len(asks[:10]) if asks else 0

        return {
            "total_bid_liquidity": total_bid_liquidity,
            "total_ask_liquidity": total_ask_liquidity,
            "total_liquidity": total_bid_liquidity + total_ask_liquidity,
            "avg_bid_size": avg_bid_size,
            "avg_ask_size": avg_ask_size,
            "bid_levels": len(bids),
            "ask_levels": len(asks)
        }

    def calculate_depth_analysis(self, bids: List, asks: List) -> Dict:
        """Calculate depth analysis at different levels"""

        depth_analysis = {}

        for level in self.calculation_levels:
            bid_depth = sum(bid["quantity"] for bid in bids[:level])
            ask_depth = sum(ask["quantity"] for ask in asks[:level])

            depth_analysis[f"level_{level}"] = {
                "bid_depth": bid_depth,
                "ask_depth": ask_depth,
                "total_depth": bid_depth + ask_depth,
                "depth_ratio": bid_depth / ask_depth if ask_depth > 0 else 0
            }

        return depth_analysis

    def calculate_imbalance_metrics(self, bids: List, asks: List) -> Dict:
        """Calculate order book imbalance metrics"""

        total_bid = sum(bid["quantity"] for bid in bids[:10])
        total_ask = sum(ask["quantity"] for ask in asks[:10])

        if total_bid + total_ask == 0:
            return {
                "imbalance_ratio": 0,
                "imbalance_percentage": 0,
                "imbalance_direction": "neutral"
            }

        imbalance_ratio = total_bid / total_ask if total_ask > 0 else 0
        imbalance_percentage = abs(total_bid - total_ask) / (total_bid + total_ask)

        if imbalance_ratio > 1.5:
            direction = "bid_heavy"
        elif imbalance_ratio < 0.67:
            direction = "ask_heavy"
        else:
            direction = "balanced"

        return {
            "imbalance_ratio": imbalance_ratio,
            "imbalance_percentage": imbalance_percentage,
            "imbalance_direction": direction,
            "total_bid": total_bid,
            "total_ask": total_ask
        }

    def calculate_spread_metrics(self, bids: List, asks: List) -> Dict:
        """Calculate spread metrics"""

        if not bids or not asks:
            return {
                "spread": 0,
                "spread_bps": 0,
                "mid_price": 0
            }

        best_bid = bids[0]["price"]
        best_ask = asks[0]["price"]

        spread = best_ask - best_bid
        mid_price = (best_bid + best_ask) / 2
        spread_bps = (spread / mid_price) * 10000 if mid_price > 0 else 0

        return {
            "spread": spread,
            "spread_bps": spread_bps,
            "mid_price": mid_price,
            "best_bid": best_bid,
            "best_ask": best_ask
        }

    def calculate_concentration_metrics(self, bids: List, asks: List) -> Dict:
        """Calculate liquidity concentration metrics"""

        if not bids or not asks:
            return {
                "bid_concentration": 0,
                "ask_concentration": 0,
                "overall_concentration": 0
            }

        # Calculate concentration using Herfindahl-Hirschman Index
        bid_concentration = self.calculate_hhi([bid["quantity"] for bid in bids[:10]])
        ask_concentration = self.calculate_hhi([ask["quantity"] for ask in asks[:10]])

        total_quantities = [bid["quantity"] for bid in bids[:10]] + [ask["quantity"] for ask in asks[:10]]
        overall_concentration = self.calculate_hhi(total_quantities)

        return {
            "bid_concentration": bid_concentration,
            "ask_concentration": ask_concentration,
            "overall_concentration": overall_concentration
        }

    def calculate_hhi(self, quantities: List[float]) -> float:
        """Calculate Herfindahl-Hirschman Index for concentration"""

        if not quantities:
            return 0

        total = sum(quantities)
        if total == 0:
            return 0

        hhi = sum((q / total) ** 2 for q in quantities)
        return hhi

    async def get_liquidity_summary(self, symbol: str) -> Dict:
        """Get liquidity summary for a symbol across venues"""

        order_books = await self.get_order_books_for_symbol(symbol)

        venue_metrics = {}
        total_metrics = {
            "total_bid_liquidity": 0,
            "total_ask_liquidity": 0,
            "total_liquidity": 0,
            "avg_spread_bps": 0,
            "venue_count": len(order_books)
        }

        for order_book in order_books:
            metrics = await self.calculate_liquidity_metrics(order_book)
            venue = order_book["venue"]
            venue_metrics[venue] = metrics

            # Aggregate metrics
            basic = metrics["basic_metrics"]
            spread = metrics["spread_metrics"]

            total_metrics["total_bid_liquidity"] += basic["total_bid_liquidity"]
            total_metrics["total_ask_liquidity"] += basic["total_ask_liquidity"]
            total_metrics["total_liquidity"] += basic["total_liquidity"]
            total_metrics["avg_spread_bps"] += spread["spread_bps"]

        if total_metrics["venue_count"] > 0:
            total_metrics["avg_spread_bps"] /= total_metrics["venue_count"]

        return {
            "symbol": symbol,
            "venue_metrics": venue_metrics,
            "total_metrics": total_metrics,
            "timestamp": datetime.now().isoformat()
        }

class LiquidityChangeDetector:
    def __init__(self, crisis_detector, stress_indicator):
        self.crisis_detector = crisis_detector
        self.stress_indicator = stress_indicator
        self.liquidity_history = {}
        self.detection_rules = self.load_detection_rules()

    async def detect_liquidity_changes(self, current_metrics: Dict, symbol: str) -> List[Dict]:
        """Detect liquidity changes and potential crises"""

        changes = []

        # Get historical metrics for comparison
        historical_metrics = self.get_historical_metrics(symbol)

        if historical_metrics:
            # Detect rapid liquidity decline
            decline_signals = await self.detect_liquidity_decline(current_metrics, historical_metrics)
            changes.extend(decline_signals)

            # Detect order book imbalance
            imbalance_signals = await self.detect_order_book_imbalance(current_metrics, historical_metrics)
            changes.extend(imbalance_signals)

            # Detect spread widening
            spread_signals = await self.detect_spread_widening(current_metrics, historical_metrics)
            changes.extend(spread_signals)

            # Detect concentration changes
            concentration_signals = await self.detect_concentration_changes(current_metrics, historical_metrics)
            changes.extend(concentration_signals)

        # Detect crisis conditions
        crisis_signals = await self.detect_crisis_conditions(current_metrics)
        changes.extend(crisis_signals)

        # Update historical metrics
        self.update_historical_metrics(symbol, current_metrics)

        return changes

    async def detect_liquidity_decline(self, current: Dict, historical: Dict) -> List[Dict]:
        """Detect rapid liquidity decline"""

        signals = []

        current_basic = current["basic_metrics"]
        historical_basic = historical["basic_metrics"]

        # Check bid liquidity decline
        bid_decline = (historical_basic["total_bid_liquidity"] - current_basic["total_bid_liquidity"]) / historical_basic["total_bid_liquidity"]
        if bid_decline > self.detection_rules["liquidity_decline_threshold"]:
            signals.append({
                "type": "bid_liquidity_decline",
                "severity": self.calculate_severity(bid_decline),
                "decline_percentage": bid_decline * 100,
                "current_value": current_basic["total_bid_liquidity"],
                "historical_value": historical_basic["total_bid_liquidity"],
                "threshold": self.detection_rules["liquidity_decline_threshold"] * 100,
                "timestamp": datetime.now().isoformat()
            })

        # Check ask liquidity decline
        ask_decline = (historical_basic["total_ask_liquidity"] - current_basic["total_ask_liquidity"]) / historical_basic["total_ask_liquidity"]
        if ask_decline > self.detection_rules["liquidity_decline_threshold"]:
            signals.append({
                "type": "ask_liquidity_decline",
                "severity": self.calculate_severity(ask_decline),
                "decline_percentage": ask_decline * 100,
                "current_value": current_basic["total_ask_liquidity"],
                "historical_value": historical_basic["total_ask_liquidity"],
                "threshold": self.detection_rules["liquidity_decline_threshold"] * 100,
                "timestamp": datetime.now().isoformat()
            })

        return signals

    async def detect_order_book_imbalance(self, current: Dict, historical: Dict) -> List[Dict]:
        """Detect order book imbalance"""

        signals = []

        current_imbalance = current["imbalance_metrics"]
        historical_imbalance = historical["imbalance_metrics"]

        # Check for severe imbalance
        if current_imbalance["imbalance_percentage"] > self.detection_rules["imbalance_threshold"]:
            signals.append({
                "type": "order_book_imbalance",
                "severity": self.calculate_severity(current_imbalance["imbalance_percentage"]),
                "imbalance_percentage": current_imbalance["imbalance_percentage"] * 100,
                "imbalance_direction": current_imbalance["imbalance_direction"],
                "threshold": self.detection_rules["imbalance_threshold"] * 100,
                "timestamp": datetime.now().isoformat()
            })

        # Check for imbalance change
        imbalance_change = abs(current_imbalance["imbalance_percentage"] - historical_imbalance["imbalance_percentage"])
        if imbalance_change > self.detection_rules["imbalance_change_threshold"]:
            signals.append({
                "type": "imbalance_change",
                "severity": self.calculate_severity(imbalance_change),
                "change_percentage": imbalance_change * 100,
                "current_imbalance": current_imbalance["imbalance_percentage"] * 100,
                "historical_imbalance": historical_imbalance["imbalance_percentage"] * 100,
                "threshold": self.detection_rules["imbalance_change_threshold"] * 100,
                "timestamp": datetime.now().isoformat()
            })

        return signals

    async def detect_spread_widening(self, current: Dict, historical: Dict) -> List[Dict]:
        """Detect spread widening"""

        signals = []

        current_spread = current["spread_metrics"]
        historical_spread = historical["spread_metrics"]

        # Check for spread widening
        spread_change = (current_spread["spread_bps"] - historical_spread["spread_bps"]) / historical_spread["spread_bps"]
        if spread_change > self.detection_rules["spread_widening_threshold"]:
            signals.append({
                "type": "spread_widening",
                "severity": self.calculate_severity(spread_change),
                "spread_change_percentage": spread_change * 100,
                "current_spread_bps": current_spread["spread_bps"],
                "historical_spread_bps": historical_spread["spread_bps"],
                "threshold": self.detection_rules["spread_widening_threshold"] * 100,
                "timestamp": datetime.now().isoformat()
            })

        # Check for absolute spread threshold
        if current_spread["spread_bps"] > self.detection_rules["absolute_spread_threshold"]:
            signals.append({
                "type": "high_spread",
                "severity": "high",
                "spread_bps": current_spread["spread_bps"],
                "threshold": self.detection_rules["absolute_spread_threshold"],
                "timestamp": datetime.now().isoformat()
            })

        return signals

    async def detect_crisis_conditions(self, current: Dict) -> List[Dict]:
        """Detect crisis conditions"""

        signals = []

        basic = current["basic_metrics"]
        spread = current["spread_metrics"]

        # Check for extremely low liquidity
        if basic["total_liquidity"] < self.detection_rules["crisis_liquidity_threshold"]:
            signals.append({
                "type": "crisis_low_liquidity",
                "severity": "critical",
                "total_liquidity": basic["total_liquidity"],
                "threshold": self.detection_rules["crisis_liquidity_threshold"],
                "timestamp": datetime.now().isoformat()
            })

        # Check for extremely wide spread
        if spread["spread_bps"] > self.detection_rules["crisis_spread_threshold"]:
            signals.append({
                "type": "crisis_wide_spread",
                "severity": "critical",
                "spread_bps": spread["spread_bps"],
                "threshold": self.detection_rules["crisis_spread_threshold"],
                "timestamp": datetime.now().isoformat()
            })

        # Check for order book collapse
        if basic["bid_levels"] < 3 or basic["ask_levels"] < 3:
            signals.append({
                "type": "order_book_collapse",
                "severity": "critical",
                "bid_levels": basic["bid_levels"],
                "ask_levels": basic["ask_levels"],
                "timestamp": datetime.now().isoformat()
            })

        return signals

    def calculate_severity(self, change_value: float) -> str:
        """Calculate severity level based on change value"""

        if change_value > 0.5:
            return "critical"
        elif change_value > 0.3:
            return "high"
        elif change_value > 0.1:
            return "medium"
        else:
            return "low"

    def get_historical_metrics(self, symbol: str) -> Optional[Dict]:
        """Get historical metrics for comparison"""

        if symbol in self.liquidity_history:
            return self.liquidity_history[symbol]
        return None

    def update_historical_metrics(self, symbol: str, metrics: Dict):
        """Update historical metrics"""

        self.liquidity_history[symbol] = metrics

    def load_detection_rules(self) -> Dict:
        """Load detection rules"""

        return {
            "liquidity_decline_threshold": 0.3,  # 30% decline
            "imbalance_threshold": 0.4,  # 40% imbalance
            "imbalance_change_threshold": 0.2,  # 20% change
            "spread_widening_threshold": 0.5,  # 50% widening
            "absolute_spread_threshold": 100,  # 100 bps
            "crisis_liquidity_threshold": 1000,  # Minimum liquidity
            "crisis_spread_threshold": 500  # 500 bps
        }

@router.get("/liquidity/status")
async def get_liquidity_status():
    """Get current liquidity status"""

    order_books = await depth_fetcher.get_all_order_books()

    return {
        "order_books": order_books,
        "timestamp": datetime.now().isoformat()
    }

@router.get("/liquidity/metrics/{symbol}")
async def get_liquidity_metrics(symbol: str):
    """Get liquidity metrics for a symbol"""

    metrics = await liquidity_calculator.get_liquidity_summary(symbol)

    return {
        "symbol": symbol,
        "metrics": metrics,
        "timestamp": datetime.now().isoformat()
    }

@router.get("/liquidity/warnings")
async def get_liquidity_warnings():
    """Get current liquidity warnings"""

    warnings = await liquidity_detector.get_current_warnings()

    return {
        "warnings": warnings,
        "total_count": len(warnings),
        "critical_count": len([w for w in warnings if w["severity"] == "critical"]),
        "timestamp": datetime.now().isoformat()
    }

@router.get("/liquidity/changes/{symbol}")
async def get_liquidity_changes(symbol: str, hours: int = 24):
    """Get liquidity changes for a symbol"""

    changes = await liquidity_detector.get_liquidity_changes(symbol, hours)

    return {
        "symbol": symbol,
        "changes": changes,
        "time_period_hours": hours,
        "timestamp": datetime.now().isoformat()
    }

@router.get("/liquidity/crisis_indicators")
async def get_crisis_indicators():
    """Get crisis indicators"""

    indicators = await crisis_detector.get_crisis_indicators()

    return {
        "indicators": indicators,
        "crisis_level": crisis_detector.get_crisis_level(),
        "timestamp": datetime.now().isoformat()
    }

@router.get("/liquidity/strategy_adjustments")
async def get_strategy_adjustments():
    """Get current strategy adjustments"""

    adjustments = await strategy_throttler.get_current_adjustments()

    return {
        "adjustments": adjustments,
        "total_strategies": len(adjustments),
        "throttled_strategies": len([a for a in adjustments if a["throttled"]]),
        "timestamp": datetime.now().isoformat()
    }

const LiquidityMonitorView: React.FC = () => {
  const [liquidityStatus, setLiquidityStatus] = useState<LiquidityStatus | null>(null);
  const [warnings, setWarnings] = useState<LiquidityWarning[]>([]);
  const [crisisIndicators, setCrisisIndicators] = useState<CrisisIndicator[]>([]);
  const [strategyAdjustments, setStrategyAdjustments] = useState<StrategyAdjustment[]>([]);

  return (
    <div className="liquidity-monitor-dashboard">
      {/* Liquidity Status Overview */}
      <LiquidityStatusPanel 
        status={liquidityStatus}
      />

      {/* Liquidity Warnings */}
      <LiquidityWarningsPanel 
        warnings={warnings}
        onWarningSelect={handleWarningSelect}
      />

      {/* Crisis Indicators */}
      <CrisisIndicatorsPanel 
        indicators={crisisIndicators}
      />

      {/* Order Book Depth */}
      <OrderBookDepthPanel 
        orderBooks={liquidityStatus?.order_books}
      />

      {/* Spread Analysis */}
      <SpreadAnalysisPanel 
        spreads={liquidityStatus?.spreads}
      />

      {/* Strategy Adjustments */}
      <StrategyAdjustmentsPanel 
        adjustments={strategyAdjustments}
      />

      {/* Liquidity Trends */}
      <LiquidityTrendsPanel 
        trends={liquidityTrends}
      />

      {/* Risk Alerts */}
      <RiskAlertsPanel 
        warnings={warnings}
        indicators={crisisIndicators}
      />
    </div>
  );
};