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 to 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) -> PricePoint:
54    """Return the price at a given point in time given by `when`. Look for the closest
55    point in time if the exact point in time is not found. Returns a `PricePoint`, i.e.,
56    a tuple of the price, the effective timestamp of the price, and the currency.
57
58    Arguments other than `when` are the same as with `price_history`.
59
60    Example call:
61    ```
62    price_point("AAPL", "2020-01-01")
63    ```
64    """
65    df, currency = price_history(query, source, currency_preference)
66    price = df.iloc[df.index.get_indexer([when], method="nearest")[0]]
67    return PricePoint(when=price.name, price=float(price), currency=currency)
68
69
70def price_point_strict(
71    query: str,
72    when: str,
73    source: SourceType = "yahoo",
74    currency_preference: str = "USD",
75) -> PricePoint:
76    """Same as `price_point` but will return either the price at the exact point in time
77    or raise a KeyError.
78    """
79    df, currency = price_history(query, source, currency_preference)
80    return PricePoint(
81        when=df.loc[when].name, price=float(df.loc[when]["close"]), currency=currency
82    )
83
84
85def price_latest(
86    query: str,
87    source: SourceType = "yahoo",
88    currency_preference: str = "USD",
89) -> PricePoint:
90    """Same as `price_point` but will return the latest price."""
91    df, currency = price_history(query, source, currency_preference)
92    return PricePoint(
93        when=df.iloc[-1].name, price=float(df.iloc[-1]["close"]), currency=currency
94    )