Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Implementation of uPUNK price feed
import datetime as dt
import math
import statistics
import web3
from umapy import connect_web3, create_cryptopunks_contract, find_cryptopunk_purchases
from umapy.util import find_oldest_block_after_ts
PUNK_FIRST_BLOCK = 3914495
def find_punkbought_prices(w3, punk, blockStart, blockEnd):
"""
Parameters
----------
w3 : web3.Web3
A web3 instance
punk : web3.Contract
The cryptopunk contract
eval_ts : int
The timestamp that you'd like to end the price computation on...
window : int
The number of seconds to include in the computation window. The
default corresponds to 30 days (30*24*60)
Return
------
punk_sales : list
A list of dictionaries that stores the relevant information from
each punkbought event. Each element will have keys of `punk_index`,
`price`, and `block_number`
"""
# Find all PunkBought events that occurred in the relevant time frame
punkbought_events = punk.events.PunkBought.getLogs(
fromBlock=blockStart, toBlock=blockEnd
)
# Create a list to store punk sales
punk_sales = []
# Iterate through each of the PunkBought events and determine
# whether it was a bid or not... If a bid, must find the bid
# events for that CryptoPunk
for punkbought in punkbought_events:
# Extract information from the event
_punkIndex = punkbought["args"]["punkIndex"]
_blockNumber = punkbought["blockNumber"]
info = {
"punkIndex": _punkIndex,
"blockNumber": _blockNumber
}
# Identify transaction
transaction = w3.eth.get_transaction(punkbought.transactionHash)
# Try decoding the input and determining whether the transaction
# happened via the `buyPunk` function or the `acceptBidForPunk`
# function
try:
# Function/arguments that generated the transaction
f, args = punk.decode_function_input(transaction["input"])
# If it was a directly bought punk, we can use the data from
# the event
if f.fn_name == punk.functions.buyPunk.fn_name:
info["price"] = punkbought["args"]["value"]
elif f.fn_name == punk.functions.acceptBidForPunk.fn_name:
# Get the bids that were entered from the "beginning of
# time" until the block that this transaction happened...
# We can ignore the fact that bids can be withdrawn because
# in order to accept a bid, a bid must be outstanding and
# only one bid can be outstanding at once
bids_entered = punk.events.PunkBidEntered.createFilter(
argument_filters={"punkIndex": _punkIndex},
fromBlock=PUNK_FIRST_BLOCK, toBlock=_blockNumber
).get_all_entries()
# Sort so that the most recent bid is the last one
# and then find the value submitted
bids_entered.sort(key=lambda x: x["blockNumber"])
info["price"] = bids_entered[-1]["args"]["value"]
else:
print(f"Transcation Hash: {punkbought.transactionHash}")
raise ValueError("I don't know how else a punk gets traded")
# If we can't figure tsuff out then just use value and cross our fingers
except:
print("Failed! The problematic transaction was:")
print(punkbought.transactionHash.hex())
info["price"] = punkbought["args"]["value"]
punk_sales.append(info)
return punk_sales
def compute_upunks_index(w3, punk, eval_ts, window=30*24*60*60):
"""
Parameters
----------
w3 : web3.Web3
A web3 instance
punk : web3.Contract
The cryptopunk contract
eval_ts : int
The timestamp that you'd like to end the price computation on...
window : int
The number of seconds to include in the computation window. The
default corresponds to 30 days (30*24*60)
Return
------
upunk_index : float
The uPUNK index value
"""
# Find the block start and end that we're interested in
blockStart = find_oldest_block_after_ts(w3, eval_ts - window)
blockEnd = find_oldest_block_after_ts(w3, eval_ts)
# Find all of the PunkBought events - This function accounts for the
# CryptoPunks bug that loses the value of PunkBought when processed
# via `acceptBidForPunk`
punk_sales = find_punkbought_prices(w3, punk, blockStart, blockEnd)
# Need to only keep track of unique sales
unique_sales = {}
for ps in punk_sales:
current_values = unique_sales.get(
ps["punkIndex"],
{"blockNumber": 0, "price": 0}
)
more_recent = ps["blockNumber"] > current_values["blockNumber"]
price_gt_0 = float(ps["price"]) > 0
if more_recent and price_gt_0:
unique_sales[ps["punkIndex"]] = {
"blockNumber": ps["blockNumber"],
"price": ps["price"]
}
# Unique executed trade prices
prices = [unique_sales[x]["price"] for x in unique_sales]
# Compute median
upunk_index = (statistics.median(prices) / w3.toWei(1, "ether")) / 1000
return upunk_index
if __name__ == "__main__":
# Create web3/cryptopunk instances
w3 = connect_web3()
punk = create_cryptopunks_contract(w3)
# Use current time to eval as default
eval_ts = int(dt.datetime.now().timestamp())
window = 30*24*60*60
upunk_index = compute_upunks_index(w3, punk, eval_ts, window)
print(upunk_index)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment