Extracted common code from WatchXYZ event sources to a common base class.

gbeced · gbeced · commit ab8bb8fc57b1 · 2026-07-11T00:23:54.000-03:00
diff --git a/basana/external/ccxt/bars.py b/basana/external/ccxt/bars.py
@@ -15,12 +15,12 @@
 # limitations under the License.
 
 from typing import Any, Dict, List, Optional, Union
-import asyncio
 import datetime
 import logging
 
 from . import helpers
-from basana.core import bar, event, logs
+from .watch_event_source import WatchEventSource
+from basana.core import bar, logs
 from basana.core.pair import Pair
 
 
@@ -44,48 +44,23 @@ def to_bar(self, pair: Pair, duration: datetime.timedelta):
         )
 
 
-class WatchOHLCVEventSource(event.FifoQueueEventSource, event.Producer):
+class WatchOHLCVEventSource(WatchEventSource):
     def __init__(self, cli: Any, pair: Pair, timeframe: str, params: Optional[Dict[str, Any]] = None):
         if timeframe not in cli.timeframes:
             raise ValueError(f"Invalid bar_duration: {timeframe}")
 
-        if not cli.has.get("watchOHLCV"):
-            raise NotImplementedError("The exchange does not support watchOHLCV")
-
-        super().__init__(producer=self)
-        self._cli = cli
-        self._pair = pair
-        self._symbol = helpers.pair_to_symbol(pair)
+        super().__init__(cli, pair, "watchOHLCV", "unWatchOHLCV")
         self._timeframe = timeframe
         self._params = dict(params or {})
         self._duration_secs = self._cli.parse_timeframe(self._timeframe)
         self._last_candle: Optional[Candle] = None
 
-    async def initialize(self):
-        await self._cli.load_markets()
-
-    async def main(self):
-        while True:
-            try:
-                ohlcv = await self._cli.watch_ohlcv(self._symbol, self._timeframe, params=self._params)
-                self._handle_ohlcv(ohlcv)
-            except Exception as e:
-                logger.error(logs.StructuredMessage(
-                    "Error watching OHLCV", pair=self._pair, timeframe=self._timeframe, error=e
-                ))
-            # Yield to the event loop to allow other tasks to run.
-            await asyncio.sleep(0)
-
-    async def finalize(self):
-        if not self._cli.has.get("unWatchOHLCV"):
-            return
+    async def watch(self):
+        ohlcv = await self._cli.watch_ohlcv(self._symbol, self._timeframe, params=self._params)
+        self._handle_ohlcv(ohlcv)
 
-        try:
-            await self._cli.un_watch_ohlcv(self._symbol, self._timeframe)
-        except Exception as error:
-            logger.error(logs.StructuredMessage(
-                "Error unwatching OHLCV", pair=self._pair, timeframe=self._timeframe, error=error
-            ))
+    async def unwatch(self):
+        await self._cli.un_watch_ohlcv(self._symbol, self._timeframe)
 
     def _handle_ohlcv(self, ohlcv: list):
         candles = list(map(Candle, ohlcv))
diff --git a/basana/external/ccxt/order_book.py b/basana/external/ccxt/order_book.py
@@ -17,17 +17,13 @@
 from dataclasses import dataclass
 from decimal import Decimal
 from typing import Any, Awaitable, Callable, Dict, List, Optional
-import asyncio
-import logging
 
 from . import helpers
-from basana.core import dt, event, logs
+from .watch_event_source import WatchEventSource
+from basana.core import dt, event
 from basana.core.pair import Pair
 
 
-logger = logging.getLogger(__name__)
-
-
 @dataclass
 class Entry:
     #: The price.
@@ -75,47 +71,23 @@ def __init__(self, when, order_book: PartialOrderBook):
         self.order_book: PartialOrderBook = order_book
 
 
-class WatchOrderBookEventSource(event.FifoQueueEventSource, event.Producer):
+class WatchOrderBookEventSource(WatchEventSource):
     def __init__(
             self, cli: Any, pair: Pair, limit: Optional[int] = None,
             params: Optional[Dict[str, Any]] = None
     ):
