from time import sleep from typing import Dict, List from bfxbot.bfxwrapper import BfxWrapper from bfxbot.currency import Symbol from bfxbot.models import SymbolStatus, Ticker, EventHandler, Strategy, Event, EventKind class BfxBot: def __init__(self, api_key: str, api_secret: str, symbols: List[Symbol], tick_duration: int = 1, ): if api_key is None: print("API_KEY is not set!") raise ValueError if api_secret is None: print("API_SECRET is not set!") raise ValueError self.__bfx: BfxWrapper = BfxWrapper(api_key, api_secret) self.__ticker: Ticker = Ticker(tick_duration) self.__status: Dict[Symbol, SymbolStatus] = {} if isinstance(symbols, Symbol): symbols = [symbols] self.symbols: List[Symbol] = symbols # init symbol statuses for s in self.symbols: self.__status[s] = SymbolStatus(s) async def __update_status__(self): active_positions = await self.__bfx.get_active_position() for symbol in self.__status: # updating tick self.__status[symbol].current_tick = self.__ticker.current_tick # updating last price last_price = await self.__bfx.get_current_prices(symbol) last_price = last_price[0] self.__status[symbol].set_tick_price(self.__ticker.current_tick, last_price) # updating positions symbol_positions = [x for x in active_positions if x.symbol == str(symbol)] for p in symbol_positions: await self.__status[Symbol.from_str(p.symbol)].add_position(p) # updating orders active_orders = await self.__bfx.get_active_orders(str(symbol)) for o in active_orders: self.__status[symbol].add_order(o) # emitting new tick event ## TODO: handle _on_new_tick() from Strategy await self.__status[symbol].__add_event__(Event(EventKind.NEW_TICK, self.__ticker.current_tick)) def close_order(self, symbol: Symbol, order_id: int): print(f"I would have closed order {order_id} for {symbol}") def close_position(self, symbol: Symbol, position_id: int): print(f"I would have closed order {position_id} for {symbol}") def set_strategy(self, symbol, strategy: Strategy): if symbol in self.__status: self.__status[symbol].strategy = strategy else: self.__status[symbol] = SymbolStatus(symbol, strategy) async def start(self): await self.__update_status__() def symbol_event_handler(self, symbol) -> EventHandler: if symbol not in self.__status: return None return self.__status[symbol].eh def symbol_status(self, symbol: Symbol) -> SymbolStatus: if symbol not in self.__status: return None return self.__status[symbol] async def update(self): sleep(self.__ticker.seconds) self.__ticker.inc() await self.__update_status__()