Setup event loop refactor

Author: MDUYN

investing_algorithm_framework/app/eventloop.py

@@ -0,0 +1,235 @@
+from typing import List
+from datetime import  datetime, timedelta, timezone
+from investing_algorithm_framework.domain import TimeUnit, ENVIRONMENT, \
+    Environment, BACKTESTING_INDEX_DATETIME
+from .strategy import TradingStrategy
+
+
+class EventLoopService:
+    """
+    A service that manages the event loop for the trading bot.
+    This service is responsible for running the trading strategy and handling
+    events in its lifecycle, such as pending orders changes, stop loss changes,
+    take profit changes, and price data updates.
+
+    The event loop runs strategies in a so called interation, which consists
+    out of various tasks. An iteration consists out of the following tasks:
+
+        - Collect all strategies and tasks that need to be
+            run (overdue on schedule)
+        - Collect all market data for the strategies
+        - Check all pending orders and update their status if needed
+        - Check all stop loss orders and update their status if needed
+        - Check all take profit orders and update their status if needed
+        - Run all tasks
+        - Run all strategies
+        - Run all on_strategy_run hooks
+        - Snapshot the portfolios based on the defined snapshot interval
+
+    The goal of this service is to provide a way to run the trading in the
+    most efficient way possible in both live trading and backtesting. This
+    is achieved by running strategies and tasks in a loop, where each
+    iteration checks which strategies and tasks are due to run based on their
+    defined intervals and time units (seconds, minutes, hours). The next run
+    times for each strategy are initialized to the current time in UTC.
+    The service also collects all data configurations from the strategies and
+    tasks, and runs them in a single iteration to avoid multiple calls to the
+    data provider service, which can be expensive in terms of performance.
+    """
+
+    def __init__(
+        self,
+        order_service,
+        portfolio_service,
+        configuration_service,
+        data_provider_service=None
+    ):
+        """
+        Initializes the event loop service with the provided algorithm.
+
+        Args:
+            order_service: The service responsible for managing orders.
+            portfolio_service: The service responsible for managing portfolios.
+            configuration_service: The service responsible for configuration.
+        """
+        self.tasks = []
+        self._algorithm = None
+        self._strategies = []
+        self._order_service = order_service
+        self._portfolio_service = portfolio_service
+        self._configuration_service = configuration_service
+        self._data_provider_service = data_provider_service
+        self._data_configurations = []
+        self.next_run_times = {}
+
+    def _get_due_strategies(self):
+        """
+        Checks which strategies are due to run based on their defined intervals
+        Returns:
+
+        """
+        environment = self._configuration_service.config[ENVIRONMENT]
+
+        if Environment.BACKTEST.equals(environment):
+            now = self._configuration_service\
+                .config[BACKTESTING_INDEX_DATETIME]
+        else:
+            now = datetime.now(timezone.utc)
+
+        due = []
+
+        for strategy in self._strategies:
+            interval = timedelta(
+                **{strategy.time_unit.value.lower(): strategy.interval}
+            )
+
+            if now >= self.next_run_times[strategy]:
+                due.append(strategy)
+                self.next_run_times[strategy] = now + interval
+
+        return due
+
+    def initialize(self, algorithm):
+        """
+        Initializes the event loop service by calculating the schedule for
+        running strategies and tasks based on their defined intervals and time
+        units (seconds, minutes, hours).
+
+        The next run times for each strategy are initialized to the current
+        time in UTC.
+
+        Args:
+            algorithm: The trading algorithm to be managed by this service.
+
+        Returns:
+            None
+        """
+        
+        self._algorithm = algorithm
+        self._strategies = algorithm.strategies
+        self.next_run_times = {
+            strategy.identifier: {
+                "next_run": datetime.now(timezone.utc)
+                "data_configurations": strategy.data_configurations
+            }
+            for strategy in self._strategies
+        }
+
+        # Collect all data configurations
+        for strategy in self._strategies:
+            self._data_configurations.append(
+                strategy.data_configurations
+            )
+
+    def start(self, number_of_iterations=None):
+        """
+        Runs the event loop for the trading algorithm.
+
+        Args:
+            number_of_iterations: Optional; the number of iterations to run.
+                If None, runs indefinitely.
+        """
+        pass
+
+    def _run_iteration_backtest(self):
+        """
+        Runs a single iteration of the event loop in backtesting mode.
+        This method collects all due strategies, fetches their data
+        configurations, and runs the strategies with the collected data.
+
+        Returns:
+            None
+        """
+        due_strategies = self._get_due_strategies()
+
+        if not due_strategies:
+            return
+
+        # Step 1: Collect all data
+        data_configurations = []
+
+        for strategy in due_strategies:
+            data_configurations.extend(strategy.data_configurations)
+
+        # Make sure we have unique data configurations
+        data_configurations = list(set(data_configurations))
+        data_object = {}
+
+        for data_config in data_configurations:
+            data_object[data_config.identifier] = \
+                self._data_provider_service.get_backest_data(
+                    symbol=data_config.symbol,
+                    data_type=data_config.data_type,
+                    date=data_config.date,
+                    market=data_config.market,
+                    time_frame=data_config.time_frame,
+                    start_date=data_config.start_date,
+                    end_date=data_config.end_date,
+                    window_size=data_config.window_size,
+                    pandas=data_config.pandas,
+                )
+
+        # Step 2: Update prices of trades
+
+        # Step 3: Check pending orders
+
+        # Step 4: Check stop loss orders
+
+        # Step 5: Check take profit orders
+
+        # Step 6: Run all tasks
+
+        # Step 7: Run all strategies
+
+    def _run_iteration(self):
+        """
+        Runs a single iteration of the event loop. This method collects all
+        due strategies, fetches their data configurations, and runs the
+        strategies with the collected data. It also checks for pending orders,
+        stop loss orders, and take profit orders, and updates their status if
+        needed. Finally, it runs all tasks and strategies, and takes a snapshot
+        of the portfolios if needed.
+
+        Returns:
+            None
+        """
+        due_strategies = self._get_due_strategies()
+
+        if not due_strategies:
+            return
+
+        # Step 1: Collect all data
+        data_configurations = []
+
+        for strategy in due_strategies:
+            data_configurations.extend(strategy.data_configurations)
+
+        # Make sure we have unique data configurations
+        data_configurations = list(set(data_configurations))
+        data_object = {}
+
+        for data_config in data_configurations:
+            data_object[data_config.identifier] = \
+                self._data_provider_service.get_data(
+                    symbol=data_config.symbol,
+                    data_type=data_config.data_type,
+                    date=data_config.date,
+                    market=data_config.market,
+                    time_frame=data_config.time_frame,
+                    start_date=data_config.start_date,
+                    end_date=data_config.end_date,
+                    window_size=data_config.window_size,
+                    pandas=data_config.pandas,
+                )
+
+        # Step 2: Update prices of trades
+
+        # Step 3: Check pending orders
+
+        # Step 4: Check stop loss orders
+
+        # Step 5: Check take profit orders
+
+        # Step 6: Run all tasks
+
+        # Step 7: Run all strategies

