Source code for flashcrashed.detector
import logging
from abc import ABC, abstractmethod
from collections import deque
log = logging.getLogger('flashcrashed')
crash_log = logging.getLogger("flashcrashed.crash")
[docs]class Detector(ABC):
def __init__(self):
self.symbol = 'Default' # for logging purposes
self.last_action = None
self._count = 0
self._log_same_every = 10
[docs] @abstractmethod
def reset(self):
raise NotImplementedError # pragma: nocover
@abstractmethod
def _predict(self, price):
raise NotImplementedError # pragma: nocover
[docs] def predict(self, price):
action = self._predict(price)
if action == 0:
crash_log.info(
'%6s - CRASH detected at %.2f' % (self.symbol, price))
elif action == 2:
crash_log.info(
'%6s - RISE detected at %.2f' % (self.symbol, price))
else:
log.info('%6s - Doing nothing at %.2f' % (self.symbol, price))
self._count += 1
if action == self.last_action == 1:
if self._count % self._log_same_every != 0:
return action
self.last_action = action
return action
[docs]class BasicDetector(Detector):
drop_ratio = 3
resell_ratio = 2.75
history_length = 150
def __init__(self):
super(BasicDetector, self).__init__()
self.prices = deque(maxlen=self.history_length)
self.bought = False
self.buy_price = 0.
[docs] def reset(self):
self.prices = deque(maxlen=self.history_length)
self.bought = False
self.buy_price = 0.
def _predict(self, price):
self.prices.append(price)
prices = list(sorted(self.prices))[-int(self.history_length/2):]
avg_price = sum(prices) / int(self.history_length/2)
if (
price <= avg_price / self.drop_ratio
and len(self.prices) == self.prices.maxlen
and not self.bought
):
self.bought = True
self.buy_price = price
return 0
if price >= self.buy_price * self.resell_ratio and self.bought:
self.bought = False
self.buy_price = 0
return 2
return 1