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

Close #17 and worked on #3

Browse Source
This commit is contained in:
Anson Biggs 2021-02-11 16:56:42 -07:00
parent 1a422b3eca
commit 9e02d692f4
3 changed files with 512 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
bot.py
View File

@ -1,78 +1,94 @@
import discord
from functions import Symbol
import os
import discord
from discord.ext import commands
from functions import Symbol
client = discord.Client()
s = Symbol(os.environ["IEX"])
DISCORD_TOKEN = os.environ["DISCORD"]
try:
IEX_TOKEN = os.environ["IEX"]
except KeyError:
IEX_TOKEN = ""
print("Starting without an IEX Token will not allow you to get market data!")
@client.event
intents = discord.Intents.default()
intents.members = True
bot = commands.Bot(
command_prefix="/",
description="Simple bot for getting stock market information.",
intents=intents,
)
@bot.event
async def on_ready():
print("We have logged in as {0.user}".format(client))
print("Starting Simple Stock Bot")
print("Logged in as")
print(bot.user.name)
print(bot.user.id)
print("------")
@client.event
@bot.event
async def on_message(message):
if message.author == client.user:
return
# Check for dividend command
if message.content.startswith("/dividend"):
replies = s.dividend_reply(s.find_symbols(message.content))
if replies:
for reply in replies.items():
if "$" in message.content:
symbols = s.find_symbols(message.content)
if symbols:
for reply in s.price_reply(symbols).items():
await message.channel.send(reply[1])
else:
await message.channel.send(
"Command requires a ticker. See /help for more information."
)
elif message.content.startswith("/news"):
replies = s.news_reply(s.find_symbols(message.content))
if replies:
for reply in replies.items():
await message.channel.send(reply[1])
else:
await message.channel.send(
"Command requires a ticker. See /help for more information."
)
elif message.content.startswith("/info"):
replies = s.info_reply(s.find_symbols(message.content))
if replies:
for reply in replies.items():
await message.channel.send(reply[1])
else:
await message.channel.send(
"Command requires a ticker. See /help for more information."
)
elif message.content.startswith("/search"):
queries = s.search_symbols(message.content[7:])[:6]
if queries:
reply = "*Search Results:*\n`$ticker: Company Name`\n"
for query in queries:
reply += "`" + query[1] + "`\n"
await message.channel.send(reply)
else:
await message.channel.send(
"Command requires a query. See /help for more information."
)
elif message.content.startswith("/help"):
"""Send link to docs when the command /help is issued."""
await message.channel.send(s.help_text)
# If no commands, check for any tickers.
else:
replies = s.price_reply(s.find_symbols(message.content))
if replies:
for reply in replies.items():
await message.channel.send(reply[1])
else:
return
client.run(os.environ["DISCORD"])
@bot.command(description="Information on how to donate.")
async def donate(ctx, cmd: str):
print("donate")
await ctx.send("donate:" + cmd)
@bot.command()
async def dividend(ctx, cmd: str):
await ctx.send("dividend:" + cmd)
@bot.command()
async def intra(ctx, cmd: str):
await ctx.send("intra:" + cmd)
@bot.command()
async def chart(ctx, cmd: str):
await ctx.send("chart:" + cmd)
@bot.command()
async def news(ctx, cmd: str):
await ctx.send("news:" + cmd)
@bot.command()
async def info(ctx, cmd: str):
await ctx.send("info:" + cmd)
@bot.command()
async def stat(ctx, cmd: str):
await ctx.send("stat:" + cmd)
# @bot.command()
# async def help(ctx, cmd: str):
# await ctx.send("help:" + cmd)
s = Symbol(IEX_TOKEN)
bot.run(DISCORD_TOKEN)

512
functions.py
View File

