Anson/Simple-Discord-Stock-Bot
1
0
Fork 0
mirror of https://gitlab.com/simple-stock-bots/simple-discord-stock-bot.git synced 2025-07-27 00:21:23 +00:00
Code Releases Activity

updated to newest version of discord and bot files

Browse Source
This commit is contained in:
Anson
2022-01-04 21:54:28 -07:00
parent 342392b604
commit eb48ab74c2
9 changed files with 238 additions and 748 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
.vscode/launch.json vendored Normal file
View File

@@ -0,0 +1,15 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Discord Bot",
"type": "python",
"request": "launch",
"program": "bot.py",
"console": "integratedTerminal"
}
]
}

7
.vscode/settings.json vendored
View File

@@ -8,13 +8,6 @@
"files.associations": {
"DockerDev": "dockerfile",
},
"workbench.colorTheme": "One Dark Pro",
"explorer.confirmDelete": false,
"editor.fontFamily": "JetBrains Mono",
"editor.letterSpacing": 1.2,
"editor.fontLigatures": true,
"editor.wordWrap": "on",
"python.formatting.provider": "black",
"python.showStartPage": false,
"editor.fontSize": 13,
}

15
Dockerfile
View File

@@ -1,7 +1,16 @@
FROM python:3.9-buster
FROM python:3.9-buster AS builder
COPY requirements.txt /requirements.txt
RUN pip install --user -r requirements.txt
FROM python:3.9-slim
ENV MPLBACKEND=Agg
COPY --from=builder /root/.local /root/.local
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

89
IEX_Symbol.py
View File

