Function collect_events

examples/top_usdt.py:11–62

()

load_dotenv()

async def collect_events():
    bearer_token = os.getenv("ENVIO_API_TOKEN")
    if not bearer_token:
        raise ValueError("ENVIO_API_TOKEN environment variable is required. Please set it in your .env file.")
    
    client = hypersync.HypersyncClient(ClientConfig(
        url="https://eth.hypersync.xyz/",
        bearer_token=bearer_token
    ))

    height = await client.get_height()

    query = hypersync.Query(
        # start from 10k blocks back
        from_block=height-int(1e4),
        # Select the logs we want
        logs=[LogSelection(
            address=["0xdAC17F958D2ee523a2206206994597C13D831ec7"],
            topics=[["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"]],
        )],
        # Select the fields and tables we want
        field_selection=FieldSelection(
            log=[
                LogField.TOPIC0,
                LogField.TOPIC1,
                LogField.TOPIC2,
                LogField.DATA,
                LogField.TRANSACTION_HASH,
            ],
            transaction=[
                TransactionField.HASH,
                TransactionField.GAS_USED,
            ]
        ),
    )

    config = hypersync.StreamConfig(
        hex_output=hypersync.HexOutput.PREFIXED,
        column_mapping=ColumnMapping(
            # map value columns to float so we can do calculations with them
            decoded_log={
                "value": DataType.FLOAT64,
            },
            transaction={
                TransactionField.GAS_USED: DataType.FLOAT64,
            },
        ),
        # give event signature so client can decode logs into decoded_logs.parquet file
        event_signature="Transfer(address indexed from, address indexed to, uint256 value)",
    )

    await client.collect_parquet("data", query, config)

def analyze_events():
    # read raw logs