""" Metrics calculation for order book analysis. """ from typing import Dict, List, Optional from ..models.core import OrderBookSnapshot, OrderBookMetrics, ImbalanceMetrics from ..utils.logging import get_logger logger = get_logger(__name__) class MetricsCalculator: """ Calculates various metrics from order book data. Metrics include: - Basic metrics (mid price, spread, volumes) - Imbalance metrics - Depth metrics - Liquidity metrics """ def __init__(self): """Initialize metrics calculator""" logger.info("Metrics calculator initialized") def calculate_orderbook_metrics(self, orderbook: OrderBookSnapshot) -> OrderBookMetrics: """ Calculate comprehensive order book metrics. Args: orderbook: Order book snapshot Returns: OrderBookMetrics: Calculated metrics """ try: # Basic calculations mid_price = self._calculate_mid_price(orderbook) spread = self._calculate_spread(orderbook) spread_percentage = (spread / mid_price * 100) if mid_price > 0 else 0.0 # Volume calculations bid_volume = sum(level.size for level in orderbook.bids) ask_volume = sum(level.size for level in orderbook.asks) # Imbalance calculation total_volume = bid_volume + ask_volume volume_imbalance = ((bid_volume - ask_volume) / total_volume) if total_volume > 0 else 0.0 # Depth calculations depth_10 = self._calculate_depth(orderbook, 10) depth_50 = self._calculate_depth(orderbook, 50) return OrderBookMetrics( symbol=orderbook.symbol, exchange=orderbook.exchange, timestamp=orderbook.timestamp, mid_price=mid_price, spread=spread, spread_percentage=spread_percentage, bid_volume=bid_volume, ask_volume=ask_volume, volume_imbalance=volume_imbalance, depth_10=depth_10, depth_50=depth_50 ) except Exception as e: logger.error(f"Error calculating order book metrics: {e}") raise def calculate_imbalance_metrics(self, orderbook: OrderBookSnapshot) -> ImbalanceMetrics: """ Calculate order book imbalance metrics. Args: orderbook: Order book snapshot Returns: ImbalanceMetrics: Calculated imbalance metrics """ try: # Volume imbalance bid_volume = sum(level.size for level in orderbook.bids) ask_volume = sum(level.size for level in orderbook.asks) total_volume = bid_volume + ask_volume volume_imbalance = ((bid_volume - ask_volume) / total_volume) if total_volume > 0 else 0.0 # Price imbalance (weighted by volume) price_imbalance = self._calculate_price_imbalance(orderbook) # Depth imbalance depth_imbalance = self._calculate_depth_imbalance(orderbook) # Momentum score (simplified - would need historical data for full implementation) momentum_score = volume_imbalance * 0.5 + price_imbalance * 0.3 + depth_imbalance * 0.2 return ImbalanceMetrics( symbol=orderbook.symbol, timestamp=orderbook.timestamp, volume_imbalance=volume_imbalance, price_imbalance=price_imbalance, depth_imbalance=depth_imbalance, momentum_score=momentum_score ) except Exception as e: logger.error(f"Error calculating imbalance metrics: {e}") raise def _calculate_mid_price(self, orderbook: OrderBookSnapshot) -> float: """Calculate mid price""" if not orderbook.bids or not orderbook.asks: return 0.0 best_bid = orderbook.bids[0].price best_ask = orderbook.asks[0].price return (best_bid + best_ask) / 2.0 def _calculate_spread(self, orderbook: OrderBookSnapshot) -> float: """Calculate bid-ask spread""" if not orderbook.bids or not orderbook.asks: return 0.0 best_bid = orderbook.bids[0].price best_ask = orderbook.asks[0].price return best_ask - best_bid def _calculate_depth(self, orderbook: OrderBookSnapshot, levels: int) -> float: """Calculate market depth for specified number of levels""" bid_depth = sum( level.size for level in orderbook.bids[:levels] ) ask_depth = sum( level.size for level in orderbook.asks[:levels] ) return bid_depth + ask_depth def _calculate_price_imbalance(self, orderbook: OrderBookSnapshot) -> float: """Calculate price-weighted imbalance""" if not orderbook.bids or not orderbook.asks: return 0.0 # Calculate volume-weighted average prices for top levels bid_vwap = self._calculate_vwap(orderbook.bids[:5]) ask_vwap = self._calculate_vwap(orderbook.asks[:5]) if bid_vwap == 0 or ask_vwap == 0: return 0.0 mid_price = (bid_vwap + ask_vwap) / 2.0 # Normalize imbalance price_imbalance = (bid_vwap - ask_vwap) / mid_price if mid_price > 0 else 0.0 return max(-1.0, min(1.0, price_imbalance)) def _calculate_depth_imbalance(self, orderbook: OrderBookSnapshot) -> float: """Calculate depth imbalance across multiple levels""" levels_to_check = [5, 10, 20] imbalances = [] for levels in levels_to_check: bid_depth = sum(level.size for level in orderbook.bids[:levels]) ask_depth = sum(level.size for level in orderbook.asks[:levels]) total_depth = bid_depth + ask_depth if total_depth > 0: imbalance = (bid_depth - ask_depth) / total_depth imbalances.append(imbalance) # Return weighted average of imbalances if imbalances: return sum(imbalances) / len(imbalances) return 0.0 def _calculate_vwap(self, levels: List) -> float: """Calculate volume-weighted average price for price levels""" if not levels: return 0.0 total_volume = sum(level.size for level in levels) if total_volume == 0: return 0.0 weighted_sum = sum(level.price * level.size for level in levels) return weighted_sum / total_volume def calculate_liquidity_score(self, orderbook: OrderBookSnapshot) -> float: """ Calculate liquidity score based on depth and spread. Args: orderbook: Order book snapshot Returns: float: Liquidity score (0.0 to 1.0) """ try: if not orderbook.bids or not orderbook.asks: return 0.0 # Spread component (lower spread = higher liquidity) spread = self._calculate_spread(orderbook) mid_price = self._calculate_mid_price(orderbook) if mid_price == 0: return 0.0 spread_pct = (spread / mid_price) * 100 spread_score = max(0.0, 1.0 - (spread_pct / 5.0)) # Normalize to 5% max spread # Depth component (higher depth = higher liquidity) total_depth = self._calculate_depth(orderbook, 10) depth_score = min(1.0, total_depth / 100.0) # Normalize to 100 units max depth # Volume balance component (more balanced = higher liquidity) bid_volume = sum(level.size for level in orderbook.bids[:10]) ask_volume = sum(level.size for level in orderbook.asks[:10]) total_volume = bid_volume + ask_volume if total_volume > 0: imbalance = abs(bid_volume - ask_volume) / total_volume balance_score = 1.0 - imbalance else: balance_score = 0.0 # Weighted combination liquidity_score = (spread_score * 0.4 + depth_score * 0.4 + balance_score * 0.2) return max(0.0, min(1.0, liquidity_score)) except Exception as e: logger.error(f"Error calculating liquidity score: {e}") return 0.0 def get_market_summary(self, orderbook: OrderBookSnapshot) -> Dict[str, float]: """ Get comprehensive market summary. Args: orderbook: Order book snapshot Returns: Dict[str, float]: Market summary metrics """ try: metrics = self.calculate_orderbook_metrics(orderbook) imbalance = self.calculate_imbalance_metrics(orderbook) liquidity = self.calculate_liquidity_score(orderbook) return { 'mid_price': metrics.mid_price, 'spread': metrics.spread, 'spread_percentage': metrics.spread_percentage, 'bid_volume': metrics.bid_volume, 'ask_volume': metrics.ask_volume, 'volume_imbalance': metrics.volume_imbalance, 'depth_10': metrics.depth_10, 'depth_50': metrics.depth_50, 'price_imbalance': imbalance.price_imbalance, 'depth_imbalance': imbalance.depth_imbalance, 'momentum_score': imbalance.momentum_score, 'liquidity_score': liquidity } except Exception as e: logger.error(f"Error generating market summary: {e}") return {}