delegate_tracking.py

import csv
import datetime
import logging
import os
import sys

import requests
from dotenv import load_dotenv
from web3 import Web3

# Load environment variables
load_dotenv()

# Etherscan API setup
etherscan_api_key = os.getenv("ETHERSCAN_API_KEY")
etherscan_base_url = "https://api.etherscan.io/api"

# Check if API key is present
if not etherscan_api_key or etherscan_api_key == "YOUR_ETHERSCAN_API_KEY":
    logging.error("Error: No Etherscan API key found. Add your API key to the .env file.")
    logging.error("If you don't have an API key, you can obtain one from https://etherscan.io/apis")
    sys.exit(1)

# Function Definitions
def get_logs(api_key, base_url, contract_address, start_block, end_block, event_hash):
    params = {
        "module": "logs",
        "action": "getLogs",
        "address": contract_address,
        "fromBlock": start_block,
        "toBlock": end_block,
        "topic0": event_hash,
        "apikey": api_key,
    }
    try:
        response = requests.get(etherscan_base_url, params=params, timeout=60)
        return response.json()
    except requests.exceptions.HTTPError:
        logging.exception("HTTP error occurred.")
    except requests.exceptions.RequestException:
        logging.exception("Error during requests to %s", base_url)

def convert_hex_to_datetime(hex_value) -> datetime:
    timestamp = int(hex_value, 16)
    # Only take date from timestamp
    return datetime.datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")

def get_closest_delegation(delegation_data, query_datetime):
    closest_delegation = 0
    query_date = query_datetime.date()  # Convert datetime to date for comparison
    for delegate, dates in delegation_data.items():
        for date_str, amounts in dates.items():
            date = datetime.datetime.strptime(date_str, "%Y-%m-%d").date()
            if date <= query_date:
                closest_delegation += amounts["net"]
    return closest_delegation

def read_csv():
    file_path = os.path.join("delegate_data", "Aligned Delegates.csv")
    try:
        with open(file_path, mode="r", encoding="utf-8") as file:
            csv_reader = csv.DictReader(file)
            contracts_data = []
            for row in csv_reader:
                contracts_data.append({
                    "name": row["Delegate Name"],
                    "contract": row["Delegate Contract"],
                    "committee": row["Aligned Voter Committee"],
                    "start_date": row["Start Date"],
                    "end_date": row["End Date"],
                    "end_reason": row["End Reason"]
                })
            return contracts_data
    except FileNotFoundError:
        logging.exception("File not found.")

def generate_dates(query_input):
    dates = []
    today = datetime.datetime.now().date()  # Get today's date

    try:
        if "to" in query_input:
            # Date range
            start_str, end_str = query_input.split(" to ")
            start_date = datetime.datetime.strptime(start_str, "%Y-%m-%d")
            end_date = datetime.datetime.strptime(end_str, "%Y-%m-%d")

            # Check if end date is in the future
            if end_date.date() > today:
                print("The application cannot see into the future.")
                confirm = input("Would you like to limit the range to today's date? (yes/no): ").strip().lower()
                if confirm == "yes":
                    end_date = datetime.datetime.now()
                else:
                    print("Exiting.")
                    sys.exit()

            while start_date <= end_date:
                dates.append(start_date)
                start_date += datetime.timedelta(days=1)

        else:
            # Single date
            single_date = datetime.datetime.strptime(query_input, "%Y-%m-%d")

            # Check if the date is in the future
            if single_date.date() > today:
                print("The application cannot see into the future.")
                confirm = input("Would you like to query today's date instead? (yes/no): ").strip().lower()
                if confirm == "yes":
                    single_date = datetime.datetime.now()
                else:
                    print("Exiting.")
                    sys.exit()

            dates.append(single_date)

    except ValueError:
        logging.exception("Invalid date format. Please try again.")
        return []

    return dates

def calculate_rankings(delegate_data):
    sorted_delegates = sorted(delegate_data.items(),
                              key=lambda x: x[1]["total"],
                              reverse=True)
    for rank, (delegate, data) in enumerate(sorted_delegates, start=1):
        delegate_data[delegate]["rank"] = rank
    return delegate_data

# Read the CSV file
contracts_data = read_csv()

# Initialize a dictionary for each contract
contract_delegations = {contract["contract"]: {} for contract in contracts_data}

