| from string import Template | |
| from pathlib import Path | |
| import sys | |
| import json | |
| import os | |
| import time | |
| from tqdm import tqdm | |
| from typing import List, Any, Optional, Union | |
| import numpy as np | |
| import pandas as pd | |
| import gc | |
| import re | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| from json.decoder import JSONDecodeError | |
| CREATOR = "0x89c5cc945dd550BcFfb72Fe42BfF002429F46Fec" | |
| BATCH_SIZE = 1000 | |
| OMEN_SUBGRAPH_URL = Template( | |
| """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/9fUVQpFwzpdWS9bq5WkAnmKbNNcoBwatMR4yZq81pbbz""" | |
| ) | |
| SCRIPTS_DIR = Path(__file__).parent | |
| ROOT_DIR = SCRIPTS_DIR.parent | |
| DATA_DIR = ROOT_DIR / "live_data" | |
| MAX_UINT_HEX = "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" | |
| SUBGRAPH_API_KEY = os.environ.get("SUBGRAPH_API_KEY", None) | |
| def measure_execution_time(func): | |
| def wrapper(*args, **kwargs): | |
| start_time = time.time() | |
| result = func(*args, **kwargs) | |
| end_time = time.time() | |
| execution_time = end_time - start_time | |
| print(f"Execution time: {execution_time:.6f} seconds") | |
| return result | |
| return wrapper | |
| def _to_content(q: str) -> dict[str, Any]: | |
| """Convert the given query string to payload content, i.e., add it under a `queries` key and convert it to bytes.""" | |
| finalized_query = { | |
| "query": q, | |
| "variables": None, | |
| "extensions": {"headers": None}, | |
| } | |
| return finalized_query | |