From 19c2daa9f9abc6cfca562b33c2ff4f7bd423982d Mon Sep 17 00:00:00 2001 From: Anson Biggs Date: Sun, 28 Feb 2021 00:44:31 -0700 Subject: [PATCH] I hate Typing --- IEX_Symbol.py | 33 +++++++------ bot.py | 20 +------- cg_Crypto.py | 20 +++++--- symbol_router.py | 124 ++++++++++++++++++++++++++++------------------- 4 files changed, 107 insertions(+), 90 deletions(-) diff --git a/IEX_Symbol.py b/IEX_Symbol.py index 48917af..fe8502d 100644 --- a/IEX_Symbol.py +++ b/IEX_Symbol.py @@ -10,6 +10,8 @@ import schedule from fuzzywuzzy import fuzz import os +from symbol_router import Stock + class IEX_Symbol: """ @@ -47,7 +49,9 @@ class IEX_Symbol: """Clears cache of chart data.""" self.charts = {} - def get_symbol_list(self, return_df=False) -> Optional[pd.DataFrame]: + def get_symbol_list( + self, return_df=False + ) -> Optional[Tuple[pd.DataFrame, datetime]]: raw_symbols = r.get( f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}" @@ -141,7 +145,7 @@ class IEX_Symbol: self.searched_symbols[search] = symbol_list return symbol_list - def price_reply(self, symbol: str) -> str: + def price_reply(self, symbol: Stock) -> str: """Returns current market price or after hours if its available for a given stock symbol. Parameters @@ -208,7 +212,7 @@ class IEX_Symbol: return message - def dividend_reply(self, symbol: str) -> str: + def dividend_reply(self, symbol: Stock) -> str: """Returns the most recent, or next dividend date for a stock symbol. Parameters @@ -259,7 +263,8 @@ class IEX_Symbol: ).days return ( - f"The next dividend for ${self.symbol_list[self.symbol_list['symbol']==symbol.upper()]['description'].item()}" + "The next dividend for" + + f"${self.symbol_list[self.symbol_list['symbol']==symbol.id.upper()]['description'].item()}" + f" is on {payment} which is in {daysDelta} days." + f" The dividend is for {price} per share." + f"\nThe dividend was declared on {declared} and the ex-dividend date is {ex}" @@ -267,7 +272,7 @@ class IEX_Symbol: return f"{symbol} either doesn't exist or pays no dividend." - def news_reply(self, symbol: str) -> str: + def news_reply(self, symbol: Stock) -> str: """Gets recent english news on stock symbols. Parameters @@ -286,7 +291,7 @@ class IEX_Symbol: if response.status_code == 200: data = response.json() if len(data): - message = f"News for **{symbol.upper()}**:\n\n" + message = f"News for **{symbol.id.upper()}**:\n\n" for news in data: if news["lang"] == "en" and not news["hasPaywall"]: message = ( @@ -300,7 +305,7 @@ class IEX_Symbol: return message - def info_reply(self, symbol: str) -> str: + def info_reply(self, symbol: Stock) -> str: """Gets information on stock symbols. Parameters @@ -329,7 +334,7 @@ class IEX_Symbol: return message - def intra_reply(self, symbol: str) -> pd.DataFrame: + def intra_reply(self, symbol: Stock) -> pd.DataFrame: """Returns price data for a symbol since the last market open. Parameters @@ -342,7 +347,7 @@ class IEX_Symbol: pd.DataFrame Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. """ - if symbol.upper() not in list(self.symbol_list["symbol"]): + if symbol.id.upper() not in list(self.symbol_list["symbol"]): return pd.DataFrame() IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/intraday-prices?token={self.IEX_TOKEN}" @@ -356,7 +361,7 @@ class IEX_Symbol: return pd.DataFrame() - def chart_reply(self, symbol: str) -> 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. @@ -372,11 +377,11 @@ class IEX_Symbol: """ schedule.run_pending() - if symbol.upper() not in list(self.symbol_list["symbol"]): + 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.upper()] + return self.charts[symbol.id.upper()] except KeyError: pass @@ -389,12 +394,12 @@ class IEX_Symbol: 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.upper()] = df + self.charts[symbol.id.upper()] = df return df return pd.DataFrame() - def stat_reply(self, symbol: str) -> str: + def stat_reply(self, symbol: Stock) -> str: """Gets key statistics for each symbol in the list Parameters diff --git a/bot.py b/bot.py index c69e734..a53bf85 100644 --- a/bot.py +++ b/bot.py @@ -67,26 +67,8 @@ def license(update: Update, context: CallbackContext): def status(update: Update, context: CallbackContext): - message = "" - try: - # Bot Status - bot_resp = ( - datetime.datetime.now(update.message.date.tzinfo) - update.message.date - ) - message += f"It took {bot_resp.total_seconds()} seconds for the bot to get your message.\n" - # IEX Status - message += s.iex_status() + "\n" - - # Message Status - message += s.message_status() - except Exception as ex: - message += ( - f"*\n\nERROR ENCOUNTERED:*\n{ex}\n\n" - + "*The bot encountered an error while attempting to find errors. Please contact the bot admin.*" - ) - - update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN) + update.message.reply_text(text=s.status(), parse_mode=telegram.ParseMode.MARKDOWN) def donate(update: Update, context: CallbackContext): diff --git a/cg_Crypto.py b/cg_Crypto.py index d005cce..714b934 100644 --- a/cg_Crypto.py +++ b/cg_Crypto.py @@ -9,6 +9,7 @@ import requests as r import schedule from fuzzywuzzy import fuzz from markdownify import markdownify +from symbol_router import Coin class cg_Crypto: @@ -39,7 +40,9 @@ class cg_Crypto: except KeyError: return "" - def get_symbol_list(self, return_df=False) -> Optional[pd.DataFrame]: + def get_symbol_list( + self, return_df=False + ) -> Optional[Tuple[pd.DataFrame, datetime]]: raw_symbols = r.get("https://api.coingecko.com/api/v3/coins/list").json() symbols = pd.DataFrame(data=raw_symbols) @@ -103,7 +106,7 @@ class cg_Crypto: self.searched_symbols[search] = symbol_list return symbol_list - def price_reply(self, symbol: str) -> str: + def price_reply(self, symbol: Coin) -> str: """Returns current market price or after hours if its available for a given coin symbol. Parameters @@ -146,7 +149,7 @@ class cg_Crypto: return message - def intra_reply(self, symbol: str) -> pd.DataFrame: + def intra_reply(self, symbol: Coin) -> pd.DataFrame: """Returns price data for a symbol since the last market open. Parameters @@ -172,7 +175,7 @@ class cg_Crypto: return pd.DataFrame() - def chart_reply(self, symbol: str) -> 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. @@ -200,7 +203,7 @@ class cg_Crypto: return pd.DataFrame() - def stat_reply(self, symbol: str) -> str: + def stat_reply(self, symbol: Coin) -> str: """Gets key statistics for each symbol in the list Parameters @@ -219,7 +222,7 @@ class cg_Crypto: if response.status_code == 200: data = response.json() - message = f""" + return f""" [{data['name']}]({data['links']['homepage'][0]}) Statistics: Maket Cap Ranking: {data.get('market_cap_rank',"Not Available")} CoinGecko Scores: @@ -228,9 +231,10 @@ class cg_Crypto: Community: {data.get('community_score','Not Available')} Public Interest: {data.get('public_interest_score','Not Available')} """ - return message + else: + return f"{symbol.symbol} returned an error." - def info_reply(self, symbol: str) -> str: + def info_reply(self, symbol: Coin) -> str: """Gets information on stock symbols. Parameters diff --git a/symbol_router.py b/symbol_router.py index 382d40f..16d8135 100644 --- a/symbol_router.py +++ b/symbol_router.py @@ -5,7 +5,7 @@ import re import requests as r import pandas as pd -from typing import List, Dict +from typing import List, Tuple, TypeVar from IEX_Symbol import IEX_Symbol from cg_Crypto import cg_Crypto @@ -19,7 +19,7 @@ class Router: self.stock = IEX_Symbol() self.crypto = cg_Crypto() - def find_symbols(self, text: str) -> Dict[str, str]: + def find_symbols(self, text: str) -> List[any]: """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? @@ -60,7 +60,7 @@ class Router: Human readable text on status of IEX API """ - def search_symbols(self, search: str) -> List[str]: + def search_symbols(self, search: str) -> List[Tuple[str, str]]: """Performs a fuzzy search to find stock symbols closest to a search term. Parameters @@ -74,9 +74,9 @@ class Router: A list tuples of every stock sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name). """ # TODO add support for crypto - return self.stock.find_symbols(search) + return self.stock.search_symbols(search) - def price_reply(self, symbols: dict) -> List[str]: + def price_reply(self, symbols: list) -> List[str]: """Returns current market price or after hours if its available for a given stock symbol. Parameters @@ -99,10 +99,10 @@ class Router: replies.append(self.crypto.price_reply(symbol)) else: print(f"{symbol} is not a Stock or Coin") - print(replies) + return replies - def dividend_reply(self, symbols: dict) -> Dict[str, str]: + def dividend_reply(self, symbols: list) -> List[str]: """Returns the most recent, or next dividend date for a stock symbol. Parameters @@ -116,15 +116,17 @@ class Router: Each symbol passed in is a key with its value being a human readable formatted string of the symbols div dates. """ replies = [] + for symbol in symbols: + if isinstance(symbol, Stock): + replies.append(self.stock.dividend_reply(symbol)) + elif isinstance(symbol, Coin): + replies.append("Cryptocurrencies do no have Dividends.") + else: + print(f"{symbol} is not a Stock or Coin") - if symbols["stocks"]: - for s in symbols["stocks"]: - replies.append(self.stock.price_reply(s)) + return replies - if symbols["crypto"]: - replies.append("Cryptocurrencies do no have Dividends.") - - def news_reply(self, symbols: dict) -> List[str]: + def news_reply(self, symbols: list) -> List[str]: """Gets recent english news on stock symbols. Parameters @@ -139,17 +141,18 @@ class Router: """ replies = [] - if symbols["stocks"]: - for s in symbols["stocks"]: - replies.append(self.stock.price_reply(s)) - - if symbols["crypto"]: - for s in symbols["crypto"]: - replies.append(self.crypto.price_reply(s)) + 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.") + else: + print(f"{symbol} is not a Stock or Coin") return replies - def info_reply(self, symbols: dict) -> List[str]: + def info_reply(self, symbols: list) -> List[str]: """Gets information on stock symbols. Parameters @@ -164,17 +167,17 @@ class Router: """ replies = [] - if symbols["stocks"]: - for s in symbols["stocks"]: - replies.append(self.stock.price_reply(s)) - - if symbols["crypto"]: - for s in symbols["crypto"]: - replies.append(self.crypto.price_reply(s)) + for symbol in symbols: + if isinstance(symbol, Stock): + replies.append(self.stock.info_reply(symbol)) + elif isinstance(symbol, Coin): + replies.append(self.crypto.info_reply(symbol)) + else: + print(f"{symbol} is not a Stock or Coin") return replies - def intra_reply(self, symbol: str, type: str) -> pd.DataFrame: + def intra_reply(self, symbol: str) -> pd.DataFrame: """Returns price data for a symbol since the last market open. Parameters @@ -187,14 +190,16 @@ class Router: pd.DataFrame Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. """ - if type == "stocks": + + if isinstance(symbol, Stock): return self.stock.intra_reply(symbol) - elif type == "crypto": + elif isinstance(symbol, Coin): return self.crypto.intra_reply(symbol) else: - raise f"Unknown type: {type}" + print(f"{symbol} is not a Stock or Coin") + return pd.DataFrame() - def chart_reply(self, symbols: str) -> pd.DataFrame: + def chart_reply(self, symbol: 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. @@ -208,12 +213,15 @@ class Router: pd.DataFrame Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. """ - if symbols["stocks"]: - return self.stock.intra_reply(symbol := symbols["stocks"][0]), symbol - if symbols["crypto"]: - return self.stock.intra_reply(symbol := symbols["crypto"][0]), symbol + if isinstance(symbol, Stock): + return self.stock.chart_reply(symbol) + elif isinstance(symbol, Coin): + return self.crypto.chart_reply(symbol) + else: + print(f"{symbol} is not a Stock or Coin") + return pd.DataFrame() - def stat_reply(self, symbols: List[str]) -> Dict[str, str]: + def stat_reply(self, symbols: List[str]) -> List[str]: """Gets key statistics for each symbol in the list Parameters @@ -228,31 +236,49 @@ class Router: """ replies = [] - if symbols["stocks"]: - for s in symbols["stocks"]: - replies.append(self.stock.price_reply(s)) + for symbol in symbols: + if isinstance(symbol, Stock): + replies.append(self.stock.stat_reply(symbol)) + elif isinstance(symbol, Coin): + replies.append(self.crypto.stat_reply(symbol)) + else: + print(f"{symbol} is not a Stock or Coin") - if symbols["crypto"]: - for s in symbols["crypto"]: - replies.append(self.crypto.price_reply(s)) + return replies + + +Sym = TypeVar("Sym", Stock, Coin) class Symbol: + """ + symbol: What the user calls it. ie tsla or btc + id: What the api expects. ie tsla or bitcoin + name: Human readable. ie Tesla or Bitcoin + """ + currency = "usd" pass + def __init__(self, symbol) -> None: + self.symbol = symbol + self.id = symbol + def __repr__(self) -> str: - return f"{self.__class__.__name__} instance of {self.id} at {id(self)}" + return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>" + + def __str__(self) -> str: + return self.id class Stock(Symbol): - def __init__(self, symbol) -> None: + def __init__(self, symbol: str) -> None: self.symbol = symbol self.id = symbol class Coin(Symbol): - def __init__(self, symbol) -> None: + def __init__(self, symbol: str) -> None: self.symbol = symbol self.get_data() @@ -263,4 +289,4 @@ class Coin(Symbol): self.name = data["name"] self.description = data["description"] - self.price = data["market_data"]["current_price"][self.currency] \ No newline at end of file + self.price = data["market_data"]["current_price"][self.currency]