38 lines
1.6 KiB
Python
38 lines
1.6 KiB
Python
from typing import Dict, List, Tuple
|
|
|
|
|
|
class OrderbookManager:
|
|
def __init__(self):
|
|
self._book_bids: Dict[float, float] = {}
|
|
self._book_asks: Dict[float, float] = {}
|
|
|
|
def apply_updates(self,
|
|
bids_updates: List[Tuple[float, float]],
|
|
asks_updates: List[Tuple[float, float]]) -> None:
|
|
self._apply_partial_updates(self._book_bids, bids_updates)
|
|
self._apply_partial_updates(self._book_asks, asks_updates)
|
|
|
|
def get_total_volume(self) -> Tuple[float, float]:
|
|
total_bids = sum(self._book_bids.values()) if self._book_bids else 0.0
|
|
total_asks = sum(self._book_asks.values()) if self._book_asks else 0.0
|
|
return total_bids, total_asks
|
|
|
|
def get_top_levels(self) -> Tuple[List[List[float]], List[List[float]]]:
|
|
bids_sorted = self._build_top_levels(self._book_bids, reverse=True)
|
|
asks_sorted = self._build_top_levels(self._book_asks, reverse=False)
|
|
return bids_sorted, asks_sorted
|
|
|
|
def _apply_partial_updates(self,
|
|
side_map: Dict[float, float],
|
|
updates: List[Tuple[float, float]]) -> None:
|
|
for price, size in updates:
|
|
if size == 0.0:
|
|
side_map.pop(price, None)
|
|
elif size > 0.0:
|
|
side_map[price] = size
|
|
|
|
def _build_top_levels(self, side_map: Dict[float, float], reverse: bool) -> List[List[float]]:
|
|
items = [(p, s) for p, s in side_map.items() if s > 0]
|
|
items.sort(key=lambda x: x[0], reverse=reverse)
|
|
return [[p, s] for p, s in items]
|