-        if not cli.has.get("watchOrderBook"):
-            raise NotImplementedError("The exchange does not support watchOrderBook")
-
-        super().__init__(producer=self)
-        self._cli = cli
-        self._pair = pair
-        self._symbol = helpers.pair_to_symbol(pair)
+        super().__init__(cli, pair, "watchOrderBook", "unWatchOrderBook")
         self._limit = limit
         self._params = dict(params or {})
 
-    async def initialize(self):
-        await self._cli.load_markets()
-
-    async def main(self):
-        while True:
-            try:
-                order_book = await self._cli.watch_order_book(
-                    self._symbol, self._limit, params=self._params
-                )
-                self._handle_order_book(order_book)
-            except Exception as e:
-                logger.error(logs.StructuredMessage(
-                    "Error watching order book", pair=self._pair, error=e
-                ))
-            await asyncio.sleep(0)
-
-    async def finalize(self):
-        if not self._cli.has.get("unWatchOrderBook"):
-            return
-
-        try:
-            await self._cli.un_watch_order_book(self._symbol, params=self._params)
-        except Exception as error:
-            logger.error(logs.StructuredMessage(
-                "Error unwatching order book", pair=self._pair, error=error
-            ))
+    async def watch(self):
+        order_book = await self._cli.watch_order_book(
+            self._symbol, self._limit, params=self._params
+        )
+        self._handle_order_book(order_book)
+
+    async def unwatch(self):
+        await self._cli.un_watch_order_book(self._symbol, params=self._params)
 
     def _handle_order_book(self, order_book: dict):
         timestamp = order_book.get("timestamp")
diff --git a/basana/external/ccxt/orders.py b/basana/external/ccxt/orders.py
@@ -16,20 +16,16 @@
 
 from decimal import Decimal
 from typing import Any, Awaitable, Callable, Dict, List, Optional
-import asyncio
 import datetime
 import functools
-import logging
 
 from . import helpers
-from basana.core import dt, event, logs
+from .watch_event_source import WatchEventSource
+from basana.core import dt, event
 from basana.core.enums import OrderOperation
 from basana.core.pair import Pair
 
 
-logger = logging.getLogger(__name__)
-
-
 class Order:
     def __init__(self, pair: Pair, raw: dict):
         #: The trading pair.
@@ -121,41 +117,17 @@ def __init__(self, when: datetime.datetime, order: Order):
         self.order: Order = order
 
 
-class WatchOrdersEventSource(event.FifoQueueEventSource, event.Producer):
+class WatchOrdersEventSource(WatchEventSource):
     def __init__(self, cli: Any, pair: Pair, params: Optional[Dict[str, Any]] = None):
-        if not cli.has.get("watchOrders"):
-            raise NotImplementedError("The exchange does not support watchOrders")
-
-        super().__init__(producer=self)
-        self._cli = cli
-        self._pair = pair
-        self._symbol = helpers.pair_to_symbol(pair)
+        super().__init__(cli, pair, "watchOrders", "unWatchOrders")
         self._params = dict(params or {})
 
-    async def initialize(self):
-        await self._cli.load_markets()
-
-    async def main(self):
-        while True:
-            try:
-                orders = await self._cli.watch_orders(self._symbol, params=self._params)
-                self._handle_orders(orders)
-            except Exception as e:
-                logger.error(logs.StructuredMessage(
-                    "Error watching orders", pair=self._pair, error=e
-                ))
-            await asyncio.sleep(0)
-
-    async def finalize(self):
-        if not self._cli.has.get("unWatchOrders"):
-            return
-
-        try:
-            await self._cli.un_watch_orders(self._symbol, params=self._params)
-        except Exception as error:
-            logger.error(logs.StructuredMessage(
-                "Error unwatching orders", pair=self._pair, error=error
-            ))
+    async def watch(self):
+        orders = await self._cli.watch_orders(self._symbol, params=self._params)
+        self._handle_orders(orders)
+
+    async def unwatch(self):
+        await self._cli.un_watch_orders(self._symbol, params=self._params)
 
     def _handle_orders(self, raw_orders: List[dict]):
         orders = list(map(functools.partial(Order, self._pair), raw_orders))
diff --git a/basana/external/ccxt/trades.py b/basana/external/ccxt/trades.py
@@ -16,18 +16,14 @@
 
 from decimal import Decimal
 from typing import Any, Awaitable, Callable, Dict, List, Optional
-import asyncio
 import datetime