investing_algorithm_framework/domain/__init__.py

@@ -20,7 +20,7 @@
     PortfolioConfiguration, Portfolio, Position, Order, TradeStatus, \
     BacktestResult, PortfolioSnapshot, StrategyProfile, \
     BacktestPosition, Trade, MarketCredential, PositionSnapshot, \
-    AppMode, BacktestDateRange, DateRange, \
+    AppMode, BacktestDateRange, DataType, DataSource, \
     MarketDataType, TradeRiskType, TradeTakeProfit, TradeStopLoss, \
     DataSource, Event, SnapshotInterval
 from .order_executor import OrderExecutor
@@ -113,7 +113,6 @@
     "RoundingService",
     "BacktestDateRange",
     "convert_polars_to_pandas",
-    "DateRange",
     "DEFAULT_LOGGING_CONFIG",
     "DATABASE_DIRECTORY_NAME",
     "BACKTESTING_INITIAL_AMOUNT",
@@ -138,5 +137,7 @@
     "SNAPSHOT_INTERVAL",
     "SnapshotInterval",
     "AWS_S3_STATE_BUCKET_NAME",
-    "AWS_LAMBDA_LOGGING_CONFIG"
+    "AWS_LAMBDA_LOGGING_CONFIG",
+    "DataType",
+    "DataSource",
 ]

investing_algorithm_framework/domain/models/__init__.py

@@ -13,11 +13,10 @@
     TradeRiskType
 from .trading_data_types import TradingDataType
 from .trading_time_frame import TradingTimeFrame
-from .date_range import DateRange
 from .market_data_type import MarketDataType
-from .data_source import DataSource
 from .snapshot_interval import SnapshotInterval
 from .event import Event
+from .data import DataSource, DataType
 
 __all__ = [
     "OrderStatus",
@@ -40,10 +39,10 @@
     "Trade",
     "MarketCredential",
     "TradeStatus",
-    "BacktestReportsEvaluation",
+    "DataType",
     "AppMode",
     "BacktestDateRange",
-    "DateRange",
+    "DataSource",
     "MarketDataType",
     "TradeStopLoss",
     "TradeTakeProfit",

investing_algorithm_framework/domain/models/data/__init__.py

@@ -0,0 +1,7 @@
+from .data_source import DataSource
+from .data_type import DataType
+
+__all__ = [
+    "DataSource",
+    "DataType",
+]

investing_algorithm_framework/domain/models/data/data_source.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+from .data_type import DataType
+
+@dataclass()
+class DataSource:
+    """
+    Base class for data sources.
+    """
+    identifier: str = None
+    data_type: DataType = None
+    symbol: str = None
+    window_size: int = None
+    time_frame: str = None

investing_algorithm_framework/domain/models/data/data_type.py

@@ -0,0 +1,46 @@
+from enum import Enum
+
+
+class DataType(Enum):
+    OHLCV = "OHLCV"
+    TICKER = "TICKER"
+    ORDER_BOOK = "ORDER_BOOK"
+    CUSTOM = "CUSTOM"
+
+    @staticmethod
+    def from_string(value: str):
+
+        if isinstance(value, str):
+
+            for entry in DataType:
+
+                if value.upper() == entry.value:
+                    return entry
+
+            raise ValueError(
+                f"Could not convert {value} to DataType"
+            )
+
+    @staticmethod
+    def from_value(value):
+
+        if isinstance(value, str):
+            return DataType.from_string(value)
+
+        if isinstance(value, DataType):
+
+            for entry in DataType:
+
+                if value == entry:
+                    return entry
+
+        raise ValueError(
+            f"Could not convert {value} to TimeFrame"
+        )
+
+    def equals(self, other):
+
+        if isinstance(other, Enum):
+            return self.value == other.value
+        else:
+            return DataType.from_string(other) == self