287 lines
8.0 KiB
Python
287 lines
8.0 KiB
Python
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__(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) -> 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]:
|
|
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 __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
|