1
0
mirror of https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot.git synced 2025-06-16 15:06:53 +00:00

just need to smash bugs and the update is done

This commit is contained in:
Anson Biggs 2021-02-25 21:37:13 -07:00
parent 487ab10933
commit 6e37e541d6
5 changed files with 160 additions and 541 deletions

View File

@ -1,9 +1,8 @@
"""Class with functions for running the bot with IEX Cloud. """Class with functions for running the bot with IEX Cloud.
""" """
import re
from datetime import datetime from datetime import datetime
from typing import Optional, List, Tuple, Dict from typing import Optional, List, Tuple
import pandas as pd import pandas as pd
import requests as r import requests as r
@ -134,23 +133,6 @@ class IEX_Symbol:
self.searched_symbols[search] = symbol_list self.searched_symbols[search] = symbol_list
return symbol_list return symbol_list
def find_symbols(self, text: str) -> List[str]:
"""Finds stock tickers starting with a dollar sign 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.
"""
return list(set(re.findall(self.SYMBOL_REGEX, text)))
def price_reply(self, symbol: str) -> str: def price_reply(self, symbol: str) -> str:
"""Returns current market price or after hours if its available for a given stock symbol. """Returns current market price or after hours if its available for a given stock symbol.
@ -417,33 +399,31 @@ class IEX_Symbol:
Dict[str, str] Dict[str, str]
Each symbol passed in is a key with its value being a human readable formatted string of the symbols statistics. Each symbol passed in is a key with its value being a human readable formatted string of the symbols statistics.
""" """
infoMessages = {}
for symbol in symbols: IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/stats?token={self.IEX_TOKEN}"
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/stats?token={self.IEX_TOKEN}" response = r.get(IEXurl)
response = r.get(IEXurl)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
[data.pop(k) for k in list(data) if data[k] == ""] [data.pop(k) for k in list(data) if data[k] == ""]
m = "" m = ""
if "companyName" in data: if "companyName" in data:
m += f"Company Name: {data['companyName']}\n" m += f"Company Name: {data['companyName']}\n"
if "marketcap" in data: if "marketcap" in data:
m += f"Market Cap: {data['marketcap']:,}\n" m += f"Market Cap: {data['marketcap']:,}\n"
if "week52high" in data: if "week52high" in data:
m += f"52 Week (high-low): {data['week52high']:,} " m += f"52 Week (high-low): {data['week52high']:,} "
if "week52low" in data: if "week52low" in data:
m += f"- {data['week52low']:,}\n" m += f"- {data['week52low']:,}\n"
if "employees" in data: if "employees" in data:
m += f"Number of Employees: {data['employees']:,}\n" m += f"Number of Employees: {data['employees']:,}\n"
if "nextEarningsDate" in data: if "nextEarningsDate" in data:
m += f"Next Earnings Date: {data['nextEarningsDate']}\n" m += f"Next Earnings Date: {data['nextEarningsDate']}\n"
if "peRatio" in data: if "peRatio" in data:
m += f"Price to Earnings: {data['peRatio']:.3f}\n" m += f"Price to Earnings: {data['peRatio']:.3f}\n"
if "beta" in data: if "beta" in data:
m += f"Beta: {data['beta']:.3f}\n" m += f"Beta: {data['beta']:.3f}\n"
return m return m
else: else:
return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."

134
T_info.py
View File

