import json import requests from web3 import Web3 from datetime import datetime, timedelta from typing import List, Dict, Optional import pandas as pd import time import requests from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry import time ADDRESSES = { "optimism": { "balancer_vault": "0xBA12222222228d8Ba445958a75a0704d566BF2C8", "uniswap_position_manager": "0xC36442b4a4522E871399CD717aBDD847Ab11FE88" }, "base": { "balancer_vault": "0xBA12222222228d8Ba445958a75a0704d566BF2C8", "uniswap_position_manager": "0x03a520b32C04BF3bEEf7BEb72E919cf822Ed34f1" } } # Defining RPC URLs and initializing Web3 instances OPTIMISM_RPC_URL = 'https://opt-mainnet.g.alchemy.com/v2/U5gnXPYxeyH43MJ9tP8ONBQHEDRav7H0' BASE_RPC_URL = 'https://base-mainnet.g.alchemy.com/v2/U5gnXPYxeyH43MJ9tP8ONBQHEDRav7H0' print("Initializing Web3 instances...") web3_optimism = Web3(Web3.HTTPProvider(OPTIMISM_RPC_URL)) web3_base = Web3(Web3.HTTPProvider(BASE_RPC_URL)) # Contract addresses for service registries contract_address_optimism = '0x3d77596beb0f130a4415df3D2D8232B3d3D31e44' contract_address_base = '0x3C1fF68f5aa342D296d4DEe4Bb1cACCA912D95fE' # Load the ABI from a local JSON file with open('./contracts/service_registry_abi.json', 'r') as abi_file: contract_abi = json.load(abi_file) # Create the contract instances service_registry_optimism = web3_optimism.eth.contract(address=contract_address_optimism, abi=contract_abi) service_registry_base = web3_base.eth.contract(address=contract_address_base, abi=contract_abi) print("Service registry contracts loaded.") def load_abi(filename): with open(filename, 'r') as file: contract_json = json.load(file) return contract_json['abi'] UNISWAP_ABI = load_abi('./contracts/NonfungiblePositionManager.json') def get_logs(api_key, chain_name, from_block, to_block, contract_address, service_safe): """Fetch logs for the given contract and wallet address with specified topics.""" base_url = { 'optimism': "https://api-optimistic.etherscan.io/api", 'base': "https://api.basescan.org/api" }.get(chain_name) if not base_url: print(f"Invalid chain name: {chain_name}") return [] print('formatted safe address for topic', f"0x000000000000000000000000{service_safe[2:].lower()}") params = { 'module': 'logs', 'action': 'getLogs', 'address': contract_address, 'fromBlock': from_block, 'toBlock': to_block, 'apikey': api_key, 'topic2': f"0x000000000000000000000000{service_safe[2:].lower()}" # Properly formatted topic2 } response = requests.get(base_url, params=params) data = response.json() if data['status'] != '1': print(f"Error: {data['message']}") return [] return data['result'] def get_block_range_for_date(chain_id, date_str, api_key, base_url): """Get the block range for a specific date.""" target_date = datetime.strptime(date_str, "%Y-%m-%d") start_of_day = datetime.combine(target_date, datetime.min.time()) end_of_day = datetime.combine(target_date, datetime.max.time()) start_timestamp = int(start_of_day.timestamp()) end_timestamp = int(end_of_day.timestamp()) # Get start block start_response = requests.get( f"{base_url}?module=block&action=getblocknobytime×tamp={start_timestamp}&closest=before&apikey={api_key}" ) if start_response.status_code == 200: start_data = start_response.json() start_block = start_data.get('result') else: print(f"Error fetching start block for {date_str} on chain {chain_id}") return None, None if start_block is None: print(f"No start block found for chain {chain_id} on {date_str}") return None, None print(f"Start block for chain {chain_id} on {date_str}: {start_block}") # Get end block end_response = requests.get( f"{base_url}?module=block&action=getblocknobytime×tamp={end_timestamp}&closest=before&apikey={api_key}" ) if end_response.status_code == 200: end_data = end_response.json() end_block = end_data.get('result') else: print(f"Error fetching end block for {date_str} on chain {chain_id}") return None, None if end_block is None: print(f"No end block found for chain {chain_id} on {date_str}") return None, None print(f"End block for chain {chain_id} on {date_str}: {end_block}") return start_block, end_block def date_range(start_date, end_date): """Generates a range of dates from start_date to end_date inclusive.""" start_dt = datetime.strptime(start_date, "%Y-%m-%d") end_dt = datetime.strptime(end_date, "%Y-%m-%d") delta = timedelta(days=1) current_dt = start_dt while current_dt <= end_dt: yield current_dt.strftime("%Y-%m-%d") current_dt += delta def parse_transfer_log(chain_name, single_date, log): # ERC-721 Transfer event signature TRANSFER_EVENT_SIGNATURE = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" if log['topics'][0].lower() == TRANSFER_EVENT_SIGNATURE.lower(): # This is a Transfer event return { "chain_name": chain_name, "date": single_date, "event": "Transfer", "topic_0": log['topics'][0], "from": "0x" + log['topics'][1][-40:], # Extract the address from the padded topic "to": "0x" + log['topics'][2][-40:], # Extract the address from the padded topic "token_id": int(log['topics'][3], 16) # Convert hex to decimal } # If no Transfer event is found return None def parse_pool_balance_log(chain_name, single_date, log): POOL_BALANCE_SIGNATURE = "0xe5ce249087ce04f05a957192435400fd97868dba0e6a4b4c049abf8af80dae78" if log['topics'][0].lower() == POOL_BALANCE_SIGNATURE.lower(): # Event PoolBalanceChanged: (bytes32 poolId, address liquidityProvider, address[] tokens, int256[] deltas, uint256[] protocolFeeAmounts) return { "chain_name": chain_name, "date": single_date, "event": "PoolBalanceChanged", "poolId": log['topics'][1], "liquidityProvider": "0x" + log['topics'][2][-40:], # Extract the address from the padded topic "tokens": [log['data'][i:i + 64] for i in range(0, len(log['data']), 64)], "deltas": [int(log['data'][i:i + 64], 16) for i in range(0, len(log['data']), 64)], "protocolFeeAmounts": [int(log['data'][i:i + 64], 16) for i in range(0, len(log['data']), 64)] } def fetch_service_safes(web3, registry_contract): print("\nFetching service safes...") total_services = registry_contract.functions.totalSupply().call() print(f"Total services: {total_services}") service_safes = set() for service_id in range(1, total_services + 1): service = registry_contract.functions.getService(service_id).call() agent_ids = service[-1] # Assuming the last element is the list of agent IDs if 25 in agent_ids: service_safe = service[1] service_safes.add(service_safe) print(f"Total service safes found: {len(service_safes)}") return service_safes def get_uniswap_v3_position(web3, contract_address, token_id): """Fetch the Uniswap V3 position details including `token0`, `token1`.""" position_manager_contract = web3.eth.contract(address=contract_address, abi=UNISWAP_ABI) position_data = position_manager_contract.functions.positions(token_id).call() return { 'token0': position_data[2], 'token1': position_data[3] } def get_uniswap_increase_liquidity(api_key, chain_name, token_id): """Fetch the IncreaseLiquidity event details including `amount0`, `amount1`.""" base_url = { 'optimism': "https://api-optimistic.etherscan.io/api", 'base': "https://api.basescan.org/api" }.get(chain_name) if not base_url: print(f"Invalid chain name: {chain_name}") return {} params = { 'module': 'logs', 'action': 'getLogs', 'address': ADDRESSES[chain_name]['uniswap_position_manager'], 'topic0': "0x3067048beee31b25b2f1681f88dac838c8bba36af25bfb2b7cf7473a5847e35f", 'topic1': f"0x{token_id:064x}", 'apikey': api_key } response = requests.get(base_url, params=params) data = response.json() if data['status'] != '1': print(f"Error: {data['message']}") return {} log = data['result'][0] if data['result'] else None if not log: return {} # Extracting amounts from the data hex-string data_hex = log['data'][2:] # Remove '0x' prefix liquidity = int(data_hex[0:64], 16) amount0 = int(data_hex[64:128], 16) amount1 = int(data_hex[128:192], 16) return { 'liquidity': liquidity, 'amount0': amount0, 'amount1': amount1 } def get_token_decimals(web3, token_address): """Fetch the number of decimals for a given ERC-20 token.""" token_contract = web3.eth.contract(address=token_address, abi=[ {"constant":True,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"payable":False,"stateMutability":"view","type":"function"} ]) return token_contract.functions.decimals().call() def requests_retry_session( retries=3, backoff_factor=0.3, status_forcelist=(500, 502, 504), session=None, ): session = session or requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=status_forcelist, ) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) return session def get_token_price_usd(chain, token_address): chain_dict = {"optimism": "optimistic-ethereum", "base": "base", "ethereum": "ethereum"} chain_name = chain_dict.get(chain, chain) time.sleep(4) url = f"https://api.coingecko.com/api/v3/simple/token_price/{chain_name}?contract_addresses={token_address}&vs_currencies=usd" headers = { "accept": "application/json", "x-cg-api-key": "CG-mf5xZnGELpSXeSqmHDLY2nNU" } try: response = requests_retry_session().get(url, headers=headers, timeout=10) response.raise_for_status() data = response.json() # Extract USD price key = token_address.lower() return data.get(key, {}).get('usd', None) except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") return None def load_existing_transactions(file_path): """Load existing transactions from a CSV file.""" try: return pd.read_csv(file_path) except FileNotFoundError: return pd.DataFrame() def get_last_processed_date(df): """Get the last processed date from the DataFrame.""" if df.empty: return None return df['date'].max() # Main function to integrate the above based on provided context def fetch_daily_value_locked(): api_keys = { 'optimism': 'XQ72JA5XZ51QC7TG1W295AAIF4KTV92K1K', 'base': '4BFQMVW1QUKEPVDA4VW711CF4462682CY8' } base_urls = { 10: "https://api-optimistic.etherscan.io/api", 8453: "https://api.basescan.org/api" } # Load existing transactions if any existing_transactions_path = 'daily_value_locked.csv' df_existing_transactions = load_existing_transactions(existing_transactions_path) last_processed_date = get_last_processed_date(df_existing_transactions) # Determine the start date based on the last processed date start_date = (pd.to_datetime(last_processed_date) + timedelta(days=1)).strftime('%Y-%m-%d') if last_processed_date else '2024-09-19' current_date = datetime.now().strftime('%Y-%m-%d') # Till present date chains = { 10: ('optimism', 'uniswap_position_manager', 'balancer_vault'), 8453: ('base', 'uniswap_position_manager', 'balancer_vault') } # Example service safe addresses - Replace these with actual addresses print("Fetching service safes for Optimism...") service_safes_optimism = fetch_service_safes(web3_optimism, service_registry_optimism) print(service_safes_optimism) print("Fetching service safes for Base...") service_safes_base = fetch_service_safes(web3_base, service_registry_base) print(service_safes_base) service_safes = { 'optimism': service_safes_optimism, 'base': service_safes_base } all_transactions = [] # List to hold all parsed logs for chain_id, (chain_name, uniswap_contract_key, balancer_contract_key) in chains.items(): base_url = base_urls[chain_id] api_key = api_keys[chain_name] for service_safe in service_safes[chain_name]: print(f"Checking service safe {service_safe} for chain {chain_name}") for single_date in date_range(start_date, current_date): start_block, end_block = get_block_range_for_date(chain_id, single_date, api_key, base_url) if start_block is None or end_block is None: print(f"Skipping date {single_date} for chain {chain_name} due to missing block data.") continue print(f"Start Block: {start_block}, End Block: {end_block} for date {single_date} on chain {chain_name} for service safe {service_safe}") # Get logs for Uniswap and Balancer contracts for contract_key, topic_key in [(uniswap_contract_key, "transfer"), (balancer_contract_key, "pool_balance_changed")]: contract_address = ADDRESSES[chain_name][contract_key] print(api_key, chain_name, start_block, end_block, contract_address, service_safe) logs = get_logs(api_key, chain_name, start_block, end_block, contract_address, service_safe) for log in logs: parsed_log = parse_pool_balance_log(chain_name, single_date, log) if topic_key == "pool_balance_changed" else parse_transfer_log(chain_name, single_date, log) if parsed_log: if topic_key == "transfer": # If the event is a Transfer event, fetch uniswap position details and increase liquidity event details uniswap_v3_data = get_uniswap_v3_position(web3_base if chain_name == 'base' else web3_optimism, contract_address, parsed_log['token_id']) increase_liquidity_data = get_uniswap_increase_liquidity(api_key, chain_name, parsed_log['token_id']) token0_address = uniswap_v3_data['token0'] token1_address = uniswap_v3_data['token1'] decimals_token0 = get_token_decimals(web3_base if chain_name == 'base' else web3_optimism, token0_address) decimals_token1 = get_token_decimals(web3_base if chain_name == 'base' else web3_optimism, token1_address) print(decimals_token0,decimals_token1) increase_liquidity_data['amount0'] /= 10**decimals_token0 increase_liquidity_data['amount1'] /= 10**decimals_token1 usd_price_token0 = get_token_price_usd(chain_name, token0_address) usd_price_token1 = get_token_price_usd(chain_name, token1_address) if usd_price_token0 is not None and usd_price_token1 is not None: increase_liquidity_data['amount0_usd'] = increase_liquidity_data['amount0'] * usd_price_token0 increase_liquidity_data['amount1_usd'] = increase_liquidity_data['amount1'] * usd_price_token1 parsed_log.update(uniswap_v3_data) parsed_log.update(increase_liquidity_data) all_transactions.append(parsed_log) # Convert to DataFrame and append to existing transactions df_new_transactions = pd.DataFrame(all_transactions) if not df_existing_transactions.empty: all_data = pd.concat([df_existing_transactions, df_new_transactions]) else: all_data = df_new_transactions all_data.to_csv(existing_transactions_path, index=False) print("Data saved to", existing_transactions_path) return all_data #print(df_transactions)