tessa.price.price

  1"""Retrieve price information."""
  2
  3from __future__ import annotations
  4from functools import lru_cache, wraps
  5from typing import Union, TYPE_CHECKING
  6import pandas as pd
  7from . import PriceHistory, PricePoint
  8from .. import sources
  9
 10if TYPE_CHECKING:
 11    from .. import SourceType
 12
 13
 14def custom_cache_wrapper(func):
 15    """To preserve the function's signature _and_ give access to `cache_clear` etc."""
 16    cached_func = lru_cache(maxsize=None)(func)
 17    wrapped_func = wraps(func)(cached_func)
 18    return wrapped_func
 19
 20
 21@custom_cache_wrapper
 22def price_history(
 23    query: str,
 24    source: SourceType = "yahoo",
 25    currency_preference: str = "USD",
 26) -> PriceHistory:
 27    """Get price history and return `PriceHistory`, i.e., a tuple of a dataframe with
 28    the price history and the effective currency. Note that the effective currency
 29    returned might differ from the currency_preference.
 30
 31    - `query`: A query string that makes sense in combination with the source. E.g.,
 32      "BTC-USD" for "yahoo" or "bitcoin" for "coingecko".
 33    - `source`: The source to query. Defaults to "yahoo".
 34    - `currency_preference`: The currency the prices should be returned in; defaults
 35      to "USD". The effective currency might differ and will be returned in the second
 36      return value.
 37    """
 38    src = sources.get_source(source)
 39    src.rate_limiter.rate_limit()
 40    df, effective_currency = src.get_price_history_bruteforcefully(
 41        query, currency_preference
 42    )
 43    return PriceHistory(df.copy(), effective_currency.upper())
 44    # (Returning a copy of the dataframe so the cached original is preserved even if it
 45    # gets modified by the caller.)
 46
 47
 48def price_point(
 49    query: str,
 50    when: Union[str, pd.Timestamp],
 51    source: SourceType = "yahoo",
 52    currency_preference: str = "USD",
 53    max_date_deviation_days: Union[int, None] = 10,
 54) -> PricePoint:
 55    """Return the price at a given point in time given by `when`. Look for the closest
 56    point in time if the exact point in time is not found. Returns a `PricePoint`, i.e.,
 57    a tuple of the price, the effective timestamp of the price, and the currency.
 58
 59    Raises a `ValueError` if the found date is more than `max_date_deviation_days` from
 60    `when`. Use `None` to disable this check.
 61
 62    Arguments other than `when` are the same as with `price_history`.
 63
 64    Example call:
 65    ```
 66    price_point("AAPL", "2020-01-01")
 67    ```
 68    """
 69    df, currency = price_history(query, source, currency_preference)
 70
 71    when = pd.Timestamp(when)
 72    if df.index.tz is not None:  # Ensure when matches the timezone of df.index
 73        when = when.tz_localize("UTC") if when.tz is None else when.tz_convert("UTC")
 74
 75    nearest_index = df.index.get_indexer([when], method="nearest")[0]
 76    found_date = df.index[nearest_index]
 77
 78    if (
 79        max_date_deviation_days is not None
 80        and abs((found_date - when).days) > max_date_deviation_days
 81    ):
 82        raise ValueError(
 83            f"Found date {found_date} is more than {max_date_deviation_days} days away "
 84            f"from requested date {when}"
 85        )
 86
 87    price = df.iloc[nearest_index]
 88    return PricePoint(when=found_date, price=float(price.iloc[0]), currency=currency)
 89
 90
 91def price_point_strict(
 92    query: str,
 93    when: str,
 94    source: SourceType = "yahoo",
 95    currency_preference: str = "USD",
 96) -> PricePoint:
 97    """Same as `price_point` but will raise a `KeyError` if no price is found. This
 98    function is offered for backwards compatibility and is largely obsolete with the
 99    introduction of `max_date_deviation_days` in `price_point`.
100    """
101    try:
102        return price_point(
103            query, when, source, currency_preference, max_date_deviation_days=0
104        )
105    except ValueError as e:
106        raise KeyError(f"No price found for {query} at {when}") from e
107
108
109def price_latest(
110    query: str,
111    source: SourceType = "yahoo",
112    currency_preference: str = "USD",
113) -> PricePoint:
114    """Same as `price_point` but will return the latest price."""
115    df, currency = price_history(query, source, currency_preference)
116    return PricePoint(
117        when=df.iloc[-1].name, price=float(df.iloc[-1]["close"]), currency=currency
118    )