import inspect import time from enum import Enum from typing import List, Dict, Tuple 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 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 add_order(self, order: Order): if self.strategy: self.strategy.order_on_new_tick(order, self) __add_to_dict_list__(self.orders, self.current_tick, order) # Applies strategy and adds position to list async def add_position(self, position: Position): # 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 = await self.__apply_strategy_to_position__(position) __add_to_dict_list__(self.positions, self.current_tick, pw) 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) -> PositionWrapper: if self.current_tick == 1: return None return next(filter(lambda x: x.position.id == pid, self.positions[self.current_tick - 1])) def set_tick_price(self, tick, price): self.prices[tick] = price 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, position: Position): if not self.strategy: return pw, events = self.strategy.position_on_new_tick(position, self) if not isinstance(pw, PositionWrapper): raise ValueError if not isinstance(events, list): raise ValueError # 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) return pw 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(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_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