@ -1,11 +1,11 @@
import json
import re
from datetime import datetime, timedelta
from datetime import datetime
from typing import Optional, List, Tuple, Dict
import pandas as pd
import requests as r
from fuzzywuzzy import fuzz
import schedule
from fuzzywuzzy import fuzz
class Symbol:
@ -16,36 +16,79 @@ class Symbol:
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)
Full documentation can be found [here.](https://simple-stock-bots.gitlab.io/site/)
Keep up with the latest news for the bot in itsTelegram Channel: https://t.me/simplestockbotnews
Full documentation on using and running your own stock bot can be found [here.](https://simple-stock-bots.gitlab.io/site)
**Commands**
- /dividend `$[symbol]` will return dividend information for the symbol.
- /news `$[symbol]` will return news about the symbol.
- /info `$[symbol]` will return general information about the symbol.
- /search `query` Takes a search string, whether a company name or ticker and returns a list of companies that are supported by the bot.
- /donate [amount in USD] 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. 🆘
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.
**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)
`Market data is provided by [IEX Cloud](https://iexcloud.io)`
If you believe the bot is not behaving properly run `/status`.
"""
def __init__(self, IEX_TOKEN: str):
donate_text = """
Simple Stock Bot is run entirely on donations[.](https://www.buymeacoffee.com/Anson)
All donations go directly towards paying for servers, and market data is provided by
[IEX Cloud](https://iexcloud.io/).
The easiest way to donate is to run the `/donate [amount in USD]` command with USdollars you would like to donate.
Example: `/donate 2` would donate 2 USD.
An alternative 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 or[anson@ansonbiggs.com](http://mailto:anson@ansonbiggs.com/)
_Donations can only be made in a chat directly with @simplestockbot_
"""
def __init__(self, IEX_TOKEN: str) -> None:
"""Creates a Symbol Object
Parameters
----------
IEX_TOKEN : str
IEX Token
"""
self.IEX_TOKEN = IEX_TOKEN
self.get_symbol_list()
schedule.every().day.do(self.get_symbol_list)
if IEX_TOKEN != "":
self.get_symbol_list()
def get_symbol_list(self, return_df=False):
"""
Fetches a list of stock market symbols from FINRA
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]:
Returns:
pd.DataFrame -- [DataFrame with columns: Symbol | Issue_Name | Primary_Listing_Mkt
datetime -- The time when the list of symbols was fetched. The Symbol list is updated every open and close of every trading day.
"""
raw_symbols = r.get(
f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}"
).json()
@ -56,16 +99,62 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
if return_df:
return symbols, datetime.now()
def search_symbols(self, search: str):
"""
Performs a fuzzy search to find stock symbols closest to a search term.
def iex_status(self) -> str:
"""Checks IEX Status dashboard for any current API issues.
Arguments:
search {str} -- String used to search, could be a company name or something close to the companies stock ticker.
Returns:
List of Tuples -- A list tuples of every stock sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
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
@ -75,13 +164,15 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
symbols = self.symbol_list
symbols["Match"] = symbols.apply(
lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()), axis=1,
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,
lambda x: fuzz.partial_ratio(search, x["name"].lower()),
axis=1,
)
symbols.sort_values(by="Match", ascending=False, inplace=True)
@ -90,28 +181,36 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
self.searched_symbols[search] = symbol_list
return symbol_list
def find_symbols(self, text: 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? -> ['tsla']
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?
Arguments:
text {str} -- Blob of text that might contain tickers with the format: $TICKER
Parameters
----------
text : str
Blob of text.
Returns:
list -- List of every found match without the dollar sign.
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):
"""
Takes a list of symbols and replies with Markdown formatted text about the symbols price change for the day.
def price_reply(self, symbols: list) -> Dict[str, str]:
"""Returns current market price or after hours if its available for a given stock symbol.
Arguments:
symbols {list} -- List of stock market symbols.
Parameters
----------
symbols : list
List of stock symbols.
Returns:
dict -- Dictionary with keys of symbols and values of markdown formatted text example: {'tsla': 'The current stock price of Tesla Motors is $**420$$, the stock price is currently **up 42%**}
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:
@ -120,15 +219,46 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
response = r.get(IEXurl)
if response.status_code == 200:
IEXData = response.json()
message = f"The current stock price of {IEXData['companyName']} is $**{IEXData['latestPrice']}**"
# Determine wording of change text
change = round(IEXData["changePercent"] * 100, 2)
if change > 0:
message += f", the stock is currently **up {change}%**"
elif change < 0:
message += f", the stock is currently **down {change}%**"
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 += ", the stock hasn't shown any movement today."
message = f"The symbol: {symbol} encountered and error. This could be due to "
else:
message = f"The symbol: {symbol} was not found."
@ -136,52 +266,95 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
return dataMessages
def dividend_reply(self, symbols: list):
divMessages = {}
def dividend_reply(self, symbol: str) -> Dict[str, str]:
"""Returns the most recent, or next dividend date for a stock symbol.
for symbol in symbols:
IEXurl = f"https://cloud.iexapis.com/stable/data-points/{symbol}/NEXTDIVIDENDDATE?token={self.IEX_TOKEN}"
response = r.get(IEXurl)
if response.status_code == 200:
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/msft/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']}"
# extract date from json
date = response.json()
# Pattern IEX uses for dividend date.
pattern = "%Y-%m-%d"
divDate = datetime.strptime(date, pattern)
daysDelta = (divDate - datetime.now()).days
datePretty = divDate.strftime("%A, %B %w")
if daysDelta < 0:
divMessages[
symbol
] = f"{symbol.upper()} dividend was on {datePretty} and a new date hasn't been announced yet."
elif daysDelta > 0:
divMessages[
symbol
] = f"{symbol.upper()} dividend is on {datePretty} which is in {daysDelta} Days."
else:
divMessages[symbol] = f"{symbol.upper()} is today."
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"
)
else:
divMessages[
symbol
] = f"{symbol} either doesn't exist or pays no dividend."
daysDelta = (
datetime.strptime(IEXData["paymentDate"], pattern) - datetime.now()
).days
return divMessages
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}"
)
def news_reply(self, symbols: list):
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/3?token={self.IEX_TOKEN}"
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()
newsMessages[symbol] = f"News for **{symbol.upper()}**:\n"
for news in data:
message = f"\t[{news['headline']}]({news['url']})\n\n"
newsMessages[symbol] = newsMessages[symbol] + message
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
@ -189,7 +362,19 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
return newsMessages
def info_reply(self, symbols: list):
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:
@ -198,9 +383,10 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
if response.status_code == 200:
data = response.json()
infoMessages[
symbol
] = f"Company Name: [{data['companyName']}]({data['website']})\nIndustry: {data['industry']}\nSector: {data['sector']}\nCEO: {data['CEO']}\nDescription: {data['description']}\n"
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[
@ -208,3 +394,153 @@ The bot also looks at every message in any chat it is in for stock symbols. Symb
] = 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 ""

7
requirements.txt
View File

@ -1,6 +1,7 @@
discord.py==1.6
requests==2.23.0
pandas==1.0.3
requests==2.25.1
pandas==1.2.1
fuzzywuzzy==0.18.0
python-Levenshtein==0.12.1
schedule==0.6.0
schedule==1.0.0
mplfinance==0.12.7a5