from abc import ABC, abstractmethod from model import Model from position import Position class Bot(ABC): def __init__(self, model: Model, nolog=False): self.previous_obs = None self.positions = list() self.num_open_pos = 0 self.nUnits_available = 0 # the number of tokens in the denominating unit that the bot can work with self.model = model self.nolog = nolog def update_num_open_pos(self): self.num_open_pos = 0 for p in self.positions: if p.open: self.num_open_pos += 1 @abstractmethod def push_new_obs(self, current_price, timestamp): """ when a new price is observed this function is called to invoke bot actions""" # self.model.add_model_data(timestamp=timestamp, newObs=current_price) # update existing open positions - maybe close them # self.update_positions(current_price=current_price, timestamp=timestamp) # check whether a new position needs to be opened pass @abstractmethod def open_position(self, current_price, timestamp): """ open a new position """ pass @abstractmethod def update_positions(self, current_price, timestamp): """ update positions in self.positions """ # for p in self.positions: # if p.open: # update position returns nUnits (size of position) if a position was closed # or None if position stays open # nUnits_pos = p.update_position(current_price=current_price, ) # if nUnits_pos: # self.nUnits += nUnits_pos def close_all_pos(self, current_price): for p in self.positions: if p.open: p.set_position_closed(current_price=current_price) self.nUnits += p.nUnits def log_msg(self, msg): """ write message to log """ if self.nolog: pass else: print(msg)