From 96321d7c07dce27ac0093fa6a83f8a51c39f27af Mon Sep 17 00:00:00 2001 From: Anson Date: Wed, 17 Feb 2021 21:09:03 -0700 Subject: [PATCH] mostly implemented router and IEX --- IEX_Symbol.py | 214 ++++++++++++++++++----------------------------- bot.py | 6 +- symbol_router.py | 212 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 296 insertions(+), 136 deletions(-) create mode 100644 symbol_router.py diff --git a/IEX_Symbol.py b/IEX_Symbol.py index c7b565f..69cc42b 100644 --- a/IEX_Symbol.py +++ b/IEX_Symbol.py @@ -151,7 +151,7 @@ class IEX_Symbol: return list(set(re.findall(self.SYMBOL_REGEX, text))) - def price_reply(self, symbols: list) -> Dict[str, str]: + def price_reply(self, symbol: str) -> str: """Returns current market price or after hours if its available for a given stock symbol. Parameters @@ -165,61 +165,60 @@ class IEX_Symbol: Each symbol passed in is a key with its value being a human readable markdown formatted string of the symbols price and movement. """ - dataMessages = {} - for symbol in symbols: - IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/quote?token={self.IEX_TOKEN}" - response = r.get(IEXurl) - if response.status_code == 200: - IEXData = response.json() - keys = ( - "isUSMarketOpen", - "extendedChangePercent", - "extendedPrice", - "companyName", - "latestPrice", - "changePercent", + IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/quote?token={self.IEX_TOKEN}" + + response = r.get(IEXurl) + if response.status_code == 200: + IEXData = response.json() + keys = ( + "isUSMarketOpen", + "extendedChangePercent", + "extendedPrice", + "companyName", + "latestPrice", + "changePercent", + ) + + if set(keys).issubset(IEXData): + + try: # Some symbols dont return if the market is open + IEXData["isUSMarketOpen"] + except KeyError: + IEXData["isUSMarketOpen"] = True + + if ( + IEXData["isUSMarketOpen"] + 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']}**" + change = round(IEXData["changePercent"] * 100, 2) + else: + message = ( + f"{IEXData['companyName']} closed at $**{IEXData['latestPrice']}**," + + f" after hours _(15 minutes delayed)_ the stock price is $**{IEXData['extendedPrice']}**" + ) + change = round(IEXData["extendedChangePercent"] * 100, 2) + + # 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 " ) - if set(keys).issubset(IEXData): + else: + message = f"The symbol: {symbol} was not found." - try: # Some symbols dont return if the market is open - IEXData["isUSMarketOpen"] - except KeyError: - IEXData["isUSMarketOpen"] = True + return message - if ( - IEXData["isUSMarketOpen"] - 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']}**" - change = round(IEXData["changePercent"] * 100, 2) - else: - message = ( - f"{IEXData['companyName']} closed at $**{IEXData['latestPrice']}**," - + f" after hours _(15 minutes delayed)_ the stock price is $**{IEXData['extendedPrice']}**" - ) - change = round(IEXData["extendedChangePercent"] * 100, 2) - - # 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." - - dataMessages[symbol] = message - - return dataMessages - - def dividend_reply(self, symbol: str) -> Dict[str, str]: + def dividend_reply(self, symbol: str) -> str: """Returns the most recent, or next dividend date for a stock symbol. Parameters @@ -278,7 +277,7 @@ class IEX_Symbol: return f"{symbol} either doesn't exist or pays no dividend." - def news_reply(self, symbols: list) -> Dict[str, str]: + def news_reply(self, symbol: str) -> str: """Gets recent english news on stock symbols. Parameters @@ -291,31 +290,27 @@ class IEX_Symbol: Dict[str, str] Each symbol passed in is a key with its value being a human readable markdown formatted string of the symbols news. """ - newsMessages = {} - for symbol in symbols: - IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/5?token={self.IEX_TOKEN}" - response = r.get(IEXurl) - if response.status_code == 200: - data = response.json() - if len(data): - newsMessages[symbol] = f"News for **{symbol.upper()}**:\n\n" - for news in data: - if news["lang"] == "en" and not news["hasPaywall"]: - message = f"*{news['source']}*: [{news['headline']}]({news['url']})\n" - newsMessages[symbol] = newsMessages[symbol] + message - else: - newsMessages[ - symbol - ] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist." + IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/5?token={self.IEX_TOKEN}" + response = r.get(IEXurl) + if response.status_code == 200: + data = response.json() + if len(data): + message = f"News for **{symbol.upper()}**:\n\n" + for news in data: + if news["lang"] == "en" and not news["hasPaywall"]: + message = ( + f"*{news['source']}*: [{news['headline']}]({news['url']})\n" + ) + message += message else: - newsMessages[ - symbol - ] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist." + 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 newsMessages + return message - def info_reply(self, symbols: List[str]) -> Dict[str, str]: + def info_reply(self, symbol: str) -> str: """Gets information on stock symbols. Parameters @@ -328,25 +323,21 @@ class IEX_Symbol: Dict[str, str] Each symbol passed in is a key with its value being a human readable formatted string of the symbols information. """ - infoMessages = {} - for symbol in symbols: - IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/company?token={self.IEX_TOKEN}" - response = r.get(IEXurl) + IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/company?token={self.IEX_TOKEN}" + response = r.get(IEXurl) - if response.status_code == 200: - data = response.json() - infoMessages[symbol] = ( - f"Company Name: [{data['companyName']}]({data['website']})\nIndustry:" - + f" {data['industry']}\nSector: {data['sector']}\nCEO: {data['CEO']}\nDescription: {data['description']}\n" - ) + if response.status_code == 200: + data = response.json() + message = ( + f"Company Name: [{data['companyName']}]({data['website']})\nIndustry:" + + f" {data['industry']}\nSector: {data['sector']}\nCEO: {data['CEO']}\nDescription: {data['description']}\n" + ) - else: - infoMessages[ - symbol - ] = f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." + else: + message = f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." - return infoMessages + return message def intra_reply(self, symbol: str) -> pd.DataFrame: """Returns price data for a symbol since the last market open. @@ -413,7 +404,7 @@ class IEX_Symbol: return pd.DataFrame() - def stat_reply(self, symbols: List[str]) -> Dict[str, str]: + def stat_reply(self, symbol: str) -> str: """Gets key statistics for each symbol in the list Parameters @@ -453,47 +444,6 @@ class IEX_Symbol: m += f"Price to Earnings: {data['peRatio']:.3f}\n" if "beta" in data: m += f"Beta: {data['beta']:.3f}\n" - infoMessages[symbol] = m + return m else: - infoMessages[ - symbol - ] = f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." - - return infoMessages - - def crypto_reply(self, pair: str) -> str: - """Returns the current price of a cryptocurrency - - Parameters - ---------- - pair : str - symbol for the cryptocurrency, sometimes with a price pair like ETHUSD - - Returns - ------- - str - Returns a human readable markdown description of the price, or an empty string if no price was found. - """ - - pair = pair.split(" ")[-1].replace("/", "").upper() - pair += "USD" if len(pair) == 3 else pair - - IEXurl = f"https://cloud.iexapis.com/stable/crypto/{pair}/quote?token={self.IEX_TOKEN}" - - response = r.get(IEXurl) - - if response.status_code == 200: - data = response.json() - - quote = f"Symbol: {data['symbol']}\n" - quote += f"Price: ${data['latestPrice']}\n" - - new, old = data["latestPrice"], data["previousClose"] - if old is not None: - change = (float(new) - float(old)) / float(old) - quote += f"Change: {change}\n" - - return quote - - else: - return "" + return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." diff --git a/bot.py b/bot.py index 24ca337..5473fa0 100644 --- a/bot.py +++ b/bot.py @@ -26,8 +26,7 @@ from telegram.ext import ( CallbackContext, ) -from IEX_Symbol import IEX_Symbol -from cg_Crypto import cg_Crypto +from symbol_router import Router from T_info import T_info TELEGRAM_TOKEN = os.environ["TELEGRAM"] @@ -43,8 +42,7 @@ except KeyError: STRIPE_TOKEN = "" print("Starting without a STRIPE Token will not allow you to accept Donations!") -s = IEX_Symbol(IEX_TOKEN) -c = cg_Crypto() +s = Router(IEX=IEX_TOKEN) t = T_info() # Enable logging diff --git a/symbol_router.py b/symbol_router.py new file mode 100644 index 0000000..3091cab --- /dev/null +++ b/symbol_router.py @@ -0,0 +1,212 @@ +"""Function that routes symbols to the correct API provider. +""" + +import re +import requests as r +import pandas as pd + +from typing import List, Dict + +from IEX_Symbol import IEX_Symbol +from cg_Crypto import cg_Crypto + + +class Router: + STOCK_REGEX = "[$]([a-zA-Z]{1,4})" + CRYPTO_REGEX = "[$$]([a-zA-Z]{1,9})" + + def __init__(self, IEX_TOKEN=""): + self.symbol = IEX_Symbol(IEX_TOKEN) + self.crypto = cg_Crypto() + + def find_symbols(self, text: str) -> Dict[str, str]: + """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. + Only returns each match once. Example: Whats the price of $tsla? + + Parameters + ---------- + text : str + Blob of text. + + Returns + ------- + List[str] + List of stock symbols as strings without dollar sign. + """ + symbols = {} + symbols["stocks"] = list(set(re.findall(self.SYMBOL_REGEX, text))) + symbols["crypto"] = list(set(re.findall(self.SYMBOL_REGEX, text))) + return symbols + + def status(self) -> str: + """Checks for any issues with APIs. + + Returns + ------- + str + Human readable text on status of IEX API + """ + + def price_reply(self, symbols: dict) -> 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 = [] + + if symbols["stocks"]: + for s in symbols["stocks"]: + replies.append(self.symbol.price_reply(s)) + + if symbols["crypto"]: + for s in symbols["crypto"]: + replies.append(self.crypto.price_reply(s)) + + return replies + + def dividend_reply(self, symbols: dict) -> Dict[str, 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 = [] + + if symbols["stocks"]: + for s in symbols["stocks"]: + replies.append(self.symbol.price_reply(s)) + + if symbols["crypto"]: + for s in symbols["crypto"]: + replies.append(self.crypto.price_reply(s)) + + def news_reply(self, symbols: dict) -> 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 = [] + + if symbols["stocks"]: + for s in symbols["stocks"]: + replies.append(self.symbol.price_reply(s)) + + if symbols["crypto"]: + for s in symbols["crypto"]: + replies.append(self.crypto.price_reply(s)) + + return replies + + def info_reply(self, symbols: dict) -> 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 = [] + + if symbols["stocks"]: + for s in symbols["stocks"]: + replies.append(self.symbol.price_reply(s)) + + if symbols["crypto"]: + for s in symbols["crypto"]: + replies.append(self.crypto.price_reply(s)) + + return replies + + def intra_reply(self, symbol: str, type: str) -> 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 type == "stocks": + return self.symbol.intra_reply(symbol) + elif type == "crypto": + return self.crypto.intra_reply(symbol) + else: + raise f"Unknown type: {type}" + + def chart_reply(self, symbol: str, type: str) -> 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 type == "stocks": + return self.symbol.intra_reply(symbol) + elif type == "crypto": + return self.crypto.intra_reply(symbol) + else: + raise f"Unknown type: {type}" + + def stat_reply(self, symbols: List[str]) -> Dict[str, 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 = [] + + if symbols["stocks"]: + for s in symbols["stocks"]: + replies.append(self.symbol.price_reply(s)) + + if symbols["crypto"]: + for s in symbols["crypto"]: + replies.append(self.crypto.price_reply(s)) \ No newline at end of file