feat loads of changes

lyra/async_client.py

@@ -0,0 +1,154 @@
+"""
+Async client for Lyra
+"""
+
+import asyncio
+import json
+import sys
+import time
+import traceback
+
+from lyra.constants import PUBLIC_HEADERS
+from lyra.enums import InstrumentType, UnderlyingCurrency
+from lyra.ws_client import WsClient as BaseClient
+
+from multiprocessing import Process
+# we need a thread safe way to collect the events
+from multiprocessing import Lock
+from threading import Thread
+
+
+class AsyncClient(BaseClient):
+    """
+    We use the async client to make async requests to the lyra API
+    We us the ws client to make async requests to the lyra ws API
+    """
+
+    current_subscriptions = {}
+
+    listener = None
+    subscribing = False
+
+
+
+    async def fetch_ticker(self, instrument_name: str):
+        """
+        Fetch the ticker for a symbol
+        """
+        id = str(int(time.time()))
+        payload = {"instrument_name": instrument_name}
+        self.ws.send(json.dumps({"method": "public/get_ticker", "params": payload, "id": id}))
+
+        # we now wait for the response
+        while True:
+            response = self.ws.recv()
+            response = json.loads(response)
+            if response["id"] == id:
+                close = float(response["result"]["best_bid_price"])  + float(response["result"]["best_ask_price"]) / 2
+                response["result"]["close"] = close
+                return response["result"]
+
+
+
+
+    async def subscribe(self, instrument_name: str, group: str = "1", depth: str = "100"):
+        """
+        Subscribe to the order book for a symbol
+        """
+
+        self.subscribing = True
+        if instrument_name not in self.current_subscriptions:
+            channel = f"orderbook.{instrument_name}.{group}.{depth}"
+            msg = json.dumps({
+                "method": "subscribe", 
+                "params": {
+                    "channels": [channel]
+                }
+
+                })
+            print(f"Subscribing with {msg}")
+            self.ws.send(msg)
+            await self.collect_events(instrument_name=instrument_name)
+            print(f"Subscribed to {instrument_name}")
+            return
+
+        while instrument_name not in self.current_subscriptions:
+            await asyncio.sleep(1)
+        return self.current_subscriptions[instrument_name]
+
+
+
+    async def collect_events(self, subscription: str = None, instrument_name: str = None):
+        """Use a thread to check the subscriptions"""
+        try:
+            response = self.ws.recv()
+            response = json.loads(response)
+            if "error" in response:
+                print(response)
+                raise Exception(response["error"])
+            if "result" in response:
+                result = response["result"]
+                if "status" in result:
+                    print(f"Succesfully subscribed to {result['status']}")
+                    for channel, value in result['status'].items():
+                        print(f"Channel {channel} has value {value}")
+                        if "error" in value:
+                            raise Exception(value["error"])
+                    self.subscribing = False
+                    return
+
+
+            channel = response["params"]["channel"]
+
+            bids = response['params']['data']['bids']
+            asks = response['params']['data']['asks']
+
+            bids = list(map(lambda x: (float(x[0]), float(x[1])), bids))
+            asks = list(map(lambda x: (float(x[0]), float(x[1])), asks))
+
+            if instrument_name in self.current_subscriptions:
+                old_params = self.current_subscriptions[instrument_name]
+                _asks, _bids = old_params["asks"], old_params["bids"]
+                if not asks:
+                    asks = _asks
+                if not bids:
+                    bids = _bids
+            self.current_subscriptions[instrument_name] = {"asks": asks, "bids": bids}
+            return self.current_subscriptions[instrument_name]
+        except Exception as e:
+            print(f"Error: {e}")
+            print(traceback.print_exc())
+            sys.exit(1)
+
+    async def watch_order_book(self, instrument_name: str, group: str = "1", depth: str = "100"):
+        """
+        Watch the order book for a symbol
+        orderbook.{instrument_name}.{group}.{depth}
+        """
+
+        if not self.subscribing:
+            await self.subscribe(instrument_name, group, depth)
+
+
+        if not self.listener:
+            print(f"Started listener for {instrument_name}")
+            self.listener = True
+
+        await self.collect_events(instrument_name=instrument_name)
+        while instrument_name not in self.current_subscriptions:
+            await asyncio.sleep(1)
+            print(f"Waiting for {instrument_name} to be in current subscriptions")
+
+        return self.current_subscriptions[instrument_name]
+
+
+    async def fetch_instruments(self, expired=False, instrument_type: InstrumentType = InstrumentType.PERP, currency: UnderlyingCurrency = UnderlyingCurrency.BTC):
+        return super().fetch_instruments(expired, instrument_type, currency)
+
+    async def close(self):
+        """
+        Close the connection
+        """
+        self.ws.close()
+        # if self.listener:
+        #     self.listener.join()

