Setup initial event loop service

MDUYN · MDUYN · commit dd9773fc18df · 2025-07-10T11:46:06.000+02:00
diff --git a/investing_algorithm_framework/__init__.py b/investing_algorithm_framework/__init__.py
@@ -12,7 +12,7 @@
     TickerMarketDataSource, MarketService, \
     RESERVED_BALANCES, APP_MODE, AppMode, DATETIME_FORMAT, \
     BacktestDateRange, convert_polars_to_pandas, \
-    DEFAULT_LOGGING_CONFIG, \
+    DEFAULT_LOGGING_CONFIG, DataType, \
     BacktestResult, TradeStatus, MarketDataType, TradeRiskType, \
     APPLICATION_DIRECTORY, DataSource, OrderExecutor, PortfolioProvider, \
     SnapshotInterval, AWS_S3_STATE_BUCKET_NAME
@@ -91,5 +91,6 @@
     "AWSS3StorageStateHandler",
     "AWS_S3_STATE_BUCKET_NAME",
     "AWS_LAMBDA_LOGGING_CONFIG",
-    'select_backtest_date_ranges'
+    'select_backtest_date_ranges',
+    'DataType'
 ]
diff --git a/investing_algorithm_framework/app/eventloop.py b/investing_algorithm_framework/app/eventloop.py
@@ -1,4 +1,5 @@
 from typing import List
+from time import sleep
 from datetime import  datetime, timedelta, timezone
 from investing_algorithm_framework.domain import TimeUnit, ENVIRONMENT, \
     Environment, BACKTESTING_INDEX_DATETIME
@@ -54,12 +55,12 @@ def __init__(
         """
         self.tasks = []
         self._algorithm = None
-        self._strategies = []
+        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.data_sources = set()
         self.next_run_times = {}
 
     def _get_due_strategies(self):
@@ -89,7 +90,7 @@ def _get_due_strategies(self):
 
         return due
 
-    def initialize(self, algorithm):
+    def initialize(self):
         """
         Initializes the event loop service by calculating the schedule for
         running strategies and tasks based on their defined intervals and time
@@ -104,21 +105,18 @@ def initialize(self, algorithm):
         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
+            strategy.strategy_id: {
+                "next_run": datetime.now(timezone.utc),
+                "data_sources": strategy.data_sources
             }
-            for strategy in self._strategies
+            for strategy in self.strategies
         }
 
