from dataclasses import dataclass from string import Template from typing import Any from datetime import datetime, timedelta, UTC from utils import SUBGRAPH_API_KEY import requests OLD_MECH_SUBGRAPH_URL = ( "https://api.thegraph.com/subgraphs/name/stakewise/ethereum-gnosis" ) # MECH_SUBGRAPH_URL = "https://api.studio.thegraph.com/query/57238/mech/0.0.2" NETWORK_SUBGRAPH_URL = Template( """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/FxV6YUix58SpYmLBwc9gEHkwjfkqwe1X5FJQjn8nKPyA""" ) SUBGRAPH_HEADERS = { "Accept": "application/json, multipart/mixed", "Content-Type": "application/json", } QUERY_BATCH_SIZE = 1000 DATETIME_60_DAYS_AGO = datetime.now(UTC) - timedelta(days=60) BLOCK_NUMBER = Template( """ { blocks( first: 1, orderBy: timestamp, orderDirection: asc, where: { timestamp_gte: "${timestamp_from}", timestamp_lte: "${timestamp_to}" } ){ id, number, } } """ ) def fetch_block_number(timestamp_from: int, timestamp_to: int) -> dict: """Get a block number by its timestamp margins.""" query = BLOCK_NUMBER.substitute( timestamp_from=timestamp_from, timestamp_to=timestamp_to ) # print(f"Sending query for the subgraph = {query}") network_subgraph_url = NETWORK_SUBGRAPH_URL.substitute( subgraph_api_key=SUBGRAPH_API_KEY ) response = requests.post( network_subgraph_url, headers=SUBGRAPH_HEADERS, json={"query": query}, timeout=300, ) result_json = response.json() print(f"Response of the query={result_json}") blocks = result_json.get("data", {}).get("blocks", "") if len(blocks) == 0: raise ValueError(f"The query {query} did not return any results") return blocks[0] def get_mech_info_2024() -> dict[str, Any]: """Query the subgraph to get the 2024 information from mech.""" date = "2024-01-01" datetime_jan_2024 = datetime.strptime(date, "%Y-%m-%d") timestamp_jan_2024 = int(datetime_jan_2024.timestamp()) margin = timedelta(seconds=5) timestamp_jan_2024_plus_margin = int((datetime_jan_2024 + margin).timestamp()) jan_block_number = fetch_block_number( timestamp_jan_2024, timestamp_jan_2024_plus_margin ) # expecting only one block jan_block_number = jan_block_number.get("number", "") if jan_block_number.isdigit(): jan_block_number = int(jan_block_number) if jan_block_number == "": raise ValueError( "Could not find a valid block number for the first of January 2024" ) MECH_TO_INFO = { # this block number is when the creator had its first tx ever, and after this mech's creation "0xff82123dfb52ab75c417195c5fdb87630145ae81": ( "old_mech_abi.json", jan_block_number, ), # this block number is when this mech was created "0x77af31de935740567cf4ff1986d04b2c964a786a": ( "new_mech_abi.json", jan_block_number, ), } return MECH_TO_INFO def get_mech_info_last_60_days() -> dict[str, Any]: """Query the subgraph to get the last 60 days of information from mech.""" timestamp_60_days_ago = int((DATETIME_60_DAYS_AGO).timestamp()) margin = timedelta(seconds=5) timestamp_60_days_ago_plus_margin = int((DATETIME_60_DAYS_AGO + margin).timestamp()) last_month_block_number = fetch_block_number( timestamp_60_days_ago, timestamp_60_days_ago_plus_margin ) # expecting only one block last_month_block_number = last_month_block_number.get("number", "") if last_month_block_number.isdigit(): last_month_block_number = int(last_month_block_number) if last_month_block_number == "": raise ValueError("Could not find a valid block number for last month data") MECH_TO_INFO = { # this block number is when the creator had its first tx ever, and after this mech's creation "0xff82123dfb52ab75c417195c5fdb87630145ae81": ( "old_mech_abi.json", last_month_block_number, ), # this block number is when this mech was created "0x77af31de935740567cf4ff1986d04b2c964a786a": ( "new_mech_abi.json", last_month_block_number, ), } return MECH_TO_INFO if __name__ == "__main__": result = get_mech_info_last_60_days() print(result)