-import logging
 
 from . import helpers
-from basana.core import event, logs
+from .watch_event_source import WatchEventSource
+from basana.core import event
 from basana.core.pair import Pair
 
 
-logger = logging.getLogger(__name__)
-
-
 class Trade:
     def __init__(self, pair: Pair, raw: dict):
         #: The trading pair.
@@ -81,42 +77,17 @@ def __init__(self, when: datetime.datetime, trade: Trade):
         self.trade: Trade = trade
 
 
-class WatchTradesEventSource(event.FifoQueueEventSource, event.Producer):
+class WatchTradesEventSource(WatchEventSource):
     def __init__(self, cli: Any, pair: Pair, params: Optional[Dict[str, Any]] = None):
-        if not cli.has.get("watchTrades"):
-            raise NotImplementedError("The exchange does not support watchTrades")
-
-        super().__init__(producer=self)
-        self._cli = cli
-        self._pair = pair
-        self._symbol = helpers.pair_to_symbol(pair)
+        super().__init__(cli, pair, "watchTrades", "unWatchTrades")
         self._params = dict(params or {})
 
-    async def initialize(self):
-        await self._cli.load_markets()
-
-    async def main(self):
-        while True:
-            try:
-                trades = await self._cli.watch_trades(self._symbol, params=self._params)
-                self._handle_trades(trades)
-            except Exception as e:
-                logger.error(logs.StructuredMessage(
-                    "Error watching trades", pair=self._pair, error=e
-                ))
-            # Yield to the event loop to allow other tasks to run.
-            await asyncio.sleep(0)
-
-    async def finalize(self):
-        if not self._cli.has.get("unWatchTrades"):
-            return
-
-        try:
-            await self._cli.un_watch_trades(self._symbol)
-        except Exception as error:
-            logger.error(logs.StructuredMessage(
-                "Error unwatching trades", pair=self._pair, error=error
-            ))
+    async def watch(self):
+        trades = await self._cli.watch_trades(self._symbol, params=self._params)
+        self._handle_trades(trades)
+
+    async def unwatch(self):
+        await self._cli.un_watch_trades(self._symbol)
 
     def _handle_trades(self, trades: List[dict]):
         trades.sort(key=lambda trade: int(trade["timestamp"]))
diff --git a/basana/external/ccxt/watch_event_source.py b/basana/external/ccxt/watch_event_source.py
@@ -0,0 +1,76 @@
+# Basana
+#
+# Copyright 2022 Gabriel Martin Becedillas Ruiz
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Any
+import abc
+import asyncio
+import logging
+
+from . import helpers
+from basana.core import event, logs
+from basana.core.pair import Pair
+
+
+logger = logging.getLogger(__name__)
+
+
+class WatchEventSource(event.FifoQueueEventSource, event.Producer):
+    def __init__(self, cli: Any, pair: Pair, watch_capability: str, unwatch_capability: str):
+        if not cli.has.get(watch_capability):
+            raise NotImplementedError(f"The exchange does not support {watch_capability}")
+
+        super().__init__(producer=self)
+        self._cli = cli
+        self._pair = pair
+        self._symbol = helpers.pair_to_symbol(pair)
+        self._watch_capability = watch_capability
+        self._unwatch_capability = unwatch_capability
+        self.err_sleep = 0.1
+
+    async def initialize(self):
+        await self._cli.load_markets()
+
+    async def main(self):
+        while True:
+            sleep = 0
+            try:
+                await self.watch()
+            except Exception as e:
+                logger.error(logs.StructuredMessage(
+                    f"Error during {self._watch_capability}", pair=self._pair, error=e
+                ))
+                sleep = self.err_sleep
+            # Yield to the event loop to allow other tasks to run.
+            await asyncio.sleep(sleep)
+
+    async def finalize(self):
+        if not self._cli.has.get(self._unwatch_capability):
+            return
+
+        try:
+            await self.unwatch()
+        except Exception as e:
+            logger.error(logs.StructuredMessage(
+                f"Error during {self._unwatch_capability}", pair=self._pair, error=e
+            ))
+
+    @abc.abstractmethod
+    async def watch(self):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    async def unwatch(self):
+        raise NotImplementedError()
diff --git a/tests/test_ccxt_exchange.py b/tests/test_ccxt_exchange.py
