moved everything from telegram over

D_info.py

@@ -0,0 +1,75 @@
+"""Functions and Info specific to the Telegram Bot
+"""
+
+import re
+
+import requests as r
+
+
+class T_info:
+    license = re.sub(
+        r"\b\n",
+        " ",
+        r.get(
+            "https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot/-/raw/master/LICENSE"
+        ).text,
+    )
+
+    help_text = """
+Thanks for using this bot, consider supporting it by [buying me a beer.](https://www.buymeacoffee.com/Anson)
+
+Keep up with the latest news for the bot in its Telegram Channel: https://t.me/simplestockbotnews
+
+Full documentation on using and running your own stock bot can be found on the bots [docs.](https://docs.simplestockbot.com)
+
+The bot detects _"Symbols"_ using either one `$` or two `$$` dollar signs before the symbol. One dollar sign is for a stock market ticker, while two is for a cryptocurrency coin. `/chart $$eth` would return a chart of the past month of data for Ethereum, while `/dividend $psec` returns dividend information for Prospect Capital stock.
+
+Simply calling a symbol in any message that the bot can see will also return the price. So a message like: `I wonder if $$btc will go to the Moon now that $tsla accepts it as payment` would return the current price for both Bitcoin and Tesla. 
+
+**Commands**
+        - `/donate [amount in USD]` to donate. 🎗️
+        - `/dividend $[symbol]` Dividend information for the symbol. 📅
+        - `/intra $[symbol]` Plot of the stocks movement since the last market open.  📈
+        - `/chart $[symbol]` Plot of the stocks movement for the past 1 month. 📊
+        - `/news $[symbol]` News about the symbol. 📰
+        - `/info $[symbol]` General information about the symbol. ℹ️
+        - `/stat $[symbol]` Key statistics about the symbol. 🔢
+        - `/cap $[symbol]` Market Capitalization of symbol. 💰
+        - `/trending` Trending Stocks and Cryptos. 💬
+        - `/help` Get some help using the bot. 🆘
+
+**Inline Features**
+    You can type @SimpleStockBot `[search]` in any chat or direct message to search for the stock bots full list of stock symbols and return the price of the ticker. Then once you select the ticker want the bot will send a message as you in that chat with the latest stock price.
+    
+    Market data is provided by [IEX Cloud](https://iexcloud.io)
+
+    If you believe the bot is not behaving properly run `/status` or [get in touch](https://docs.simplestockbot.com/contact).
+    """
+
+    donate_text = """
+Simple Stock Bot is run entirely on donations[.](https://www.buymeacoffee.com/Anson)
+All donations go directly towards paying for servers, and market data is provided by
+[IEX Cloud](https://iexcloud.io/).
+
+The easiest way to donate is to run the `/donate [amount in USD]` command with US dollars you would like to donate.
+
+Example: `/donate 2` would donate 2 USD.
+
+An alternative way to donate is through https://www.buymeacoffee.com/Anson which requires no account and accepts Paypal or Credit card.
+If you have any questions see the [website](https:docs.simplestockbot.com)
+
+    """
+
+
+commands = """
+donate - Donate to the bot 🎗️
+help - Get some help using the bot. 🆘
+info - $[symbol] General information about the symbol. ℹ️
+news - $[symbol] News about the symbol. 📰
+stat - $[symbol] Key statistics about the symbol. 🔢
+cap - $[symbol] Market Capitalization of symbol. 💰
+dividend - $[symbol] Dividend info 📅
+trending - Trending Stocks and Cryptos. 💬
+intra - $[symbol] Plot since the last market open. 📈
+chart - $[chart] Plot of the past month. 📊
+"""  # Not used by the bot but for updaing commands with BotFather

Dockerfile

@@ -3,6 +3,6 @@ FROM python:3.8-buster
 COPY requirements.txt ./
 RUN pip install --no-cache-dir -r requirements.txt
 
-COPY . .
+# COPY . .
 
-CMD [ "python", "./bot.py" ]
+# CMD [ "python", "./bot.py" ]

IEX_Symbol.py

