mirror of
https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot.git
synced 2025-06-16 15:06:53 +00:00
Merge branch 'IEX-2.0' into 'master'
Iex 2.0 See merge request simple-stock-bots/simple-telegram-bot!12
This commit is contained in:
commit
4f5f56f874
77
bot.py
77
bot.py
@ -3,11 +3,10 @@ import logging
|
||||
import os
|
||||
|
||||
import telegram
|
||||
from functions import *
|
||||
from telegram.ext import CommandHandler, Filters, MessageHandler, Updater
|
||||
|
||||
from functions import *
|
||||
|
||||
TOKEN = os.environ["TELEGRAM"]
|
||||
TELEGRAM_TOKEN = os.environ["TELEGRAM"]
|
||||
TICKER_REGEX = "[$]([a-zA-Z]{1,4})"
|
||||
|
||||
# Enable logging
|
||||
@ -38,72 +37,36 @@ def tickerDetect(bot, update):
|
||||
"""
|
||||
message = update.message.text
|
||||
chat_id = update.message.chat_id
|
||||
|
||||
# Let user know bot is working
|
||||
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
|
||||
|
||||
tickers = getTickers(message)
|
||||
|
||||
data = tickerData(tickers) if tickers else {}
|
||||
if tickers:
|
||||
# Let user know bot is working
|
||||
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
|
||||
|
||||
for ticker in data:
|
||||
for symbol, reply in tickerDataReply(tickers).items():
|
||||
|
||||
# Keep track of which tickers had a return from tickerData()
|
||||
if ticker.lower() in tickers:
|
||||
tickers.remove(ticker.lower())
|
||||
|
||||
reply = tickerDataReply(data[ticker])
|
||||
update.message.reply_text(text=reply, parse_mode=telegram.ParseMode.MARKDOWN)
|
||||
|
||||
# For any tickers that didnt have data, return that they don't exist.
|
||||
for ticker in tickers:
|
||||
update.message.reply_text(
|
||||
ticker.upper()
|
||||
+ " does not exist, you should search for a real stock like $PSEC"
|
||||
)
|
||||
|
||||
|
||||
def news(bot, update):
|
||||
"""
|
||||
/news
|
||||
Returns a small snippet of general information, and any news articles that are found.
|
||||
"""
|
||||
message = update.message.text
|
||||
chat_id = update.message.chat_id
|
||||
|
||||
# Let user know bot is working
|
||||
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
|
||||
|
||||
tickers = getTickers(message)
|
||||
|
||||
news = tickerNews(tickers) if tickers else {}
|
||||
|
||||
for ticker in news:
|
||||
|
||||
reply = tickerNewsReply(news[ticker])
|
||||
update.message.reply_text(text=reply, parse_mode=telegram.ParseMode.MARKDOWN)
|
||||
|
||||
# Keep track of which tickers had a return from tickerData()
|
||||
if ticker.lower() in tickers:
|
||||
tickers.remove(ticker.lower())
|
||||
update.message.reply_text(
|
||||
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
|
||||
)
|
||||
|
||||
|
||||
def dividend(bot, update):
|
||||
"""
|
||||
This Functions is incomplete.
|
||||
waits for /dividend or /div command and then finds dividend info on that ticker.
|
||||
"""
|
||||
message = update.message.text
|
||||
chat_id = update.message.chat_id
|
||||
tickers = getTickers(message)
|
||||
|
||||
# regex to find tickers in messages, looks for up to 4 word characters following a dollar sign and captures the 4 word characters
|
||||
tickers = re.findall(TICKER_REGEX, message)
|
||||
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
|
||||
|
||||
for ticker in tickers:
|
||||
message = tickerDividend(ticker)
|
||||
update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN)
|
||||
if tickers:
|
||||
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
|
||||
|
||||
for symbol, reply in tickerDividend(tickers).items():
|
||||
|
||||
update.message.reply_text(
|
||||
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
|
||||
)
|
||||
|
||||
def error(bot, update, error):
|
||||
"""Log Errors caused by Updates."""
|
||||
logger.warning('Update "%s" caused error "%s"', update, error)
|
||||
@ -112,7 +75,7 @@ def error(bot, update, error):
|
||||
def main():
|
||||
"""Start the bot."""
|
||||
# Create the EventHandler and pass it your bot's token.
|
||||
updater = Updater(TOKEN)
|
||||
updater = Updater(TELEGRAM_TOKEN)
|
||||
|
||||
# Get the dispatcher to register handlers
|
||||
dp = updater.dispatcher
|
||||
@ -120,8 +83,8 @@ def main():
|
||||
# on different commands - answer in Telegram
|
||||
dp.add_handler(CommandHandler("start", start))
|
||||
dp.add_handler(CommandHandler("help", help))
|
||||
dp.add_handler(CommandHandler("news", news))
|
||||
dp.add_handler(CommandHandler("dividend", dividend))
|
||||
dp.add_handler(CommandHandler("div", dividend))
|
||||
|
||||
# on noncommand i.e message - echo the message on Telegram
|
||||
dp.add_handler(MessageHandler(Filters.text, tickerDetect))
|
||||
|
||||
229
functions.py
229
functions.py
@ -1,10 +1,11 @@
|
||||
import urllib.request
|
||||
import json
|
||||
from datetime import datetime
|
||||
import time
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
IEX_TOKEN = os.environ["IEX"]
|
||||
def getTickers(text: str):
|
||||
"""
|
||||
Takes a blob of text and returns any stock tickers found.
|
||||
@ -15,181 +16,63 @@ def getTickers(text: str):
|
||||
return list(set(re.findall(TICKER_REGEX, text)))
|
||||
|
||||
|
||||
def tickerData(tickers: list):
|
||||
def tickerDataReply(tickers: list):
|
||||
"""
|
||||
Takes a list of tickers and returns a dictionary of information on ticker.
|
||||
example:
|
||||
input list: ["aapl","tsla"]
|
||||
returns:
|
||||
{
|
||||
"AAPL": {"name": "Apple Inc.", "price": 200.72, "change": -0.01074},
|
||||
"TSLA": {"name": "Tesla Inc.", "price": 241.98, "change": -0.01168},
|
||||
}
|
||||
Takes a list of tickers and returns a list of strings with information about the ticker.
|
||||
"""
|
||||
tickerReplies = {}
|
||||
for ticker in tickers:
|
||||
IEXURL = (
|
||||
f"https://cloud.iexapis.com/stable/stock/{ticker}/quote?token={IEX_TOKEN}"
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(IEXURL) as url:
|
||||
IEXData = json.loads(url.read().decode())
|
||||
|
||||
stockData = {}
|
||||
IEXURL = (
|
||||
"https://api.iextrading.com/1.0/stock/market/batch?symbols="
|
||||
+ ",".join(tickers)
|
||||
+ "&types=quote"
|
||||
)
|
||||
with urllib.request.urlopen(IEXURL) as url:
|
||||
IEXData = json.loads(url.read().decode())
|
||||
reply = 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:
|
||||
reply += f", the stock is currently **up {change}%**"
|
||||
elif change < 0:
|
||||
reply += f", the stock is currently **down {change}%**"
|
||||
else:
|
||||
reply += ", the stock hasn't shown any movement today."
|
||||
except:
|
||||
reply = f"The ticker: {ticker} was not found."
|
||||
|
||||
tickerReplies[ticker] = reply
|
||||
|
||||
return tickerReplies
|
||||
|
||||
|
||||
def tickerDividend(tickers: list):
|
||||
messages = {}
|
||||
|
||||
for ticker in tickers:
|
||||
ticker = ticker.upper()
|
||||
IEXurl = f"https://cloud.iexapis.com/stable/stock/{ticker}/dividends/next?token={IEX_TOKEN}"
|
||||
with urllib.request.urlopen(IEXurl) as url:
|
||||
data = json.loads(url.read().decode())
|
||||
if data:
|
||||
# Pattern IEX uses for dividend date.
|
||||
pattern = "%Y-%m-%d"
|
||||
|
||||
# Makes sure ticker exists before populating a dictionary
|
||||
if ticker in IEXData:
|
||||
stockData[ticker] = {}
|
||||
stockData[ticker]["name"] = IEXData[ticker]["quote"]["companyName"]
|
||||
stockData[ticker]["price"] = IEXData[ticker]["quote"]["latestPrice"]
|
||||
stockData[ticker]["change"] = IEXData[ticker]["quote"]["changePercent"]
|
||||
# Convert divDate to seconds, and subtract it from current time.
|
||||
dividendSeconds = datetime.strptime(
|
||||
data["paymentDate"], pattern
|
||||
).timestamp()
|
||||
difference = dividendSeconds - int(time.time())
|
||||
|
||||
return stockData
|
||||
# Calculate (d)ays, (h)ours, (m)inutes, and (s)econds
|
||||
d, h = divmod(difference, 86400)
|
||||
h, m = divmod(h, 3600)
|
||||
m, s = divmod(m, 60)
|
||||
|
||||
messages[
|
||||
ticker
|
||||
] = f"{data['description']}\n\nThe dividend is in: {d:.0f} Days {h:.0f} Hours {m:.0f} Minutes {s:.0f} Seconds."
|
||||
else:
|
||||
messages[ticker] = f"{ticker} either doesn't exist or pays no dividend."
|
||||
|
||||
def tickerDataReply(ticker: dict):
|
||||
"""
|
||||
Takes a dictionary, likely produced from tickerData(), and returns a markdown string with information on the ticker.
|
||||
example:
|
||||
input dict: {"AAPL": {"name": "Apple Inc.", "price": 200.72, "change": -0.01074}}
|
||||
returns:
|
||||
The current stock price of Apple Inc. is $**200.72**, the stock is currently **down -1.07**%
|
||||
"""
|
||||
reply = f"The current stock price of {ticker['name']} is $**{ticker['price']}**"
|
||||
|
||||
# Determine wording of change text
|
||||
change = round(ticker["change"] * 100, 2)
|
||||
if change > 0:
|
||||
changeText = f", the stock is currently **up {change}%**"
|
||||
elif change < 0:
|
||||
changeText = f", the stock is currently **down {change}%**"
|
||||
else:
|
||||
changeText = ", the stock hasn't shown any movement today."
|
||||
|
||||
return reply + changeText
|
||||
|
||||
|
||||
def tickerNews(tickers: list):
|
||||
"""
|
||||
Takes a list of ticker and returns a dictionary of dictionarys
|
||||
with a list of tuples under news that contains (title, url)
|
||||
example:
|
||||
input list: ["aapl"]
|
||||
returns:
|
||||
{
|
||||
"AAPL": {
|
||||
"name": "Apple Inc.",
|
||||
"price": 200.72,
|
||||
"change": -0.01074,
|
||||
"news": [
|
||||
(
|
||||
"April Dividend Income Report - Beating Records, Holding Steady",
|
||||
"https://api.iextrading.com/1.0/stock/aapl/article/8556681822768653",
|
||||
),
|
||||
(
|
||||
"Is Vanguard's VIG Better Than Its WisdomTree Counterpart?",
|
||||
"https://api.iextrading.com/1.0/stock/aapl/article/7238581261167527",
|
||||
),
|
||||
],
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
data = tickerData(tickers)
|
||||
|
||||
for ticker in tickers:
|
||||
ticker = ticker.upper()
|
||||
IEXNews = f"https://api.iextrading.com/1.0/stock/{ticker}/news/last/5"
|
||||
with urllib.request.urlopen(IEXNews) as url:
|
||||
newsData = json.loads(url.read().decode())
|
||||
|
||||
data[ticker]["news"] = []
|
||||
for index, story in enumerate(newsData):
|
||||
tup = (newsData[index]["headline"], newsData[index]["url"])
|
||||
data[ticker]["news"].append(tup)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def tickerNewsReply(ticker: dict):
|
||||
"""
|
||||
Takes a dictionary, likely produced from tickerNews(), and returns a markdown string with news information on the ticker.
|
||||
example:
|
||||
{
|
||||
"name": "Apple Inc.",
|
||||
"price": 200.72,
|
||||
"change": -0.01074,
|
||||
"news": [
|
||||
(
|
||||
"April Dividend Income Report - Beating Records, Holding Steady",
|
||||
"https://api.iextrading.com/1.0/stock/aapl/article/8556681822768653",
|
||||
),
|
||||
(
|
||||
"Is Vanguard's VIG Better Than Its WisdomTree Counterpart?",
|
||||
"https://api.iextrading.com/1.0/stock/aapl/article/7238581261167527",
|
||||
),
|
||||
],
|
||||
}
|
||||
returns:
|
||||
The current stock price of Apple Inc. is $**200.72**, the stock is currently **down -1.07%**
|
||||
[April Dividend Income Report - Beating Records, Holding Steady](https://api.iextrading.com/1.0/stock/aapl/article/8556681822768653)
|
||||
[Is Vanguard's VIG Better Than Its WisdomTree Counterpart?](https://api.iextrading.com/1.0/stock/aapl/article/7238581261167527)
|
||||
"""
|
||||
reply = tickerDataReply(ticker)
|
||||
if len(ticker["news"]) == 0:
|
||||
return reply + f"\n\tNo News was found for {ticker['name']}"
|
||||
for title, url in ticker["news"]:
|
||||
reply = reply + f"\n\t[{title}]({url})"
|
||||
return reply
|
||||
|
||||
|
||||
# Below Functions are incomplete
|
||||
|
||||
|
||||
def tickerInfo(ticker):
|
||||
infoURL = f"https://api.iextrading.com/1.0/stock/{ticker}/stats"
|
||||
|
||||
with urllib.request.urlopen(infoURL) as url:
|
||||
data = json.loads(url.read().decode())
|
||||
|
||||
info = {}
|
||||
|
||||
info["companyName"] = data["companyName"]
|
||||
info["marketCap"] = data["marketcap"]
|
||||
info["yearHigh"] = data["week52high"]
|
||||
info["yearLow"] = data["week52low"]
|
||||
info["divRate"] = data["dividendRate"]
|
||||
info["divYield"] = data["dividendYield"]
|
||||
info["divDate"] = data["exDividendDate"]
|
||||
|
||||
return info
|
||||
|
||||
|
||||
def tickerDividend(ticker):
|
||||
data = tickerInfo(ticker)
|
||||
if data["divDate"] == 0:
|
||||
return "{} has no dividend.".format(data["companyName"])
|
||||
|
||||
dividendInfo = "{} current dividend yield is: {:.3f}%, or ${:.3f} per share.".format(
|
||||
data["companyName"], data["divRate"], data["divYield"]
|
||||
)
|
||||
|
||||
divDate = data["divDate"]
|
||||
|
||||
# Pattern IEX uses for dividend date.
|
||||
pattern = "%Y-%m-%d %H:%M:%S.%f"
|
||||
|
||||
# Convert divDate to seconds, and subtract it from current time.
|
||||
divSeconds = datetime.strptime(divDate, pattern).timestamp()
|
||||
difference = divSeconds - int(time.time())
|
||||
|
||||
# Calculate (d)ays, (h)ours, (m)inutes, and (s)econds
|
||||
d, h = divmod(difference, 86400)
|
||||
h, m = divmod(h, 3600)
|
||||
m, s = divmod(m, 60)
|
||||
|
||||
countdownMessage = f"\n\nThe dividend is in: {d:.0f} Days {h:.0f} Hours {m:.0f} Minutes {s:.0f} Seconds."
|
||||
|
||||
return dividendInfo + countdownMessage
|
||||
return messages
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user