# Event hashes
lock_event_hash = "0x625fed9875dada8643f2418b838ae0bc78d9a148a18eee4ee1979ff0f3f5d427"
free_event_hash = "0xce6c5af8fd109993cb40da4d5dc9e4dd8e61bc2e48f1e3901472141e4f56f293"

# Hardcoding the first block of 2023-01-01
start_block = "16308190"
end_block = "latest"

# Process each contract from CSV data
for i, contract in enumerate(contracts_data):
    print(f"Processing contract {i + 1}")
    address = contract["contract"]
    # Initialize storage for each contract and delegate
    contract_delegations[address] = {}

    # Retrieve logs for 'Lock' and 'Free' events
    for event_hash in [lock_event_hash, free_event_hash]:
        logs = get_logs(etherscan_api_key, etherscan_base_url, address, start_block, end_block, event_hash)
        for log in logs["result"]:
            hex_data = log["data"]
            decimal_data = Web3.to_int(hexstr=log["data"])
            ether_value = Web3.from_wei(decimal_data, "ether")
            delegate_address = Web3.to_checksum_address(log["topics"][1][-40:])
            timestamp = convert_hex_to_datetime(log["timeStamp"])

            # Initialize delegate storage
            if delegate_address not in contract_delegations[address]:
                contract_delegations[address][delegate_address] = {}

            # Initialize date storage
            if timestamp not in contract_delegations[address][delegate_address]:
                contract_delegations[address][delegate_address][timestamp] = {"net": 0}

            # Adjust net delegation based on event type
            if event_hash == lock_event_hash:
                contract_delegations[address][delegate_address][timestamp]["net"] += ether_value
            elif event_hash == free_event_hash:
                contract_delegations[address][delegate_address][timestamp]["net"] -= ether_value

# Initialize a dictionary to track delegate and committee delegations
delegate_committee_delegations = {}

# Loop to allow multiple date queries
while True:
    query_input = input("\nEnter the date to query (YYYY-MM-DD), or a range (YYYY-MM-DD to YYYY-MM-DD):")
    query_dates = generate_dates(query_input)

    results = {}  # Initialize results list

    for query_date in query_dates:
        query_date_str = query_date.strftime("%Y-%m-%d")  # Define query_date_str

        delegate_committee_delegations = {}  # Reset the dictionary for each date

        # Process each contract and update delegate_committee_delegations
        for contract in contracts_data:
            delegate_name = contract["name"]
            committee_name = contract["committee"]
            address = contract["contract"]

            # Retrieve delegation data for this contract
            delegation_data = contract_delegations.get(address, {})
            delegation_at_query = get_closest_delegation(delegation_data, query_date)

            # Initialize delegate data if not already present
            if delegate_name not in delegate_committee_delegations:
                delegate_committee_delegations[delegate_name] = {"total": 0, "committees": {}}

            # Update total delegation and committee-specific delegation for the delegate
            delegate_committee_delegations[delegate_name]["total"] += delegation_at_query
            delegate_committee_delegations[delegate_name]["committees"][committee_name] = delegation_at_query

            # Calculate and include rankings
            delegate_committee_delegations = calculate_rankings(delegate_committee_delegations)

            # Display results and store results for each date
            print(f"\nDelegations on {query_date_str}:")
            results[query_date_str] = []
            for delegate, data in delegate_committee_delegations.items():
                print(f"\nDelegate {delegate}:")
                print(f"Total Delegation: {data['total']} MKR")
                print(f"Ranking: {data['rank']}")
                for committee, delegation in data["committees"].items():
                    print(f"   - {committee}: {delegation} MKR")
                results[query_date_str].append({
                    "Delegate": delegate,
                    "Total Delegation": data["total"],
                    "Rank": data["rank"],
                    "Date": query_date_str,
                })

    export_to_csv = input("\nDo you want to export the results to a CSV file? (yes/no): ").strip().lower()
    if export_to_csv == "yes":
        csv_filename = f"delegation_data_{query_date_str}.csv"
        with open(csv_filename, "w", newline="") as csvfile:
            fieldnames = ["Delegate", "Total Delegation", "Rank", "Date"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            for query_date_str, date_results in results.items():
                for row in date_results:
                    writer.writerow(row)
        print(f"Data exported to {csv_filename}")

    continue_query = input("\nDo you want to query another date? (yes/no): ").strip().lower()
    if continue_query != "yes":
        break