Simple-Telegram-Stock-Bot/MarketData.py

"""Class with functions for running the bot with IEX Cloud.
"""

import logging
import os
from datetime import datetime
from logging import warning
from typing import List, Optional, Tuple

import pandas as pd
import requests as r
import schedule

from Symbol import Stock


class MarketData:
    """
    Functions for finding stock market information about symbols from MarkData.app
    """

    SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})"

    searched_symbols = {}
    otc_list = []
    charts = {}
    trending_cache = None

    def __init__(self) -> None:
        """Creates a Symbol Object

        Parameters
        ----------
        MARKETDATA_TOKEN : str
            MarketData.app API Token
        """
        try:
            self.MARKETDATA_TOKEN = os.environ["MarketData"]

            if self.MARKETDATA_TOKEN == "TOKEN":
                self.MARKETDATA_TOKEN = ""
        except KeyError:
            self.MARKETDATA_TOKEN = ""
            warning(
                "Starting without an MarketData.app Token will not allow you to get market data!"
            )

        if self.MARKETDATA_TOKEN != "":
            self.get_symbol_list()

            schedule.every().day.do(self.get_symbol_list)
            schedule.every().day.do(self.clear_charts)

    def get(self, endpoint, params: dict = {}, timeout=5) -> dict:
        url = "https://api.marketdata.app/v1/" + endpoint

        # set token param if it wasn't passed.
        params["token"] = params.get("token", self.MARKETDATA_TOKEN)

        resp = r.get(url, params=params, timeout=timeout)

        # Make sure API returned a proper status code
        try:
            resp.raise_for_status()
        except r.exceptions.HTTPError as e:
            logging.error(e)
            return {}

        # Make sure API returned valid JSON
        try:
            resp_json = resp.json()

            match resp_json["s"]:
                case "ok":
                    return resp_json
                case "no_data":
                    return resp_json
                case "error":
                    logging.error("MarketData Error:\n" + resp_json["errmsg"])
                    return {}

        except r.exceptions.JSONDecodeError as e:
            logging.error(e)
            return {}

    def clear_charts(self) -> None:
        """
        Clears cache of chart data.
        Charts are cached so that only 1 API call per 24 hours is needed since the
            chart data is expensive and a large download.
        """
        self.charts = {}

    def get_symbol_list(
        self, return_df=False
    ) -> Optional[Tuple[pd.DataFrame, datetime]]:
        """Gets list of all symbols supported by IEX

        Parameters
        ----------
        return_df : bool, optional
            return the dataframe of all stock symbols, by default False

        Returns
        -------
        Optional[Tuple[pd.DataFrame, datetime]]
            If `return_df` is set to `True` returns a dataframe, otherwise returns `None`.
        """

        reg_symbols = self.get("/ref-data/symbols")
        otc_symbols = self.get("/ref-data/otc/symbols")

        reg = pd.DataFrame(data=reg_symbols)
        otc = pd.DataFrame(data=otc_symbols)
        self.otc_list = set(otc["symbol"].to_list())

        symbols = pd.concat([reg, otc])

        # IEX uses backtick ` as apostrophe which breaks telegram markdown parsing
        symbols["name"] = symbols["name"].str.replace("`", "'")

        symbols["description"] = "$" + symbols["symbol"] + ": " + symbols["name"]
        symbols["id"] = symbols["symbol"]
        symbols["type_id"] = "$" + symbols["symbol"].str.lower()

        symbols = symbols[["id", "symbol", "name", "description", "type_id"]]
        self.symbol_list = symbols
        if return_df:
            return symbols, datetime.now()

    def status(self) -> str:
        return "status isnt implemented by marketdata.app"

    def price_reply(self, symbol: Stock) -> str:
        """Returns price movement of Stock for the last market day, or after hours.

        Parameters
        ----------
        symbol : Stock

        Returns
        -------
        str
            Formatted markdown
        """

        if quoteResp := self.get(f"/stocks/quotes/{symbol}/"):
            return f"The current price of {quoteResp['symbol']} is ${quoteResp['last']}"

        else:
            return f"Getting a quote for {symbol} encountered an error."

    def chart_reply(self, symbol: Stock) -> pd.DataFrame:
        """Returns price data for a symbol of the past month up until the previous trading days close.
        Also caches multiple requests made in the same day.

        Parameters
        ----------
        symbol : str
            Stock symbol.

        Returns
        -------
        pd.DataFrame
            Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
        """
        schedule.run_pending()

        try:
            return self.charts[symbol.id.upper()]
        except KeyError:
            pass

        to_date = datetime.today().strftime("%Y-%m-%d")
        from_date = (datetime.today() - datetime.timedelta(days=30)).strftime(
            "%Y-%m-%d"
        )

        if data := self.get(
            f"/stocks/candles/h/{symbol}",
            params={
                "chartInterval": 3,
                "to": to_date,
                "from": from_date,
            },
        ):
            df = pd.DataFrame(
                data, columns=["Date", "Open", "High", "Low", "Close"]
            ).dropna()
            df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
            df = df.set_index("DT")
            self.charts[symbol.id.upper()] = df
            return df

        return pd.DataFrame()