From 0bac96cbd77cbe9d1ee5cab511ea7fb8991fe3d7 Mon Sep 17 00:00:00 2001 From: Anson Date: Thu, 26 Aug 2021 10:26:14 -0700 Subject: [PATCH 01/13] Closes #73 --- bot.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bot.py b/bot.py index 169911c..f21a789 100644 --- a/bot.py +++ b/bot.py @@ -116,12 +116,12 @@ def donate(update: Update, context: CallbackContext): except ValueError: update.message.reply_text(f"{amount} is not a valid donation amount or number.") return - info(f"Donation amount: {price}") + info(f"Donation amount: {price} by {update.message.chat.username}") context.bot.send_invoice( chat_id=chat_id, title="Simple Stock Bot Donation", - description=f"Simple Stock Bot Donation of ${amount}", + description=f"Simple Stock Bot Donation of ${amount} by {update.message.chat.username}", payload=f"simple-stock-bot-{chat_id}", provider_token=STRIPE_TOKEN, currency="USD", From f34fd3759751d33b5b853a3c9a52b5b4086de1fa Mon Sep 17 00:00:00 2001 From: Anson Date: Thu, 26 Aug 2021 17:46:13 +0000 Subject: [PATCH 02/13] /close #74 --- IEX_Symbol.py | 1005 +++++++++++++++++++++++++------------------------ cg_Crypto.py | 4 +- 2 files changed, 506 insertions(+), 503 deletions(-) diff --git a/IEX_Symbol.py b/IEX_Symbol.py index 3eff964..f4940df 100644 --- a/IEX_Symbol.py +++ b/IEX_Symbol.py @@ -1,502 +1,503 @@ -"""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 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 get(self, endpoint, params: dict = {}, timeout=5) -> dict: - - url = "https://cloud.iexapis.com/stable" + endpoint - - # set token param if it wasn't passed. - params["token"] = params.get("token", self.IEX_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() - return resp_json - 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]) - - 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=15, - ) - - 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 - """ - - if IEXData := self.get(f"/stock/{symbol.id}/quote"): - - 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." - - if resp := self.get(f"/stock/{symbol.id}/dividends/next"): - try: - IEXData = resp[0] - except IndexError as e: - return f"${symbol.id.upper()} either doesn't exist or pays no dividend." - 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." - - if data := self.get(f"/stock/{symbol.id}/news/last/15"): - line = [] - - for news in data: - if news["lang"] == "en" and not news["hasPaywall"]: - line.append( - f"*{news['source']}*: [{news['headline']}]({news['url']})" - ) - - return f"News for **{symbol.id.upper()}**:\n" + "\n".join(line[:5]) - - else: - return f"No news found for: {symbol.id}\nEither today is boring or the symbol does not exist." - - 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." - - if data := self.get(f"/stock/{symbol.id}/company"): - [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." - - if data := self.get(f"/stock/{symbol.id}/stats"): - [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, symbol: Stock) -> str: - """Get the Market Cap of a stock""" - - if data := self.get(f"/stable/stock/{symbol.id}/stats"): - - try: - cap = data["marketcap"] - except KeyError: - return f"{symbol.id} returned an error." - - message = f"The current market cap of {symbol.name} is $**{cap:,.2f}**" - - else: - message = f"The Stock: {symbol.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() - - if data := self.get(f"/stock/{symbol.id}/intraday-prices"): - df = pd.DataFrame(data) - 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 - - if data := self.get( - f"/stock/{symbol.id}/chart/1mm", - params={"chartInterval": 3, "includeToday": "false"}, - ): - df = pd.DataFrame(data) - 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% - """ - - if data := self.get(f"/stock/market/list/mostactive"): - return [ - f"`${s['symbol']}`: {s['companyName']}, {100*s['changePercent']:.2f}%" - for s in data - ] - else: - return ["Trending Stocks Currently Unavailable."] +"""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 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 = {} + trending_cache = ["Trending Stocks Currently Unavailable."] + + 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 get(self, endpoint, params: dict = {}, timeout=5) -> dict: + + url = "https://cloud.iexapis.com/stable" + endpoint + + # set token param if it wasn't passed. + params["token"] = params.get("token", self.IEX_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() + return resp_json + 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]) + + 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=15, + ) + + 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 + """ + + if IEXData := self.get(f"/stock/{symbol.id}/quote"): + + 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." + + if resp := self.get(f"/stock/{symbol.id}/dividends/next"): + try: + IEXData = resp[0] + except IndexError as e: + return f"${symbol.id.upper()} either doesn't exist or pays no dividend." + 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." + + if data := self.get(f"/stock/{symbol.id}/news/last/15"): + line = [] + + for news in data: + if news["lang"] == "en" and not news["hasPaywall"]: + line.append( + f"*{news['source']}*: [{news['headline']}]({news['url']})" + ) + + return f"News for **{symbol.id.upper()}**:\n" + "\n".join(line[:5]) + + else: + return f"No news found for: {symbol.id}\nEither today is boring or the symbol does not exist." + + 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." + + if data := self.get(f"/stock/{symbol.id}/company"): + [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." + + if data := self.get(f"/stock/{symbol.id}/stats"): + [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, symbol: Stock) -> str: + """Get the Market Cap of a stock""" + + if data := self.get(f"/stable/stock/{symbol.id}/stats"): + + try: + cap = data["marketcap"] + except KeyError: + return f"{symbol.id} returned an error." + + message = f"The current market cap of {symbol.name} is $**{cap:,.2f}**" + + else: + message = f"The Stock: {symbol.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() + + if data := self.get(f"/stock/{symbol.id}/intraday-prices"): + df = pd.DataFrame(data) + 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 + + if data := self.get( + f"/stock/{symbol.id}/chart/1mm", + params={"chartInterval": 3, "includeToday": "false"}, + ): + df = pd.DataFrame(data) + 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% + """ + + if data := self.get(f"/stock/market/list/mostactive"): + self.trending_cache = [ + f"`${s['symbol']}`: {s['companyName']}, {100*s['changePercent']:.2f}%" + for s in data + ] + + return self.trending_cache diff --git a/cg_Crypto.py b/cg_Crypto.py index 17de1e3..a511a4f 100644 --- a/cg_Crypto.py +++ b/cg_Crypto.py @@ -22,6 +22,7 @@ class cg_Crypto: vs_currency = "usd" # simple/supported_vs_currencies for list of options searched_symbols = {} + trending_cache = ["Trending Coins Currently Unavailable."] def __init__(self) -> None: """Creates a Symbol Object @@ -368,8 +369,9 @@ class cg_Crypto: except Exception as e: logging.warning(e) - trending = ["Trending Coins Currently Unavailable."] + return self.trending_cache + self.trending_cache = trending return trending def batch_price(self, coins: list[Coin]) -> list[str]: From 082c42c39ff59da903ac43e594ec0049929b11bc Mon Sep 17 00:00:00 2001 From: Anson Date: Thu, 26 Aug 2021 19:12:38 +0000 Subject: [PATCH 03/13] Bot now keeps track of symbols for own trending --- Symbol.py | 4 ++++ symbol_router.py | 21 +++++++++++++++++++-- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/Symbol.py b/Symbol.py index db04b4a..e0629a3 100644 --- a/Symbol.py +++ b/Symbol.py @@ -8,6 +8,7 @@ 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 + tag: Uppercase tag to call the symbol. ie $TSLA or $$BTC """ currency = "usd" @@ -17,6 +18,7 @@ class Symbol: self.symbol = symbol self.id = symbol self.name = symbol + self.tag = "$" + symbol def __repr__(self) -> str: return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>" @@ -32,6 +34,7 @@ class Stock(Symbol): self.symbol = symbol self.id = symbol self.name = "$" + symbol.upper() + self.tag = "$" + symbol.upper() # Used by Coin to change symbols for ids @@ -44,6 +47,7 @@ class Coin(Symbol): @functools.cache def __init__(self, symbol: str) -> None: self.symbol = symbol + self.tag = "$$" + symbol.upper() self.get_data() def get_data(self) -> None: diff --git a/symbol_router.py b/symbol_router.py index 047a2f2..0560e7f 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -18,6 +18,7 @@ class Router: STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z.]{1,6})" CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,20})" searched_symbols = {} + trending_count = {} def __init__(self): self.stock = IEX_Symbol() @@ -54,8 +55,12 @@ class Router: if symbols: info(symbols) + for symbol in symbols: + self.trending_count[symbol.tag] = ( + self.trending_count.get(symbol.tag, 0) + 1 + ) - return symbols + return symbols def status(self, bot_resp) -> str: """Checks for any issues with APIs. @@ -364,7 +369,19 @@ class Router: stocks = self.stock.trending() coins = self.crypto.trending() - reply = "Trending Stocks:\n" + reply = "" + + reply += "Trending on the Stock Bot:\n" + reply += "-" * len("Trending on the Stock Bot:") + "\n" + + sorted_trending = [ + s[0] for s in sorted(self.trending_count.items(), key=lambda item: item[1]) + ][::-1][0:5] + + for t in sorted_trending: + reply += self.price_reply(self.find_symbols(t))[0] + "\n" + + reply += "\n\nTrending Stocks:\n" reply += "-" * len("Trending Stocks:") + "\n" for stock in stocks: reply += stock + "\n" From 42bb17dd4051295f03b065ad85af984e6647391d Mon Sep 17 00:00:00 2001 From: Anson Date: Thu, 26 Aug 2021 19:13:51 +0000 Subject: [PATCH 04/13] #18 --- symbol_router.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/symbol_router.py b/symbol_router.py index 0560e7f..5da0a0d 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -200,7 +200,7 @@ class Router: elif isinstance(symbol, Coin): replies.append("Cryptocurrencies do no have Dividends.") else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return replies @@ -229,7 +229,7 @@ class Router: "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") + debug(f"{symbol} is not a Stock or Coin") return replies @@ -255,7 +255,7 @@ class Router: elif isinstance(symbol, Coin): replies.append(self.crypto.info_reply(symbol)) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return replies @@ -279,7 +279,7 @@ class Router: elif isinstance(symbol, Coin): return self.crypto.intra_reply(symbol) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return pd.DataFrame() def chart_reply(self, symbol: Symbol) -> pd.DataFrame: @@ -302,7 +302,7 @@ class Router: elif isinstance(symbol, Coin): return self.crypto.chart_reply(symbol) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return pd.DataFrame() def stat_reply(self, symbols: list[Symbol]) -> list[str]: @@ -327,7 +327,7 @@ class Router: elif isinstance(symbol, Coin): replies.append(self.crypto.stat_reply(symbol)) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return replies @@ -353,7 +353,7 @@ class Router: elif isinstance(symbol, Coin): replies.append(self.crypto.cap_reply(symbol)) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") return replies @@ -429,7 +429,7 @@ class Router: elif isinstance(symbol, Coin): coins.append(symbol) else: - print(f"{symbol} is not a Stock or Coin") + debug(f"{symbol} is not a Stock or Coin") if stocks: # IEX batch endpoint doesnt seem to be working right now From e56679f29f66ecdc2c88d9c8d898fd4afa9a1691 Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Mon, 30 Aug 2021 12:52:46 -0700 Subject: [PATCH 05/13] more idiomatic setup for Dockerfile --- DockerDev => .devcontainer/Dockerfile | 0 .devcontainer/devcontainer.json | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) rename DockerDev => .devcontainer/Dockerfile (100%) diff --git a/DockerDev b/.devcontainer/Dockerfile similarity index 100% rename from DockerDev rename to .devcontainer/Dockerfile diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 1587f43..efc4208 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -5,7 +5,7 @@ // Sets the run context to one level up instead of the .devcontainer folder. "context": "..", // Update the 'dockerFile' property if you aren't using the standard 'Dockerfile' filename. - "dockerFile": "../DockerDev", + "dockerFile": "Dockerfile", // Set *default* container specific settings.json values on container create. "settings": {}, // Add the IDs of extensions you want installed when the container is created. From 99fdd6e8beb282f731bf094387cfce8815b9d56d Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Mon, 30 Aug 2021 13:21:44 -0700 Subject: [PATCH 06/13] documentation fix --- symbol_router.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/symbol_router.py b/symbol_router.py index 5da0a0d..ca64eb8 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -35,8 +35,8 @@ class Router: Returns ------- - list[str] - List of stock symbols as strings without dollar sign. + list[Symbol] + List of stock symbols as Symbol objects """ symbols = [] stocks = set(re.findall(self.STOCK_REGEX, text)) From c7f083b2b1bff6f42b71fa26e763a3dcde3afdc7 Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Mon, 30 Aug 2021 13:21:56 -0700 Subject: [PATCH 07/13] trending stocks now decay #75 --- symbol_router.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/symbol_router.py b/symbol_router.py index ca64eb8..b926b24 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -7,6 +7,7 @@ import re from logging import critical, debug, error, info, warning import pandas as pd +import schedule from fuzzywuzzy import fuzz from cg_Crypto import cg_Crypto @@ -24,6 +25,20 @@ class Router: self.stock = IEX_Symbol() self.crypto = cg_Crypto() + schedule.every().hour().do(self.trending_decay) + + def trending_decay(self, decay=0.5): + """Decays the value of each trending stock by a multiplier""" + + info("Decaying trending symbols.") + if self.trending_count: + for key in self.trending_count.keys(): + if self.trending_count[key] < 0.01: + # This just makes sure were not keeping around keys that havent been called in a very long time. + self.trending_count.pop(key, None) + else: + self.trending_count[key] = self.trending_count[key] * decay + 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. @@ -38,6 +53,8 @@ class Router: list[Symbol] List of stock symbols as Symbol objects """ + schedule.run_pending() + symbols = [] stocks = set(re.findall(self.STOCK_REGEX, text)) for stock in stocks: From 0649789bf81957905287824b0ed9a8e73400ff30 Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Mon, 30 Aug 2021 20:35:04 +0000 Subject: [PATCH 08/13] bug in decay schedule #75 --- symbol_router.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/symbol_router.py b/symbol_router.py index b926b24..641bcc9 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -25,7 +25,7 @@ class Router: self.stock = IEX_Symbol() self.crypto = cg_Crypto() - schedule.every().hour().do(self.trending_decay) + schedule.every().hour.do(self.trending_decay) def trending_decay(self, decay=0.5): """Decays the value of each trending stock by a multiplier""" From 0877949a8945fdacb3c48058da1215ac955eba30 Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Mon, 30 Aug 2021 13:46:12 -0700 Subject: [PATCH 09/13] trending only returns if data was retrieved. #75 --- IEX_Symbol.py | 2 +- cg_Crypto.py | 2 +- symbol_router.py | 922 ++++++++++++++++++++++++----------------------- 3 files changed, 467 insertions(+), 459 deletions(-) diff --git a/IEX_Symbol.py b/IEX_Symbol.py index f4940df..62a680f 100644 --- a/IEX_Symbol.py +++ b/IEX_Symbol.py @@ -25,7 +25,7 @@ class IEX_Symbol: searched_symbols = {} otc_list = [] charts = {} - trending_cache = ["Trending Stocks Currently Unavailable."] + trending_cache = None def __init__(self) -> None: """Creates a Symbol Object diff --git a/cg_Crypto.py b/cg_Crypto.py index a511a4f..d4c0d82 100644 --- a/cg_Crypto.py +++ b/cg_Crypto.py @@ -22,7 +22,7 @@ class cg_Crypto: vs_currency = "usd" # simple/supported_vs_currencies for list of options searched_symbols = {} - trending_cache = ["Trending Coins Currently Unavailable."] + trending_cache = None def __init__(self) -> None: """Creates a Symbol Object diff --git a/symbol_router.py b/symbol_router.py index 641bcc9..66cb8be 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -1,458 +1,466 @@ -"""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 -import schedule -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 = {} - trending_count = {} - - def __init__(self): - self.stock = IEX_Symbol() - self.crypto = cg_Crypto() - +"""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 +import schedule +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 = {} + trending_count = {} + + def __init__(self): + self.stock = IEX_Symbol() + self.crypto = cg_Crypto() + schedule.every().hour.do(self.trending_decay) - - def trending_decay(self, decay=0.5): - """Decays the value of each trending stock by a multiplier""" - - info("Decaying trending symbols.") - if self.trending_count: - for key in self.trending_count.keys(): - if self.trending_count[key] < 0.01: - # This just makes sure were not keeping around keys that havent been called in a very long time. - self.trending_count.pop(key, None) - else: - self.trending_count[key] = self.trending_count[key] * decay - - 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[Symbol] - List of stock symbols as Symbol objects - """ - schedule.run_pending() - - 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") - - if symbols: - info(symbols) - for symbol in symbols: - self.trending_count[symbol.tag] = ( - self.trending_count.get(symbol.tag, 0) + 1 - ) - - 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: - debug(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: - debug(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: - debug(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: - debug(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: - debug(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: - debug(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: - debug(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 = "" - - reply += "Trending on the Stock Bot:\n" - reply += "-" * len("Trending on the Stock Bot:") + "\n" - - sorted_trending = [ - s[0] for s in sorted(self.trending_count.items(), key=lambda item: item[1]) - ][::-1][0:5] - - for t in sorted_trending: - reply += self.price_reply(self.find_symbols(t))[0] + "\n" - - reply += "\n\nTrending 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: - debug(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 + + def trending_decay(self, decay=0.5): + """Decays the value of each trending stock by a multiplier""" + + info("Decaying trending symbols.") + if self.trending_count: + for key in self.trending_count.keys(): + if self.trending_count[key] < 0.01: + # This just makes sure were not keeping around keys that havent been called in a very long time. + self.trending_count.pop(key, None) + else: + self.trending_count[key] = self.trending_count[key] * decay + + 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[Symbol] + List of stock symbols as Symbol objects + """ + schedule.run_pending() + + 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") + + if symbols: + info(symbols) + for symbol in symbols: + self.trending_count[symbol.tag] = ( + self.trending_count.get(symbol.tag, 0) + 1 + ) + + 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: + debug(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: + debug(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: + debug(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: + debug(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: + debug(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: + debug(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: + debug(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 = "" + + if self.trending_count: + reply += "šŸ”„Trending on the Stock Bot:\n`" + reply += "ā”" * len("Trending on the Stock Bot:") + "`\n" + + sorted_trending = [ + s[0] + for s in sorted(self.trending_count.items(), key=lambda item: item[1]) + ][::-1][0:5] + + for t in sorted_trending: + reply += self.price_reply(self.find_symbols(t))[0] + "\n" + + if stocks: + reply += "\n\nšŸ’µTrending Stocks:\n`" + reply += "ā”" * len("Trending Stocks:") + "`\n" + for stock in stocks: + reply += stock + "\n" + + if coins: + reply += "\n\nšŸ¦ŽTrending Crypto:\n`" + reply += "ā”" * len("Trending Crypto:") + "`\n" + for coin in coins: + reply += coin + "\n" + + if reply: + return reply + else: + warning("Failed to collect trending data.") + return "Trending data is not currently available." + + 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: + debug(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 From dcce2cb95a8ffcddcfe8c80976b5ef851990a91c Mon Sep 17 00:00:00 2001 From: Anson Date: Wed, 1 Sep 2021 01:51:18 +0000 Subject: [PATCH 10/13] #18 --- cg_Crypto.py | 825 ++++++++++++++++++++++++++------------------------- 1 file changed, 413 insertions(+), 412 deletions(-) diff --git a/cg_Crypto.py b/cg_Crypto.py index d4c0d82..00bcb75 100644 --- a/cg_Crypto.py +++ b/cg_Crypto.py @@ -1,412 +1,413 @@ -"""Class with functions for running the bot with IEX Cloud. -""" - -import logging -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 = {} - trending_cache = None - - 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 get(self, endpoint, params: dict = {}, timeout=10) -> dict: - - url = "https://api.coingecko.com/api/v3" + endpoint - 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() - return resp_json - except r.exceptions.JSONDecodeError as e: - logging.error(e) - return {} - - 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 = self.get("/coins/list") - 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, - ) - - try: - status.raise_for_status() - return f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds." - except: - return f"CoinGecko API returned an error code {status.status_code} 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. - """ - - if resp := self.get( - "/simple/price", - params={ - "ids": coin.id, - "vs_currencies": self.vs_currency, - "include_24hr_change": "true", - }, - ): - try: - data = resp[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 price for {coin.name} is not available. If you suspect this is an error run `/status`" - - 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. - """ - - if resp := self.get( - f"/coins/{symbol.id}/ohlc", - params={"vs_currency": self.vs_currency, "days": 1}, - ): - df = pd.DataFrame( - resp, 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. - """ - - if resp := self.get( - f"/coins/{symbol.id}/ohlc", - params={"vs_currency": self.vs_currency, "days": 30}, - ): - df = pd.DataFrame( - resp, 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. - """ - - if data := self.get( - f"/coins/{symbol.id}", - params={ - "localization": "false", - }, - ): - - 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. - """ - - if resp := self.get( - f"/simple/price", - params={ - "ids": coin.id, - "vs_currencies": self.vs_currency, - "include_market_cap": "true", - }, - ): - print(resp) - try: - data = resp[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. - """ - - if data := self.get( - f"/coins/{symbol.id}", - params={"localization": "false"}, - ): - 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 = self.get("/search/trending") - try: - trending = [] - for coin in coins["coins"]: - c = coin["item"] - - sym = c["symbol"].upper() - name = c["name"] - change = self.get( - f"/simple/price", - params={ - "ids": c["id"], - "vs_currencies": self.vs_currency, - "include_24hr_change": "true", - }, - )[c["id"]]["usd_24h_change"] - - msg = f"`$${sym}`: {name}, {change:.2f}%" - - trending.append(msg) - - except Exception as e: - logging.warning(e) - return self.trending_cache - - self.trending_cache = trending - 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 = self.get( - f"/simple/price", - params={ - "ids": query, - "vs_currencies": self.vs_currency, - "include_24hr_change": "true", - }, - ) - - 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 +"""Class with functions for running the bot with IEX Cloud. +""" + +import logging +from datetime import datetime +from logging import critical, debug, error, info, warning +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 = {} + trending_cache = None + + 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 get(self, endpoint, params: dict = {}, timeout=10) -> dict: + + url = "https://api.coingecko.com/api/v3" + endpoint + 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() + return resp_json + except r.exceptions.JSONDecodeError as e: + logging.error(e) + return {} + + 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 = self.get("/coins/list") + 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, + ) + + try: + status.raise_for_status() + return f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds." + except: + return f"CoinGecko API returned an error code {status.status_code} 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. + """ + + if resp := self.get( + "/simple/price", + params={ + "ids": coin.id, + "vs_currencies": self.vs_currency, + "include_24hr_change": "true", + }, + ): + try: + data = resp[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 price for {coin.name} is not available. If you suspect this is an error run `/status`" + + 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. + """ + + if resp := self.get( + f"/coins/{symbol.id}/ohlc", + params={"vs_currency": self.vs_currency, "days": 1}, + ): + df = pd.DataFrame( + resp, 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. + """ + + if resp := self.get( + f"/coins/{symbol.id}/ohlc", + params={"vs_currency": self.vs_currency, "days": 30}, + ): + df = pd.DataFrame( + resp, 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. + """ + + if data := self.get( + f"/coins/{symbol.id}", + params={ + "localization": "false", + }, + ): + + 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. + """ + + if resp := self.get( + f"/simple/price", + params={ + "ids": coin.id, + "vs_currencies": self.vs_currency, + "include_market_cap": "true", + }, + ): + debug(resp) + try: + data = resp[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. + """ + + if data := self.get( + f"/coins/{symbol.id}", + params={"localization": "false"}, + ): + 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 = self.get("/search/trending") + try: + trending = [] + for coin in coins["coins"]: + c = coin["item"] + + sym = c["symbol"].upper() + name = c["name"] + change = self.get( + f"/simple/price", + params={ + "ids": c["id"], + "vs_currencies": self.vs_currency, + "include_24hr_change": "true", + }, + )[c["id"]]["usd_24h_change"] + + msg = f"`$${sym}`: {name}, {change:.2f}%" + + trending.append(msg) + + except Exception as e: + logging.warning(e) + return self.trending_cache + + self.trending_cache = trending + 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 = self.get( + f"/simple/price", + params={ + "ids": query, + "vs_currencies": self.vs_currency, + "include_24hr_change": "true", + }, + ) + + 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 From a583cad741da13d81b4af4c7d772628c82c83954 Mon Sep 17 00:00:00 2001 From: Anson Date: Tue, 31 Aug 2021 20:03:59 -0700 Subject: [PATCH 11/13] Mostly wrapped up #75 --- IEX_Symbol.py | 19 ++++++++++++++++++- cg_Crypto.py | 12 ++++++++++++ symbol_router.py | 27 ++++++++++++++++++++++++++- 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/IEX_Symbol.py b/IEX_Symbol.py index 62a680f..b40c87b 100644 --- a/IEX_Symbol.py +++ b/IEX_Symbol.py @@ -36,7 +36,7 @@ class IEX_Symbol: IEX API Token """ try: - self.IEX_TOKEN = os.environ["IEX"] + self.IEX_TOKEN = "pk_3c39d940736e47dabfdd47eb689a65be" except KeyError: self.IEX_TOKEN = "" warning( @@ -485,6 +485,23 @@ class IEX_Symbol: return pd.DataFrame() + def spark_reply(self, symbol: Stock) -> str: + quote = self.get(f"/stock/{symbol.id}/quote") + + open_change = quote.get("changePercent", 0) + after_change = quote.get("extendedChangePercent", 0) + + change = 0 + + if open_change: + change = change + open_change + if after_change: + change = change + after_change + + change = change * 100 + + return f"`{symbol.tag}`: {quote['companyName']}, {change:.2f}%" + def trending(self) -> list[str]: """Gets current coins trending on IEX. Only returns when market is open. diff --git a/cg_Crypto.py b/cg_Crypto.py index 00bcb75..155475a 100644 --- a/cg_Crypto.py +++ b/cg_Crypto.py @@ -338,6 +338,18 @@ class cg_Crypto: return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." + def spark_reply(self, symbol: Coin) -> str: + change = self.get( + f"/simple/price", + params={ + "ids": symbol.id, + "vs_currencies": self.vs_currency, + "include_24hr_change": "true", + }, + )[symbol.id]["usd_24h_change"] + + return f"`{symbol.tag}`: {symbol.name}, {change:.2f}%" + def trending(self) -> list[str]: """Gets current coins trending on coingecko diff --git a/symbol_router.py b/symbol_router.py index 66cb8be..6af6969 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -374,6 +374,31 @@ class Router: return replies + def spark_reply(self, symbols: list[Symbol]) -> list[str]: + """Gets change for each symbol and returns it in a compact format + + Parameters + ---------- + symbols : list[str] + List of stock symbols + + Returns + ------- + list[str] + List of human readable strings. + """ + replies = [] + + for symbol in symbols: + if isinstance(symbol, Stock): + replies.append(self.stock.spark_reply(symbol)) + elif isinstance(symbol, Coin): + replies.append(self.crypto.spark_reply(symbol)) + else: + debug(f"{symbol} is not a Stock or Coin") + + return replies + def trending(self) -> str: """Checks APIs for trending symbols. @@ -398,7 +423,7 @@ class Router: ][::-1][0:5] for t in sorted_trending: - reply += self.price_reply(self.find_symbols(t))[0] + "\n" + reply += self.spark_reply(self.find_symbols(t))[0] + "\n" if stocks: reply += "\n\nšŸ’µTrending Stocks:\n`" From 6e21565f2b42ffeb12d1051ccc651443ec076ac8 Mon Sep 17 00:00:00 2001 From: Anson Date: Tue, 31 Aug 2021 20:04:44 -0700 Subject: [PATCH 12/13] #75 enough calls to justify making cmd async --- bot.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bot.py b/bot.py index f21a789..f2496f0 100644 --- a/bot.py +++ b/bot.py @@ -560,7 +560,7 @@ def main(): dp.add_handler(CommandHandler("stat", stat)) dp.add_handler(CommandHandler("stats", stat)) dp.add_handler(CommandHandler("cap", cap)) - dp.add_handler(CommandHandler("trending", trending)) + dp.add_handler(CommandHandler("trending", trending, run_async=True)) dp.add_handler(CommandHandler("search", search)) dp.add_handler(CommandHandler("random", rand_pick)) dp.add_handler(CommandHandler("donate", donate)) From 4f5c5585888e260397f823eb91b8651cf5ae063b Mon Sep 17 00:00:00 2001 From: Anson Date: Wed, 1 Sep 2021 03:11:58 +0000 Subject: [PATCH 13/13] small easteregg --- symbol_router.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/symbol_router.py b/symbol_router.py index 6af6969..cc1597e 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -437,6 +437,9 @@ class Router: for coin in coins: reply += coin + "\n" + if "`$GME" in reply: + reply = reply.replace("šŸ”„", "šŸ¦") + if reply: return reply else: