Merge pull request #38 from coding-kitties/hot_fix_executor

Hot fix executor

Author: coding-kitties

README.md

@@ -1,4 +1,4 @@
-[![Build Status](https://travis-ci.org/investing-algorithms/investing-algorithm-framework.svg?branch=master)](https://travis-ci.org/investing-algorithms/investing-algorithm-framework)
+[![Build Status](https://travis-ci.org/coding-kitties/investing-algorithm-framework.svg?branch=master)](https://travis-ci.org/investing-algorithms/investing-algorithm-framework)
 
 # Investing Algorithm Framework
 

investing_algorithm_framework/core/executors/executor.py

@@ -1,117 +1,49 @@
-from queue import Queue
-from typing import List, Dict
-from wrapt import synchronized
-from abc import abstractmethod, ABC
+from typing import List
+from abc import ABC
+from concurrent.futures import ThreadPoolExecutor
 
 from investing_algorithm_framework.core.workers import Worker
-from investing_algorithm_framework.core.exceptions import OperationalException
-from investing_algorithm_framework.core.utils import StoppableThread
 from investing_algorithm_framework.core.events.observer import Observer
 from investing_algorithm_framework.core.events.observable import Observable
 from investing_algorithm_framework.configuration.config_constants import \
     DEFAULT_MAX_WORKERS
 
 
-class Executor(Observable, Observer, ABC):
+class Executor(Observable, ABC):
     """
-    Executor class: functions as an abstract class that will handle the
+    Executor class: functions as a thread executor that will handle the
     executions of workers in asynchronous order.
+
+    It will make use of the concurrent library for execution of the workers.
+    Also the executor functions as an observable instance.
     """
 
-    def __init__(self,  max_workers: int = DEFAULT_MAX_WORKERS):
+    def __init__(
+            self,
+            workers: List[Worker],
+            max_concurrent_workers: int = DEFAULT_MAX_WORKERS
+    ) -> None:
         super(Executor, self).__init__()
 
-        self._max_workers = max_workers
-        self._pending_workers: Queue = None
-        self._running_threads: Dict[Worker, StoppableThread] = {}
+        self._workers = workers
+        self._max_concurrent_workers = max_concurrent_workers
 
     def start(self) -> None:
         """
-        Main entry for the executor.
-        """
-
-        self._initialize()
-        self.run_jobs()
-
-    def stop(self) -> None:
-        """
-        Function that will stop all running workers.
-        """
-
-        for worker in self._running_threads:
-            self.stop_running_worker(worker)
-
-        self.clean_up()
-
-    def clean_up(self):
-        """
-        Clean ups the resources.
-        """
-
-        self._pending_workers: Queue = None
-        self._running_threads: Dict[Worker, StoppableThread] = {}
-
-    def _initialize(self):
-        """
-        Functions that initializes the pending workers.
-        """
-
-        workers = self.create_workers()
-
-        if not workers or len(workers) == 0:
-            raise OperationalException("There where no workers initialized "
-                                       "for the executor instance")
-
-        self._pending_workers = Queue()
-
-        for worker in workers:
-            self._pending_workers.put(worker)
-
-    @abstractmethod
-    def create_workers(self) -> List[Worker]:
-        """
-        Abstract function that will create the workers.
-        """
-        pass
+        Main entry for the executor. The executor creates a ThreadPoolExecutor
+        with the given amount of max_workers.
 
-    def run_jobs(self) -> None:
+        It will then pass all the workers to the ThreadPoolExecutor. When
+        finished it will update all its observers
         """
-        Function that will start all the workers.
-        """
-
-        worker_iteration = self._max_workers - len(
-            self._running_threads.keys()
-        )
 
-        while worker_iteration > 0 and not self._pending_workers.empty():
-            worker = self._pending_workers.get()
-            worker_iteration -= 1
-            thread = StoppableThread(target=worker.start)
-            worker.add_observer(self)
-            self._running_threads[worker] = thread
-            thread.start()
+        with ThreadPoolExecutor(max_workers=self._max_concurrent_workers) as \
+                executor:
 
-    @synchronized
-    def update(self, observable, **kwargs) -> None:
-        """
-        Observer implementation.
-        """
+            for worker in self.workers:
+                executor.submit(worker.start)
 
-        if observable in self._running_threads:
-            del self._running_threads[observable]
-
-        if not self.processing:
-            self.notify_observers()
-        else:
-            self.run_jobs()
-
-    def stop_running_worker(self, worker: Worker) -> None:
-        """
-        Function that will stop a running worker.
-        """
-
-        thread = self._running_threads[worker]
-        thread.kill()
+        self.notify_observers()
 
     def add_observer(self, observer: Observer) -> None:
         super(Executor, self).add_observer(observer)
@@ -120,12 +52,5 @@ def remove_observer(self, observer: Observer) -> None:
         super(Executor, self).remove_observer(observer)
 
     @property
-    def processing(self) -> bool:
-        """
-        Property that will show if the executor is running.
-        """
-
-        return (self._pending_workers is not None
-                and not self._pending_workers.empty()) or \
-               (self._running_threads is not None
-                and len(self._running_threads.keys()) > 0)
+    def workers(self) -> List[Worker]:
+        return self._workers

investing_algorithm_framework/core/state/state.py

@@ -6,9 +6,9 @@
 
 class State(ABC):
     """
-    Represents a state of the Bot, these state are use by the BotContext.
-    Each implemented state represents a work mode for the
-    investing_algorithm_framework.
+    Represents a state of the context, these state are use by the Context.
+    Each implemented state represents a work mode for a application created
+    with investing_algorithm_framework.
     """
 
     # Transition state for the next BotState
@@ -24,7 +24,7 @@ def __init__(self, context) -> None:
     def start(self):
 
         # Will stop the state if pre-conditions are not met
-        if not self.validate_state():
+        if not self.validate_state(pre_state=True):
             return
 
         while True:

investing_algorithm_framework/core/workers/__init__.py

@@ -1,5 +1,7 @@
 from investing_algorithm_framework.core.workers.worker import Worker
 from investing_algorithm_framework.core.workers.scheduled_worker \
     import ScheduledWorker
+from investing_algorithm_framework.core.workers.relational_worker import \
+    RelationalWorker
 
-__all__ = ['Worker', 'ScheduledWorker']
+__all__ = ['Worker', 'ScheduledWorker', 'RelationalWorker']

investing_algorithm_framework/core/workers/relational_worker.py

@@ -0,0 +1,39 @@
+from abc import ABC
+from typing import Dict, Any
+
+from investing_algorithm_framework.core.workers.worker import Worker
+from investing_algorithm_framework.core.exceptions import OperationalException
+
+
+class RelationalWorker(Worker, ABC):
+    """
+    RelationalWorker will start after it's relational worker has run.
+
+    It will check if the related worked had run, and if this is
+    true it will start itself. Use this worker if you want to create
+    chains of workers that are depended on each other.
+    """
+    run_after: Worker
+
+    def start(self, **kwargs: Dict[str, Any]) -> None:
+
+        # Only run if the last time this worker stared is before
+        # the last time the 'run_after' worker had finished.
+        if not self.run_after:
+            raise OperationalException(
+                'The run_after worker is not set, make sure you set this '
+                'attribute to let the RelationalWorker run properly.'
+            )
+
+        if self.last_run is not None:
+
+            if self.run_after.last_run is None:
+                raise OperationalException(
+                    "Relational Worker has run before 'run_after' worker."
+                )
+
+            if self.run_after.last_run > self.last_run:
+                super(RelationalWorker, self).start()
+
+        elif self.run_after.last_run is not None:
+            super(RelationalWorker, self).start()

investing_algorithm_framework/core/workers/scheduled_worker.py

@@ -1,24 +1,58 @@
 from abc import ABC
+from datetime import datetime
+from typing import Dict, Any
 
 from investing_algorithm_framework.core.utils import TimeUnit
 from investing_algorithm_framework.core.workers.worker import Worker
 
 
 class ScheduledWorker(Worker, ABC):
+    time_unit: TimeUnit = None
+    time_interval: int = None
+
+    def start(self, **kwargs: Dict[str, Any]) -> None:
+
+        # If the worker has never run, run it
+        if self.last_run is None:
+            super(ScheduledWorker, self).start()
+
+        else:
+            # Get the current time
+            elapsed_time = datetime.now() - self.last_run
+
+            # Second evaluation
+            if self.get_time_unit() is TimeUnit.SECOND:
+                seconds = elapsed_time.total_seconds()
+
+                if seconds > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
+
+            # Minute evaluation
+            elif self.get_time_unit() is TimeUnit.MINUTE:
+                minutes = divmod(elapsed_time.total_seconds(), 60)
+
+                if minutes > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
+
+            # Hour evaluation
+            elif self.get_time_unit() is TimeUnit.HOUR:
+                hours = divmod(elapsed_time.total_seconds(), 3600)
+
+                if hours > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
 
     def get_time_unit(self) -> TimeUnit:
         assert getattr(self, 'time_unit', None) is not None, (
             "{} should either include a time_unit attribute, or override the "
-            "`get_time_unit()`, method.".format(self.__class__.__name__)
+            "`get_time_unit()` method.".format(self.__class__.__name__)
         )
 
         return getattr(self, 'time_unit')
 
     def get_time_interval(self) -> int:
         assert getattr(self, 'time_interval', None) is not None, (
-            "{} should either include a time_interval attribute, or "
-            "override the `get_time_interval()`, "
-            "method.".format(self.__class__.__name__)
+            "{} should either include a time_interval attribute, or override "
+            "the `get_time_interval()` method.".format(self.__class__.__name__)
         )
 
         return getattr(self, 'time_interval')

investing_algorithm_framework/core/workers/worker.py

@@ -1,9 +1,14 @@
+import logging
 from abc import abstractmethod, ABC
 from typing import Dict, Any
+from wrapt import synchronized
+from datetime import datetime
 
 from investing_algorithm_framework.core.events.observable import Observable
 from investing_algorithm_framework.core.events.observer import Observer
 
+logger = logging.getLogger('investing_algorithm_framework')
+
 
 class Worker(Observable, ABC):
     """
@@ -12,15 +17,19 @@ class Worker(Observable, ABC):
     """
 
     id = None
+    last_run: datetime = None
 
     def start(self, **kwargs: Dict[str, Any]) -> None:
         """
         Function that will start the worker, and notify its observers when
         it is finished
         """
 
+        logger.info("Starting worker {}".format(self.get_id()))
         self.work(**kwargs)
         self.notify_observers()
+        self.update_last_run()
+        logger.info("Worker {} finished".format(self.get_id()))
 
     @abstractmethod
     def work(self, **kwargs: Dict[str, Any]) -> None:
@@ -42,3 +51,17 @@ def get_id(self) -> str:
         )
 
         return getattr(self, 'id')
+
+    @classmethod
+    @synchronized
+    def update_last_run(cls) -> None:
+        """
+        Update last run, this function is synchronized, which means that
+        different instances can update the last_run attribute from different
+        threads.
+        """
+        cls.last_run = datetime.now()
+
+    @classmethod
+    def get_last_run(cls):
+        return cls.last_run

investing_algorithm_framework/management/command.py

@@ -89,6 +89,8 @@ def execute(self, *args, **options) -> Any:
         if response is None and self.success_message is not None:
             return self.success_message
 
+        return response
+
     @abstractmethod
     def handle(self, *args, **options) -> Any:
         """

investing_algorithm_framework/management/command_manager.py

@@ -42,6 +42,7 @@ def execute(self) -> None:
             ]:
                 response = self.fetch_command(sub_command).\
                     run_from_argv(self.argv)
+                response = format_success_message(response)
             else:
                 # Help for sub command
                 if len(self.argv) > 2:
@@ -55,10 +56,12 @@ def execute(self) -> None:
                     self.argv[2] = option
                     response = self.fetch_command(sub_command)\
                         .run_from_argv(self.argv)
+                    response = format_success_message(response)
                 else:
                     # Show general help command
                     command = HelpCommand()
                     response = command.run_from_argv(self.argv)
+                    response = format_success_message(response)
         except Exception as e:
             response = format_error_message(str(e))