Orca (#55)

* Orca

Author: m5l14i11

core/actors/collector/_data.py

@@ -32,14 +32,16 @@ async def start(self, msg: Event):
 
     async def stop(self):
         self._stop_event.set()
-        self._queue.put_nowait(STOP)
+        await self._queue.put(STOP)
 
         for task in self._tasks:
             task.cancel()
 
+        tasks_to_cancel = [task for task in self._tasks if not task.done()]
+
         try:
             await asyncio.wait_for(
-                asyncio.gather(*self._tasks, return_exceptions=True), timeout=5
+                asyncio.gather(*tasks_to_cancel, return_exceptions=True), timeout=5
             )
         except asyncio.TimeoutError:
             logger.warning("Timeout while waiting for tasks to finish.")
@@ -74,6 +76,7 @@ async def _run_consumer(self, consumer):
             while not self._stop_event.is_set():
                 data = await self._queue.get()
                 if data is STOP:
+                    self._queue.task_done()
                     break
                 try:
                     await consumer(data)

core/actors/state/_signal.py

@@ -1,3 +1,4 @@
+import asyncio
 from typing import Any, Callable, List
 
 
@@ -9,5 +10,17 @@ def connect(self, subscriber: Callable[..., Any]) -> None:
         self._subscribers.append(subscriber)
 
     def emit(self, *args, **kwargs) -> None:
+        tasks = []
+
         for subscriber in self._subscribers:
-            subscriber(*args, **kwargs)
+            if asyncio.iscoroutinefunction(subscriber):
+                tasks.append(subscriber(*args, **kwargs))
+            else:
+                subscriber(*args, **kwargs)
+
+        if tasks:
+            asyncio.create_task(self._run_async_subscribers(tasks))
+
+    @staticmethod
+    async def _run_async_subscribers(tasks: List[asyncio.Future]) -> None:
+        await asyncio.gather(*tasks)

core/event_decorators.py

@@ -42,8 +42,10 @@ async def run(self, task, *args, **kwargs):
             return await self._dispatcher.run(task, *args, **kwargs)
 
         async def wait(self):
-            await asyncio.sleep(1)
-            await self._dispatcher.wait()
+            try:
+                await asyncio.wait_for(self._dispatcher.wait(), timeout=10)
+            except asyncio.TimeoutError:
+                pass
 
         def _unregister(self):
             for event_type, handler in self._registered_handlers:

core/models/entity/ohlcv.py

@@ -67,44 +67,6 @@ def from_dict(cls, data: Dict) -> "OHLCV":
 
         return cls.from_list([data[key] for key in ohlcv_keys])
 
-    @property
-    def real_body(self) -> float:
-        return abs(self.open - self.close)
-
-    @property
-    def upper_shadow(self) -> float:
-        return self.high - max(self.open, self.close)
-
-    @property
-    def lower_shadow(self) -> float:
-        return min(self.open, self.close) - self.low
-
-    @property
-    def price_range(self) -> float:
-        return self.high - self.low
-
-    @property
-    def price_movement(self) -> float:
-        return self.close - self.open
-
-    @property
-    def body_range_ratio(self) -> float:
-        return self.real_body / self.price_range if self.price_range != 0 else 0
-
-    @property
-    def body_shadow_ratio(self) -> float:
-        total_shadow = self.upper_shadow + self.lower_shadow
-        return self.real_body / total_shadow if total_shadow != 0 else 0
-
-    @property
-    def shadow_range_ratio(self) -> float:
-        total_shadow = self.upper_shadow + self.lower_shadow
-        return total_shadow / self.price_range if self.price_range != 0 else 0
-
-    @property
-    def real_body_normalized(self) -> float:
-        return self.real_body / self.price_range if self.price_range != 0 else 0
-
     @property
     def type(self) -> CandleType:
         if self.price_movement > 0:

core/tasks/_base.py

@@ -17,7 +17,11 @@ def set_task(self, task: asyncio.Task):
 
     async def wait_for_finishing(self) -> asyncio.Task:
         await self._task_event.wait()
-        result = await self._task
+
+        if not self._task.done():
+            result = await self._task
+        else:
+            result = None
 
         if not self._task.done():
             self._task.cancel()

factor/generator/_population.py

@@ -247,7 +247,7 @@ def __next__(self) -> Individual:
             symbol, timeframe, strategy = next(self.iter)
             return Individual(symbol, timeframe, strategy)
         except StopIteration:
-            raise StopIteration
+            raise StopIteration from None
 
     def _init(self) -> Iterator:
         sampled_symbols = self._generate_symbols()

feed/_historical.py

@@ -30,6 +30,7 @@ def __init__(
         super().__init__(symbol, timeframe, datasource)
         self.datasource_factory = datasource_factory
         self.config = config_service.get("feed")
+        self.bp = asyncio.Semaphore(10)
 
     async def on_receive(self, msg: StartHistoricalFeed):
         await self.collector.start(msg)
@@ -64,18 +65,18 @@ async def _handle_market(self, batch: List[Bar]) -> None:
             await self.tell(
                 NewMarketDataReceived(self.symbol, self.timeframe, self.datasource, bar)
             )
-        await asyncio.sleep(0.0001)
 
     async def _outbox(self, batch: List[Bar]) -> None:
-        tasks = [
-            self.ask(
-                IngestMarketData(self.symbol, self.timeframe, self.datasource, bar)
-            )
-            for bar in batch
-            if bar.closed
-        ]
+        async with self.bp:
+            tasks = [
+                self.ask(
+                    IngestMarketData(self.symbol, self.timeframe, self.datasource, bar)
+                )
+                for bar in batch
+                if bar.closed
+            ]
 
-        await asyncio.gather(*tasks)
+            await asyncio.gather(*tasks)
 
     @staticmethod
     async def batched(stream: AsyncIterator[Bar], batch_size: int):

feed/_realtime.py

@@ -47,58 +47,53 @@ async def on_receive(self, msg: StartRealtimeFeed):
 
     @Producer
     async def _kline_producer(self, msg: StartRealtimeFeed):
-        async with AsyncRealTimeData(
-            self.datasource_factory.create(
-                msg.datasource, ProtocolType.WS, WSType.PUBLIC
-            ),
+        async for bars in self._stream_producer(
             KlineStreamStrategy(self.timeframe, self.symbol),
-        ) as stream:
-            async for bars in stream:
-                yield bars
+            ProtocolType.WS,
+            WSType.PUBLIC,
+            msg,
+        ):
+            yield bars
 
     @Producer
     async def _ob_producer(self, msg: StartRealtimeFeed):
-        async with AsyncRealTimeData(
-            self.datasource_factory.create(
-                msg.datasource, ProtocolType.WS, WSType.PUBLIC
-            ),
+        async for orders in self._stream_producer(
             OrderBookStreamStrategy(self.symbol, self.config.get("dom", 15)),
-        ) as stream:
-            async for orders in stream:
-                yield orders
+            ProtocolType.WS,
+            WSType.PUBLIC,
+            msg,
+        ):
+            yield orders
 
     @Producer
     async def _liquidation_producer(self, msg: StartRealtimeFeed):
-        async with AsyncRealTimeData(
-            self.datasource_factory.create(
-                msg.datasource, ProtocolType.WS, WSType.PUBLIC
-            ),
+        async for liquidations in self._stream_producer(
             LiquidationStreamStrategy(self.symbol),
-        ) as stream:
-            async for liquidations in stream:
-                yield liquidations
+            ProtocolType.WS,
+            WSType.PUBLIC,
+            msg,
+        ):
+            yield liquidations
 
     @Producer
     async def _order_producer(self, msg: StartRealtimeFeed):
-        async with AsyncRealTimeData(
-            self.datasource_factory.create(
-                msg.datasource, ProtocolType.WS, WSType.PRIVATE
-            ),
+        async for order in self._stream_producer(
             OrderStreamStrategy(self.symbol),
-        ) as stream:
-            async for order in stream:
-                yield order
+            ProtocolType.WS,
+            WSType.PRIVATE,
+            msg,
+        ):
+            yield order
 
     @Producer
     async def _position_producer(self, msg: StartRealtimeFeed):
-        async with AsyncRealTimeData(
-            self.datasource_factory.create(
-                msg.datasource, ProtocolType.WS, WSType.PRIVATE
-            ),
+        async for position in self._stream_producer(
             PositionStreamStrategy(self.symbol),
-        ) as stream:
-            async for position in stream:
-                yield position
+            ProtocolType.WS,
+            WSType.PRIVATE,
+            msg,
+        ):
+            yield position
 
     @Consumer
     async def _consumer(self, data: List[Bar]):
@@ -129,3 +124,11 @@ async def _process_orders(self, orders: List[Order]):
                 )
             )
             logger.info(f"{self.symbol}_{self.timeframe}:{order}")
