PriceOracle

Documentation for eth_defi.price_oracle.oracle.PriceOracle Python 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`([block_number])	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

block_number (int) –
timestamp (datetime.datetime) –

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

now – Override the real-time clock for testing stale data.
now_ (Optional[datetime.datetime]) –

Raises

PriceCalculationError – If we have data quality issues

calculate_price(block_number=None)[source]

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

Raises: PriceCalculationError – If we have data quality issues.
Parameters: block_number (Optional[int]) –
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

https://docs.python.org/3/library/heapq.html

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