get_buyers_range_time_bitquery_v1.py

import time
import requests
from web3 import Web3

infura_url = 'https://eth.llamarpc.com'
web3 = Web3(Web3.HTTPProvider(infura_url))

bitquery_api = '<bitquery_api>'

api_url = "https://graphql.bitquery.io/"

# token contract address to get early buyers from
token_address = '0x292fcDD1B104DE5A00250fEBbA9bC6A5092A0076'

time_since = '2024-04-02T18:15:00Z'
time_till = '2024-04-02T19:30:00Z'

graphql_query = """
{
  ethereum(network: ethereum) {
    dexTrades(
      options: { asc: ["block.height", "tradeIndex"], limit: <limit>, offset: <offset> }
      baseCurrency: { is: "<token_address>" }
      time: { since: "<time_since>" till: "<time_till>"}
      buyAmount: {gt: 0}
    ) {
      block {
        timestamp {
          time(format: "%Y-%m-%d %H:%M:%S")
        }
        height
      }
      side
      tradeIndex
      protocol
      exchange {
        fullName
        address {
          address
        }
      }
      smartContract {
        address {
          address
          annotation
        }
      }
      baseAmount
      base_amount_usd: baseAmount(in: USD)
      baseCurrency {
        address
        symbol
      }
      quoteAmount
      quote_amount_usd: quoteAmount(in: USD)
      quoteCurrency {
        address
        symbol
      }
      maker {
        address
      }
      taker {
        address
      }
      transaction {
        hash
        txFrom {
          address
        }
      }
    }
  }
}
"""

# Prepare the header with the API Key
headers = {
    'Content-Type': 'application/json',
    'X-API-KEY': bitquery_api
}

# Prepare the payload with the query and variables

traders = set()
count = 100
offset = 0
check_is_contract = False


def is_contract(address: str):
    return web3.eth.get_code(web3.to_checksum_address(address)).hex() != '0x'


while True:
    # Send the POST request
    query = (graphql_query
             .replace('<token_address>', token_address)
             .replace('<limit>', str(count))
             .replace('<offset>', str(offset))
             .replace('<time_since>', time_since)
             .replace('<time_till>', time_till))

    payload = {
        'query': query,
        'variables': {'address': token_address}
    }
    response = requests.post(api_url, json=payload, headers=headers)

    # Check if the request was successful
    if response.ok:
        # Parse the JSON response
        data = response.json()
        if 'data' not in data:
            raise ValueError('')
        data = data['data']
        if 'ethereum' not in data:
            raise ValueError('')
        data = data['ethereum']
        if 'dexTrades' not in data:
            raise ValueError('')

        trades = data['dexTrades']

        for trade in trades:
            side = trade['side']
            tx_hash = trade['transaction']['hash']
            taker_address = trade['taker']['address']
            maker_address = trade['maker']['address']
            from_address = trade['transaction']['txFrom']['address']
            ex_name = trade['exchange']['fullName']
            base_amount = trade['baseAmount']
            base_amount_usd = trade['base_amount_usd']
            quote_amount = trade['quoteAmount']
            usd_spent = trade['quote_amount_usd']

            if taker_address != from_address:
                # likely a bot interaction?
                # in my experience still from_address is the trader address
                pass

            if check_is_contract and is_contract(from_address):
                continue

            print(f'{side} {from_address} - {round(usd_spent, 2)}$ - {ex_name} - {tx_hash}')

            traders.add(from_address)

        if len(trades) >= count:
            offset += len(trades)
        else:
            # no more trades left
            break

        time.sleep(3)
    else:
        print("Error:", response.text)

print(traders)