""".#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 = 0,
|
Sell = 1
|
|
|
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
|
|
def __init__(self, price: decimal.Decimal(3), side: OrderSide,
|
volume: int):
|
"""Initialise class."""
|
self.price = price
|
self.side = side
|
self.remaining_volume = volume
|
self.filled_volume = 0
|
|
|
class Level:
|
"""Level class."""
|
|
price: decimal.Decimal
|
volume: int
|
|
def __init__(self, price: decimal.Decimal(3), volume):
|
"""Initialise class."""
|
self.priority = int(price)
|
self.price = price
|
self.volume = volume
|
|
def __lt__(self, other):
|
"""Less than comparator."""
|
if (self.price < other.price):
|
return True
|
else:
|
return False
|
|
def __gt__(self, other):
|
"""Greater than comparator."""
|
if (self.price > other.price):
|
return True
|
else:
|
return False
|
|
def __le__(self, other):
|
"""Less than or equal compatator."""
|
if (self.price <= other.price):
|
return True
|
else:
|
return False
|
|
def __ge__(self, other):
|
"""Greater than or equal comparator."""
|
if (self.price >= other.price):
|
return True
|
else:
|
return False
|
|
def __eq__(self, other):
|
"""Equalty compatator."""
|
if (self.price == other.price):
|
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)
|
|
|
class Side:
|
"""Side class."""
|
|
levels: [Level]
|
|
def __init__(self):
|
"""Initialise class."""
|
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][1].price >= order.price):
|
if (self.bidSide.levels[0][1].volume > order.remaining_volume):
|
temp = self.bidSide.levels[0][1].volume
|
order.filled_volume += self.bidSide.levels[0][1].volume
|
self.bidSide.levels[0][1].volume -= order.remaining_volume
|
order.remaining_volume -= temp
|
break
|
else:
|
order.remaining_volume -= self.bidSide.levels[0][1].volume
|
order.filled_volume += self.bidSide.levels[0][1].volume
|
heapq.heappop(self.bidSide.levels)
|
if (order.remaining_volume > 0):
|
if (len(self.askSide.levels) > 0 and
|
self.askSide.levels[0][1].price == order.price):
|
self.askSide.levels[0][1].addVolume(order.remaining_volume)
|
else:
|
heapq.heappush(self.askSide.levels,
|
(order.price,
|
Level(order.price, order.remaining_volume)))
|
|
def bid(self, order: Order):
|
"""Add bid to book."""
|
while(len(self.askSide.levels) > 0 and
|
self.askSide.levels[0][1].price <= order.price):
|
if (self.askSide.levels[0][1].volume > order.remaining_volume):
|
temp = self.askSide.levels[0][1].volume
|
order.filled_volume += self.askSide.levels[0][1].volume
|
self.askSide.levels[0][1].volume -= order.remaining_volume
|
order.remaining_volume -= temp
|
break
|
else:
|
order.remaining_volume -= self.askSide.levels[0][1].volume
|
order.filled_volume -= self.askSide.levels[0][1].volume
|
heapq.heappop(self.askSide.levels)
|
if (order.remaining_volume > 0):
|
if (len(self.bidSide.levels) > 0 and
|
self.bidSide.levels[0][1].price == order.price):
|
self.bidSide.levels[0][1].addVolume(order.remaining_volume)
|
else:
|
heapq.heappush(self.bidSide.levels,
|
(100 - order.price,
|
Level(order.price, order.remaining_volume)))
|
|
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):
|
print(i[1])
|
print("Buy side")
|
for i in sorted(self.bidSide.levels):
|
print(str(i[1]))
|
|
|
def testBook():
|
"""Test an example book."""
|
b = Book()
|
for i in range(1, 10):
|
b.addOrder(Order(i, OrderSide.Buy, 10))
|
for i in range(11, 20):
|
b.addOrder(Order(i, OrderSide.Sell, 10))
|
b.printBook()
|
