""" In-memory fast temporary storage """ import bisect from collections import defaultdict, deque from time import monotonic from typing import List class Quote: """ How much something costs at which time since the process started """ __slots__ = ["time", "price"] def __init__(self, time: float, price: float): self.time = time self.price = price class MemStorage: """ In-memory storage and notifier class """ def __init__(self): # FIFO queue where old records get discarded self.__time_ordered_queues = defaultdict(deque) # Discarded records get looked up in an ordered list with O(log(n)) and value is removed # ordered list is maintained ordered self.__value_ordered_lists = defaultdict(list) # we remember what's hot self.__hot_stuff = set() def notify_hot(self, isin: str): """ Symbol isin is "hot". Send real-time notifications """ def process_quote(self, isin: str, price=float): """ Fast in-memory storage of values """ now = monotonic() q: deque[Quote] = self.__time_ordered_queues[isin] ol: List[float] = self.__value_ordered_lists[isin] while now - q[0].time > 600: item = q.popleft() index = bisect.bisect_left(ol, item.price) ol.pop(index) q.append(Quote(now, price)) insert_at = bisect.bisect_left(ol, price) ol.insert(insert_at, price) min_val = ol[0] max_val = ol[-1] is_hot = (max_val - min_val) / min_val > 0.1 if is_hot and isin not in self.__hot_stuff: self.__hot_stuff.add(isin) self.notify_hot(isin) elif not is_hot and isin in self.__hot_stuff: self.__hot_stuff.remove(isin)