""".#book.py package for handling book.""" import decimal import heapq from enum import Enum import time decimal.getcontext().prec = 3 class OrderSide(Enum): """Order side enum.""" Buy = "BUY", Sell = "SELL" class ProductTypeEnum(Enum): """Product type enum.""" TEST = 0, FUTURE = 1, SPREAD = 2, CALL = 3, PUT = 4 class UnknownProductTypeError(Exception): """Exception for unknown product type.""" pass class Order: """Order class.""" price: decimal.Decimal side: OrderSide remaining_volume: int filled_volume: int timestamp: float id: str def __init__(self, price: decimal.Decimal(3), side: OrderSide, volume: int, timestamp: float, id: str): """Initialise class.""" self.price = price self.side = side self.remaining_volume = volume self.filled_volume = 0 self.timestamp = timestamp self.id = id class Level: """Level class.""" price: decimal.Decimal volume: int side: OrderSide timestamp: float id: str def __init__(self, order: Order): """Initialise class.""" self.price = order.price self.volume = order.remaining_volume self.side = order.side self.timestamp = order.timestamp self.id = order.id def __lt__(self, other): """Less than comparator.""" if (self.side == OrderSide.Sell): if (self.price < other.price): return True elif (self.price == other.price and self.timestamp < other.timestamp): return True else: return False else: if (self.price > other.price): return True elif (self.price == other.price and self.timestamp < other.timestamp): return True else: return False def __gt__(self, other): """Greater than comparator.""" if (self.side == OrderSide.Sell): if (self.price > other.price): return True elif (self.price == other.price and self.timestamp < other.timestamp): return True else: return False else: if (self.price < other.price): return True elif (self.price == other.price and self.timestamp < other.timestamp): return True else: return False def __le__(self, other): """Less than or equal compatator.""" if (self.side == OrderSide.Sell): if (self.price <= other.price): return True elif (self.price == other.price and self.timestamp <= other.timestamp): return True else: return False else: if (self.price >= other.price): return True elif (self.price == other.price and self.timestamp <= other.timestamp): return True else: return False def __ge__(self, other): """Greater than or equal comparator.""" if (self.side == OrderSide.Sell): if (self.price >= other.price): return True elif (self.price == other.price and self.timestamp <= other.timestamp): return True else: return False else: if (self.price <= other.price): return True elif (self.price == other.price and self.timestamp <= other.timestamp): return True else: return False def __eq__(self, other): """Equalty compatator.""" if (self.price == other.price and self.timestamp == other.timestamp): return True else: return False def addVolume(self, volume: int): """Add volume to price level.""" self.volume += volume def __str__(self): """Turn into str for printing.""" return ("Price: " + str(self.price) + ", volume: " + str(self.volume) + ", time: " + str(self.timestamp) + ", id: " + self.id) class Side: """Side class.""" levels: [Level] def __init__(self): """Initialise class.""" self.levels = [] def delete(self, orderId: str): """Delete order from side.""" for i in self.levels: if (i.id == orderId): self.levels.remove(i) heapq.heapify(self.levels) class Book: """Book class.""" bidSide: Side askSide: Side productType: ProductTypeEnum product: str stationId: str unit: str expiry: time.struct_time aggFee: decimal.Decimal pasFee: decimal.Decimal broFee: decimal.Decimal def __init__(self, productType: str = "TEST", product: str = "a", stationId: str = "b", unit: str = "c", expiry: str = time.strftime("%Y-%m-%d %H:%M%z", time.localtime(0)), aggFee: decimal.Decimal = 1, pasFee: decimal.Decimal = -1, broFee: decimal.Decimal = 2): """Initialise module.""" self.bidSide = Side() self.askSide = Side() if productType == "FUTURE": self.productType = ProductTypeEnum.FUTURE elif productType == "SPREAD": self.productType = ProductTypeEnum.SPREAD elif productType == "CALL": self.productType = ProductTypeEnum.CALL elif productType == "PUT": self.productType = ProductTypeEnum.PUT elif productType == "TEST": self.productType = ProductTypeEnum.TEST else: raise UnknownProductTypeError(productType) self.product = product self.stationId = stationId self.unit = unit self.expiry = time.strptime(expiry, "%Y-%m-%d %H:%M%z") self.aggFee = aggFee self.pasFee = pasFee self.broFee = broFee def addOrder(self, order: Order): """Add order to book.""" if (order.side == OrderSide.Buy): self.bid(order) else: self.ask(order) def ask(self, order: Order): """Add ask to book.""" while(len(self.bidSide.levels) > 0 and self.bidSide.levels[0].price >= order.price): if (self.bidSide.levels[0].volume > order.remaining_volume): temp = self.bidSide.levels[0].volume order.filled_volume += self.bidSide.levels[0].volume self.bidSide.levels[0].volume -= order.remaining_volume order.remaining_volume -= temp break else: order.remaining_volume -= self.bidSide.levels[0].volume order.filled_volume += self.bidSide.levels[0].volume heapq.heappop(self.bidSide.levels) if (order.remaining_volume > 0): heapq.heappush(self.askSide.levels, Level(order)) def bid(self, order: Order): """Add bid to book.""" while(len(self.askSide.levels) > 0 and self.askSide.levels[0].price <= order.price): if (self.askSide.levels[0].volume > order.remaining_volume): temp = self.askSide.levels[0].volume order.filled_volume += self.askSide.levels[0].volume self.askSide.levels[0].volume -= order.remaining_volume order.remaining_volume -= temp break else: order.remaining_volume -= self.askSide.levels[0].volume order.filled_volume -= self.askSide.levels[0].volume heapq.heappop(self.askSide.levels) if (order.remaining_volume > 0): heapq.heappush(self.bidSide.levels, Level(order)) def printBook(self): """Test print book. Book is a heap so not strictly sorted.""" print("Sell side") for i in sorted(self.askSide.levels, reverse=True)[-10:]: print(i) print("Buy side") for i in sorted(self.bidSide.levels)[:10]: print(str(i)) def testBook(orders: int = 10, printBook=True): """Test an example book.""" b = Book() time = 1 for i in range(1, orders): b.addOrder(Order(price=i, side=OrderSide.Buy, volume=10, timestamp=time, id="a")) time += 1 for i in range(orders + 1, 2 * orders): b.addOrder(Order(i, OrderSide.Sell, 10, time, "a")) time += 1 if (printBook): b.printBook() return b