@@ -0,0 +1,531 @@
+"""Class with functions for running the bot with IEX Cloud.
+"""
+
+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 fuzzywuzzy import fuzz
+
+from Symbol import Stock
+
+
+class IEX_Symbol:
+    """
+    Functions for finding stock market information about symbols.
+    """
+
+    SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})"
+
+    searched_symbols = {}
+    otc_list = []
+    charts = {}
+
+    def __init__(self) -> None:
+        """Creates a Symbol Object
+
+        Parameters
+        ----------
+        IEX_TOKEN : str
+            IEX API Token
+        """
+        try:
+            self.IEX_TOKEN = os.environ["IEX"]
+        except KeyError:
+            self.IEX_TOKEN = ""
+            warning(
+                "Starting without an IEX Token will not allow you to get market data!"
+            )
+
+        if self.IEX_TOKEN != "":
+            self.get_symbol_list()
+
+            schedule.every().day.do(self.get_symbol_list)
+            schedule.every().day.do(self.clear_charts)
+
+    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 = r.get(
+            f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}",
+            timeout=5,
+        ).json()
+        otc_symbols = r.get(
+            f"https://cloud.iexapis.com/stable/ref-data/otc/symbols?token={self.IEX_TOKEN}",
+            timeout=5,
+        ).json()
+
+        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])
+
+        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:
+        """Checks IEX Status dashboard for any current API issues.
+
+        Returns
+        -------
+        str
+            Human readable text on status of IEX API
+        """
+        resp = r.get(
+            "https://pjmps0c34hp7.statuspage.io/api/v2/status.json",
+            timeout=5,
+        )
+
+        if resp.status_code == 200:
+            status = resp.json()["status"]
+        else:
+            return "IEX Cloud did not respond. Please check their status page for more information. https://status.iexapis.com"
+
+        if status["indicator"] == "none":
+            return "IEX Cloud is currently not reporting any issues with its API."
+        else:
+            return (
+                f"{status['indicator']}: {status['description']}."
+                + " Please check the status page for more information. https://status.iexapis.com"
+            )
+
+    def search_symbols(self, search: str) -> List[Tuple[str, str]]:
+        """Performs a fuzzy search to find stock symbols closest to a search term.
+
+        Parameters
+        ----------
+        search : str
+            String used to search, could be a company name or something close to the companies stock ticker.
+
+        Returns
+        -------
+        List[tuple[str, str]]
+            A list tuples of every stock sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
+        """
+
+        schedule.run_pending()
+        search = search.lower()
+        try:  # https://stackoverflow.com/a/3845776/8774114
+            return self.searched_symbols[search]
+        except KeyError:
+            pass
+
+        symbols = self.symbol_list
+        symbols["Match"] = symbols.apply(
+            lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
+            axis=1,
+        )
+
+        symbols.sort_values(by="Match", ascending=False, inplace=True)
+        if symbols["Match"].head().sum() < 300:
+            symbols["Match"] = symbols.apply(
+                lambda x: fuzz.partial_ratio(search, x["name"].lower()),
+                axis=1,
+            )
+
+            symbols.sort_values(by="Match", ascending=False, inplace=True)
+        symbols = symbols.head(10)
+        symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
+        self.searched_symbols[search] = symbol_list
+        return symbol_list
+
+    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
+        """
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol.id}/quote?token={self.IEX_TOKEN}"
+
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+        if response.status_code == 200:
+            IEXData = response.json()
+
+            if symbol.symbol.upper() in self.otc_list:
+                return f"OTC - {symbol.symbol.upper()}, {IEXData['companyName']} most recent price is: $**{IEXData['latestPrice']}**"
+
+            keys = (
+                "extendedChangePercent",
+                "extendedPrice",
+                "companyName",
+                "latestPrice",
+                "changePercent",
+            )
+
+            if set(keys).issubset(IEXData):
+
+                if change := IEXData.get("changePercent", 0):
+                    change = round(change * 100, 2)
+                else:
+                    change = 0
+
+                if (
+                    IEXData.get("isUSMarketOpen", True)
+                    or (IEXData["extendedChangePercent"] is None)
+                    or (IEXData["extendedPrice"] is None)
+                ):  # Check if market is open.
+                    message = f"The current stock price of {IEXData['companyName']} is $**{IEXData['latestPrice']}**"
+                else:
+                    message = (
+                        f"{IEXData['companyName']} closed at $**{IEXData['latestPrice']}** with a change of {change}%,"
+                        + f" after hours _(15 minutes delayed)_ the stock price is $**{IEXData['extendedPrice']}**"
+                    )
+                    if change := IEXData.get("extendedChangePercent", 0):
+                        change = round(change * 100, 2)
+                    else:
+                        change = 0
+
+                # Determine wording of change text
+                if change > 0:
+                    message += f", the stock is currently **up {change}%**"
+                elif change < 0:
+                    message += f", the stock is currently **down {change}%**"
+                else:
+                    message += ", the stock hasn't shown any movement today."
+            else:
+                message = (
+                    f"The symbol: {symbol} encountered and error. This could be due to "
+                )
+
+        else:
+            message = f"The symbol: {symbol} was not found."
+
+        return message
+
+    def dividend_reply(self, symbol: Stock) -> str:
+        """Returns the most recent, or next dividend date for a stock symbol.
+
+        Parameters
+        ----------
+        symbol : Stock
+
+        Returns
+        -------
+        str
+            Formatted markdown
+        """
+        if symbol.symbol.upper() in self.otc_list:
+            return "OTC stocks do not currently support any commands."
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/dividends/next?token={self.IEX_TOKEN}"
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+        if response.status_code == 200 and response.json():
+            IEXData = response.json()[0]
+            keys = (
+                "amount",
+                "currency",
+                "declaredDate",
+                "exDate",
+                "frequency",
+                "paymentDate",
+                "flag",
+            )
+
+            if set(keys).issubset(IEXData):
+
+                if IEXData["currency"] == "USD":
+                    price = f"${IEXData['amount']}"
+                else:
+                    price = f"{IEXData['amount']} {IEXData['currency']}"
+
+                # Pattern IEX uses for dividend date.
+                pattern = "%Y-%m-%d"
+
+                declared = datetime.strptime(IEXData["declaredDate"], pattern).strftime(
+                    "%A, %B %w"
+                )
+                ex = datetime.strptime(IEXData["exDate"], pattern).strftime("%A, %B %w")
+                payment = datetime.strptime(IEXData["paymentDate"], pattern).strftime(
+                    "%A, %B %w"
+                )
+
+                daysDelta = (
+                    datetime.strptime(IEXData["paymentDate"], pattern) - datetime.now()
+                ).days
+
+                return (
+                    "The next dividend for "
+                    + f"{self.symbol_list[self.symbol_list['symbol']==symbol.id.upper()]['description'].item()}"  # Get full name without api call
+                    + f" is on {payment} which is in {daysDelta} days."
+                    + f" The dividend is for {price} per share."
+                    + f"\n\nThe dividend was declared on {declared} and the ex-dividend date is {ex}"
+                )
+
+        return f"${symbol.id.upper()} either doesn't exist or pays no dividend."
+
+    def news_reply(self, symbol: Stock) -> str:
+        """Gets most recent, english, non-paywalled news
+
+        Parameters
+        ----------
+        symbol : Stock
+
+        Returns
+        -------
+        str
+            Formatted markdown
+        """
+        if symbol.symbol.upper() in self.otc_list:
+            return "OTC stocks do not currently support any commands."
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/15?token={self.IEX_TOKEN}"
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+        if response.status_code == 200:
+            data = response.json()
+            if data:
+                line = []
+
+                for news in data:
+                    if news["lang"] == "en" and not news["hasPaywall"]:
+                        line.append(
+                            f"*{news['source']}*: [{news['headline']}]({news['url']})"
+                        )
+
+            else:
+                return f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
+        else:
+            return f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
+
+        return f"News for **{symbol.id.upper()}**:\n" + "\n".join(line[:5])
+
+    def info_reply(self, symbol: Stock) -> str:
+        """Gets description for Stock
+
+        Parameters
+        ----------
+        symbol : Stock
+
+        Returns
+        -------
+        str
+            Formatted text
+        """
+        if symbol.symbol.upper() in self.otc_list:
+            return "OTC stocks do not currently support any commands."
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/company?token={self.IEX_TOKEN}"
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+
+        if response.status_code == 200:
+            data = response.json()
+            [data.pop(k) for k in list(data) if data[k] == ""]
+
+            if "description" in data:
+                return data["description"]
+
+        return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
+
+    def stat_reply(self, symbol: Stock) -> str:
+        """Key statistics on a Stock
+
+        Parameters
+        ----------
+        symbol : Stock
+
+        Returns
+        -------
+        str
+            Formatted markdown
+        """
+        if symbol.symbol.upper() in self.otc_list:
+            return "OTC stocks do not currently support any commands."
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/stats?token={self.IEX_TOKEN}"
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+
+        if response.status_code == 200:
+            data = response.json()
+            [data.pop(k) for k in list(data) if data[k] == ""]
+
+            m = ""
+            if "companyName" in data:
+                m += f"Company Name: {data['companyName']}\n"
+            if "marketcap" in data:
+                m += f"Market Cap: ${data['marketcap']:,}\n"
+            if "week52high" in data:
+                m += f"52 Week (high-low): {data['week52high']:,} "
+            if "week52low" in data:
+                m += f"- {data['week52low']:,}\n"
+            if "employees" in data:
+                m += f"Number of Employees: {data['employees']:,}\n"
+            if "nextEarningsDate" in data:
+                m += f"Next Earnings Date: {data['nextEarningsDate']}\n"
+            if "peRatio" in data:
+                m += f"Price to Earnings: {data['peRatio']:.3f}\n"
+            if "beta" in data:
+                m += f"Beta: {data['beta']:.3f}\n"
+            return m
+        else:
+            return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
+
+    def cap_reply(self, stock: Stock) -> str:
+        """Get the Market Cap of a stock"""
+        response = r.get(
+            f"https://cloud.iexapis.com/stable/stock/{stock.id}/stats?token={self.IEX_TOKEN}",
+            timeout=5,
+        )
+        if response.status_code == 200:
+
+            try:
+                data = response.json()
+
+                cap = data["marketcap"]
+            except KeyError:
+                return f"{stock.id} returned an error."
+
+            message = f"The current market cap of {stock.name} is $**{cap:,.2f}**"
+
+        else:
+            message = f"The Coin: {stock.name} was not found or returned and error."
+
+        return message
+
+    def intra_reply(self, symbol: Stock) -> pd.DataFrame:
+        """Returns price data for a symbol since the last market open.
+
+        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.
+        """
+        if symbol.symbol.upper() in self.otc_list:
+            return pd.DataFrame()
+
+        if symbol.id.upper() not in list(self.symbol_list["symbol"]):
+            return pd.DataFrame()
+
+        IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/intraday-prices?token={self.IEX_TOKEN}"
+        response = r.get(
+            IEXurl,
+            timeout=5,
+        )
+        if response.status_code == 200:
+            df = pd.DataFrame(response.json())
+            df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
+            df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
+            df = df.set_index("DT")
+            return df
+
+        return pd.DataFrame()
+
+    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()
+
+        if symbol.symbol.upper() in self.otc_list:
+            return pd.DataFrame()
+
+        if symbol.id.upper() not in list(self.symbol_list["symbol"]):
+            return pd.DataFrame()
+
+        try:  # https://stackoverflow.com/a/3845776/8774114
+            return self.charts[symbol.id.upper()]
+        except KeyError:
+            pass
+
+        response = r.get(
+            f"https://cloud.iexapis.com/stable/stock/{symbol}/chart/1mm?token={self.IEX_TOKEN}&chartInterval=3&includeToday=false",
+            timeout=5,
+        )
+
+        if response.status_code == 200:
+            df = pd.DataFrame(response.json())
+            df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
+            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()
+
+    def trending(self) -> list[str]:
+        """Gets current coins trending on IEX. Only returns when market is open.
+
+        Returns
+        -------
+        list[str]
+            list of $ID: NAME, CHANGE%
+        """
+
+        stocks = r.get(
+            f"https://cloud.iexapis.com/stable/stock/market/list/mostactive?token={self.IEX_TOKEN}",
+            timeout=5,
+        )
+        if stocks.status_code == 200:
+            return [
+                f"`${s['symbol']}`: {s['companyName']}, {100*s['changePercent']:.2f}%"
+                for s in stocks.json()
+            ]
+        else:
+            return ["Trending Stocks Currently Unavailable."]

Symbol.py

@@ -0,0 +1,58 @@
+import functools
+
+import requests as r
+
+
+class Symbol:
+    """
+    symbol: What the user calls it. ie tsla or btc
+    id: What the api expects. ie tsla or bitcoin
+    name: Human readable. ie Tesla or Bitcoin
+    """
+
+    currency = "usd"
+    pass
+
+    def __init__(self, symbol) -> None:
+        self.symbol = symbol
+        self.id = symbol
+        self.name = symbol
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>"
+
+    def __str__(self) -> str:
+        return self.id
+
+
+class Stock(Symbol):
+    """Stock Market Object. Gets data from IEX Cloud"""
+
+    def __init__(self, symbol: str) -> None:
+        self.symbol = symbol
+        self.id = symbol
+        self.name = "$" + symbol.upper()
+
+
+# Used by Coin to change symbols for ids
+coins = r.get("https://api.coingecko.com/api/v3/coins/list").json()
+
+
+class Coin(Symbol):
+    """Cryptocurrency Object. Gets data from CoinGecko."""
+
+    @functools.cache
+    def __init__(self, symbol: str) -> None:
+        self.symbol = symbol
+        self.get_data()
+
+    def get_data(self) -> None:
+        self.id = list(filter(lambda coin: coin["symbol"] == self.symbol, coins))[0][
+            "id"
+        ]
+        data = r.get("https://api.coingecko.com/api/v3/coins/" + self.id).json()
+        self.data = data
+
+        self.name = data["name"]
+        self.description = data["description"]
+        # self.price = data["market_data"]["current_price"][self.currency]

bot.py

@@ -9,7 +9,7 @@ from discord.ext import commands
 
 from functions import Symbol
 
-DISCORD_TOKEN = os.environ["DISCORD"]
+DISCORD_TOKEN = "NjAxNjYwNzM1ODQzNDY3Mjg0.XTFiOQ.DMEKIug2mhX1WGGVSD1bYuYXVro"
 
 try:
     IEX_TOKEN = os.environ["IEX"]

cg_Crypto.py

@@ -0,0 +1,384 @@
+"""Class with functions for running the bot with IEX Cloud.
+"""
+
+from datetime import datetime
+from typing import List, Optional, Tuple
+
+import pandas as pd
+import requests as r
+import schedule
+from fuzzywuzzy import fuzz
+from markdownify import markdownify
+
+from Symbol import Coin
+
+
+class cg_Crypto:
+    """
+    Functions for finding crypto info
+    """
+
+    vs_currency = "usd"  # simple/supported_vs_currencies for list of options
+
+    searched_symbols = {}
+
+    def __init__(self) -> None:
+        """Creates a Symbol Object
+
+        Parameters
+        ----------
+        IEX_TOKEN : str
+            IEX Token
+        """
+        self.get_symbol_list()
+        schedule.every().day.do(self.get_symbol_list)
+
+    def symbol_id(self, symbol) -> str:
+        try:
+            return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[
+                0
+            ]
+        except KeyError:
+            return ""
+
+    def get_symbol_list(
+        self, return_df=False
+    ) -> Optional[Tuple[pd.DataFrame, datetime]]:
+
+        raw_symbols = r.get(
+            "https://api.coingecko.com/api/v3/coins/list",
+            timeout=5,
+        ).json()
+        symbols = pd.DataFrame(data=raw_symbols)
+
+        symbols["description"] = (
+            "$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
+        )
+        symbols = symbols[["id", "symbol", "name", "description"]]
+        symbols["type_id"] = "$$" + symbols["id"]
+
+        self.symbol_list = symbols
+        if return_df:
+            return symbols, datetime.now()
+
+    def status(self) -> str:
+        """Checks CoinGecko /ping endpoint for API issues.
+
+        Returns
+        -------
+        str
+            Human readable text on status of CoinGecko API
+        """
+        status = r.get(
+            "https://api.coingecko.com/api/v3/ping",
+            timeout=5,
+        )
+
+        if status.status_code == 200:
+            return f"CoinGecko API responded that it was OK in {status.elapsed.total_seconds()} Seconds."
+        else:
+            return f"CoinGecko API returned an error in {status.elapsed.total_seconds()} Seconds."
+
+    def search_symbols(self, search: str) -> List[Tuple[str, str]]:
+        """Performs a fuzzy search to find coin symbols closest to a search term.
+
+        Parameters
+        ----------
+        search : str
+            String used to search, could be a company name or something close to the companies coin ticker.
+
+        Returns
+        -------
+        List[tuple[str, str]]
+            A list tuples of every coin sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
+        """
+        schedule.run_pending()
+        search = search.lower()
+        try:  # https://stackoverflow.com/a/3845776/8774114
+            return self.searched_symbols[search]
+        except KeyError:
+            pass
+
+        symbols = self.symbol_list
+        symbols["Match"] = symbols.apply(
+            lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
+            axis=1,
+        )
+
+        symbols.sort_values(by="Match", ascending=False, inplace=True)
+        if symbols["Match"].head().sum() < 300:
+            symbols["Match"] = symbols.apply(
+                lambda x: fuzz.partial_ratio(search, x["name"].lower()),
+                axis=1,
+            )
+
+            symbols.sort_values(by="Match", ascending=False, inplace=True)
+        symbols = symbols.head(10)
+        symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
+        self.searched_symbols[search] = symbol_list
+        return symbol_list
+
+    def price_reply(self, coin: Coin) -> str:
+        """Returns current market price or after hours if its available for a given coin symbol.
+
+        Parameters
+        ----------
+        symbols : list
+            List of coin symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+            markdown formatted string of the symbols price and movement.
+        """
+
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/simple/price?ids={coin.id}&vs_currencies={self.vs_currency}&include_24hr_change=true",
+            timeout=5,
+        )
+        if response.status_code == 200:
+
+            try:
+                data = response.json()[coin.id]
+
+                price = data[self.vs_currency]
+                change = data[self.vs_currency + "_24h_change"]
+                if change is None:
+                    change = 0
+            except KeyError:
+                return f"{coin.id} returned an error."
+
+            message = f"The current price of {coin.name} is $**{price:,}**"
+
+            # Determine wording of change text
+            if change > 0:
+                message += f", the coin is currently **up {change:.3f}%** for today"
+            elif change < 0:
+                message += f", the coin is currently **down {change:.3f}%** for today"
+            else:
+                message += ", the coin hasn't shown any movement today."
+
+        else:
+            message = f"The Coin: {coin.name} was not found."
+
+        return message
+
+    def intra_reply(self, symbol: Coin) -> pd.DataFrame:
+        """Returns price data for a symbol since the last market open.
+
+        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.
+        """
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/coins/{symbol.id}/ohlc?vs_currency=usd&days=1",
+            timeout=5,
+        )
+        if response.status_code == 200:
+            df = pd.DataFrame(
+                response.json(), columns=["Date", "Open", "High", "Low", "Close"]
+            ).dropna()
+            df["Date"] = pd.to_datetime(df["Date"], unit="ms")
+            df = df.set_index("Date")
+            return df
+
+        return pd.DataFrame()
+
+    def chart_reply(self, symbol: Coin) -> 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.
+        """
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/coins/{symbol.id}/ohlc?vs_currency=usd&days=30",
+            timeout=5,
+        )
+
+        if response.status_code == 200:
+            df = pd.DataFrame(
+                response.json(), columns=["Date", "Open", "High", "Low", "Close"]
+            ).dropna()
+            df["Date"] = pd.to_datetime(df["Date"], unit="ms")
+            df = df.set_index("Date")
+            return df
+
+        return pd.DataFrame()
+
+    def stat_reply(self, symbol: Coin) -> str:
+        """Gathers key statistics on coin. Mostly just CoinGecko scores.
+
+        Parameters
+        ----------
+        symbol : Coin
+
+        Returns
+        -------
+        str
+            Preformatted markdown.
+        """
+
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/coins/{symbol.id}?localization=false",
+            timeout=5,
+        )
+        if response.status_code == 200:
+            data = response.json()
+
+            return f"""
+                [{data['name']}]({data['links']['homepage'][0]}) Statistics:
+                Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,}
+                Market Cap Ranking: {data.get('market_cap_rank',"Not Available")}
+                CoinGecko Scores:
+                    Overall: {data.get('coingecko_score','Not Available')}
+                    Development: {data.get('developer_score','Not Available')}
+                    Community: {data.get('community_score','Not Available')}
+                    Public Interest: {data.get('public_interest_score','Not Available')}
+                    """
+        else:
+            return f"{symbol.symbol} returned an error."
+
+    def cap_reply(self, coin: Coin) -> str:
+        """Gets market cap for Coin
+
+        Parameters
+        ----------
+        coin : Coin
+
+        Returns
+        -------
+        str
+            Preformatted markdown.
+        """
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/simple/price?ids={coin.id}&vs_currencies={self.vs_currency}&include_market_cap=true",
+            timeout=5,
+        )
+        if response.status_code == 200:
+
+            try:
+                data = response.json()[coin.id]
+
+                price = data[self.vs_currency]
+                cap = data[self.vs_currency + "_market_cap"]
+            except KeyError:
+                return f"{coin.id} returned an error."
+
+            if cap == 0:
+                return f"The market cap for {coin.name} is not available for unknown reasons."
+
+            message = f"The current price of {coin.name} is $**{price:,}** and its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}"
+
+        else:
+            message = f"The Coin: {coin.name} was not found or returned and error."
+
+        return message
+
+    def info_reply(self, symbol: Coin) -> str:
+        """Gets coin description
+
+        Parameters
+        ----------
+        symbol : Coin
+
+        Returns
+        -------
+        str
+            Preformatted markdown.
+        """
+
+        response = r.get(
+            f"https://api.coingecko.com/api/v3/coins/{symbol.id}?localization=false",
+            timeout=5,
+        )
+        if response.status_code == 200:
+            data = response.json()
+            try:
+                return markdownify(data["description"]["en"])
+            except KeyError:
+                return f"{symbol} does not have a description available."
+
+        return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
+
+    def trending(self) -> list[str]:
+        """Gets current coins trending on coingecko
+
+        Returns
+        -------
+        list[str]
+            list of $$ID: NAME, CHANGE%
+        """
+
+        coins = r.get(
+            "https://api.coingecko.com/api/v3/search/trending",
+            timeout=5,
+        )
+        try:
+            trending = []
+            if coins.status_code == 200:
+                for coin in coins.json()["coins"]:
+                    c = coin["item"]
+
+                    sym = c["symbol"].upper()
+                    name = c["name"]
+                    change = r.get(
+                        f"https://api.coingecko.com/api/v3/simple/price?ids={c['id']}&vs_currencies={self.vs_currency}&include_24hr_change=true"
+                    ).json()[c["id"]]["usd_24h_change"]
+
+                    msg = f"`$${sym}`: {name}, {change:.2f}%"
+
+                    trending.append(msg)
+
+        except Exception as e:
+            print(e)
+            trending = ["Trending Coins Currently Unavailable."]
+
+        return trending
+
+    def batch_price(self, coins: list[Coin]) -> list[str]:
+        """Gets price of a list of coins all in one API call
+
+        Parameters
+        ----------
+        coins : list[Coin]
+
+        Returns
+        -------
+        list[str]
+            returns preformatted list of strings detailing price movement of each coin passed in.
+        """
+        query = ",".join([c.id for c in coins])
+
+        prices = r.get(
+            f"https://api.coingecko.com/api/v3/simple/price?ids={query}&vs_currencies=usd&include_24hr_change=true",
+            timeout=5,
+        ).json()
+
+        replies = []
+        for coin in coins:
+            if coin.id in prices:
+                p = prices[coin.id]
+
+                if p.get("usd_24h_change") is None:
+                    p["usd_24h_change"] = 0
+
+                replies.append(
+                    f"{coin.name}: ${p.get('usd',0):,} and has moved {p.get('usd_24h_change',0.0):.2f}% in the past 24 hours."
+                )
+
+        return replies

