feat: Publishing offchain future assets

.gitignore

@@ -62,4 +62,7 @@ cover/
 devnet.pkl
 
 .idea/
-.vscode/
+.vscode/
+
+# Vscode workspaces
+*.code-workspace

pragma/core/entry.py

@@ -9,20 +9,30 @@
 
 class Entry(abc.ABC):
     @abc.abstractmethod
-    def serialize(self) -> Dict[str, str]: ...
+    def to_tuple(self) -> Tuple: ...
 
     @abc.abstractmethod
-    def to_tuple(self) -> Tuple: ...
+    def serialize(self) -> Dict[str, str]: ...
+
+    @abc.abstractclassmethod
+    def offchain_serialize(self) -> Dict[str, str]: ...
 
     @staticmethod
     def serialize_entries(entries: List[Entry]) -> List[Dict[str, int]]:
         serialized_entries = [
-            entry.serialize() for entry in entries if issubclass(entry, Entry)
+            entry.serialize() for entry in entries if isinstance(entry, Entry)
         ]
         return list(filter(lambda item: item is not None, serialized_entries))
 
     @staticmethod
-    def flatten_entries(entries: List[SpotEntry]) -> List[int]:
+    def offchain_serialize_entries(entries: List[Entry]) -> List[Dict[str, int]]:
+        serialized_entries = [
+            entry.offchain_serialize() for entry in entries if isinstance(entry, Entry)
+        ]
+        return list(filter(lambda item: item is not None, serialized_entries))
+
+    @staticmethod
+    def flatten_entries(entries: List[Entry]) -> List[int]:
         """This flattens entriees to tuples.  Useful when you need the raw felt array"""
         expanded = [entry.to_tuple() for entry in entries]
         flattened = [x for entry in expanded for x in entry]
@@ -167,30 +177,6 @@
             autoscale_volume=False,
         )
 
-    @staticmethod
-    def serialize_entries(entries: List[SpotEntry]) -> List[Dict[str, int]]:
-        """serialize entries to a List of dictionaries"""
-        # TODO (#000): log errors
-        serialized_entries = [
-            entry.serialize()
-            for entry in entries
-            # TODO (#000): This needs to be much more resilient to publish errors
-            if isinstance(entry, SpotEntry)
-        ]
-        return list(filter(lambda item: item is not None, serialized_entries))
-
-    @staticmethod
-    def offchain_serialize_entries(entries: List[SpotEntry]) -> List[Dict[str, int]]:
-        """serialize entries to a List of dictionaries for off-chain consumption"""
-        # TODO (#000): log errors
-        serialized_entries = [
-            entry.offchain_serialize()
-            for entry in entries
-            # TODO (#000): This needs to be much more resilient to publish errors
-            if isinstance(entry, SpotEntry)
-        ]
-        return list(filter(lambda item: item is not None, serialized_entries))
-
     def __repr__(self):
         return (
             f'SpotEntry(pair_id="{felt_to_str(self.pair_id)}", '
@@ -204,14 +190,17 @@
     """
     Represents a Future Entry.
 
+    Also used to represent a Perp Entry - the only difference is that a perpetual future has no
+    expiry timestamp.
+
     ⚠️ By default, the constructor will autoscale the provided volume to be quoted in the base asset.
     This behavior can be overwritten witht the `autoscale_volume` parameter.
     """
 
     base: BaseEntry
     pair_id: int
     price: int
-    expiry_timestamp: int
+    expiry_timestamp: Optional[int]
     volume: int
 
     def __init__(
@@ -221,7 +210,7 @@
         timestamp: int,
         source: Union[str, int],
         publisher: Union[str, int],
-        expiry_timestamp: int,
+        expiry_timestamp: Optional[int] = None,
         volume: Optional[float] = 0,
         autoscale_volume: bool = True,
     ):
@@ -297,31 +286,19 @@
             "volume": self.volume,
         }
 
