''' This file houses the main simulation code that is going to test the algorithms. It needs to be able to manage the data available to the algorithm, help the brokerage simulator with stock data requests, and iterate through time. ''' from datetime import datetime, timedelta from pytz import timezone EASTERN = timezone('US/Eastern') from Historical_Data_Accessor import HistoricalDataAccessor from Orders import GenericLimitOrder, BuyOrder, SellOrder, OrderFilledResponse from Common import TickData from Brokerage_Simulation import BrokerageSimulation, AccountInformation from typing import TYPE_CHECKING if TYPE_CHECKING: from strategies.Strategy_Template import StrategyTemplate from strategies.Random_Strategy import RandomStrategy class ModelSimulation(): '''Simulation to test a model over historical data. Only tests a model with 1 stock symbol as provided.''' def __init__(self, symbol: str, datetime_start: datetime, datetime_end: datetime, datetime_delta: timedelta, strategy: 'StrategyTemplate', starting_cash: float): '''Specify the datetime to start the simulation and end it''' self._symbol = symbol self._datetime_start = datetime_start self._datetime_end = datetime_end self._datetime_delta = datetime_delta self._strategy = strategy self._status = 0 self._historical_data_accessor: HistoricalDataAccessor = HistoricalDataAccessor() if self._historical_data_accessor.status != 0: # Something went wrong self._status = 1 if self.check_datetimes() == False: # Not valid datetimes self._status = 1 # Make a BrokerageSimulation instance self._brokerage_simulation: BrokerageSimulation = BrokerageSimulation() self._brokerage_simulation.deposit_cash(starting_cash) self._brokerage_simulation.attach_strategy(self._strategy) self._strategy.attach_brokerage_simulation(self._brokerage_simulation) @property def status(self) -> int: return self._status def check_datetimes(self) -> bool: '''Check the start and end datetimes to make sure we have datapoints for them. Also, make sure start is before end. Return True for good datetimes and data available, false for bad datetimes or no data''' # Check if there is a datetime provided. If not, set it to eastern if self._datetime_start.tzinfo is None: self._datetime_start = EASTERN.localize(self._datetime_start) if self._datetime_end.tzinfo is None: self._datetime_end = EASTERN.localize(self._datetime_end) # Remove the seconds and microseconds self._datetime_start = self._datetime_start.replace(second=0, microsecond=0) self._datetime_end = self._datetime_end.replace(second=0, microsecond=0) # Check start is before end if self._datetime_start >= self._datetime_end: return False # Request if data is available using the data accessor if (self._historical_data_accessor.is_datapoint_available(self._symbol, self._datetime_start) == False or self._historical_data_accessor.is_datapoint_available(self._symbol, self._datetime_end) == False): return False # Datetimes are correctly ordered and we have data for them return True def run(self) -> None: '''Main function to run''' current_datetime = self._datetime_start while current_datetime <= self._datetime_end: # Pull the current datetime's data from HistoricalDataAccessor new_data = self._historical_data_accessor.get_single_tick_data(self._symbol, current_datetime) # Give the strategy and broker simulator the data for this datetime self._brokerage_simulation.set_current_tick_data(new_data) # Have the broker simulator check open orders for execution # If there are executed orders, relay the respose to the strategy filled_responses = self._brokerage_simulation.check_open_orders() # Set the new datetime for the strategy self._strategy.set_current_datetime(current_datetime) # Have the strategy send new buy/sell orders to the broker simulator for this datetime new_orders = self._strategy.react_to_market() # Add data regarding the account and brokerage simulator to an array for data analysis at the end # Include things like account values, open orders, executed orders, etc # The strategy should do its own data storage for important parts about its analysis # TODO self._brokerage_simulation.print_account_information() # Iterate to next datetime current_datetime += self._datetime_delta # With the iterations done, do some calculations on the performance of the model # Average monthly ROI, minimums, maximums, how it beats the underlying performance, etc # Save the data to files # Output general information from the sim def main(): datetime_first = EASTERN.localize(datetime(year=2020, month=1, day=8, hour=12, minute=30)) datetime_last = EASTERN.localize(datetime(year=2020, month=1, day=8, hour=13, minute=30)) symbol = "AMD" strategy = RandomStrategy(symbol) model_simulation = ModelSimulation( symbol=symbol, datetime_start=datetime_first, datetime_end=datetime_last, datetime_delta=timedelta(minutes=1), strategy=strategy, starting_cash=10_000 ) model_simulation.run() if __name__ == "__main__": main()