137 lines
5.6 KiB
Python
137 lines
5.6 KiB
Python
'''
|
|
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() |