@@ -10,7 +10,6 @@ from typing import List, Optional, Tuple
import pandas as pd
import requests as r
import schedule
from fuzzywuzzy import fuzz
from Symbol import Stock
@@ -25,6 +24,7 @@ class IEX_Symbol:
searched_symbols = {}
otc_list = []
charts = {}
trending_cache = None
def __init__(self) -> None:
"""Creates a Symbol Object
@@ -36,6 +36,9 @@ class IEX_Symbol:
"""
try:
self.IEX_TOKEN = os.environ["IEX"]
if self.IEX_TOKEN == "TOKEN":
self.IEX_TOKEN = ""
except KeyError:
self.IEX_TOKEN = ""
warning(
@@ -67,6 +70,13 @@ class IEX_Symbol:
# Make sure API returned valid JSON
try:
resp_json = resp.json()
# IEX uses backtick ` as apostrophe which breaks telegram markdown parsing
if type(resp_json) is dict:
resp_json["companyName"] = resp_json.get("companyName", "").replace(
"`", "'"
)
return resp_json
except r.exceptions.JSONDecodeError as e:
logging.error(e)
@@ -105,6 +115,9 @@ class IEX_Symbol:
symbols = pd.concat([reg, otc])
# IEX uses backtick ` as apostrophe which breaks telegram markdown parsing
symbols["name"] = symbols["name"].str.replace("`", "'")
symbols["description"] = "$" + symbols["symbol"] + ": " + symbols["name"]
symbols["id"] = symbols["symbol"]
symbols["type_id"] = "$" + symbols["symbol"].str.lower()
@@ -122,6 +135,10 @@ class IEX_Symbol:
str
Human readable text on status of IEX API
"""
if self.IEX_TOKEN == "":
return "The `IEX_TOKEN` is not set so Stock Market data is not available."
resp = r.get(
"https://pjmps0c34hp7.statuspage.io/api/v2/status.json",
timeout=15,
@@ -140,46 +157,6 @@ class IEX_Symbol:
+ " 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.
@@ -265,7 +242,8 @@ class IEX_Symbol:
try:
IEXData = resp[0]
except IndexError as e:
return f"${symbol.id.upper()} either doesn't exist or pays no dividend."
logging.info(e)
return f"Getting dividend information for ${symbol.id.upper()} encountered an error. The provider for upcoming dividend information has been having issues recently which has likely caused this error. It is also possible that the stock has no dividend or does not exist."
keys = (
"amount",
"currency",
@@ -306,7 +284,7 @@ class IEX_Symbol:
+ 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."
return f"Getting dividend information for ${symbol.id.upper()} encountered an error. The provider for upcoming dividend information has been having issues recently which has likely caused this error. It is also possible that the stock has no dividend or does not exist."
def news_reply(self, symbol: Stock) -> str:
"""Gets most recent, english, non-paywalled news
@@ -402,7 +380,7 @@ class IEX_Symbol:
def cap_reply(self, symbol: Stock) -> str:
"""Get the Market Cap of a stock"""
if data := self.get(f"/stable/stock/{symbol.id}/stats"):
if data := self.get(f"/stock/{symbol.id}/stats"):
try:
cap = data["marketcap"]
@@ -484,6 +462,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.
@@ -494,9 +489,9 @@ class IEX_Symbol:
"""
if data := self.get(f"/stock/market/list/mostactive"):
return [
self.trending_cache = [
f"`${s['symbol']}`: {s['companyName']}, {100*s['changePercent']:.2f}%"
for s in data
]
else:
return ["Trending Stocks Currently Unavailable."]
return self.trending_cache

44
Symbol.py
View File

@@ -1,6 +1,5 @@
import functools
import requests as r
import pandas as pd
import logging
class Symbol:
@@ -8,6 +7,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 +17,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)}>"
@@ -28,31 +29,26 @@ class Symbol:
class Stock(Symbol):
"""Stock Market Object. Gets data from IEX Cloud"""
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.id = symbol
self.name = "$" + symbol.upper()
def __init__(self, symbol: pd.DataFrame) -> None:
if len(symbol) > 1:
logging.info(f"Crypto with shared id:\n\t{symbol.id}")
symbol = symbol.head(1)
# Used by Coin to change symbols for ids
coins = r.get("https://api.coingecko.com/api/v3/coins/list").json()
self.symbol = symbol.symbol.values[0]
self.id = symbol.id.values[0]
self.name = symbol.name.values[0]
self.tag = symbol.type_id.values[0].upper()
class Coin(Symbol):
"""Cryptocurrency Object. Gets data from CoinGecko."""
@functools.cache
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.get_data()
def __init__(self, symbol: pd.DataFrame) -> None:
if len(symbol) > 1:
logging.info(f"Crypto with shared id:\n\t{symbol.id}")
symbol = symbol.head(1)
def get_data(self) -> None:
self.id = list(filter(lambda coin: coin["symbol"] == self.symbol, coins))[0][
"id"
]
data = r.get("https://api.coingecko.com/api/v3/coins/" + self.id).json()
self.data = data
self.name = data["name"]
self.description = data["description"]
# self.price = data["market_data"]["current_price"][self.currency]
self.symbol = symbol.symbol.values[0]
self.id = symbol.id.values[0]
self.name = symbol.name.values[0]
self.tag = symbol.type_id.values[0].upper()

64
cg_Crypto.py
View File

@@ -3,12 +3,12 @@
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
@@ -22,6 +22,7 @@ class cg_Crypto:
vs_currency = "usd" # simple/supported_vs_currencies for list of options
searched_symbols = {}
trending_cache = None
def __init__(self) -> None:
"""Creates a Symbol Object
@@ -68,11 +69,14 @@ class cg_Crypto:
raw_symbols = self.get("/coins/list")
symbols = pd.DataFrame(data=raw_symbols)
# Removes all binance-peg symbols
symbols = symbols[~symbols["id"].str.contains("binance-peg")]
symbols["description"] = (
"$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
)
symbols = symbols[["id", "symbol", "name", "description"]]
symbols["type_id"] = "$$" + symbols["id"]
symbols["type_id"] = "$$" + symbols["symbol"]
self.symbol_list = symbols
if return_df:
@@ -97,45 +101,6 @@ class cg_Crypto:
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.
@@ -293,7 +258,7 @@ class cg_Crypto:
"include_market_cap": "true",
},
):
print(resp)
debug(resp)
try:
data = resp[coin.id]
@@ -336,6 +301,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
@@ -368,8 +345,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]:

545
functions.py
View File

