Created
April 2, 2021 15:34
-
-
Save miohtama/7179d140e7893851939d401a0f242a20 to your computer and use it in GitHub Desktop.
SQLAlchemy crypto price feed class and find the nearest price to a timepoint
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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