rewriting gql query such that it uses existing purchasing pattern w/ …

…lastPriceValue rather than predictionsSubscriptions. Either should work.
oceanprotocol · Dec 22, 2023 · 44f5fc9 · 44f5fc9
1 parent 8d4ff34
commit 44f5fc9
Show file tree

Hide file tree

Showing 4 changed files with 55 additions and 41 deletions.
diff --git a/pdr_backend/contract/subscription.py b/pdr_backend/contract/subscription.py
@@ -14,6 +14,7 @@ def __init__(
         source: str,
         timestamp: int,  # timestamp == subscription purchased timestamp
         tx_id: str,
+        amt: float,
         user: str,
     ) -> None:
         self.ID = ID
@@ -22,6 +23,7 @@ def __init__(
         self.source = source
         self.timestamp = timestamp
         self.tx_id = tx_id
+        self.amt = amt
         self.user = user
 
 
@@ -37,6 +39,7 @@ def mock_subscription(subscription_tuple: tuple) -> Subscription:
         source,
         timestamp,
         tx_id,
+        amt,
         event_index,
         user,
     ) = subscription_tuple
@@ -49,6 +52,7 @@ def mock_subscription(subscription_tuple: tuple) -> Subscription:
         source=source,
         timestamp=timestamp,
         tx_id=tx_id,
+        amt=amt,
         user=user,
     )
 
@@ -68,6 +72,7 @@ def mock_subscriptions() -> List[Subscription]:
         "binance",
         1698850800,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809592",
+        2.01,
         98,
         "0x2433e002ed10b5d6a3d8d1e0c5d2083be9e37f1d",
     ),
@@ -77,6 +82,7 @@ def mock_subscriptions() -> List[Subscription]:
         "kraken",
         1698937200,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809593",
+        2.01,
         99,
         "0xabcdef0123456789abcdef0123456789abcdef01",
     ),
@@ -86,6 +92,7 @@ def mock_subscriptions() -> List[Subscription]:
         "kraken",
         1699110000,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809594",
+        2.01,
         100,
         "0x123456789abcdef0123456789abcdef01234567",
     ),
@@ -95,6 +102,7 @@ def mock_subscriptions() -> List[Subscription]:
         "binance",
         1699110000,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809595",
+        2.01,
         101,
         "0xabcdef0123456789abcdef0123456789abcdef02",
     ),
@@ -104,6 +112,7 @@ def mock_subscriptions() -> List[Subscription]:
         "kraken",
         1699110000,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809596",
+        2.01,
         102,
         "0xabcdef0123456789abcdef0123456789abcdef03",
     ),
@@ -113,6 +122,7 @@ def mock_subscriptions() -> List[Subscription]:
         "kraken",
         1699200000,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809597",
+        2.01,
         103,
         "0xabcdef0123456789abcdef0123456789abcdef04",
     ),
@@ -122,6 +132,7 @@ def mock_subscriptions() -> List[Subscription]:
         "binance",
         1699200000,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809598",
+        2.01,
         104,
         "0xabcdef0123456789abcdef0123456789abcdef05",
     ),
@@ -131,6 +142,7 @@ def mock_subscriptions() -> List[Subscription]:
         "kraken",
         1699286400,
         "0x01d3285e0e3b83a4c029142477c0573c3be5317ff68223703696093b27809599",
+        2.01,
         105,
         "0xabcdef0123456789abcdef0123456789abcdef06",
     ),

diff --git a/pdr_backend/lake/table_pdr_subscriptions.py b/pdr_backend/lake/table_pdr_subscriptions.py
@@ -2,7 +2,7 @@
 
 import polars as pl
 from enforce_typing import enforce_types
-from polars import Int64, Utf8
+from polars import Int64, Utf8, Float32
 
 from pdr_backend.subgraph.subgraph_subscriptions import (
     fetch_filtered_subscriptions,
@@ -20,7 +20,7 @@
     "timeframe": Utf8,
     "source": Utf8,
     "tx_id": Utf8,
-    "event_index": Int64,
+    "amt": Float32,
     "timestamp": Int64,
     "user": Utf8,
 }

