core/bfxbot/models.py

import inspect
import time
from enum import Enum
from typing import List, Dict

from bfxapi import Order, Position

from bfxbot.currency import Symbol


def __add_to_dict_list__(dict: Dict[int, List], k, v):
    if k not in dict:
        dict[k] = [v]
        return

    dict[k].append(v)


class PositionState(Enum):
    CRITICAL = -1,
    LOSS = 0,
    BREAK_EVEN = 1,
    MINIMUM_PROFIT = 2,
    PROFIT = 3

    def color(self) -> str:
        if self == self.LOSS or self == self.CRITICAL:
            return "red"
        elif self == self.BREAK_EVEN:
            return "yellow"
        else:
            return "green"


class EventKind(Enum):
    NEW_MINIMUM = 1,
    NEW_MAXIMUM = 2,
    REACHED_LOSS = 3,
    REACHED_BREAK_EVEN = 4,
    REACHED_MIN_PROFIT = 5,
    REACHED_GOOD_PROFIT = 6,
    REACHED_MAX_LOSS = 7,
    CLOSE_POSITION = 8,
    TRAILING_STOP_SET = 9,
    TRAILING_STOP_MOVED = 10,
    ORDER_SUBMITTED = 11,
    NEW_TICK = 12


class Ticker:
    def __init__(self, sec) -> None:
        self.seconds: int = sec
        self.start_time = time.time()
        self.current_tick: int = 1

    def inc(self):
        self.current_tick += 1


class Event:
    def __init__(self, kind: EventKind, pid: int, tick: int) -> None:
        self.kind: EventKind = kind
        self.tick: int = tick
        # position ID
        self.pid: int = pid

    def __repr__(self) -> str:
        return f"[{self.pid}]: {self.kind.name} @ Tick {self.tick}"


class PositionWrapper:
    def __init__(self, position: Position):
        self.position: Position = position
        self.stop_percentage: float = None
        self.state: PositionState = PositionState.LOSS

    def get_stop_percentage(self) -> float:
        return self.stop_percentage

    def get_state(self) -> PositionState:
        return self.state

    def set_stop_percentage(self, perc: float):
        self.stop_percentage = perc

    def set_state(self, state: PositionState):
        if not isinstance(state, PositionState):
            return

        self.state = state


class SymbolStatus:
    def __init__(self, symbol: Symbol, strategy=None):
        self.symbol = symbol
        self.eh = EventHandler()
        self.prices: Dict[int, float] = {}
        self.events: List[Event] = []
        self.orders: Dict[int, List[Order]] = {}
        self.positions: Dict[int, List[PositionWrapper]] = {}
        self.current_tick: int = 1
        self.strategy: Strategy = strategy

    def last_events(self, n) -> List[Event]:
        return self.events[-n:]

    def last_positions(self) -> List[PositionWrapper]:
        return self.positions[self.current_tick]

    # Applies strategy and adds position to list
    async def add_position(self, position: Position):
        pw = PositionWrapper(position)

        if self.strategy:
            await self.__apply_strategy_to_position__(pw)
        __add_to_dict_list__(self.positions, self.current_tick, pw)

    def add_order(self, order: Order):
        if self.strategy:
            self.strategy.order_on_tick(order, self)
        __add_to_dict_list__(self.orders, self.current_tick, order)

    def previous_position_w(self, pid: int) -> PositionWrapper:
        if self.current_tick == 1:
            return None

        return next(filter(lambda x: x.pid == pid, self.positions[self.current_tick - 1]))

    async def __add_event__(self, event: Event):
        self.events.append(event)
        await self.eh.call_event(self, event)

    async def __apply_strategy_to_position__(self, pw: PositionWrapper):
        (new_state, events) = self.strategy.position_on_tick(pw.position, self)

        if isinstance(new_state, PositionState):
            await self.__update_position_state__(pw, new_state)

        if isinstance(events, List):
            for e in events:
                if isinstance(e, Event):
                    await self.__add_event__(e)

    async def __update_position_state__(self, pw: PositionWrapper, state: PositionState):
        pw.set_state(state)
        await self.eh.call_state(self, pw)

    def set_price(self, tick, price):
        self.prices[tick] = price

class Strategy:
    """
    Defines new position state and events after tick.
    """

    def position_on_tick(self, position: Position, ss: SymbolStatus) -> (PositionState, List[Event]):
        pass

    """
    Defines new order state and events after tick.
    """

    def order_on_tick(self, order: Order, ss: SymbolStatus):
        pass


class EventHandler:
    def __init__(self):
        self.event_handlers = {}
        self.state_handlers = {}
        self.any_events = []
        self.any_state = []

    async def call_event(self, status: SymbolStatus, event: Event):
        value = event.kind.value

        # print("CALLING EVENT: {}".format(event))
        if value in self.event_handlers:
            for h in self.event_handlers[value]:
                if inspect.iscoroutinefunction(h):
                    await h(event, status)
                else:
                    h(event, status)

        for h in self.any_events:
            if inspect.iscoroutinefunction(h):
                await h(event, status)
            else:
                h(event, status)

    async def call_state(self, status: SymbolStatus, pw: PositionWrapper):
        state = pw.state

        if state in self.state_handlers:
            for h in self.state_handlers[state]:
                if inspect.iscoroutinefunction(h):
                    await h(status)
                else:
                    h(status)

        for h in self.any_state:
            if inspect.iscoroutinefunction(h):
                await h(status)
            else:
                h(status)

    def on_event(self, kind: EventKind):
        value = kind.value

        def registerhandler(handler):
            if value in self.event_handlers:
                self.event_handlers[value].append(handler)
            else:
                self.event_handlers[value] = [handler]
            return handler

        return registerhandler

    def on_state(self, state: PositionState):
        def registerhandler(handler):
            if state in self.state_handlers:
                self.state_handlers[state].append(handler)
            else:
                self.state_handlers[state] = [handler]
            return handler

        return registerhandler

    def on_any_event(self):
        def registerhandle(handler):
            self.any_events.append(handler)
            return handler
        return registerhandle

    def on_any_state(self):
        def registerhandle(handler):
            self.any_state.append(handler)
            return handler
        return registerhandle