mirror of
https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot.git
synced 2025-06-16 15:06:53 +00:00
343 lines
13 KiB
Python
343 lines
13 KiB
Python
import json
|
||
import os
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
|
||
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 Telegram 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. 📅
|
||
- /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 you 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)
|
||
"""
|
||
|
||
def __init__(self, IEX_TOKEN: str):
|
||
self.IEX_TOKEN = IEX_TOKEN
|
||
self.get_symbol_list()
|
||
|
||
schedule.every().monday.do(self.get_symbol_list)
|
||
schedule.every().day.do(self.clear_charts)
|
||
|
||
def clear_charts(self):
|
||
self.charts = {}
|
||
|
||
def get_symbol_list(self, return_df=False):
|
||
"""
|
||
Fetches a list of stock market symbols from FINRA
|
||
|
||
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()
|
||
symbols = pd.DataFrame(data=raw_symbols)
|
||
|
||
symbols["description"] = symbols["symbol"] + ": " + symbols["name"]
|
||
self.symbol_list = symbols
|
||
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.
|
||
|
||
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).
|
||
"""
|
||
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):
|
||
"""
|
||
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']
|
||
|
||
Arguments:
|
||
text {str} -- Blob of text that might contain tickers with the format: $TICKER
|
||
|
||
Returns:
|
||
list -- List of every found match without the 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.
|
||
|
||
Arguments:
|
||
symbols {list} -- List of stock market 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%**}
|
||
"""
|
||
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()
|
||
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}%**"
|
||
else:
|
||
message += ", the stock hasn't shown any movement today."
|
||
else:
|
||
message = f"The symbol: {symbol} was not found."
|
||
|
||
dataMessages[symbol] = message
|
||
|
||
return dataMessages
|
||
|
||
def dividend_reply(self, symbols: list):
|
||
divMessages = {}
|
||
|
||
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:
|
||
|
||
# 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."
|
||
|
||
else:
|
||
divMessages[
|
||
symbol
|
||
] = f"{symbol} either doesn't exist or pays no dividend."
|
||
|
||
return divMessages
|
||
|
||
def news_reply(self, symbols: list):
|
||
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):
|
||
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: {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):
|
||
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
|
||
|
||
def chart_reply(self, symbol: str):
|
||
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
|
||
|
||
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/chart/1mm?token={self.IEX_TOKEN}&chartInterval=3&includeToday=false"
|
||
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")
|
||
self.charts[symbol.upper()] = df
|
||
return df
|
||
|
||
def stat_reply(self, symbols: list):
|
||
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):
|
||
"""Get quote for a cryptocurrency pair.
|
||
|
||
Args:
|
||
pair (string): Cryptocurrency
|
||
"""
|
||
|
||
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 False
|