@ -3,34 +3,9 @@
import re import re
import requests as r import requests as r
import pandas as pd
from typing import List, Dict
class T_info: class T_info:
STOCK_REGEX = "[$]([a-zA-Z]{1,4})"
CRYPTO_REGEX = "[$$]([a-zA-Z]{1,9})"
def find_symbols(self, text: str) -> List[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 = list(set(re.findall(self.SYMBOL_REGEX, text)))
crypto = list(set(re.findall(self.SYMBOL_REGEX, text)))
return
license = re.sub( license = re.sub(
r"\b\n", r"\b\n",
" ", " ",
@ -80,112 +55,3 @@ If you have any questions get in touch: @MisterBiggs or[anson@ansonbiggs.com](ht
_Donations can only be made in a chat directly with @simplestockbot_ _Donations can only be made in a chat directly with @simplestockbot_
""" """
def status(self) -> str:
"""Checks IEX Status dashboard for any current API issues.
Returns
-------
str
Human readable text on status of IEX API
"""
def price_reply(self, symbols: list) -> Dict[str, 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.
"""
def dividend_reply(self, symbol: str) -> 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.
"""
def news_reply(self, symbols: list) -> Dict[str, 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.
"""
def info_reply(self, symbols: List[str]) -> Dict[str, 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.
"""
def intra_reply(self, symbol: 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.
"""
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.
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.
"""
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.
"""

View File

@ -1,14 +1,14 @@
"""Class with functions for running the bot with CoinGecko """Class with functions for running the bot with IEX Cloud.
""" """
import re
from datetime import datetime from datetime import datetime
from typing import Optional, List, Tuple, Dict from typing import Optional, List, Tuple
import pandas as pd import pandas as pd
import requests as r import requests as r
import schedule import schedule
from fuzzywuzzy import fuzz from fuzzywuzzy import fuzz
from markdownify import markdownify
class cg_Crypto: class cg_Crypto:
@ -16,12 +16,11 @@ class cg_Crypto:
Functions for finding crypto info Functions for finding crypto info
""" """
SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})" vs_currency = "usd" # simple/supported_vs_currencies for list of options
searched_symbols = {} searched_symbols = {}
charts = {}
def __init__(self, IEX_TOKEN: str) -> None: def __init__(self) -> None:
"""Creates a Symbol Object """Creates a Symbol Object
Parameters Parameters
@ -29,22 +28,20 @@ class cg_Crypto:
IEX_TOKEN : str IEX_TOKEN : str
IEX Token IEX Token
""" """
self.IEX_TOKEN = IEX_TOKEN self.get_symbol_list()
if IEX_TOKEN != "": schedule.every().day.do(self.get_symbol_list)
self.get_symbol_list()
schedule.every().day.do(self.get_symbol_list) def symbol_id(self, symbol) -> str:
schedule.every().day.do(self.clear_charts) try:
return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[
def clear_charts(self) -> None: 0
"""Clears cache of chart data.""" ]
self.charts = {} except KeyError:
return ""
def get_symbol_list(self, return_df=False) -> Optional[pd.DataFrame]: def get_symbol_list(self, return_df=False) -> Optional[pd.DataFrame]:
raw_symbols = r.get( raw_symbols = r.get("https://api.coingecko.com/api/v3/coins/list").json()
f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}"
).json()
symbols = pd.DataFrame(data=raw_symbols) symbols = pd.DataFrame(data=raw_symbols)
symbols["description"] = symbols["symbol"] + ": " + symbols["name"] symbols["description"] = symbols["symbol"] + ": " + symbols["name"]
@ -52,62 +49,34 @@ class cg_Crypto:
if return_df: if return_df:
return symbols, datetime.now() return symbols, datetime.now()
def iex_status(self) -> str: def cg_status(self) -> str:
"""Checks IEX Status dashboard for any current API issues. """Checks CoinGecko /ping endpoint for API issues.
Returns Returns
------- -------
str str
Human readable text on status of IEX API Human readable text on status of CoinGecko API
""" """
status = r.get("https://pjmps0c34hp7.statuspage.io/api/v2/status.json").json()[ status = r.get("https://api.coingecko.com/api/v3/ping")
"status"
]
if status["indicator"] == "none": if status.status_code == 200:
return "IEX Cloud is currently not reporting any issues with its API." return "CoinGecko API responded that it was OK in {status.elapsed.total_seconds()} Seconds."
else: else:
return ( return "CoinGecko API returned an error in {status.elapsed.total_seconds()} Seconds."
f"{status['indicator']}: {status['description']}."
+ " Please check the status page for more information. https://status.iexapis.com"
)
def message_status(self) -> str:
"""Checks to see if the bot has available IEX Credits
Returns
-------
str
Human readable text on status of IEX Credits.
"""
usage = r.get(
f"https://cloud.iexapis.com/stable/account/metadata?token={self.IEX_TOKEN}"
).json()
try:
if (
usage["messagesUsed"] >= usage["messageLimit"] - 10000
and not usage["payAsYouGoEnabled"]
):
return "Bot may be out of IEX Credits."
else:
return "Bot has available IEX Credits."
except KeyError:
return "**IEX API could not be reached.**"
def search_symbols(self, search: str) -> List[Tuple[str, str]]: def search_symbols(self, search: str) -> List[Tuple[str, str]]:
"""Performs a fuzzy search to find stock symbols closest to a search term. """Performs a fuzzy search to find coin symbols closest to a search term.
Parameters Parameters
---------- ----------
search : str search : str
String used to search, could be a company name or something close to the companies stock ticker. String used to search, could be a company name or something close to the companies coin ticker.
Returns Returns
------- -------
List[tuple[str, str]] List[tuple[str, str]]
A list tuples of every stock sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name). A list tuples of every coin sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
""" """
schedule.run_pending() schedule.run_pending()
search = search.lower() search = search.lower()
try: # https://stackoverflow.com/a/3845776/8774114 try: # https://stackoverflow.com/a/3845776/8774114
@ -134,30 +103,13 @@ class cg_Crypto:
self.searched_symbols[search] = symbol_list self.searched_symbols[search] = symbol_list
return symbol_list return symbol_list
def find_symbols(self, text: str) -> List[str]: def price_reply(self, symbol: str) -> str:
"""Finds stock tickers starting with a dollar sign in a blob of text and returns them in a list. """Returns current market price or after hours if its available for a given coin symbol.
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.
"""
return list(set(re.findall(self.SYMBOL_REGEX, text)))
def price_reply(self, symbols: list) -> Dict[str, str]:
"""Returns current market price or after hours if its available for a given stock symbol.
Parameters Parameters
---------- ----------
symbols : list symbols : list
List of stock symbols. List of coin symbols.
Returns Returns
------- -------
@ -165,188 +117,34 @@ class cg_Crypto:
Each symbol passed in is a key with its value being a human readable Each symbol passed in is a key with its value being a human readable
markdown formatted string of the symbols price and movement. 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) response = r.get(
if response.status_code == 200: f"https://api.coingecko.com/api/v3/coins/{symbol}?localization=false"
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 "
else:
message = f"The symbol: {symbol} was not found."
dataMessages[symbol] = message
return dataMessages
def dividend_reply(self, symbol: str) -> 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.
"""
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/dividends/next?token={self.IEX_TOKEN}"
response = r.get(IEXurl)
if response.status_code == 200: if response.status_code == 200:
IEXData = response.json()[0] data = response.json()
keys = (
"amount",
"currency",
"declaredDate",
"exDate",
"frequency",
"paymentDate",
"flag",
)
if set(keys).issubset(IEXData): try:
name = data["name"]
price = data["market_data"]["current_price"][self.vs_currency]
change = data["market_data"]["price_change_percentage_24h"]
except KeyError:
return f"{symbol} returned an error."
if IEXData["currency"] == "USD": message = f"The current coin price of {name} is $**{price}**"
price = f"${IEXData['amount']}"
else:
price = f"{IEXData['amount']} {IEXData['currency']}"
# Pattern IEX uses for dividend date. # Determine wording of change text
pattern = "%Y-%m-%d" if change > 0:
message += f", the coin is currently **up {change}%**"
declared = datetime.strptime(IEXData["declaredDate"], pattern).strftime( elif change < 0:
"%A, %B %w" message += f", the coin is currently **down {change}%**"
)
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 (
f"The next dividend for ${self.symbol_list[self.symbol_list['symbol']==symbol.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}"
)
return f"{symbol} either doesn't exist or pays no dividend."
def news_reply(self, symbols: list) -> Dict[str, 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.
"""
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."
else: else:
newsMessages[ message += ", the coin hasn't shown any movement today."
symbol
] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
return newsMessages else:
message = f"The symbol: {symbol} was not found."
def info_reply(self, symbols: List[str]) -> Dict[str, str]: return message
"""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.
"""
infoMessages = {}
for symbol in symbols:
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"
)
else:
infoMessages[
symbol
] = f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
return infoMessages
def intra_reply(self, symbol: str) -> pd.DataFrame: def intra_reply(self, symbol: str) -> pd.DataFrame:
"""Returns price data for a symbol since the last market open. """Returns price data for a symbol since the last market open.
@ -361,16 +159,15 @@ class cg_Crypto:
pd.DataFrame pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty 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"]): response = r.get(
return pd.DataFrame() "https://api.coingecko.com/api/v3/coins/{symbol}/ohlc?vs_currency=usd&days=1"
)
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/intraday-prices?token={self.IEX_TOKEN}"
response = r.get(IEXurl)
if response.status_code == 200: if response.status_code == 200:
df = pd.DataFrame(response.json()) df = pd.DataFrame(
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"]) response.json(), columns=["Date", "Open", "High", "Low", "Close"]
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"]) ).dropna()
df = df.set_index("DT") df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date")
return df return df
return pd.DataFrame() return pd.DataFrame()
@ -389,111 +186,71 @@ class cg_Crypto:
pd.DataFrame pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty 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.upper() not in list(self.symbol_list["symbol"]):
return pd.DataFrame()
try: # https://stackoverflow.com/a/3845776/8774114
return self.charts[symbol.upper()]
except KeyError:
pass
response = r.get( response = r.get(
f"https://cloud.iexapis.com/stable/stock/{symbol}/chart/1mm?token={self.IEX_TOKEN}&chartInterval=3&includeToday=false" "https://api.coingecko.com/api/v3/coins/{symbol}/ohlc?vs_currency=usd&days=30"
) )
if response.status_code == 200: if response.status_code == 200:
df = pd.DataFrame(response.json()) df = pd.DataFrame(
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"]) response.json(), columns=["Date", "Open", "High", "Low", "Close"]
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"]) ).dropna()
df = df.set_index("DT") df["Date"] = pd.to_datetime(df["Date"], unit="ms")
self.charts[symbol.upper()] = df df = df.set_index("Date")
return df return df
return pd.DataFrame() 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 """Gets key statistics for each symbol in the list
Parameters Parameters
---------- ----------
symbols : List[str] symbols : List[str]
List of stock symbols List of coin symbols
Returns Returns
------- -------
Dict[str, str] Dict[str, str]
Each symbol passed in is a key with its value being a human readable formatted string of the symbols statistics. Each symbol passed in is a key with its value being a human readable formatted string of the symbols statistics.
""" """
infoMessages = {} response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol}?localization=false"
for symbol in symbols: )
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/stats?token={self.IEX_TOKEN}"
response = r.get(IEXurl)
if response.status_code == 200:
data = response.json()
[data.pop(k) for k in list(data) if data[k] == ""]
m = ""
if "companyName" in data:
m += f"Company Name: {data['companyName']}\n"
if "marketcap" in data:
m += f"Market Cap: {data['marketcap']:,}\n"
if "week52high" in data:
m += f"52 Week (high-low): {data['week52high']:,} "
if "week52low" in data:
m += f"- {data['week52low']:,}\n"
if "employees" in data:
m += f"Number of Employees: {data['employees']:,}\n"
if "nextEarningsDate" in data:
m += f"Next Earnings Date: {data['nextEarningsDate']}\n"
if "peRatio" in data:
m += f"Price to Earnings: {data['peRatio']:.3f}\n"
if "beta" in data:
m += f"Beta: {data['beta']:.3f}\n"
infoMessages[symbol] = 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: if response.status_code == 200:
data = response.json() data = response.json()
quote = f"Symbol: {data['symbol']}\n" message = f"""
quote += f"Price: ${data['latestPrice']}\n" [{data['name']}]({data['links']['homepage'][0]}) Statistics:
Maket 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')}
"""
return message
new, old = data["latestPrice"], data["previousClose"] def info_reply(self, symbol: str) -> str:
if old is not None: """Gets information on stock symbols.
change = (float(new) - float(old)) / float(old)
quote += f"Change: {change}\n"
return quote Parameters
----------
symbols : List[str]
List of stock symbols.
else: Returns
return "" -------
Dict[str, str]
Each symbol passed in is a key with its value being a human readable formatted string of the symbols information.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol}?localization=false"
)
if response.status_code == 200:
data = response.json()
try:
return markdownify(data["description"])
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."

View File

@ -4,4 +4,5 @@ pandas==1.2.1
fuzzywuzzy==0.18.0 fuzzywuzzy==0.18.0
python-Levenshtein==0.12.1 python-Levenshtein==0.12.1
schedule==1.0.0 schedule==1.0.0
mplfinance==0.12.7a5 mplfinance==0.12.7a5
markdownify==0.6.5

View File

@ -1,8 +1,7 @@
"""Function that routes symbols to the correct API provider. """Function that routes symbols to the correct API provider.
""" """
import re import re
import requests as r
import pandas as pd import pandas as pd
from typing import List, Dict from typing import List, Dict
@ -12,10 +11,10 @@ from cg_Crypto import cg_Crypto
class Router: class Router:
STOCK_REGEX = "[$]([a-zA-Z]{1,4})" STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z]{1,4})"
CRYPTO_REGEX = "[$$]([a-zA-Z]{1,9})" CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,9})"
def __init__(self, IEX_TOKEN=""): def __init__(self, IEX_TOKEN):
self.symbol = IEX_Symbol(IEX_TOKEN) self.symbol = IEX_Symbol(IEX_TOKEN)
self.crypto = cg_Crypto() self.crypto = cg_Crypto()
@ -35,8 +34,11 @@ class Router:
List of stock symbols as strings without dollar sign. List of stock symbols as strings without dollar sign.
""" """
symbols = {} symbols = {}
symbols["stocks"] = list(set(re.findall(self.SYMBOL_REGEX, text))) symbols["stocks"] = list(set(re.findall(self.STOCK_REGEX, text)))
symbols["crypto"] = list(set(re.findall(self.SYMBOL_REGEX, text))) symbols["crypto"] = [
self.crypto.symbol_id(c) for c in set(re.findall(self.CRYPTO_REGEX, text))
]
return symbols return symbols
def status(self) -> str: def status(self) -> str:
@ -48,6 +50,22 @@ class Router:
Human readable text on status of IEX API Human readable text on status of IEX API
""" """
def search_symbols(self, search: str) -> List[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).
"""
# TODO add support for crypto
return self.symbol.find_symbols(str)
def price_reply(self, symbols: dict) -> List[str]: def price_reply(self, symbols: dict) -> List[str]:
"""Returns current market price or after hours if its available for a given stock symbol. """Returns current market price or after hours if its available for a given stock symbol.
@ -94,8 +112,7 @@ class Router:
replies.append(self.symbol.price_reply(s)) replies.append(self.symbol.price_reply(s))
if symbols["crypto"]: if symbols["crypto"]:
for s in symbols["crypto"]: replies.append("Cryptocurrencies do no have Dividends.")
replies.append(self.crypto.price_reply(s))
def news_reply(self, symbols: dict) -> List[str]: def news_reply(self, symbols: dict) -> List[str]:
"""Gets recent english news on stock symbols. """Gets recent english news on stock symbols.
@ -167,7 +184,7 @@ class Router:
else: else:
raise f"Unknown type: {type}" raise f"Unknown type: {type}"
def chart_reply(self, symbol: str, type: str) -> pd.DataFrame: def chart_reply(self, symbols: str) -> pd.DataFrame:
"""Returns price data for a symbol of the past month up until the previous trading days close. """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. Also caches multiple requests made in the same day.
@ -181,12 +198,10 @@ class Router:
pd.DataFrame pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
""" """
if type == "stocks": if symbols["stocks"]:
return self.symbol.intra_reply(symbol) return self.symbol.intra_reply(symbol := symbols["stocks"][0]), symbol
elif type == "crypto": if symbols["crypto"]:
return self.crypto.intra_reply(symbol) return self.symbol.intra_reply(symbol := symbols["crypto"][0]), symbol
else:
raise f"Unknown type: {type}"
def stat_reply(self, symbols: List[str]) -> Dict[str, str]: def stat_reply(self, symbols: List[str]) -> Dict[str, str]:
"""Gets key statistics for each symbol in the list """Gets key statistics for each symbol in the list
@ -209,4 +224,4 @@ class Router:
if symbols["crypto"]: if symbols["crypto"]:
for s in symbols["crypto"]: for s in symbols["crypto"]:
replies.append(self.crypto.price_reply(s)) replies.append(self.crypto.price_reply(s))