@@ -1,545 +0,0 @@
import re
from datetime import datetime
from typing import Optional, List, Tuple, Dict
import pandas as pd
import requests as r
import schedule
from fuzzywuzzy import fuzz
class Symbol:
"""
Functions for finding stock market information about symbols.
"""
SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})"
searched_symbols = {}
charts = {}
license = re.sub(
r"\b\n",
" ",
r.get(
"https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot/-/raw/master/LICENSE"
).text,
)
help_text = """
Thanks for using this bot, consider supporting it by buying me a beer: https://www.buymeacoffee.com/Anson)
Keep up with the latest news for the bot in its Discord Server: https://discord.gg/VtEHyyTBAK
Full documentation on using and running your own stock bot can be found here: https://simple-stock-bots.gitlab.io/site
**Commands**
- /donate information on how to donate. 🎗️
- /dividend $[symbol] will return dividend information for the symbol. 📅
- /intra $[symbol] Plot of the stocks movement since the last market open. 📈
- /chart $[symbol] Plot of the stocks movement for the past 1 month. 📊
- /news $[symbol] News about the symbol. 📰
- /info $[symbol] General information about the symbol. ℹ️
- /stat $[symbol] Key statistics about the symbol. 🔢
- /help Get some help using the bot. 🆘
**Inline Features**
You can type @SimpleStockBot `[search]` in any chat or direct message to search for the stock bots
full list of stock symbols and return the price of the ticker. Then once you select the ticker
want the bot will send a message as you in that chat with the latest stock price.
The bot also looks at every message in any chat it is in for stock symbols.Symbols start with a
`$` followed by the stock symbol. For example:$tsla would return price information for Tesla Motors.
Market data is provided by [IEX Cloud](https://iexcloud.io)
If you believe the bot is not behaving properly run `/status`.
"""
donate_text = """
Simple Stock Bot is run entirely on donations.
All donations go directly towards paying for servers, and market data is provided by
[IEX Cloud](https://iexcloud.io/).
The best way to donate is through https://www.buymeacoffee.com/Anson which accepts Paypal or Credit card.
If you have any questions get in touch: MisterBiggs#0465 or[anson@ansonbiggs.com](http://mailto:anson@ansonbiggs.com/)
"""
def __init__(self, IEX_TOKEN: str) -> None:
"""Creates a Symbol Object
Parameters
----------
IEX_TOKEN : str
IEX Token
"""
self.IEX_TOKEN = IEX_TOKEN
if IEX_TOKEN != "":
self.get_symbol_list()
schedule.every().day.do(self.get_symbol_list)
schedule.every().day.do(self.clear_charts)
def clear_charts(self) -> None:
"""Clears cache of chart data."""
self.charts = {}
def get_symbol_list(self, return_df=False) -> Optional[pd.DataFrame]:
raw_symbols = r.get(
f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}"
).json()
symbols = pd.DataFrame(data=raw_symbols)
symbols["description"] = symbols["symbol"] + ": " + symbols["name"]
self.symbol_list = symbols
if return_df:
return symbols, datetime.now()
def iex_status(self) -> str:
"""Checks IEX Status dashboard for any current API issues.
Returns
-------
str
Human readable text on status of IEX API
"""
status = r.get("https://pjmps0c34hp7.statuspage.io/api/v2/status.json").json()[
"status"
]
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 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]]:
"""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 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, 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.
"""
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",
)
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."
if symbol.upper() == "GME":
message += "\n\n🙌💎Power to the Players💎🙌"
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:
IEXData = response.json()[0]
keys = (
"amount",
"currency",
"declaredDate",
"exDate",
"frequency",
"paymentDate",
"flag",
)
if set(keys).issubset(IEXData):
if IEXData["currency"] == "USD":
price = f"${IEXData['amount']}"
else:
price = f"{IEXData['amount']} {IEXData['currency']}"
# Pattern IEX uses for dividend date.
pattern = "%Y-%m-%d"
declared = datetime.strptime(IEXData["declaredDate"], pattern).strftime(
"%A, %B %w"
)
ex = datetime.strptime(IEXData["exDate"], pattern).strftime("%A, %B %w")
payment = datetime.strptime(IEXData["paymentDate"], pattern).strftime(
"%A, %B %w"
)
daysDelta = (
datetime.strptime(IEXData["paymentDate"], pattern) - datetime.now()
).days
return (
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:
newsMessages[
symbol
] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
return newsMessages
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.
"""
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:
"""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.upper() not in list(self.symbol_list["symbol"]):
return pd.DataFrame()
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/intraday-prices?token={self.IEX_TOKEN}"
response = r.get(IEXurl)
if response.status_code == 200:
df = pd.DataFrame(response.json())
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
df = df.set_index("DT")
return df
return pd.DataFrame()
def chart_reply(self, symbol: 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.
"""
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(
f"https://cloud.iexapis.com/stable/stock/{symbol}/chart/1mm?token={self.IEX_TOKEN}&chartInterval=3&includeToday=false"
)
if response.status_code == 200:
df = pd.DataFrame(response.json())
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
df = df.set_index("DT")
self.charts[symbol.upper()] = df
return df
return 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.
"""
infoMessages = {}
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:
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 ""

5
requirements.txt
View File

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

198
symbol_router.py
View File

@@ -2,12 +2,14 @@
"""
import datetime
import logging
import random
import re
from logging import critical, debug, error, info, warning
import pandas as pd
from fuzzywuzzy import fuzz
import schedule
from cachetools import TTLCache, cached
from cg_Crypto import cg_Crypto
from IEX_Symbol import IEX_Symbol
@@ -17,13 +19,33 @@ 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()
def find_symbols(self, text: str) -> list[Symbol]:
schedule.every().hour.do(self.trending_decay)
def trending_decay(self, decay=0.5):
"""Decays the value of each trending stock by a multiplier"""
t_copy = {}
dead_keys = []
if self.trending_count:
t_copy = self.trending_count.copy()
for key in t_copy.keys():
if t_copy[key] < 0.01:
# This just makes sure were not keeping around keys that havent been called in a very long time.
dead_keys.append(key)
else:
t_copy[key] = t_copy[key] * decay
for dead in dead_keys:
t_copy.pop(dead)
self.trending_count = t_copy.copy()
info("Decayed trending symbols.")
def find_symbols(self, text: str, *, trending_weight: int = 1) -> 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.
@@ -34,25 +56,42 @@ class Router:
Returns
-------
list[str]
List of stock symbols as strings without dollar sign.
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))
sym = self.stock.symbol_list[
self.stock.symbol_list["symbol"].str.fullmatch(stock, case=False)
]
if ~sym.empty:
print(sym)
symbols.append(Stock(sym))
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()))
sym = self.crypto.symbol_list[
self.crypto.symbol_list["symbol"].str.fullmatch(
coin.lower(), case=False
)
]
if ~sym.empty:
symbols.append(Coin(sym))
else:
info(f"{coin} is not in list of coins")
info(symbols)
return symbols
if symbols:
info(symbols)
for symbol in symbols:
self.trending_count[symbol.tag] = (
self.trending_count.get(symbol.tag, 0) + trending_weight
)
return symbols
def status(self, bot_resp) -> str:
"""Checks for any issues with APIs.
@@ -78,45 +117,7 @@ class Router:
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]]:
def inline_search(self, search: str, matches: int = 5) -> pd.DataFrame:
"""Searches based on the shortest symbol that contains the same string as the search.
Should be very fast compared to a fuzzy search.
@@ -133,16 +134,16 @@ class Router:
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
search = search.lower()
df = df[
df["description"].str.contains(search, regex=False, case=False)
].sort_values(by="type_id", key=lambda x: x.str.len())
df = df[df["type_id"].str.contains(search, regex=False)].sort_values(
by="type_id", key=lambda x: x.str.len()
symbols = df.head(matches)
symbols["price_reply"] = symbols["type_id"].apply(
lambda sym: self.price_reply(self.find_symbols(sym, trending_weight=0))[0]
)
symbols = df.head(20)
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
self.searched_symbols[search] = symbol_list
return symbol_list
return symbols
def price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol.
@@ -192,7 +193,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
@@ -221,7 +222,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
@@ -247,7 +248,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
@@ -271,7 +272,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:
@@ -294,7 +295,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]:
@@ -319,7 +320,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
@@ -345,10 +346,36 @@ 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
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
@cached(cache=TTLCache(maxsize=1024, ttl=600))
def trending(self) -> str:
"""Checks APIs for trending symbols.
@@ -361,17 +388,40 @@ class Router:
stocks = self.stock.trending()
coins = self.crypto.trending()
reply = "Trending Stocks:\n"
reply += "-" * len("Trending Stocks:") + "\n"
for stock in stocks:
reply += stock + "\n"
reply = ""
reply += "\n\nTrending Crypto:\n"
reply += "-" * len("Trending Crypto:") + "\n"
for coin in coins:
reply += coin + "\n"
if self.trending_count:
reply += "🔥Trending on the Stock Bot:\n`"
reply += "━" * len("Trending on the Stock Bot:") + "`\n"
return reply
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.spark_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 "`$GME" in reply:
reply = reply.replace("🔥", "🦍")
if reply:
return reply
else:
warning("Failed to collect trending data.")
return "Trending data is not currently available."
def random_pick(self) -> str:
@@ -409,7 +459,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