diff --git a/pdr_backend/lake/test/test_table_subscriptions.py b/pdr_backend/lake/test/test_table_subscriptions.py
@@ -184,4 +184,5 @@ def test_load_and_verify_schema(
 
     assert len(gql_dfs) == 1
     assert len(gql_dfs[pdr_subscriptions_record]) == 7
+    assert round(gql_dfs[pdr_subscriptions_record]["amt"].sum(), 2) == 14.07
     assert gql_dfs[pdr_subscriptions_record].schema == subscriptions_schema
diff --git a/pdr_backend/subgraph/subgraph_subscriptions.py b/pdr_backend/subgraph/subgraph_subscriptions.py
@@ -49,39 +49,36 @@ def fetch_filtered_subscriptions(
     contracts = [f.lower() for f in contracts]
 
     # pylint: disable=line-too-long
-    if len(contracts) > 0:
-        where_clause = f", where: {{predictContract_: {{id_in: {json.dumps(contracts)}, timestamp_gt: {start_ts}, timestamp_lt: {end_ts}}}}}"
-    else:
-        where_clause = f", where: {{timestamp_gt: {start_ts}, timestamp_lt: {end_ts}}}"
-
     while True:
         query = f"""
             {{
-                predictSubscriptions(skip: {offset}, first: {chunk_size} {where_clause}) {{
-                    id
-                    txId
-                    timestamp
-                    expireTime
-                    user {{
-                        id
-                    }}
-                    predictContract {{
+                predictContracts(first: 1000, where: {{id_in {json.dumps(contracts)}}}) {{
+                    id	
+                    token {{
                         id
-                        token {{
-                            id
-                            name
-                            nft{{
-                                nftData {{
+                        name
+                        symbol
+                        nft {{
+                            owner {{
+                                id
+                            }}
+                            nftData {{
                                 key
                                 value
-                                }}
                             }}
                         }}
-                        token {{
+                        orders(where: {{createdTimestamp_gt: {start_ts}, createdTimestamp_lt: {end_ts} first: {chunk_size}, skip: {offset}}}) {{
+                            createdTimestamp
+                            consumer {{
+                                id
+                            }}
                             lastPriceValue
+                            txId
                         }}
-                        secondsPerSubscription
                     }}
+                    secondsPerEpoch
+                    secondsPerSubscription
+                    truevalSubmitTimeout
                 }}
             }}"""
 
@@ -97,29 +94,33 @@ def fetch_filtered_subscriptions(
         if not "data" in result:
             break
 
-        data = result["data"]["predictSubscriptions"]
+        data = result["data"]["predictContracts"]
         if len(data) == 0:
             break
 
-        for subscription_sg_dict in data:
-            info725 = subscription_sg_dict["predictContract"]["token"]["nft"]["nftData"]
+        for contract_sg_dict in data:
+            info725 = contract_sg_dict["token"]["nft"]["nftData"]
             info = info725_to_info(info725)
             pair = info["pair"]
             timeframe = info["timeframe"]
             source = info["source"]
-            timestamp = subscription_sg_dict["timestamp"]
-            tx_id = subscription_sg_dict["txId"]
-            user = subscription_sg_dict["user"]["id"]
-
-            subscription = Subscription(
-                ID=subscription_sg_dict["id"],
-                pair=pair,
-                timeframe=timeframe,
-                source=source,
-                timestamp=timestamp,
-                tx_id=tx_id,
-                user=user,
-            )
-            subscriptions.append(subscription)
+
+            for subscription_sg_dict in contract_sg_dict["token"]["orders"]:
+                timestamp = subscription_sg_dict["createdTimestamp"]
+                tx_id = subscription_sg_dict["txId"]
+                user = subscription_sg_dict["consumer"]["id"]
+                amt = float(subscription_sg_dict["lastPriceValue"]) * 1.201
+
+                subscription = Subscription(
+                    ID=subscription_sg_dict["id"],
+                    pair=pair,
+                    timeframe=timeframe,
+                    source=source,
+                    timestamp=timestamp,
+                    tx_id=tx_id,
+                    amt=amt,
+                    user=user,
+                )
+                subscriptions.append(subscription)
 
     return subscriptions