+
+    async def _stream_producer(self, strategy, protocol_type, ws_type, msg):
+        async with AsyncRealTimeData(
+            self.datasource_factory.create(msg.datasource, protocol_type, ws_type),
+            strategy,
+        ) as stream:
+            async for data in stream:
+                yield data

infrastructure/event_dispatcher/event_worker.py

@@ -1,6 +1,9 @@
 import asyncio
+from collections import deque
 from typing import Any, AsyncIterable, Dict, Tuple
 
+import numpy as np
+
 from core.events._base import Event
 
 from .event_handler import EventHandler
@@ -11,19 +14,28 @@ def __init__(
         self,
         event_handler: EventHandler,
         cancel_event: asyncio.Event,
+        task_duration_limit: int = 100,
     ):
         self._event_handler = event_handler
         self._cancel_event = cancel_event
         self._queue = asyncio.Queue()
+        self._task_durations = deque(maxlen=task_duration_limit)
 
     @property
-    def queue_size(self):
-        return self._queue.qsize()
+    def score(self):
+        return self._queue.qsize(), (
+            np.mean(self._task_durations) if len(self._task_durations) > 2 else 0.0
+        )
 
     async def run(self):
         async for event, args, kwargs in self._get_event_stream():
+            start_time = asyncio.get_event_loop().time()
+
             await self._event_handler.handle_event(event, *args, **kwargs)
 
