Add filter for paused contracts in subgraph queries

pdr_backend/analytics/check_network.py

@@ -114,7 +114,7 @@ def do_query_network(subgraph_url: str, lookback_hours: int):
     start_ut = cur_ut - lookback_hours * 60 * 60
     query = """
             {
-                predictContracts{
+                predictContracts(where: {paused: false}){
                     id
                     token{
                         name

pdr_backend/cli/cli_arguments.py

@@ -24,7 +24,7 @@
   pdr predictoor PPSS_FILE NETWORK
   pdr dashboard PPSS_FILE NETWORK
   pdr trader APPROACH PPSS_FILE NETWORK
-  pdr claim_payouts PPSS_FILE
+  pdr claim_payouts PPSS_FILE [--include_paused]
   pdr claim_ROSE PPSS_FILE
 """
 
@@ -276,6 +276,17 @@ def add_argument_NATIVE_TOKEN(self):
         )
 
 
+@enforce_types
+class INCLUDE_PAUSED_Mixin:
+    def add_argument_INCLUDE_PAUSED(self):
+        self.add_argument(
+            "--include_paused",
+            action="store_true",
+            default=False,
+            help="Include paused contracts when querying for payouts",
+        )
+
+
 # ========================================================================
 # argparser base classes
 class CustomArgParser(NestedArgParser):
@@ -554,7 +565,15 @@ def print_args(arguments: Namespace, nested_args: dict):
 SimArgParser = _ArgParser_PPSS
 PredictoorArgParser = _ArgParser_PPSS_NETWORK
 TraderArgParser = _ArgParser_APPROACH_PPSS_NETWORK
-ClaimOceanArgParser = _ArgParser_PPSS
+
+
+@enforce_types
+class ClaimOceanArgParser(CustomArgParser, PPSS_Mixin, INCLUDE_PAUSED_Mixin):
+    def __init__(self, description: str, command_name: str):
+        super().__init__(description=description)
+        self.add_arguments_bulk(command_name, ["PPSS", "INCLUDE_PAUSED"])
+
+
 ClaimRoseArgParser = _ArgParser_PPSS
 
 # power tools

pdr_backend/cli/cli_module.py

@@ -129,7 +129,8 @@ def do_claim_payouts(args, nested_args=None):
         network="sapphire-mainnet",
         nested_override_args=nested_args,
     )
-    do_ocean_payout(ppss)
+    include_paused = args.include_paused
+    do_ocean_payout(ppss, include_paused=include_paused)
 
 
 @enforce_types

pdr_backend/cli/test/test_cli_module.py

@@ -110,14 +110,30 @@ class _NATIVE_TOKEN:
     NATIVE_TOKEN = True
 
 
+class _INCLUDE_PAUSED:
+    include_paused = False
+
+
+class _INCLUDE_PAUSED_TRUE:
+    include_paused = True
+
+
 class _Base:
     def __init__(self, *args, **kwargs):
         pass
 
 
 class MockArgParser_PPSS(_Base):
     def parse_args(self):
-        class MockArgs(Namespace, _PPSS, _PPSS_OBJ):
+        class MockArgs(Namespace, _PPSS, _PPSS_OBJ, _INCLUDE_PAUSED):
+            pass
+
+        return MockArgs()
+
+
+class MockArgParser_PPSS_WITH_PAUSED(_Base):
+    def parse_args(self):
+        class MockArgs(Namespace, _PPSS, _PPSS_OBJ, _INCLUDE_PAUSED_TRUE):
             pass
 
         return MockArgs()
@@ -265,6 +281,19 @@ def test_do_claim_payouts(monkeypatch):
 
     do_claim_payouts(MockArgParser_PPSS().parse_args())
     mock_f.assert_called()
+    # Verify it was called with include_paused=False by default
+    assert mock_f.call_args[1]["include_paused"] is False
+
+
+@enforce_types
+def test_do_claim_payouts_with_include_paused(monkeypatch):
+    mock_f = Mock()
+    monkeypatch.setattr(f"{_CLI_PATH}.do_ocean_payout", mock_f)
+
+    do_claim_payouts(MockArgParser_PPSS_WITH_PAUSED().parse_args())
+    mock_f.assert_called()
+    # Verify it was called with include_paused=True
+    assert mock_f.call_args[1]["include_paused"] is True
 
 
 @enforce_types

pdr_backend/payout/payout.py

