original progress from a few weekends ago

MarketData.py

@@ -0,0 +1,196 @@
+"""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()