-    @staticmethod
-    def from_dict(entry_dict: Dict[str, str]) -> "FutureEntry":
-        base = dict(entry_dict["base"])
-        return FutureEntry(
-            entry_dict["pair_id"],
-            entry_dict["price"],
-            base["timestamp"],
-            base["source"],
-            base["publisher"],
-            entry_dict["expiration_timestamp"],
-            volume=entry_dict["volume"],
-            autoscale_volume=False,
-        )
-
-    @staticmethod
-    def serialize_entries(entries: List[FutureEntry]) -> List[Dict[str, int]]:
-        """serialize entries to a List of dictionaries"""
-        # TODO (#000): log errors
-        serialized_entries = [
-            entry.serialize()
-            for entry in entries
-            # TODO (#000): This needs to be much more resilient to publish errors
-            if isinstance(entry, FutureEntry)
-        ]
-        return list(filter(lambda item: item is not None, serialized_entries))
+    def offchain_serialize(self) -> Dict[str, str]:
+        serialized = {
+            "base": {
+                "timestamp": self.base.timestamp,
+                "source": felt_to_str(self.base.source),
+                "publisher": felt_to_str(self.base.publisher),
+            },
+            "pair_id": felt_to_str(self.pair_id),
+            "price": self.price,
+            "volume": self.volume,
+            "expiration_timestamp": self.expiry_timestamp,
+        }
+        return serialized
 
     def __repr__(self):
         return (
@@ -334,58 +311,16 @@
             f'expiry_timestamp={self.expiry_timestamp})")'
         )
 
-
-class GenericEntry(Entry):
-    base: BaseEntry
-    key: int
-    value: int
-
-    def __init__(
-        self,
-        timestamp: int,
-        source: Union[str, int],
-        publisher: Union[str, int],
-        key: Union[str, int],
-        value: int,
-    ):
-        if isinstance(publisher, str):
-            publisher = str_to_felt(publisher)
-
-        if isinstance(source, str):
-            source = str_to_felt(source)
-
-        if isinstance(key, str):
-            key = str_to_felt(key)
-
-        self.base = BaseEntry(timestamp, source, publisher)
-        self.key = key
-        self.value = value
-
-    def serialize(self) -> Dict[str, str]:
-        return {
-            "base": {
-                "timestamp": self.base.timestamp,
-                "source": self.base.source,
-                "publisher": self.base.publisher,
-            },
-            "key": self.key,
-            "value": self.value,
-        }
-
-    def to_tuple(self) -> Tuple:
-        return (
-            self.base.timestamp,
-            self.base.source,
-            self.base.publisher,
-            self.key,
-            self.value,
-        )
-
-    def __repr__(self):
-        return (
-            f'GenericEntry(key="{felt_to_str(self.key)}", '
-            f"value={self.value}, "
-            f"timestamp={self.base.timestamp}, "
-            f'source="{felt_to_str(self.base.source)}", '
-            f'publisher="{felt_to_str(self.base.publisher)}")'
+    @staticmethod
+    def from_dict(entry_dict: Dict[str, str]) -> "FutureEntry":
+        base = dict(entry_dict["base"])
+        return FutureEntry(
+            entry_dict["pair_id"],
+            entry_dict["price"],
+            base["timestamp"],
+            base["source"],
+            base["publisher"],
+            entry_dict["expiration_timestamp"],
+            volume=entry_dict["volume"],
+            autoscale_volume=False,
         )

pragma/core/mixins/offchain.py

@@ -10,8 +10,9 @@
 from starknet_py.net.client import Client
 from starknet_py.utils.typed_data import TypedData
 
-from pragma.core.entry import SpotEntry
+from pragma.core.entry import Entry, FutureEntry, SpotEntry
 from pragma.core.types import AggregationMode
+from pragma.core.utils import exclude_none_and_exceptions
 
 logger = logging.getLogger(__name__)
 
@@ -26,25 +27,14 @@
     ],
 )
 
-"""
-{'base': 
-{'publisher': 88314212732225, 
-'source': 5787760245619121969, 'timestamp': 1697147959}, 
-'pair_id': 19514442401534788, '
-price': 1000, 
-'volume': 0}
-"""
 