+            end_time = asyncio.get_event_loop().time()
+            self._task_durations.append(end_time - start_time)
+
     async def _get_event_stream(
         self,
     ) -> AsyncIterable[Tuple[Event, Tuple[Any], Dict[str, Any]]]:

infrastructure/event_dispatcher/worker_pool.py

@@ -18,6 +18,8 @@ def __init__(
         num_piority_groups: int,
         event_handler: EventHandler,
         cancel_event: asyncio.Event,
+        alpha: float = 0.7,
+        beta: float = 0.3,
     ):
         self.workers = []
         self.load_balancer = LoadBalancer(num_piority_groups)
@@ -26,6 +28,8 @@ def __init__(
         self.cancel_event = cancel_event
         self._num_priority_groups = num_piority_groups
         self._initialize_workers(num_workers)
+        self.alpha = alpha
+        self.beta = beta
 
     async def dispatch_to_worker(self, event: Event, *args, **kwargs) -> None:
         priority_group = self.load_balancer.determine_priority_group(
@@ -60,14 +64,19 @@ def _distribute_workers(self, priority_group: int) -> List[EventWorker]:
         return self.workers[group_start:group_end]
 
     def _choose_worker(self, group_workers: List[EventWorker]) -> EventWorker:
-        not_busy_workers = [
-            worker for worker in group_workers if worker.queue_size == 0
-        ]
+        scores = np.array([worker.score for worker in group_workers])
+        queue_sizes, median_times = scores[:, 0], scores[:, 1]
+
+        norm_queue_sizes = (queue_sizes - queue_sizes.min() + 1) / (
+            queue_sizes.max() - queue_sizes.min() + 1
+        )
+        norm_median_times = (median_times - median_times.min() + 1) / (
+            median_times.max() - median_times.min() + 1
+        )
 
-        if not_busy_workers:
-            return np.random.choice(not_busy_workers)
+        combined_scores = self.alpha * norm_queue_sizes + self.beta * norm_median_times
 
-        weights = np.array([1 / (worker.queue_size + 1) for worker in group_workers])
+        weights = 1 / (combined_scores + np.finfo(float).eps)
         total_weight = sum(weights)
 
         choice_point = np.random.uniform(0, total_weight)