app_value_locked.py

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
import os 

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 = os.getenv('OPTIMISM_RPC_URL')
BASE_RPC_URL = os.getenv('BASE_RPC_URL')
ETH_RPC_URL = os.getenv('ETH_RPC_URL')

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&timestamp={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&timestamp={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 40 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)