requirements.txt

@@ -1,7 +1,8 @@
-discord.py==1.6
+discord.py==1.7.3
 requests==2.25.1
 pandas==1.2.1
 fuzzywuzzy==0.18.0
 python-Levenshtein==0.12.1
 schedule==1.0.0
 mplfinance==0.12.7a5
+markdownify==0.6.5

symbol_router.py

@@ -0,0 +1,421 @@
+"""Function that routes symbols to the correct API provider.
+"""
+
+import datetime
+import random
+import re
+from logging import critical, debug, error, info, warning
+
+import pandas as pd
+from fuzzywuzzy import fuzz
+
+from cg_Crypto import cg_Crypto
+from IEX_Symbol import IEX_Symbol
+from Symbol import Coin, Stock, Symbol
+
+
+class Router:
+    STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z.]{1,6})"
+    CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,20})"
+    searched_symbols = {}
+
+    def __init__(self):
+        self.stock = IEX_Symbol()
+        self.crypto = cg_Crypto()
+
+    def find_symbols(self, text: str) -> list[Symbol]:
+        """Finds stock tickers starting with a dollar sign, and cryptocurrencies with two dollar signs
+        in a blob of text and returns them in a list.
+
+        Parameters
+        ----------
+        text : str
+            Blob of text.
+
+        Returns
+        -------
+        list[str]
+            List of stock symbols as strings without dollar sign.
+        """
+        symbols = []
+        stocks = set(re.findall(self.STOCK_REGEX, text))
+        for stock in stocks:
+            if stock.upper() in self.stock.symbol_list["symbol"].values:
+                symbols.append(Stock(stock))
+            else:
+                info(f"{stock} is not in list of stocks")
+
+        coins = set(re.findall(self.CRYPTO_REGEX, text))
+        for coin in coins:
+            if coin.lower() in self.crypto.symbol_list["symbol"].values:
+                symbols.append(Coin(coin.lower()))
+            else:
+                info(f"{coin} is not in list of coins")
+        info(symbols)
+        return symbols
+
+    def status(self, bot_resp) -> str:
+        """Checks for any issues with APIs.
+
+        Returns
+        -------
+        str
+            Human readable text on status of the bot and relevant APIs
+        """
+
+        stats = f"""
+        Bot Status:
+        {bot_resp}
+
+        Stock Market Data:
+        {self.stock.status()}
+
+        Cryptocurrency Data:
+        {self.crypto.status()}
+        """
+
+        warning(stats)
+
+        return stats
+
+    def search_symbols(self, search: str) -> list[tuple[str, str]]:
+        """Performs a fuzzy search to find stock symbols closest to a search term.
+
+        Parameters
+        ----------
+        search : str
+            String used to search, could be a company name or something close to the companies stock ticker.
+
+        Returns
+        -------
+        list[tuple[str, str]]
+            A list tuples of every stock sorted in order of how well they match.
+                Each tuple contains: (Symbol, Issue Name).
+        """
+
+        df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
+
+        search = search.lower()
+
+        df["Match"] = df.apply(
+            lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
+            axis=1,
+        )
+
+        df.sort_values(by="Match", ascending=False, inplace=True)
+        # if df["Match"].head().sum() < 300:
+        #     df["Match"] = df.apply(
+        #         lambda x: fuzz.partial_ratio(search, x["name"].lower()),
+        #         axis=1,
+        #     )
+
+        #     df.sort_values(by="Match", ascending=False, inplace=True)
+
+        symbols = df.head(20)
+        symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
+        self.searched_symbols[search] = symbol_list
+        return symbol_list
+
+    def inline_search(self, search: str) -> list[tuple[str, str]]:
+        """Searches based on the shortest symbol that contains the same string as the search.
+        Should be very fast compared to a fuzzy search.
+
+        Parameters
+        ----------
+        search : str
+            String used to match against symbols.
+
+        Returns
+        -------
+        list[tuple[str, str]]
+            Each tuple contains: (Symbol, Issue Name).
+        """
+
+        df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
+
+        search = search.lower()
+
+        df = df[df["type_id"].str.contains(search, regex=False)].sort_values(
+            by="type_id", key=lambda x: x.str.len()
+        )
+
+        symbols = df.head(20)
+        symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
+        self.searched_symbols[search] = symbol_list
+        return symbol_list
+
+    def price_reply(self, symbols: list[Symbol]) -> list[str]:
+        """Returns current market price or after hours if its available for a given stock symbol.
+
+        Parameters
+        ----------
+        symbols : list
+            List of stock symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+            markdown formatted string of the symbols price and movement.
+        """
+        replies = []
+
+        for symbol in symbols:
+            info(symbol)
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.price_reply(symbol))
+            elif isinstance(symbol, Coin):
+                replies.append(self.crypto.price_reply(symbol))
+            else:
+                info(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def dividend_reply(self, symbols: list) -> list[str]:
+        """Returns the most recent, or next dividend date for a stock symbol.
+
+        Parameters
+        ----------
+        symbols : list
+            List of stock symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+                formatted string of the symbols div dates.
+        """
+        replies = []
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.dividend_reply(symbol))
+            elif isinstance(symbol, Coin):
+                replies.append("Cryptocurrencies do no have Dividends.")
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def news_reply(self, symbols: list) -> list[str]:
+        """Gets recent english news on stock symbols.
+
+        Parameters
+        ----------
+        symbols : list
+            List of stock symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human
+                readable markdown formatted string of the symbols news.
+        """
+        replies = []
+
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.news_reply(symbol))
+            elif isinstance(symbol, Coin):
+                # replies.append(self.crypto.news_reply(symbol))
+                replies.append(
+                    "News is not yet supported for cryptocurrencies. If you have any suggestions for news sources please contatct @MisterBiggs"
+                )
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def info_reply(self, symbols: list) -> list[str]:
+        """Gets information on stock symbols.
+
+        Parameters
+        ----------
+        symbols : list[str]
+            List of stock symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable formatted
+                string of the symbols information.
+        """
+        replies = []
+
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.info_reply(symbol))
+            elif isinstance(symbol, Coin):
+                replies.append(self.crypto.info_reply(symbol))
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def intra_reply(self, symbol: Symbol) -> pd.DataFrame:
+        """Returns price data for a symbol since the last market open.
+
+        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.
+        """
+
+        if isinstance(symbol, Stock):
+            return self.stock.intra_reply(symbol)
+        elif isinstance(symbol, Coin):
+            return self.crypto.intra_reply(symbol)
+        else:
+            print(f"{symbol} is not a Stock or Coin")
+            return pd.DataFrame()
+
+    def chart_reply(self, symbol: Symbol) -> 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.
+        """
+        if isinstance(symbol, Stock):
+            return self.stock.chart_reply(symbol)
+        elif isinstance(symbol, Coin):
+            return self.crypto.chart_reply(symbol)
+        else:
+            print(f"{symbol} is not a Stock or Coin")
+            return pd.DataFrame()
+
+    def stat_reply(self, symbols: list[Symbol]) -> list[str]:
+        """Gets key statistics for each symbol in the list
+
+        Parameters
+        ----------
+        symbols : list[str]
+            List of stock symbols
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+                formatted string of the symbols statistics.
+        """
+        replies = []
+
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.stat_reply(symbol))
+            elif isinstance(symbol, Coin):
+                replies.append(self.crypto.stat_reply(symbol))
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def cap_reply(self, symbols: list[Symbol]) -> list[str]:
+        """Gets market cap for each symbol in the list
+
+        Parameters
+        ----------
+        symbols : list[str]
+            List of stock symbols
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+                formatted string of the symbols market cap.
+        """
+        replies = []
+
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                replies.append(self.stock.cap_reply(symbol))
+            elif isinstance(symbol, Coin):
+                replies.append(self.crypto.cap_reply(symbol))
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        return replies
+
+    def trending(self) -> str:
+        """Checks APIs for trending symbols.
+
+        Returns
+        -------
+        list[str]
+            List of preformatted strings to be sent to user.
+        """
+
+        stocks = self.stock.trending()
+        coins = self.crypto.trending()
+
+        reply = "Trending Stocks:\n"
+        reply += "-" * len("Trending Stocks:") + "\n"
+        for stock in stocks:
+            reply += stock + "\n"
+
+        reply += "\n\nTrending Crypto:\n"
+        reply += "-" * len("Trending Crypto:") + "\n"
+        for coin in coins:
+            reply += coin + "\n"
+
+        return reply
+
+    def random_pick(self) -> str:
+
+        choice = random.choice(
+            list(self.stock.symbol_list["description"])
+            + list(self.crypto.symbol_list["description"])
+        )
+        hold = (
+            datetime.date.today() + datetime.timedelta(random.randint(1, 365))
+        ).strftime("%b %d, %Y")
+
+        return f"{choice}\nBuy and hold until: {hold}"
+
+    def batch_price_reply(self, symbols: list[Symbol]) -> list[str]:
+        """Returns current market price or after hours if its available for a given stock symbol.
+
+        Parameters
+        ----------
+        symbols : list
+            List of stock symbols.
+
+        Returns
+        -------
+        Dict[str, str]
+            Each symbol passed in is a key with its value being a human readable
+            markdown formatted string of the symbols price and movement.
+        """
+        replies = []
+        stocks = []
+        coins = []
+
+        for symbol in symbols:
+            if isinstance(symbol, Stock):
+                stocks.append(symbol)
+            elif isinstance(symbol, Coin):
+                coins.append(symbol)
+            else:
+                print(f"{symbol} is not a Stock or Coin")
+
+        if stocks:
+            # IEX batch endpoint doesnt seem to be working right now
+            for stock in stocks:
+                replies.append(self.stock.price_reply(stock))
+        if coins:
+            replies = replies + self.crypto.batch_price(coins)
+
+        return replies