''' The brokerage simulator needs to simulate what IBKR does for the user. 1. Submitting "orders" to the market 2. Determining when orders would have been filled 3. Keeping track of account information such as positions, cash balances, etc 4. This brokerage simulation is not responsible for feeding data to the algorithm It does need to get data from the ''' from datetime import datetime from ibapi.contract import Contract from Account_Information import AccountInformation from Orders import GenericLimitOrder, BuyOrder, SellOrder, OrderFilledResponse from typing import TYPE_CHECKING if TYPE_CHECKING: from Common import TickData from strategies.Strategy_Template import StrategyTemplate class BrokerageSimulation(): def __init__(self): self._account_information = AccountInformation() self._current_tick_data: 'TickData' = None self._strategy: 'StrategyTemplate' = None self._current_order_id: int = 0 # Internal functions def calculate_commission(self, order: GenericLimitOrder) -> float: return 0.0 def make_filled_response(self, order: GenericLimitOrder) -> OrderFilledResponse: '''Execute an order. Return an OrderFilledResponse for the order. Assume the stock filled at the limit price of the order placed''' order_filled_response = OrderFilledResponse( symbol = order.symbol, matching_order_id = order.id, datetime_filled = TickData.open_datetime, quantity = order.quantity, price = order.limit_price, commission = self.calculate_commission(order) ) return order_filled_response # General functions def print_account_information(self) -> None: current_datetime = self._current_tick_data.open_datetime print(f"Account information on {current_datetime}") print(self._account_information) print() def get_current_account_info(self) -> AccountInformation: return self._account_information # Functions for the ModelSimulator to interact with def attach_strategy(self, strategy: 'StrategyTemplate') -> None: self._strategy = strategy # Add the internal account_information so that the strategy has access to it without constantly pulling self._strategy.attach_account_information(self._account_information) def set_current_tick_data(self, tick_data: 'TickData') -> None: self._current_tick_data = tick_data def check_open_orders(self) -> list[OrderFilledResponse]: '''For every open order, check if it can be filled. If so, fill it. Send the list to the attached strategy. Return a list of OrderFilledResponse for every order filled ''' filled_responses: list[OrderFilledResponse] = [] open_orders = self._account_information.open_orders for order in open_orders: if order.is_fillable(tick_data=self._current_tick_data): order_id = order.id # Generate the filled response to send to the strategy filled_response = self.make_filled_response(order) filled_responses.append(filled_response) # Move the order from open_orders to filled_orders status = self._account_information.fill_open_order(order_id=order_id) # Add the OrderFilledResponse to the AccountInformation._cumulative_stock_positions[symbol] self._account_information.add_position_from_filled_response(filled_response) # Send the filled_responses to the strategy so it knows the order was filled return filled_responses # Functions for the strategy to interact with def deposit_cash(self, amount: float) -> None: self._account_information.deposit_cash(amount) def submit_order(self, order: GenericLimitOrder) -> OrderFilledResponse: '''Adds an order to open_orders''' order_id = self._current_order_id self._current_order_id += 1 order.change_id(order_id) self._account_information.add_open_order(order) def cancel_order(self, order_id: int) -> int: '''Cancels a current order that is open based on the order_id. Returns 0 on success, 1 on order_id cannot be found, 2 on the order_id has already been filled ''' pass