Skip to content

Instantly share code, notes, and snippets.

@jnmclarty
Created August 3, 2023 13:43
Show Gist options
  • Save jnmclarty/e10abfc04fd5c66964f21ad9a5ff7b28 to your computer and use it in GitHub Desktop.
Save jnmclarty/e10abfc04fd5c66964f21ad9a5ff7b28 to your computer and use it in GitHub Desktop.
Pull UniV3 Topics
# -*- coding: utf-8 -*-
import json
import datetime as dt
import requests as r
MINT_TOPIC = '0x7a53080ba414158be7ec69b987b5fb7d07dee101fe85488f0853ae16239d0bde'
BURN_TOPIC = '0x0c396cd989a39f4459b5fa1aed6a9a8dcdbc45908acfd67e028cd568da98982c'
# start = 12375
# end = 17830
# size = 1000
start = 1237
end = 1783
size = 10000
import requests as r
def get_logs(topic, from_block, to_block):
url = "https://mainnet.infura.io/v3/..."
headers = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"method": "eth_getLogs",
"params": [{"fromBlock" : hex(from_block),
"toBlock" : hex(to_block),
"topics":[topic]}],
"id": 1
}
response = r.post(url, headers=headers, data=json.dumps(data)).json()
if 'error' in response:
error = response['error']
if error['code'] == -32005: # More than 10K, passing infura's limits.
mid_point = int((to_block - from_block) * 0.5)
mid_point_offset = mid_point + 1
return get_logs(topic, from_block, from_block + mid_point) + get_logs(topic, from_block + mid_point + 1, to_block)
else:
print(response)
try:
return response['result']
except:
print(response)
raise
start_t = dt.datetime.now()
cnt = 0
DO_TOPIC = MINT_TOPIC
for blk in range(start, end):
print(f"Processing {blk * size}")
from_blk = blk * size
to_blk = (blk * size) + size - 1
logs = get_logs(DO_TOPIC, from_blk, to_blk)
with open(f"D:\\data\\univ3\\{DO_TOPIC}\\{from_blk}.json", 'w') as f:
f.write(json.dumps(logs))
cnt = cnt + 1
tot_time = (dt.datetime.now() - start_t)
avg_time = tot_time / cnt
print(f"Avg time per {size} blocks: {avg_time}, total time: {tot_time}, percent complete: {cnt / (end - start)}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment