[Exchanges] fix kline parsing

Author: Guillaume De Saint Martin

octobot_trading/exchanges/adapters/abstract_adapter.py

@@ -49,6 +49,11 @@ def adapt_ohlcv(self, raw, **kwargs):
         fixed = self.fix_ohlcv(raw, **kwargs)
         return self.parse_ohlcv(fixed, **kwargs)
 
+    @_adapter
+    def adapt_kline(self, raw, **kwargs):
+        fixed = self.fix_kline(raw, **kwargs)
+        return self.parse_kline(fixed, **kwargs)
+
     @_adapter
     def adapt_ticker(self, raw, **kwargs):
         fixed = self.fix_ticker(raw, **kwargs)
@@ -124,6 +129,13 @@ def fix_ohlcv(self, raw, **kwargs):
     def parse_ohlcv(self, fixed, **kwargs):
         raise NotImplementedError("parse_ohlcv is not implemented")
 
+    def fix_kline(self, raw, **kwargs):
+        # add generic logic if necessary
+        return raw
+
+    def parse_kline(self, fixed, **kwargs):
+        raise NotImplementedError("parse_kline is not implemented")
+
     def fix_ticker(self, raw, **kwargs):
         # add generic logic if necessary
         return raw

octobot_trading/exchanges/connectors/ccxt/ccxt_adapter.py

@@ -23,6 +23,7 @@
 import octobot_trading.enums as enums
 from octobot_trading.enums import ExchangeConstantsOrderColumns as ecoc
 import octobot_commons.enums as common_enums
+import octobot_commons.constants as common_constants
 
 
 class CCXTAdapter(adapters.AbstractAdapter):
@@ -41,14 +42,23 @@ def parse_order(self, fixed, **kwargs):
         fixed.pop(ecoc.INFO.value)
         return fixed
 
+    def _fix_ohlcv_prices(self, ohlcv):
+        for index, value in enumerate(ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value + 1:]):
+            ohlcv[index + 1] = float(value)
+
     def fix_ohlcv(self, raw, **kwargs):
         fixed = super().fix_ohlcv(raw, **kwargs)
+        # ensure open time is not the current time but the actual candle open time
+        # time_frame kwarg has to be passed to parse candle time
+        candles_s = 1
+        if "time_frame" in kwargs:
+            candles_s = common_enums.TimeFramesMinutes[common_enums.TimeFrames(kwargs["time_frame"])] * \
+                        common_constants.MINUTE_TO_SECONDS
         for ohlcv in fixed:
             try:
-                ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value] = \
-                    int(self.get_uniformized_timestamp(ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value]))
-                for index, value in enumerate(ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value + 1:]):
-                    ohlcv[index + 1] = float(value)
+                int_val = int(self.get_uniformized_timestamp(ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value]))
+                ohlcv[common_enums.PriceIndexes.IND_PRICE_TIME.value] = int_val - (int_val % candles_s)
+                self._fix_ohlcv_prices(ohlcv)
             except KeyError as e:
                 self.logger.error(f"Fail to fix ohlcv ({e})")
         return fixed
@@ -57,6 +67,21 @@ def parse_ohlcv(self, fixed, **kwargs):
         # CCXT standard ohlcv parsing logic
         return fixed
 
+    def fix_kline(self, raw, **kwargs):
+        fixed = super().fix_kline(raw, **kwargs)
+        for kline in fixed:
+            try:
+                kline[common_enums.PriceIndexes.IND_PRICE_TIME.value] = \
+                    int(self.get_uniformized_timestamp(kline[common_enums.PriceIndexes.IND_PRICE_TIME.value]))
+                self._fix_ohlcv_prices(kline)
+            except KeyError as e:
+                self.logger.error(f"Fail to fix kline ({e})")
+        return fixed
+
+    def parse_kline(self, fixed, **kwargs):
+        # CCXT standard kline parsing logic
+        return fixed
+
     def fix_ticker(self, raw, **kwargs):
         fixed = super().fix_ticker(raw, **kwargs)
         # CCXT standard ticker fixing logic

octobot_trading/exchanges/connectors/ccxt/ccxt_connector.py

@@ -211,9 +211,10 @@ async def get_kline_price(self,
                               time_frame: octobot_commons.enums.TimeFrames,
                               **kwargs: dict) -> typing.Optional[list]:
         try:
-            # default implementation
-            # already adapted in get_symbol_prices
-            return await self.get_symbol_prices(symbol, time_frame, limit=1, **kwargs)
+            with self.error_describer():
+                return self.adapter.adapt_kline(
+                    await self.client.fetch_ohlcv(symbol, time_frame.value, limit=1, params=kwargs)
+                )
         except ccxt.NotSupported:
             raise octobot_trading.errors.NotSupported
         except ccxt.BaseError as e:

octobot_trading/exchanges/connectors/ccxt/ccxt_websocket_connector.py

@@ -728,6 +728,7 @@ async def candle(self, candles: list, symbol=None, timeframe=None, **kwargs):
         :param kwargs: the feed kwargs
         """
         time_frame = commons_enums.TimeFrames(timeframe)
+        kline = self.adapter.adapt_kline([copy.deepcopy(candles[-1])])[0]
         adapted = self.adapter.adapt_ohlcv(candles, time_frame=time_frame)
         last_candle = adapted[-1]
         if symbol not in self.watched_pairs:
@@ -765,23 +766,23 @@ async def candle(self, candles: list, symbol=None, timeframe=None, **kwargs):
                 trading_constants.KLINE_CHANNEL,
                 time_frame,
                 symbol,
-                last_candle
+                kline
             )
 
         # Push a new ticker if necessary : only push on the min timeframe
         if time_frame is self.min_timeframe:
             ticker = {
-                Ectc.HIGH.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_HIGH.value],
-                Ectc.LOW.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_LOW.value],
+                Ectc.HIGH.value: kline[commons_enums.PriceIndexes.IND_PRICE_HIGH.value],
+                Ectc.LOW.value: kline[commons_enums.PriceIndexes.IND_PRICE_LOW.value],
                 Ectc.BID.value: None,
                 Ectc.BID_VOLUME.value: None,
                 Ectc.ASK.value: None,
                 Ectc.ASK_VOLUME.value: None,
-                Ectc.OPEN.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_OPEN.value],
-                Ectc.CLOSE.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_CLOSE.value],
-                Ectc.LAST.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_CLOSE.value],
+                Ectc.OPEN.value: kline[commons_enums.PriceIndexes.IND_PRICE_OPEN.value],
+                Ectc.CLOSE.value: kline[commons_enums.PriceIndexes.IND_PRICE_CLOSE.value],
+                Ectc.LAST.value: kline[commons_enums.PriceIndexes.IND_PRICE_CLOSE.value],
                 Ectc.PREVIOUS_CLOSE.value: None,
-                Ectc.BASE_VOLUME.value: last_candle[commons_enums.PriceIndexes.IND_PRICE_VOL.value],
+                Ectc.BASE_VOLUME.value: kline[commons_enums.PriceIndexes.IND_PRICE_VOL.value],
                 Ectc.TIMESTAMP.value: self.exchange.get_exchange_current_time(),
             }
             await self.push_to_channel(