@@ -55,7 +55,9 @@ def request_payout_batches(
 
 
 @enforce_types
-def find_slots_and_payout_with_mgr(pred_submitter_mgr, ppss, query_old_slots=False):
+def find_slots_and_payout_with_mgr(
+    pred_submitter_mgr, ppss, query_old_slots=False, include_paused=False
+):
     # we only need to query in one direction, since both predict on the same slots
     # query_old_slots is by default false to improve bot speed,
     # running the command line argument will set it to true
@@ -65,7 +67,9 @@ def find_slots_and_payout_with_mgr(pred_submitter_mgr, ppss, query_old_slots=Fal
     logger.info("Starting payout")
     wait_until_subgraph_syncs(web3_config, subgraph_url)
     logger.info("Finding pending payouts")
-    pending_slots = query_pending_payouts(subgraph_url, up_addr, query_old_slots)
+    pending_slots = query_pending_payouts(
+        subgraph_url, up_addr, query_old_slots, include_paused
+    )
     payout_batch_size = ppss.predictoor_ss.payout_batch_size
     shared_slots = find_shared_slots(pending_slots, payout_batch_size)
     unique_slots = count_unique_slots(shared_slots)
@@ -88,7 +92,9 @@ def find_slots_and_payout_with_mgr(pred_submitter_mgr, ppss, query_old_slots=Fal
 
 
 @enforce_types
-def do_ocean_payout(ppss: PPSS, check_network: bool = True):
+def do_ocean_payout(
+    ppss: PPSS, check_network: bool = True, include_paused: bool = False
+):
     web3_config = ppss.web3_pp.web3_config
     if check_network:
         assert ppss.web3_pp.network == "sapphire-mainnet"
@@ -97,7 +103,7 @@ def do_ocean_payout(ppss: PPSS, check_network: bool = True):
     pred_submitter_mgr_addr = ppss.predictoor_ss.pred_submitter_mgr
     pred_submitter_mgr = PredSubmitterMgr(ppss.web3_pp, pred_submitter_mgr_addr)
 
-    find_slots_and_payout_with_mgr(pred_submitter_mgr, ppss, True)
+    find_slots_and_payout_with_mgr(pred_submitter_mgr, ppss, True, include_paused)
 
 
 @enforce_types

pdr_backend/subgraph/legacy/subgraph_slot.py

@@ -58,6 +58,7 @@ def get_predict_slots_query(
                 slot_lte: {initial_slot}
                 slot_gte: {last_slot}
                 predictContract_in: {asset_ids_str}
+                predictContract_: {{paused: false}}
             }}
                 ) {{
                 id

pdr_backend/subgraph/subgraph_consume_so_far.py

@@ -24,7 +24,7 @@ def get_consume_so_far_per_contract(
     while True:
         query = """
         {
-            predictSubscriptions(where: {timestamp_gt:%s, user_:{id: "%s"}}, first: %s, skip: %s){
+            predictSubscriptions(where: {timestamp_gt:%s, user_:{id: "%s"}, predictContract_: {paused: false}}, first: %s, skip: %s){
                 id
                 timestamp
                 user {

pdr_backend/subgraph/subgraph_dfbuyer.py

@@ -17,7 +17,7 @@ def get_consume_so_far(
     while True:  # pylint: disable=too-many-nested-blocks
         query = """
         {
-            predictContracts(skip:%s, first:%s){
+            predictContracts(where: {paused: false}, skip:%s, first:%s){
                 id
                 token{
                     orders(where: {createdTimestamp_gt:%s, consumer_in:["%s"]}){

pdr_backend/subgraph/subgraph_feed_contracts.py

@@ -40,7 +40,7 @@ def query_feed_contracts(
     while True:
         query = """
         {
-            predictContracts(skip:%s, first:%s){
+            predictContracts(where: {paused: false}, skip:%s, first:%s){
                 id
                 token {
                     id

pdr_backend/subgraph/subgraph_payout.py

@@ -41,7 +41,8 @@ def get_payout_query(
                 {
                     timestamp_gte: %s,
                     timestamp_lte: %s,
-                    prediction_contains: "%s"
+                    prediction_contains: "%s",
+                    prediction_: {slot_: {predictContract_: {paused: false}}}
                 }
             """
             % (start_ts, end_ts, asset_id)

pdr_backend/subgraph/subgraph_pending_payouts.py

@@ -11,20 +11,28 @@
 
 
 def _fetch_subgraph_payouts(
-    subgraph_url: str, addr: str, slot_filter: str, chunk_size: int
+    subgraph_url: str,
+    addr: str,
+    slot_filter: str,
+    chunk_size: int,
+    include_paused: bool = False,
 ) -> List[Dict[str, Any]]:
     """
     slot_filter: string inside slot_{ ... } e.g.
         'status_in: ["Paying","Canceled"]'
         or 'status_in: ["Paying","Canceled","Pending"], slot_gte: %d, slot_lt: %d' % (a,b)
+    include_paused: if True, include paused contracts in the query
     """
     results = []
     offset = 0
     while True:
+        # Conditionally add the paused filter
+        paused_filter = "" if include_paused else ", predictContract_: {paused: false}"
+
         query = """
         {
             predictPredictions(
-            where: { user: "%s", payout: null, slot_: { %s } },
+            where: { user: "%s", payout: null, slot_: { %s%s } },
             first: %d,
             skip: %d
             ) {
@@ -40,6 +48,7 @@ def _fetch_subgraph_payouts(
         """ % (
             addr,
             slot_filter,
+            paused_filter,
             chunk_size,
             offset,
         )
@@ -69,8 +78,18 @@ def _fetch_subgraph_payouts(
 
 @enforce_types
 def query_pending_payouts(
-    subgraph_url: str, addr: str, query_old_slots=False
+    subgraph_url: str, addr: str, query_old_slots=False, include_paused=False
 ) -> Dict[str, List[UnixTimeS]]:
+    """
+    Fetch pending payouts for a given address.
+        Parameters:
+    subgraph_url (str): The URL of the subgraph to query.
+    addr (str): The address to fetch pending payouts for.
+    query_old_slots (bool): Whether to query old slots (older than 3 days).
+    include_paused (bool): Whether to include paused contracts in the query.
+        Returns:
+    Dict[str, List[UnixTimeS]]: contract addresses to lists of pending slot timestamps.
+    """
     chunk_size = 1000
     pending_slots: Dict[str, List[UnixTimeS]] = {}
     addr = addr.lower()
@@ -92,6 +111,7 @@ def query_pending_payouts(
         addr=addr,
         slot_filter='status_in: ["Paying", "Canceled"]',
         chunk_size=chunk_size,
+        include_paused=include_paused,
     )
 
     query2_results = []
@@ -101,6 +121,7 @@ def query_pending_payouts(
             addr=addr,
             slot_filter='status_in: ["Pending"], slot_lt: %d' % (ts_end),
             chunk_size=chunk_size,
+            include_paused=include_paused,
         )
 
     merged = query1_results + query2_results

pdr_backend/subgraph/subgraph_pending_slots.py

@@ -33,7 +33,7 @@ def get_pending_slots(
     while True:
         query = """
         {
-            predictSlots(where: {slot_gt: %s, slot_lte: %s, status: "Pending"}, skip:%s, first:%s){
+            predictSlots(where: {slot_gt: %s, slot_lte: %s, status: "Pending", predictContract_: {paused: false}}, skip:%s, first:%s){
                 id
                 slot
                 status

pdr_backend/subgraph/subgraph_predictions.py

@@ -71,7 +71,7 @@ def fetch_filtered_predictions(
     filters = [f.lower() for f in addresses]
 
     # pylint: disable=line-too-long
-    where_clause = f", where: {{timestamp_gt: {start_ts}, timestamp_lt: {end_ts}, slot_: {{predictContract_in: {json.dumps(filters)}}}}}"
+    where_clause = f", where: {{timestamp_gt: {start_ts}, timestamp_lt: {end_ts}, slot_: {{predictContract_in: {json.dumps(filters)}, predictContract_: {{paused: false}}}}}}"
 
     query = f"""
         {{
@@ -242,6 +242,7 @@ def fetch_contract_id_and_spe(
         {
             predictContracts(where: {
                 id_in: %s
+                paused: false
             }){
                 id
                 secondsPerEpoch

pdr_backend/subgraph/subgraph_slot.py

@@ -41,6 +41,7 @@ def get_predict_slots_query(
                 slot_lte: %s
                 slot_gte: %s
                 predictContract_in: %s
+                predictContract_: {paused: false}
             },
             orderBy: slot,
             orderDirection: asc

pdr_backend/subgraph/subgraph_subscriptions.py

@@ -51,9 +51,9 @@ def fetch_filtered_subscriptions(
 
     # 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}}}"
+        where_clause = f", where: {{predictContract_: {{id_in: {json.dumps(contracts)}, paused: false}}, timestamp_gt: {start_ts}, timestamp_lt: {end_ts}}}"
     else:
-        where_clause = f", where: {{timestamp_gt: {start_ts}, timestamp_lt: {end_ts}}}"
+        where_clause = f", where: {{timestamp_gt: {start_ts}, timestamp_lt: {end_ts}, predictContract_: {{paused: false}}}}"
 
     # pylint: disable=line-too-long
     query = f"""

pdr_backend/subgraph/subgraph_sync.py

@@ -13,7 +13,7 @@
 def block_number_is_synced(subgraph_url: str, block_number: int) -> bool:
     query = """
         {
-            predictContracts(block:{number:%s}){
+            predictContracts(block:{number:%s}, where: {paused: false}){
                 id
             }
         }

pdr_backend/subgraph/subgraph_trueval.py

@@ -40,7 +40,7 @@ def get_truevals_query(
             predictTrueVals (
                 first: %s
                 skip: %s
-                where: { timestamp_gte: %s, timestamp_lte: %s, slot_: {predictContract_in: %s}},
+                where: { timestamp_gte: %s, timestamp_lte: %s, slot_: {predictContract_in: %s, predictContract_: {paused: false}}},
                 orderBy: timestamp,
                 orderDirection: asc
             ) {

system_tests/test_ocean_payout.py

@@ -70,7 +70,7 @@ def checksum_mock(_, y):
 
         # Additional assertions
         mock_query_pending_payouts.assert_called_with(
-            mock_web3_pp.subgraph_url, "0x1", True
+            mock_web3_pp.subgraph_url, "0x1", True, False
         )
         print(mock_contract.get_payout.call_args_list)
         mock_contract.get_payout.assert_any_call([1, 2, 3], ["0x1"])