-        # Collect all data configurations
-        for strategy in self._strategies:
-            self._data_configurations.append(
-                strategy.data_configurations
+        # Collect all data sources
+        for strategy in self.strategies:
+            self.data_sources = self.data_sources.union(
+                set(strategy.data_sources)
             )
 
     def start(self, number_of_iterations=None):
@@ -129,7 +127,14 @@ def start(self, number_of_iterations=None):
             number_of_iterations: Optional; the number of iterations to run.
                 If None, runs indefinitely.
         """
-        pass
+
+        # if number_of_iterations is None:
+        try:
+            self._run_iteration()
+            number_of_iterations_since_last_orders_check += 1
+            sleep(1)
+        except KeyboardInterrupt:
+            exit(0)
 
     def _run_iteration_backtest(self):
         """
diff --git a/investing_algorithm_framework/app/strategy.py b/investing_algorithm_framework/app/strategy.py
@@ -23,16 +23,17 @@ class TradingStrategy:
         worker_id (optional): str - the id of the worker
         strategy_id (optional): str - the id of the strategy
         decorated (optional): function - the decorated function
-        market_data_sources (optional): list - the list of market data
-            sources to use for the strategy. This will be passed to the
-            run_strategy function.
+        data_sources (List[DataSource] optional):the list of data
+            sources to use for the strategy. The data sources will be used
+            to indetify data providers that will be called to gather data
+            and pass to the strategy before its run.
     """
     time_unit: str = None
     interval: int = None
     worker_id: str = None
     strategy_id: str = None
     decorated = None
-    market_data_sources = None
+    data_sources = None
     traces = None
     context: Context = None
 
@@ -41,7 +42,7 @@ def __init__(
         strategy_id=None,
         time_unit=None,
         interval=None,
-        market_data_sources=None,
+        data_sources=None,
         worker_id=None,
         decorated=None
     ):
@@ -55,8 +56,8 @@ def __init__(
         if time_unit is not None:
             self.time_unit = TimeUnit.from_value(time_unit)
 
-        if market_data_sources is not None:
-            self.market_data_sources = market_data_sources
+        if data_sources is not None:
+            self.data_sources = data_sources
 
         if decorated is not None:
             self.decorated = decorated
@@ -146,7 +147,7 @@ def strategy_profile(self):
             strategy_id=self.worker_id,
             interval=self.interval,
             time_unit=self.time_unit,
-            market_data_sources=self.market_data_sources
+            data_sources=self.data_sources
         )
 
     def _update_trades_and_orders(self, market_data):
diff --git a/investing_algorithm_framework/dependency_container.py b/investing_algorithm_framework/dependency_container.py
@@ -13,7 +13,7 @@
     PortfolioConfigurationService, MarketDataSourceService, BacktestService, \
     ConfigurationService, PortfolioSnapshotService, PositionSnapshotService, \
     MarketCredentialService, TradeService, PortfolioSyncService, \
-    OrderExecutorLookup, PortfolioProviderLookup
+    OrderExecutorLookup, PortfolioProviderLookup, DataProviderService
 
 
 def setup_dependency_container(app, modules=None, packages=None):
@@ -162,6 +162,11 @@ class DependencyContainer(containers.DeclarativeContainer):
         strategy_orchestrator_service=strategy_orchestrator_service,
         portfolio_snapshot_service=portfolio_snapshot_service,
     )
+    data_provider_service = providers.ThreadSafeSingleton(
+        DataProviderService,
+        configuration_service=configuration_service,
+        market_credentials_service=market_credential_service
+    )
     context = providers.Factory(
         Context,
         configuration_service=configuration_service,
diff --git a/investing_algorithm_framework/domain/__init__.py b/investing_algorithm_framework/domain/__init__.py
@@ -18,9 +18,9 @@
 from .models import OrderStatus, OrderSide, OrderType, TimeInterval, \
     TimeUnit, TimeFrame, TradingTimeFrame, TradingDataType, \
     PortfolioConfiguration, Portfolio, Position, Order, TradeStatus, \
-    BacktestResult, PortfolioSnapshot, StrategyProfile, \
-    BacktestPosition, Trade, MarketCredential, PositionSnapshot, \
-    AppMode, BacktestDateRange, DataType, DataSource, \
+    PortfolioSnapshot, StrategyProfile, \
+    Trade, MarketCredential, PositionSnapshot, \
+    AppMode, DataType, DataSource, \
     PortfolioSnapshot, StrategyProfile, \
     Trade, MarketCredential, PositionSnapshot, AppMode, \
     MarketDataType, TradeRiskType, TradeTakeProfit, TradeStopLoss, \
diff --git a/investing_algorithm_framework/domain/models/data/data_source.py b/investing_algorithm_framework/domain/models/data/data_source.py
@@ -1,13 +1,14 @@
 from dataclasses import dataclass
 from .data_type import DataType
 
-@dataclass()
+@dataclass(frozen=True)
 class DataSource:
     """
     Base class for data sources.
     """
-    identifier: str = None
+    data_provider_identifier: str = None
     data_type: DataType = None
     symbol: str = None
     window_size: int = None
     time_frame: str = None
+    market: str = None
diff --git a/tests/app/test_eventloop.py b/tests/app/test_eventloop.py
@@ -0,0 +1,117 @@
+from unittest import TestCase
+from datetime import datetime, timezone, timedelta
+
+from investing_algorithm_framework.app.eventloop import EventLoopService
+from investing_algorithm_framework import TradingStrategy, DataSource, \
+    DataType, MarketCredential, PortfolioConfiguration
+
+from tests.resources import TestBase
+
+
+class StragyForTesting(TradingStrategy):
+    data_sources = [
+        DataSource(
+            data_type=DataType.OHLCV,
+            window_size=200,
+            symbol="DOT/EUR",
+            time_frame="2h",
+            market="bitvavo"
+        )
+    ]
+    time_unit = "hour"
+    interval = 2
+
+    def run_strategy(self, context, market_data):
+        pass
+
+
+class StrategyForTestingTwo(TradingStrategy):
+    data_sources = [
+        DataSource(
+            data_type=DataType.OHLCV,
+            window_size=200,
+            symbol="ETH/EUR",
+            time_frame="2h",
+            market="bitvavo"
+        ),
+        DataSource(
+            data_type=DataType.CUSTOM,
+            data_provider_identifier="custom_feed_data"
+        ),
+    ]
+    time_unit = "hour"
+    interval = 4
+
+    def run_strategy(self, context, market_data):
+        pass
+
+
+class StrategyForTestingThree(TradingStrategy):
+    data_sources = [
+        DataSource(
+            data_type=DataType.OHLCV,
+            window_size=200,
+            symbol="BTC/EUR",
+            time_frame="2h",
+            market="bitvavo"
+        ),
+        DataSource(
+            data_type=DataType.CUSTOM,
+            data_provider_identifier="twitter_data"
+        ),
+    ]
+    time_unit = "day"
+    interval = 1
+
+    def run_strategy(self, context, market_data):
+        pass
+
+
+class TestEventloopService(TestBase):
+    storage_repo_type = "pandas"
+    market_credentials = [
+        MarketCredential(
+            market="binance",
+            api_key="api_key",
+            secret_key="secret_key",
+        )
+    ]
+    portfolio_configurations = [
+        PortfolioConfiguration(
+            market="binance",
+            trading_symbol="EUR"
+        )
+    ]
+    external_balances = {
+        "EUR": 1000
+    }
+
+    def test_initialize(self):
+        event_loop_service = EventLoopService(
+            order_service=self.app.container.order_service(),
+            portfolio_service=self.app.container.portfolio_service(),
+            configuration_service=self.app.container.configuration_service(),
+            data_provider_service=self.app.container.data_provider_service()
+        )
+        event_loop_service.strategies = [
+            StragyForTesting(),
+            StrategyForTestingTwo(),
+            StrategyForTestingThree()
+        ]
+        event_loop_service.initialize()
+        self.assertEqual(len(event_loop_service.next_run_times), 3)
+        self.assertEqual(len(event_loop_service.data_sources), 5)
+
+        # Each next run time should be set to the current datatime
+        # because no runs have been executed yet
+        for strategy in event_loop_service.strategies:
+            self.assertIn(
+                strategy.strategy_id,
+                event_loop_service.next_run_times
+            )
+            self.assertAlmostEqual(
+                event_loop_service\
+                    .next_run_times[strategy.strategy_id]["next_run"],
+                datetime.now(tz=timezone.utc),
+                delta=timedelta(seconds=10)
+            )
diff --git a/tests/app/test_strategy.py b/tests/app/test_strategy.py
@@ -0,0 +1,33 @@
+from unittest import TestCase
+from investing_algorithm_framework import TradingStrategy, DataSource, \
+    DataType, TimeUnit
+
+
+class StrategyForTesting(TradingStrategy):
+    data_sources = [
+        DataSource(
+            data_type=DataType.OHLCV,
+            window_size=200,
+            symbol="BTC/EUR",
+            time_frame="2h",
+            market="bitvavo"
+        ),
+        DataSource(
+            data_type=DataType.CUSTOM,
+            data_provider_identifier="twitter_data"
+        ),
+    ]
+    time_unit = "hour"
+    interval = 2
+
+    def run_strategy(self, context, market_data):
+        pass
+
+
+class TestStrategy(TestCase):
+
+    def test_configurations(self):
+        strategy = StrategyForTesting()
+        self.assertEqual(len(strategy.data_sources), 2)
+        self.assertTrue(TimeUnit.HOUR.equals(strategy.time_unit))
+        self.assertEqual(strategy.interval, 2)
diff --git a/tests/domain/models/data/test_data_source.py b/tests/domain/models/data/test_data_source.py
diff --git a/tests/domain/models/data/test_data_type.py b/tests/domain/models/data/test_data_type.py
