core/bfxbot/models.py

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

from bfxapi import Order, Position

from bfxbot.currency import Symbol

OFFER_PERC = 0.008
TAKER_FEE = 0.2
MAKER_FEE = 0.1


def __add_to_dict_list__(dictionary: Dict[int, List], k, v) -> Dict[int, List]:
    if k not in dictionary:
        dictionary[k] = [v]
    else:
        dictionary[k].append(v)

    return dictionary


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 EventMetadata:
    def __init__(self, position_id: int = None, order_id: int = None):
        self.position_id: int = position_id
        self.order_id: int = order_id


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

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

    def __str__(self):
        return f"{self.name}"

    def __repr__(self):
        return self.__str__()


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, tick: int, metadata: EventMetadata = None) -> None:
        self.kind: EventKind = kind
        self.tick: int = tick
        self.metadata: EventMetadata = metadata

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

    def has_metadata(self) -> bool:
        return self.metadata is not None


class PositionWrapper:
    def __init__(self, position: Position, state: PositionState = PositionState.UNDEFINED,
                 net_profit_loss: float = None,
                 net_profit_loss_percentage: float = None):
        self.position: Position = position
        self.__net_profit_loss: float = net_profit_loss
        self.__net_profit_loss_percentage: float = net_profit_loss_percentage
        self.__state: PositionState = state

    def net_profit_loss(self) -> float:
        return self.__net_profit_loss

    def net_profit_loss_percentage(self) -> float:
        return self.__net_profit_loss_percentage

    def set_state(self, state: PositionState):
        self.__state = state

    def state(self) -> PositionState:
        return self.__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 __init_tick__(self, tick: int):
        self.current_tick = tick
        self.prices[self.current_tick] = None
        self.orders[self.current_tick] = []
        self.positions[self.current_tick] = []

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

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

    # Applies strategy and adds position to list
    async def add_position(self, position: Position):
        events = []

        # if a strategy is defined then the strategy takes care of creating a PW for us
        if not self.strategy:
            pw = PositionWrapper(position)
        else:
            pw, events = await self.__apply_strategy_to_position__(position)
        self.positions = __add_to_dict_list__(self.positions, self.current_tick, pw)

        # triggering state callbacks
        await self.__trigger_position_state_callbacks__(pw)

        # triggering events callbacks
        for e in events:
            if not isinstance(e, Event):
                raise ValueError
            await self.add_event(e)

    def all_prices(self) -> List[float]:
        return list(map(lambda x: self.prices[x], range(1, self.current_tick + 1)))

    def all_ticks(self) -> List[int]:
        return [x for x in range(1, self.current_tick + 1)]

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

    def current_price(self):
        return self.prices[self.current_tick]

    def previous_pw(self, pid: int) -> Optional[PositionWrapper]:
        if self.current_tick == 1:
            return None

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

    def active_position_wrapper_from_id(self, position_id: int) -> Optional[PositionWrapper]:
        if self.current_tick in self.positions:
            for pw in self.positions[self.current_tick]:
                if pw.position.id == position_id:
                    return pw
        return None

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

    async def __apply_strategy_to_position__(self, position: Position) -> Tuple[PositionWrapper, List[Event]]:
        pw, events = self.strategy.position_on_new_tick(position, self)

        if not isinstance(pw, PositionWrapper):
            raise ValueError

        if not isinstance(events, list):
            raise ValueError

        return pw, events

    async def __trigger_position_state_callbacks__(self, pw: PositionWrapper):
        await self.eh.call_position_state(self, pw)


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

    def position_on_new_tick(self, position: Position, ss: SymbolStatus) -> Tuple[PositionWrapper, List[Event]]:
        pass

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

    def order_on_new_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_position_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(pw, status)
                else:
                    h(pw, status)

        for h in self.any_state:
            if inspect.iscoroutinefunction(h):
                await h(pw, status)
            else:
                h(pw, 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_position_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_position_state(self):
        def registerhandle(handler):
            self.any_state.append(handler)
            return handler

        return registerhandle