Skip to content

Instantly share code, notes, and snippets.

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}"
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:
: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, - 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
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