""".#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
|
