Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
SQLAlchemy crypto price feed class and find the nearest price to a timepoint
import enum
import datetime
import sqlalchemy as sa
from sqlalchemy import case, union_all
from sqlalchemy.orm import Session, aliased
from .utils import TimeStampedBaseModel
class PriceTooFar(Exception):
"""Could not find a dollar conversion rate close enough to the desired timestamp."""
class PriceFeedPair(enum.Enum):
usd_usd = "USDUSD"
btc_usd = "BTCUSD"
eth_usd = "ETHUSD"
class PriceFeedSource(enum.Enum):
coingecko = "coingecko"
uniswap_pool = "uniswap_pool"
class PriceFeedEntry(TimeStampedBaseModel):
"""A price conversion data either from Coingecko or Uniswap pool.
Allows looking for the nearest price for any time point
using PostgreSQL BRIN index andmax distance method.
"""
__tablename__ = "price_feed"
__table_args__ = (
sa.UniqueConstraint('price_at', 'pair', 'source'),
)
# When this price was taken, as UNIX timestamp
price_at = sa.Column(sa.TIMESTAMP, nullable=False)
pair = sa.Column(sa.Enum(PriceFeedPair), nullable=False, index=True)
source = sa.Column(sa.Enum(PriceFeedSource), nullable=False, index=True)
value = sa.Column(sa.Float, nullable=False)
def __str__(self):
return f"{self.pair} is {self.value} at {self.price_at} from {self.source}"
@staticmethod
def get_conversion_rate(dbsession: Session, price_feed_pair: PriceFeedPair, when: datetime.datetime, max_distance=24*3600) -> float:
"""Get the rate to convert a cryptocurrency to USD.
Originally based on the work of Ilja Everila: https://stackoverflow.com/a/42553618/315168
:param dbsession: SQLAlchemy session
:param price_feed_pair: One of pairs we track
:param when: UNIX timestamp
:param max_distance: If the price we have is further than this many seconds raise PriceTooFar
:return: How many dollars is one cryptocurrency unit
"""
assert isinstance(when, datetime.datetime), "Use datetimes"
when_dt = when.replace(tzinfo=None)
col = PriceFeedEntry.price_at
greater = dbsession.query(PriceFeedEntry).filter_by(pair=price_feed_pair).filter(col >= when_dt).order_by(col.asc()).limit(1).subquery().select()
lesser = dbsession.query(PriceFeedEntry).filter_by(pair=price_feed_pair).filter(col < when_dt).order_by(col.desc()).limit(1).subquery().select()
the_union = union_all(lesser, greater).alias()
the_alias = aliased(PriceFeedEntry, the_union)
the_diff = getattr(the_alias, col.name) - when_dt
abs_diff = case([(the_diff < datetime.timedelta(0), -the_diff)], else_=the_diff)
entry = dbsession.query(the_alias).order_by(abs_diff.asc()).first()
assert entry, f"No price data for {price_feed_pair}" # We must have at least one row in the table
if abs((entry.price_at - when_dt).total_seconds()) > max_distance:
raise PriceTooFar(f"Could not get price for {price_feed_pair} because the nearest price for {when} is too far {entry.price_at}")
return entry.value
# Use BRIN for efficient time range indexing
sa.Index("price_feed_price_at", PriceFeedEntry.__table__.c.price_at, postgresql_using="brin")
def translate_to_price_feed_symbol(symbol: str) -> PriceFeedPair:
"""Translate a ERC-20 symbol to underlying asset.
World has multiple US stablecoins, wrapped BTCs and such.
Map these to their underlying assets.
"""
if symbol in ("DAI", "USDC", "USDT", "cUSDC", "cUSDT", "sUSD", "aDAI", "DAI"):
return PriceFeedPair.usd_usd
elif symbol in ("WETH",):
return PriceFeedPair.eth_usd
elif symbol in ("WBTC", "renBTC"):
return PriceFeedPair.btc_usd
else:
raise RuntimeError(f"Unknown quote token {symbol}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment