Refactor integrations to improve type hints and error handling

- Added type hints for better clarity in bulbaswap_integration.py and evaa_integration.py.
- Enhanced error handling in evaa_integration.py and affluent_integration.py to return empty dictionaries on exceptions.
- Updated variable types for consistency and clarity in bulbaswap_integration.py.

Author: ElliotWood13

constants/uniswap_v4.py

@@ -1,10 +1,12 @@
 import json
+from typing import cast
 from web3 import Web3
 from web3.contract import Contract
+from web3.types import HexStr
 from utils.web3_utils import w3
 
 UNISWAP_V4_USDE_POOL = Web3.to_bytes(
-    hexstr="0x63bb22f47c7ede6578a25c873e77eb782ec8e4c19778e36ce64d37877b5bd1e7"
+    hexstr=cast(HexStr, "0x63bb22f47c7ede6578a25c873e77eb782ec8e4c19778e36ce64d37877b5bd1e7")
 )
 
 UNISWAP_V4_STATE_VIEW = Web3.to_checksum_address(

integrations/affluent_integration.py

@@ -46,6 +46,7 @@ def get_l2_block_balances(
             err_msg = f"Error fetching Affluent balances at block {target_block}: {e}"
             print(err_msg)
             slack_message(err_msg)
+            return {}
 
     def get_token_symbol(self):
          return self.integration_id.get_token()

integrations/bulbaswap_integration.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional, Set
+from typing import Callable, Dict, List, Optional, Set
 import requests
 from eth_typing import ChecksumAddress
 from web3 import Web3
@@ -22,7 +22,7 @@ def __init__(
         excluded_addresses: Optional[Set[ChecksumAddress]] = None,
         end_block: Optional[int] = None,
         ethereal_multiplier: int = 0,
-        ethereal_multiplier_func: Optional[callable] = None,
+        ethereal_multiplier_func: Optional[Callable[..., int]] = None,
     ):
         super().__init__(
             integration_id,
@@ -64,7 +64,7 @@ def get_block_balances(
                 result[block] = cached_data[block]
                 continue
 
-            block_data = {}
+                block_data: Dict[ChecksumAddress, float] = {}
 
             try:
                 # Fetch data for both token addresses
@@ -79,8 +79,8 @@ def get_block_balances(
                                 "blockNumber": block,
                                 "page": page,
                                 "limit": self.page_size
-                            }
-                        )
+                            },
+                        )  # type: ignore[arg-type]
                         data = response.json()
 
                         if data["code"] == 200 and data["data"]["status"] == 0:
@@ -91,7 +91,7 @@ def get_block_balances(
                                 user_address = Web3.to_checksum_address(item["userAddress"])
 
                                 # Sum up liquidity for all pools
-                                total_liquidity = 0
+                                total_liquidity: float = 0.0
                                 for pool_data in item["userPositions"].values():
                                     if float(pool_data["liquidityUSD"]) > 0:
                                         total_liquidity += float(pool_data["liquidityUSD"])

integrations/cork_susde.py

@@ -220,7 +220,7 @@ def update_pair_config(
         # For each pair, update term config...
         for pair_id, pair_config in pair_config_by_id.items():
             start_block = max(from_block, pair_config.start_block)
-            if len(new_lpt_events) > 0:
+            if len(list(new_lpt_events)) > 0:
                 # print(f"Found {len(new_lpt_events)} new LPT events")
                 # Update the LP token address for each term
                 for term_id, term_config in pair_config.terms.items():
@@ -684,7 +684,7 @@ def get_block_balances(
                 for pair_id in pair_ids
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=vault_calls,
                 block_identifier=block,
@@ -724,7 +724,7 @@ def get_block_balances(
                 for amm_pool in amm_pools
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=amm_calls,
                 block_identifier=block,
@@ -761,7 +761,7 @@ def get_block_balances(
                 for term_id in pair_config.terms
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=psm_calls,
                 block_identifier=block,

integrations/cork_usde.py

@@ -220,7 +220,7 @@ def update_pair_config(
         # For each pair, update term config...
         for pair_id, pair_config in pair_config_by_id.items():
             start_block = max(from_block, pair_config.start_block)
-            if len(new_lpt_events) > 0:
+            if len(list(new_lpt_events)) > 0:
                 # print(f"Found {len(new_lpt_events)} new LPT events")
                 # Update the LP token address for each term
                 for term_id, term_config in pair_config.terms.items():
@@ -684,7 +684,7 @@ def get_block_balances(
                 for pair_id in pair_ids
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=vault_calls,
                 block_identifier=block,
@@ -724,7 +724,7 @@ def get_block_balances(
                 for amm_pool in amm_pools
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=amm_calls,
                 block_identifier=block,
@@ -761,7 +761,7 @@ def get_block_balances(
                 for term_id in pair_config.terms
             ]
             multicall_results = multicall_by_address(
-                w3=self.w3,
+                wb3=self.w3,
                 multical_address=MULTICALL_ADDRESS_BY_CHAIN[self.chain],
                 calls=psm_calls,
                 block_identifier=block,

integrations/echelon_integration.py

@@ -3,7 +3,7 @@
 import json
 import requests
 
-from typing import Dict, List, Set
+from typing import Dict, List, Optional, Set
 from dotenv import load_dotenv
 from constants.summary_columns import SummaryColumn
 from constants.example_integrations import (
@@ -95,7 +95,8 @@ def get_participants_data(self, block, user_addresses=[]):
             print(f"Unexpected error: {e}")
             raise
 
-    def get_participants(self, block: int) -> List[str]:
+    def get_participants(self, blocks: Optional[List[int]] = None) -> Set[str]:
+        block = blocks[0] if blocks else 0
         try:
             response = requests.get(
                 f"{ETHENA_ADDRESS_API_URL}?block={block}",
@@ -106,16 +107,16 @@ def get_participants(self, block: int) -> List[str]:
             data = response.json()['data']
             if not isinstance(data, list):
                 logging.warning(f"Unexpected response format from API: {data}")
-                return []
+                return set()
 
-            return [addr for addr in data if isinstance(addr, str)]
+            return set(addr for addr in data if isinstance(addr, str))
 
         except requests.RequestException as e:
             logging.error(f"Request failed for block {block}: {str(e)}")
-            return []
+            return set()
         except Exception as e:
             logging.error(f"Error processing participants for block {block}: {str(e)}")
-            return []
+            return set()
 
 if __name__ == "__main__":
     example_integration = EchelonAptosIntegration(

integrations/echelon_xlpt_integration.py

@@ -3,7 +3,7 @@
 import json
 import requests
 
-from typing import Dict, List, Set
+from typing import Dict, List, Optional, Set
 from dotenv import load_dotenv
 from constants.summary_columns import SummaryColumn
 from constants.example_integrations import (
@@ -97,7 +97,8 @@ def get_participants_data(self, block, user_addresses=[]):
             print(f"Unexpected error: {e}")
             raise
 
-    def get_participants(self, block: int) -> List[str]:
+    def get_participants(self, blocks: Optional[List[int]] = None) -> Set[str]:
+        block = blocks[0] if blocks else 0
         try:
             response = requests.get(
                 f"{ETHENA_SUSDE_USDC_XLPT_ADDRESS_API_URL}?block={block}",
@@ -108,16 +109,16 @@ def get_participants(self, block: int) -> List[str]:
             data = response.json()['data']
             if not isinstance(data, list):
                 logging.warning(f"Unexpected response format from API: {data}")
-                return []
+                return set()
 
-            return [addr for addr in data if isinstance(addr, str)]
+            return set(addr for addr in data if isinstance(addr, str))
 
         except requests.RequestException as e:
             logging.error(f"Request failed for block {block}: {str(e)}")
-            return []
+            return set()
         except Exception as e:
             logging.error(f"Error processing participants for block {block}: {str(e)}")
-            return []
+            return set()
 
 if __name__ == "__main__":
     example_integration = EchelonAptosIntegration(

integrations/evaa_integration.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 from constants.evaa import EVAA_POOLS_MAP, EVAA_ENDPOINT, EVAA_USDE_START_BLOCK, EVAA_SUSDE_START_BLOCK
 
@@ -42,28 +42,29 @@ def get_l2_block_balances(
             for block_number in blocks:
                 block_balances[block_number] = {}
                 block_data = self.get_participants_data(block_number)
-                for participant in block_data:
+                participants: List[Dict[str, Any]] = block_data if isinstance(block_data, list) else []
+                for participant in participants:
                     ton_address = participant["ton_address"]
-                    balance = participant["balance"]
+                    balance = float(participant["balance"])
                     block_balances[block_number][ton_address] = balance
 
             return block_balances
         except Exception as e:
-            err_msg = f"Error fetching EVAA balances at block {block_number}: {e}"
+            err_msg = f"Error fetching EVAA balances: {e}"
             print(err_msg)
             slack_message(err_msg)
+            return {}
 
     def get_token_symbol(self):
          return self.integration_id.get_token()
 
-    def get_participants_data(self, block: int) -> Dict[str, float]:
+    def get_participants_data(self, block: int) -> List[Dict[str, Any]]:
         """
-        Returns a list of "ton_address": "balance"
+        Returns a list of dicts with "ton_address" and "balance".
         """
-
         token = self.get_token_symbol()
         pools_list = EVAA_POOLS_MAP[token]
-        block_data: Dict[str, float] = {}
+        block_data: List[Dict[str, Any]] = []
         target_date = get_block_date(block, self.chain, adjustment=3600, fmt="%Y-%m-%dT%H:%M:%S")
 
 
@@ -84,7 +85,10 @@ def get_participants_data(self, block: int) -> Dict[str, float]:
                 if payload is None:
                     raise Exception(f"Error getting participants data for EVAA Protocol token {token} at block {block}: empty response")
 
-                block_data = payload
+                if isinstance(payload, list):
+                    block_data.extend(payload)
+                else:
+                    block_data.append(payload)
 
         except Exception as e:
                 err_msg = f"Error getting participants data for EVAA Protocol at block {block}: {e}"

integrations/merchantmoe_lbt_integration.py

@@ -84,7 +84,7 @@ def get_participants(self, blocks: list[int] | None) -> set[str]:
             )
 
             logging.info(
-                f"[{self.name}] Scanning blocks {start_block} to {to_block}, received {len(transfers)} mETH/USDe transfer events"
+                f"[{self.name}] Scanning blocks {start_block} to {to_block}, received {len(list(transfers))} mETH/USDe transfer events"
             )
             for transfer in transfers:
                 from_address = transfer["args"]["from"]

integrations/nuri.py

@@ -128,7 +128,7 @@ def get_participants(self, blocks: list[int] | None) -> set[str]:
                     start_block,
                     to_block,
                 )
-                print(f"Fetched {len(mint_events)} Mint events")
+                print(f"Fetched {len(list(mint_events))} Mint events")
 
                 for event in mint_events:
                     tx_hash = event["transactionHash"]