File size: 4,537 Bytes
c811726 16f8a8e c811726 16f8a8e c811726 16f8a8e c811726 16f8a8e c811726 16f8a8e c811726 acc99f2 c811726 cf2f2ff c811726 16f8a8e c811726 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 |
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)
|