-
-def build_publish_message(entries: List[SpotEntry], now: int, expiry: int) -> TypedData:
+def build_publish_message(entries: List[Entry], for_future_entries: bool) -> TypedData:
     message = {
         "domain": {"name": "Pragma", "version": "1"},
         "primaryType": "Request",
         "message": {
             "action": "Publish",
-            "entries": SpotEntry.serialize_entries(entries),
-            # "timestamp": now,
-            # "expiration": expiry,
+            "entries": Entry.serialize_entries(entries),
         },
         "types": {
             "StarkNetDomain": [
@@ -68,7 +58,10 @@
             ],
         },
     }
-
+    if for_future_entries:
+        message["types"]["Entry"] = message["types"]["Entry"] + [
+            {"name": "expiration_timestamp", "type": "felt"},
+        ]
     return message
 
 
@@ -79,13 +72,11 @@
     ssl_context: ssl.SSLContext
     api_key: str
 
-    def sign_publish_message(
-        self, entries: List[SpotEntry], now: int, expiry: int
-    ) -> (List[int], int):
+    def sign_publish_message(self, entries: List[Entry]) -> (List[int], int):
         """
         Sign a publish message
         """
-        message = build_publish_message(entries, now, expiry)
+        message = build_publish_message(entries)
         hash_ = TypedData.from_dict(message).message_hash(self.account.address)
         sig = self.account.sign_message(message)
 
@@ -104,20 +95,37 @@
 
     async def publish_data(
         self,
-        entries: List[SpotEntry],
+        entries: List[Entry],
     ):
         """
         Publish data to PragmAPI
 
         Args:
-            entries (List[SpotEntry]): List of SpotEntry to publish
+            entries (List[Entry]): List of Entry to publish
         """
+        # TODO: We sometimes have some Error types in entries, we need to
+        # investigate why and don't push them in our entries.
+        # Currently, we only exclude them from here.
+        entries = exclude_none_and_exceptions(entries)
+
+        spot_entries = [entry for entry in entries if isinstance(entry, SpotEntry)]
+        future_entries = [entry for entry in entries if isinstance(entry, FutureEntry)]
+
+        spot_response = self._publish_entries(spot_entries)
+        future_response = self._publish_entries(future_entries, is_future=True)
+
+        return spot_response, future_response
+
+    async def _publish_entries(self, entries: List[Entry], is_future: bool = False):
+        # Check if all entries are of the same type
+        EntryClass = type(entries[0])
+        assert all(isinstance(entry, EntryClass) for entry in entries)
 
         now = int(time.time())
         expiry = now + 24 * 60 * 60
 
         # Sign message
-        sig, _ = self.sign_publish_message(entries, now, expiry)
+        sig, _ = self.sign_publish_message(entries, is_future)
 
         # Add headers
         headers: Dict = {
@@ -128,10 +136,12 @@
 
         body = {
             "signature": [str(s) for s in sig],
-            "entries": SpotEntry.offchain_serialize_entries(entries),
+            "entries": EntryClass.offchain_serialize_entries(entries),
         }
 
         url = self.api_url + "/v1/data/publish"
+        if isinstance(entries[0], FutureEntry):
+            url += "_future"
 
         logger.info(f"POST {url}")
         logger.info(f"Headers: {headers}")

pragma/core/mixins/oracle.py

@@ -78,7 +78,8 @@
             return
 
         invocations = []
-        serialized_spot_entries = SpotEntry.serialize_entries(entries)
+        spot_entries = [entry for entry in entries if isinstance(entry, SpotEntry)]
+        serialized_spot_entries = SpotEntry.serialize_entries(spot_entries)
         if pagination:
             index = 0
             while index < len(serialized_spot_entries):
@@ -112,7 +113,8 @@
                 hex(invocation.hash),
             )
 
-        serialized_future_entries = FutureEntry.serialize_entries(entries)
+        future_entries = [entry for entry in entries if isinstance(entry, FutureEntry)]
+        serialized_future_entries = FutureEntry.serialize_entries(future_entries)
         if pagination:
             index = 0
             while index < len(serialized_future_entries):