Enable on strategy run hooks

Author: MDUYN

investing_algorithm_framework/app/app.py

@@ -595,6 +595,13 @@ def run(self, number_of_iterations: int = None):
         """
         self.initialize_config()
 
+        # Run all on_initialize hooks
+        for hook in self._on_initialize_hooks:
+            logger.info(
+                f"Running on_initialize hook: {hook.__class__.__name__}"
+            )
+            hook.on_run(self.context)
+
         # Load the state if a state handler is provided
         if self._state_handler is not None:
             logger.info("Detected state handler, loading state")
@@ -607,8 +614,12 @@ def run(self, number_of_iterations: int = None):
         event_loop_service = None
 
         try:
-            # Run all on_initialize hooks
-            for hook in self._on_initialize_hooks:
+            # Run all on_after_initialize hooks
+            for hook in self._on_after_initialize_hooks:
+                logger.info(
+                    f"Running on_after_initialize "
+                    f"hook: {hook.__class__.__name__}"
+                )
                 hook.on_run(self.context)
 
             algorithm = self.get_algorithm()

investing_algorithm_framework/app/eventloop.py

@@ -1,6 +1,7 @@
 from datetime import datetime, timedelta, timezone
 from time import sleep
 from typing import List, Set, Dict
+from logging import getLogger
 
 import polars as pl
 
@@ -13,6 +14,8 @@
 from .algorithm import Algorithm
 from .strategy import TradingStrategy
 
+logger = getLogger("investing_algorithm_framework")
+
 
 class EventLoopService:
     """
@@ -256,7 +259,7 @@ def _snapshot(
     def initialize(
         self,
         algorithm: Algorithm,
-        trade_order_evaluator: TradeOrderEvaluator
+        trade_order_evaluator: TradeOrderEvaluator,
     ):
         """
         Initializes the event loop service by calculating the schedule for
@@ -279,6 +282,7 @@ def initialize(
         """
         self._algorithm = algorithm
         self.strategies = algorithm.strategies
+        self.tasks = algorithm.tasks
 
         if len(self.strategies) == 0:
             raise OperationalException(
@@ -367,9 +371,10 @@ def start(
                         INDEX_DATETIME, current_time
                     )
                     strategy_ids = schedule[current_time]["strategy_ids"]
-                    # task_ids = schedule[current_time]["task_ids"]
                     strategies = self._get_strategies(strategy_ids)
-                    self._run_iteration(strategies=strategies, tasks=[])
+                    self._run_iteration(
+                        strategies=strategies
+                    )
 
             else:
                 for current_time in sorted_times:
@@ -379,14 +384,18 @@ def start(
                     strategy_ids = schedule[current_time]["strategy_ids"]
                     # task_ids = schedule[current_time]["task_ids"]
                     strategies = self._get_strategies(strategy_ids)
-                    self._run_iteration(strategies=strategies, tasks=[])
+                    self._run_iteration(
+                        strategies=strategies
+                    )
         else:
             if number_of_iterations is None:
                 try:
                     config = self._configuration_service.config
                     current_time = config[INDEX_DATETIME]
                     strategies = self._get_due_strategies(current_time)
-                    self._run_iteration(strategies)
+                    self._run_iteration(
+                        strategies=strategies, tasks=self.tasks
+                    )
                     current_time = datetime.now(timezone.utc)
                     self._configuration_service.add_value(
                         INDEX_DATETIME, current_time
@@ -405,7 +414,9 @@ def start(
                             config = self._configuration_service.config
                             current_time = config[INDEX_DATETIME]
                             strategies = self._get_due_strategies(current_time)
-                            self._run_iteration(strategies)
+                            self._run_iteration(
+                                strategies=strategies, tasks=self.tasks
+                            )
                             current_time = datetime.now(timezone.utc)
                             self._configuration_service.add_value(
                                 INDEX_DATETIME, current_time
@@ -419,7 +430,9 @@ def start(
                         config = self._configuration_service.config
                         current_time = config[INDEX_DATETIME]
                         strategies = self._get_due_strategies(current_time)
-                        self._run_iteration(strategies)
+                        self._run_iteration(
+                            strategies=strategies, tasks=self.tasks
+                        )
                         current_time = datetime.now(timezone.utc)
                         self._configuration_service.add_value(
                             INDEX_DATETIME, current_time
@@ -526,6 +539,9 @@ def _run_iteration(
         if not strategies:
             return
 
+        for task in self.tasks:
+            logger.info(f"Running task {task.__class__.__name__}")
+
         for strategy in strategies:
 
             if strategy.data_sources is not None:
@@ -537,13 +553,9 @@ def _run_iteration(
             else:
                 data = {}
 
-            # Select data for the strategy
+            logger.info(f"Running strategy {strategy.strategy_id}")
             strategy.run_strategy(context=self.context, data=data)
 
-        # # Step 6: Run all on_strategy_run hooks
-        # for strategy in due_strategies:
-        #     strategy.run_on_strategy_run_hooks(context=self.context)
-
         # Step 7: Snapshot the portfolios if needed and update history
         created_orders = self._order_service.get_all(
             {