feat(safe): halve API calls and rotate keys

Author: spalen0

.env.example

@@ -36,8 +36,9 @@ AAVE_ALERT_THRESHOLD=0.95
 AAVE_CRITICAL_THRESHOLD=0.98
 AAVE_ENABLE_NOTIFICATIONS=true
 
-# Safe
+# Safe (two keys rotated to stay within API rate limits)
 SAFE_API_KEY=your-api-key
+SAFE_API_KEY_2=your-second-api-key
 
 # Tenderly (for transaction simulation)
 TENDERLY_API_KEY=your-tenderly-api-key

.github/workflows/_run-monitoring.yml

@@ -121,6 +121,7 @@ env:
   COINGECKO_API_KEY: ${{ secrets.COINGECKO_API_KEY }}
   TENDERLY_API_KEY: ${{ secrets.TENDERLY_API_KEY }}
   SAFE_API_KEY: ${{ secrets.SAFE_API_KEY }}
+  SAFE_API_KEY_2: ${{ secrets.SAFE_API_KEY_2 }}
   LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
   LLM_MODEL: ${{ secrets.LLM_MODEL }}
   LLM_PROVIDER: ${{ secrets.LLM_PROVIDER }}

safe/main.py

@@ -1,3 +1,4 @@
+import itertools
 import os
 import time
 
@@ -21,6 +22,12 @@
 provider_url_mainnet = os.getenv("PROVIDER_URL_MAINNET")
 provider_url_arb = os.getenv("PROVIDER_URL_ARBITRUM")
 
+# Round-robin iterator over available Safe API keys.
+_api_keys: list[str] = [k for k in [os.getenv("SAFE_API_KEY"), os.getenv("SAFE_API_KEY_2")] if k]
+if not _api_keys:
+    raise ValueError("At least one SAFE_API_KEY must be set.")
+_api_key_cycle = itertools.cycle(_api_keys)
+
 safe_address_network_prefix = {
     "mainnet": "eth",
     "arbitrum-main": "arb1",
@@ -188,9 +195,7 @@ def get_safe_transactions(
     if executed is not None:
         params["executed"] = str(executed).lower()
 
-    api_key = os.getenv("SAFE_API_KEY")
-    if not api_key:
-        raise ValueError("SAFE_API_KEY environment variable not set.")
+    api_key = next(_api_key_cycle)
 
     headers = {
         "Authorization": f"Bearer {api_key}",
@@ -230,36 +235,23 @@ def get_safe_transactions(
     return []
 
 
-def get_last_executed_nonce(safe_address: str, network_name: str) -> int:
-    executed_txs = get_safe_transactions(safe_address, network_name, executed=True, limit=1)
-    if executed_txs:
-        return int(executed_txs[0]["nonce"])
-    return -1  # Return -1 if no executed transactions found
-
-
-def get_pending_transactions_after_last_executed(safe_address: str, network_name: str) -> list[dict]:
-    last_executed_nonce = get_last_executed_nonce(safe_address, network_name)
+def get_pending_transactions(safe_address: str, network_name: str) -> list[dict]:
+    """Fetch pending transactions with nonce higher than the last cached nonce."""
+    last_cached_nonce = get_last_executed_nonce_from_file(safe_address)
     pending_txs = get_safe_transactions(safe_address, network_name, executed=False)
-
-    if pending_txs:
-        return [tx for tx in pending_txs if int(tx["nonce"]) > last_executed_nonce]
-    return []
+    return [tx for tx in pending_txs if int(tx["nonce"]) > last_cached_nonce]
 
 
 def get_safe_url(safe_address: str, network_name: str) -> str:
     return f"{SAFE_WEBSITE_URL}{safe_address_network_prefix[network_name]}:{safe_address}"
 
 
 def check_for_pending_transactions(safe_address: str, network_name: str, protocol: str) -> None:
-    pending_transactions = get_pending_transactions_after_last_executed(safe_address, network_name)
+    pending_transactions = get_pending_transactions(safe_address, network_name)
 
     if pending_transactions:
         for tx in pending_transactions:
             nonce = int(tx["nonce"])
-            # skip tx if the nonce is already processed
-            if nonce <= get_last_executed_nonce_from_file(safe_address):
-                logger.info("Skipping tx with nonce %s as it is already processed.", nonce)
-                continue
 
             target_contract = tx["to"]
 
@@ -355,7 +347,7 @@ def main():
         logger.info("Running for %s on %s", safe[0], safe[1])
         last_api_call_time, request_counter = check_api_limit(last_api_call_time, request_counter)
         run_for_network(safe[1], safe[2], safe[0])
-        request_counter += 2
+        request_counter += 1
 
 
 if __name__ == "__main__":