lyra/autonomy/packages/eightballer/protocols/markets/__init__.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+# ------------------------------------------------------------------------------
+#
+#   Copyright 2023 eightballer
+#
+#   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.
+#
+# ------------------------------------------------------------------------------
+
+"""
+This module contains the support resources for the markets protocol.
+
+It was created with protocol buffer compiler version `libprotoc 3.19.4` and aea protocol generator version `1.0.0`.
+"""
+
+from packages.eightballer.protocols.markets.message import MarketsMessage
+from packages.eightballer.protocols.markets.serialization import MarketsSerializer
+
+MarketsMessage.serializer = MarketsSerializer

lyra/autonomy/packages/eightballer/protocols/markets/custom_types.py

@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+# ------------------------------------------------------------------------------
+#
+#   Copyright 2023 eightballer
+#
+#   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.
+#
+# ------------------------------------------------------------------------------
+
+"""This module contains class representations corresponding to every custom type in the protocol specification."""
+
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+
+class ErrorCode(Enum):
+    """This class represents an instance of ErrorCode."""
+
+    UNSUPPORTED_PROTOCOL = 0
+    DECODING_ERROR = 1
+    INVALID_MESSAGE = 2
+    UNSUPPORTED_SKILL = 3
+    INVALID_DIALOGUE = 4
+
+    @staticmethod
+    def encode(error_code_protobuf_object: Any, error_code_object: "ErrorCode") -> None:
+        """
+        Encode an instance of this class into the protocol buffer object.
+        The protocol buffer object in the error_code_protobuf_object argument is matched with the instance of this class in the 'error_code_object' argument.
+        :param error_code_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :param error_code_object: an instance of this class to be encoded in the protocol buffer object.
+        """
+        error_code_protobuf_object.error_code = error_code_object.value
+
+    @classmethod
+    def decode(cls, error_code_protobuf_object: Any) -> "ErrorCode":
+        """
+        Decode a protocol buffer object that corresponds with this class into an instance of this class.
+        A new instance of this class is created that matches the protocol buffer object in the 'error_code_protobuf_object' argument.
+        :param error_code_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :return: A new instance of this class that matches the protocol buffer object in the 'error_code_protobuf_object' argument.
+        """
+        enum_value_from_pb2 = error_code_protobuf_object.error_code
+        return ErrorCode(enum_value_from_pb2)
+
+
+@dataclass
+class Market:
+    """
+    This class represents an instance of Market.
+    """
+
+    id: str
+    lowercaseId: Optional[str] = None
+    symbol: Optional[str] = None
+    base: Optional[str] = None
+    quote: Optional[str] = None
+    settle: Optional[str] = None
+    baseId: Optional[str] = None
+    quoteId: Optional[str] = None
+    settleId: Optional[str] = None
+    type: Optional[str] = None
+    spot: Optional[bool] = None
+    margin: Optional[bool] = None
+    swap: Optional[bool] = None
+    future: Optional[bool] = None
+    option: Optional[bool] = None
+    active: Optional[bool] = None
+    contract: Optional[bool] = None
+    linear: Optional[bool] = None
+    inverse: Optional[bool] = None
+    taker: Optional[float] = None
+    maker: Optional[float] = None
+    contractSize: Optional[float] = None
+    expiry: Optional[float] = None
+    expiryDatetime: Optional[str] = None
+    strike: Optional[float] = None
+    optionType: Optional[str] = None
+    precision: Optional[float] = None
+    limits: Optional[str] = None
+    info: Optional[Dict[str, Any]] = None
+    exchange_id: Optional[str] = None
+    created: Optional[str] = None
+
+    @staticmethod
+    def encode(market_protobuf_object, market_object: "Market") -> None:
+        """
+        Encode an instance of this class into the protocol buffer object.
+
+        The protocol buffer object in the market_protobuf_object argument is matched with the instance of this class in the 'market_object' argument.
+
+        :param market_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :param market_object: an instance of this class to be encoded in the protocol buffer object.
+        """
+        for (
+            attribute
+        ) in Market.__dataclass_fields__.keys():  # pylint: disable=no-member
+            if hasattr(market_object, attribute):
+                value = getattr(market_object, attribute)
+                setattr(market_protobuf_object.Market, attribute, value)
+            else:
+                setattr(market_protobuf_object.Market, attribute, None)
+
+    @classmethod
+    def decode(cls, market_protobuf_object) -> "Market":
+        """
+        Decode a protocol buffer object that corresponds with this class into an instance of this class.
+
+        A new instance of this class is created that matches the protocol buffer object in the 'market_protobuf_object' argument.
+
+        :param market_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :return: A new instance of this class that matches the protocol buffer object in the 'market_protobuf_object' argument.
+        """
+        attribute_dict = dict()
+        for (
+            attribute
+        ) in Market.__dataclass_fields__.keys():  # pylint: disable=no-member
+            if hasattr(market_protobuf_object.Market, attribute):
+                if getattr(market_protobuf_object.Market, attribute) is not None:
+                    attribute_dict[attribute] = getattr(
+                        market_protobuf_object.Market, attribute
+                    )
+        return cls(**attribute_dict)
+
+    def __eq__(self, other):
+        if isinstance(other, Market):
+            set_of_self_attributue = set(
+                i
+                for i in Market.__dataclass_fields__.keys()  # pylint: disable=no-member
+                if getattr(self, i) is not None
+            )
+            set_of_other_attributue = set(
+                i
+                for i in Market.__dataclass_fields__.keys()  # pylint: disable=no-member
+                if getattr(other, i) is not None
+            )
+            return set_of_self_attributue == set_of_other_attributue
+        return False
+
+    def to_json(self):
+        """TO a pretty dictionary string."""
+        result = {}
+        for (
+            attribute
+        ) in Market.__dataclass_fields__.keys():  # pylint: disable=no-member
+            if hasattr(self, attribute):
+                value = getattr(self, attribute)
+                if value is not None:
+                    result[attribute] = value
+        return result
+
+
+@dataclass
+class Markets:
+    """This class represents an instance of Markets."""
+
+    markets: List[Market]
+
+    @staticmethod
+    def encode(markets_protobuf_object, markets_object: "Markets") -> None:
+        """
+        Encode an instance of this class into the protocol buffer object.
+
+        The protocol buffer object in the markets_protobuf_object argument is matched with the instance of this class in the 'markets_object' argument.
+
+        :param markets_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :param markets_object: an instance of this class to be encoded in the protocol buffer object.
+        """
+        if markets_protobuf_object is None:
+            raise ValueError(
+                "The protocol buffer object 'markets_protobuf_object' is not initialized."
+            )
+        markets_protobuf_object.Markets.markets = markets_object.markets
+
+    @classmethod
+    def decode(cls, markets_protobuf_object) -> "Markets":
+        """
+        Decode a protocol buffer object that corresponds with this class into an instance of this class.
+
+        A new instance of this class is created that matches the protocol buffer object in the 'markets_protobuf_object' argument.
+
+        :param markets_protobuf_object: the protocol buffer object whose type corresponds with this class.
+        :return: A new instance of this class that matches the protocol buffer object in the 'markets_protobuf_object' argument.
+        """
+        return cls(markets_protobuf_object.Markets.markets)
+
+    def __eq__(self, other):
+        if isinstance(other, Markets):
+            set_of_self_attributue = set(
+                i
+                for i in Markets.__dataclass_fields__.keys()  # pylint: disable=no-member
+                if getattr(self, i) is not None
+            )
+            set_of_other_attributue = set(
+                i
+                for i in Markets.__dataclass_fields__.keys()  # pylint: disable=no-member
+                if getattr(other, i) is not None
+            )
+            return set_of_self_attributue == set_of_other_attributue
+        return False