price_oracle.oracle

Documentation for eth_defi.price_oracle.oracle Python module.

Price oracle core functionality.

This core mechanism is used by outside event feeders, like eth_defi.uniswap_v2.oracle.

Functions

time_weighted_average_price(events)

Calculate TWAP price over all entries in the buffer.

Classes

`BasePriceOracle`	Base class for price oracles.
`FixedPriceOracle`	Always use the same hardcoded exchange rate.
`PriceEntry`	A single source entry for price calculations.
`PriceFunction`	A callable for calcualte
`PriceOracle`	Price oracle core.
`PriceSource`	Different price entry sources.
`TrustedStablecoinOracle`	Return a price for a token we trust we can always redeem for 1 USD.

Exceptions

`DataPeriodTooShort`	We do not have enough events for a longer period of time.
`DataTooOld`	The price buffer data does not have recent enough entries..
`NotEnoughData`	The price buffer does not have enough data.
`PriceCalculationError`	Something wrong with price calculation.

class BasePriceOracle

Bases: object

Base class for price oracles.

abstract calculate_price(block_number=None)

Get a price for the current block.

Parameters: block_number (Optional[int]) – Hint of what is the current block. We do not support prices for historical blocks, but we may cache the result of the previous block calculation for speedups.
Return type: decimal.Decimal

exception DataPeriodTooShort

Bases: eth_defi.price_oracle.oracle.PriceCalculationError

We do not have enough events for a longer period of time.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

exception DataTooOld

Bases: eth_defi.price_oracle.oracle.PriceCalculationError

The price buffer data does not have recent enough entries..

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

class FixedPriceOracle

Bases: eth_defi.price_oracle.oracle.BasePriceOracle

Always use the same hardcoded exchange rate.

Most useful for unit testing.

__init__(exchange_rate)

Parameters: exchange_rate (decimal.Decimal) –

calculate_price(block_number=None)

Get a price for the current block.

Parameters: block_number (Optional[int]) – Hint of what is the current block. We do not support prices for historical blocks, but we may cache the result of the previous block calculation for speedups.
Return type: decimal.Decimal

exception NotEnoughData

Bases: eth_defi.price_oracle.oracle.PriceCalculationError

The price buffer does not have enough data.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

exception PriceCalculationError

Bases: Exception

Something wrong with price calculation.

__init__(*args, **kwargs)

__new__(**kwargs)

add_note(note, /): Add a note to the exception

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

class PriceEntry

Bases: object

A single source entry for price calculations.

PriceOracle maintains a buffer of these to calculate a smoothed out price, like py:func:time_weighted_average_price.

Price entry can be sourced from:

Manually entered price
Price from Uniswap v2 sync events
Price from some other event

__init__(timestamp, price, source, volume=None, pool_contract_address=None, block_number=None, tx_hash=None, block_hash=None, first_seen_at_block_number=None)

Parameters

timestamp (datetime.datetime) –
price (decimal.Decimal) –
source (eth_defi.price_oracle.oracle.PriceSource) –
volume (Optional[decimal.Decimal]) –
pool_contract_address (Optional[str]) –
block_number (Optional[int]) –
tx_hash (Optional[str]) –
block_hash (Optional[str]) –
first_seen_at_block_number (Optional[int]) –

Return type

None

update_chain_reorg(new_entry)

Update entry data in the case of chain reorganisation.

TODO: We are not yet dealing with the situation if the transaction gets reorganisated and rejected.

Parameters: new_entry (eth_defi.price_oracle.oracle.PriceEntry) –

class PriceFunction

Bases: Protocol

A callable for calcualte

You can give different function for

Volume weighted average
Time weighted average

__init__(*args, **kwargs)

class PriceOracle

Bases: eth_defi.price_oracle.oracle.BasePriceOracle

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"))

Create a new price oracle.

The security parameters are set for a simple defaults.

Parameters

price_function – What function we use to calculate the price based on the events. Defaults to time-weighted average price.
target_time_window – 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 – 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 – The minimum number of entries we want to have to calculate the price reliably.

__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)

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) –

add_price_entry(evt)

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)

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

calculate_price(block_number=None)

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

check_data_quality(now_=None)

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

feed_simple_data(data, source=PriceSource.unknown)

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]) –

get_buffer_duration()

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

Return type: datetime.timedelta

get_by_transaction_hash(tx_hash)

Get an event by transaction hash.

Parameters: tx_hash (str) –
Return type: Optional[eth_defi.price_oracle.oracle.PriceEntry]

get_last_refreshed()

When the oracle data was refreshed last time.

To figure out max age in real time tracking mode.

Return type: datetime.datetime

get_newest()

Return the newest price entry.

Return type: Optional[eth_defi.price_oracle.oracle.PriceEntry]

get_oldest()

Return the oldest price entry.

Return type: Optional[eth_defi.price_oracle.oracle.PriceEntry]

truncate_buffer(current_timestamp)

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

update_last_refresh(block_number, timestamp)

Update the last seen block.

Parameters

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

class PriceSource

Bases: enum.Enum

Different price entry sources.

class TrustedStablecoinOracle

Bases: eth_defi.price_oracle.oracle.BasePriceOracle

Return a price for a token we trust we can always redeem for 1 USD.

calculate_price(block_number=None)

Get a price for the current block.

Parameters: block_number (Optional[int]) – Hint of what is the current block. We do not support prices for historical blocks, but we may cache the result of the previous block calculation for speedups.
Return type: decimal.Decimal

time_weighted_average_price(events)

Calculate TWAP price over all entries in the buffer.

Calculates the price using statistics.mean().