Implementation of uPUNK price feed

upunk_index_price.py

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)

    # Note: These two lines set the evaluation time and the time window.
    # If you would like to evaluate this at a timestamp other than the current one then
    # you would need to change the value of `eval_ts`. Similarly, if you'd like to use a
    # window different than 30 days, you'd need to change `window` to be the corresponding
    # number of seconds.
    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)