tessa.price.coingecko

 1"""Everything coingecko-related (other than search)."""
 2
 3import pandas as pd
 4from pycoingecko import CoinGeckoAPI
 5from .types import (
 6    PriceHistory,
 7    SymbolNotFoundError,
 8    CurrencyPreferenceNotFoundError,
 9    RateLimitHitError,
10)
11
12
13def dataframify_price_list(prices: list) -> pd.DataFrame:
14    """Turn price list returned by Coingecko API into a pricing dataframe in the form
15    that we use.
16    """
17    df = pd.DataFrame(prices)
18    df.columns = ["date", "close"]
19    df["date"] = pd.to_datetime(df["date"] / 1000, unit="s", utc=True)
20    df = df.set_index("date")
21    return df
22
23
24def get_price_history(query: str, currency_preference: str = "USD") -> PriceHistory:
25    """Get price history for a given cryptocurrency."""
26    try:
27        res = CoinGeckoAPI().get_coin_market_chart_by_id(
28            id=query,
29            vs_currency=currency_preference,
30            days="max",  # FIXME Add some lower default bound similar to yahoo?
31            interval="daily",
32        )["prices"]
33    except ValueError as exc:
34        if "invalid vs_currency" in str(exc):
35            raise CurrencyPreferenceNotFoundError(
36                source="coingecko", cur_pref=currency_preference
37            ) from exc
38        if "429" in str(exc):
39            # (pycoingecko masks the underlying HTTPError.)
40            raise RateLimitHitError(source="coingecko") from exc
41        raise SymbolNotFoundError(source="coingecko", query=query) from exc
42
43    return PriceHistory(dataframify_price_list(res), currency_preference)