PriceOracle

Documentation for eth_defi.price_oracle.oracle.PriceOracle class.

class PriceOracle[source]

Price oracle core.

  • Suitable for real-time price calculation for data coming over WebSockets

  • Suitable for point of time calculation using historical data

  • Sample data over multiple events

  • Rotate ring buffer of events when new data comes in. Uses Python heapq for this.

Example:

# Randomly chosen block range.
# 100 blocks * 3 sec / block = ~300 seconds
start_block = 14_000_000
end_block = 14_000_100

pair_details = fetch_pair_details(web3, bnb_busd_address)
assert pair_details.token0.symbol == "WBNB"
assert pair_details.token1.symbol == "BUSD"

oracle = PriceOracle(
    time_weighted_average_price,
    max_age=PriceOracle.ANY_AGE,  # We are dealing with historical data
    min_duration=datetime.timedelta(minutes=1),
)

update_price_oracle_with_sync_events_single_thread(
    oracle,
    web3,
    bnb_busd_address,
    start_block,
    end_block
)

assert oracle.calculate_price() == pytest.approx(Decimal('523.8243566658033237353702655'))

Attributes summary

ANY_AGE

An "infinite" place holder for max age

Methods summary

__init__(price_function[, ...])

Create a new price oracle.

add_price_entry(evt)

Add price entry to the ring buffer.

add_price_entry_reorg_safe(evt)

Add price entry to the ring buffer with support for fixing chain reorganisations.

calculate_price()

Calculate the price based on the data in the price data buffer.

check_data_quality([now_])

Raises one of PriceCalculationError subclasses if our data is not good enough to calculate the oracle price.

feed_simple_data(data[, source])

Feed sample data to the price oracle from a Python dict.

get_buffer_duration()

How long time is the time we have price events in the buffer for.

get_by_transaction_hash(tx_hash)

Get an event by transaction hash.

get_last_refreshed()

When the oracle data was refreshed last time.

get_newest()

Return the newest price entry.

get_oldest()

Return the oldest price entry.

truncate_buffer(current_timestamp)

Delete old data in the buffer that is no longer relevant for our price calculation.

update_last_refresh(block_number, timestamp)

Update the last seen block.

ANY_AGE = datetime.timedelta(days=36500)

An “infinite” place holder for max age

__init__(price_function, target_time_window=datetime.timedelta(seconds=300), min_duration=datetime.timedelta(seconds=3600), max_age=datetime.timedelta(seconds=14400), min_entries=8)[source]

Create a new price oracle.

The security parameters are set for a simple defaults.

Parameters
  • price_function (eth_defi.price_oracle.oracle.PriceFunction) – What function we use to calculate the price based on the events. Defaults to time-weighted average price.

  • target_time_window (datetime.timedelta) – What is the target time window for us to calculate the time function. Truncation will discard older data. Only relevant for real-time price oracles.

  • exchange_rate_oracle – If we depend on the secondary price data to calculate the price. E.g. converting AAVE/ETH rate to AAVE/USD using ETH/USDC pool price oracle.

  • max_age (datetime.timedelta) – A trip wire to detect corruption in real time data feeds. If the most recent entry in the buffer is older than this, throw an exception. This usually means we have stale data in our buffer and some price source pool has stopped working.

  • min_entries (int) – The minimum number of entries we want to have to calculate the price reliably.

  • min_duration (datetime.timedelta) –

get_last_refreshed()[source]

When the oracle data was refreshed last time.

To figure out max age in real time tracking mode.

Return type

datetime.datetime

update_last_refresh(block_number, timestamp)[source]

Update the last seen block.

Parameters
check_data_quality(now_=None)[source]

Raises one of PriceCalculationError subclasses if our data is not good enough to calculate the oracle price.

See PriceCalculationError

Parameters
Raises

PriceCalculationError – If we have data quality issues

calculate_price()[source]

Calculate the price based on the data in the price data buffer.

Raises

PriceCalculationError – If we have data quality issues.

Return type

decimal.Decimal

add_price_entry(evt)[source]

Add price entry to the ring buffer.

Note

It is not safe to call this function multiple times for the same event.

Further reading

Parameters

evt (eth_defi.price_oracle.oracle.PriceEntry) –

add_price_entry_reorg_safe(evt)[source]

Add price entry to the ring buffer with support for fixing chain reorganisations.

Transactions may hop between different blocks when the chain tip reorganises, getting a new timestamp. In this case, we update the

Note

It is safe to call this function multiple times for the same event.

Returns

True if the transaction hopped to a different block

Parameters

evt (eth_defi.price_oracle.oracle.PriceEntry) –

Return type

bool

get_by_transaction_hash(tx_hash)[source]

Get an event by transaction hash.

Parameters

tx_hash (str) –

Return type

Optional[eth_defi.price_oracle.oracle.PriceEntry]

get_newest()[source]

Return the newest price entry.

Return type

Optional[eth_defi.price_oracle.oracle.PriceEntry]

get_oldest()[source]

Return the oldest price entry.

Return type

Optional[eth_defi.price_oracle.oracle.PriceEntry]

get_buffer_duration()[source]

How long time is the time we have price events in the buffer for.

Return type

datetime.timedelta

feed_simple_data(data, source=PriceSource.unknown)[source]

Feed sample data to the price oracle from a Python dict.

This method is mostly for testing: for actual implementation construct your PriceEntry instances yourself.

Example:

price_data = {
    datetime.datetime(2021, 1, 3): Decimal(100),
    datetime.datetime(2021, 1, 2): Decimal(150),
    datetime.datetime(2021, 1, 1): Decimal(120),
}

oracle = PriceOracle(
    time_weighted_average_price,
)

oracle.feed_simple_data(price_data)
Parameters

data (Dict[datetime.datetime, decimal.Decimal]) –

truncate_buffer(current_timestamp)[source]

Delete old data in the buffer that is no longer relevant for our price calculation.

Returns

Numbers of items that where discared

Parameters

current_timestamp (datetime.datetime) –

Return type

int