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

rewrote entire bot code. Functionally identical.

This commit is contained in:
Anson 2019-05-10 00:21:56 -07:00
parent d7d0962eea
commit 1f474dca46
4 changed files with 315 additions and 287 deletions

135
bot.py Normal file
View File

@ -0,0 +1,135 @@
# Work with Python 3.7
import logging
import re
import telegram
from telegram.ext import CommandHandler, Filters, MessageHandler, Updater
import credentials
from functions import *
TOKEN = credentials.secrets["TELEGRAM_TOKEN"]
TICKER_REGEX = "[$]([a-zA-Z]{1,4})"
# Enable logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
print("Bot Online")
# Define a few command handlers. These usually take the two arguments bot and
# update. Error handlers also receive the raised TelegramError object in error.
def start(bot, update):
"""Send a message when the command /start is issued."""
update.message.reply_text("I am started and ready to go!")
def help(bot, update):
"""Send link to docs when the command /help is issued."""
message = "[Please see the docs for Bot information](https://misterbiggs.gitlab.io/simple-telegram-bot)"
update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN)
def tickerDetect(bot, update):
"""
Runs on any message that doesn't have a command and searches for tickers, then returns the prices of any tickers found.
"""
message = update.message.text
chat_id = update.message.chat_id
tickers = re.findall(TICKER_REGEX, message)
data = tickerData(tickers) if tickers else {}
for ticker in data:
# 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
tickers = re.findall(TICKER_REGEX, 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())
def dividend(bot, update):
"""
This Functions is incomplete.
"""
message = update.message.text
chat_id = update.message.chat_id
# 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
)
def error(bot, update, error):
"""Log Errors caused by Updates."""
logger.warning('Update "%s" caused error "%s"', update, error)
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
updater = Updater(TOKEN)
# Get the dispatcher to register handlers
dp = updater.dispatcher
# 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))
# on noncommand i.e message - echo the message on Telegram
dp.add_handler(MessageHandler(Filters.text, tickerDetect))
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == "__main__":
main()

180
functions.py Normal file
View File

@ -0,0 +1,180 @@
import urllib.request
import json
from datetime import datetime
import time
def tickerData(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},
}
"""
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())
for ticker in tickers:
ticker = ticker.upper()
# 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"]
return stockData
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

View File

@ -1,182 +0,0 @@
# Work with Python 3.7
import json
import logging
import re
import urllib.request
import telegram
from telegram.ext import CommandHandler, Filters, MessageHandler, Updater
import credentials
import tickerInfo
TOKEN = credentials.secrets["TELEGRAM_TOKEN"]
TICKER_REGEX = "[$]([a-zA-Z]{1,4})"
# Enable logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
print("Bot Online")
# Define a few command handlers. These usually take the two arguments bot and
# update. Error handlers also receive the raised TelegramError object in error.
def start(bot, update):
"""Send a message when the command /start is issued."""
update.message.reply_text("I am started and ready to go!")
def help(bot, update):
"""Send link to docs when the command /help is issued."""
message = "[Please see the docs for Bot information](https://misterbiggs.gitlab.io/simple-telegram-bot)"
update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN)
def news(bot, update):
"""Send a message when the /news command is issued."""
message = update.message.text
chat_id = update.message.chat_id
try:
# 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)
## Checks if a ticker was passed in
if tickers == []:
message = "No Ticker, showing Market News:"
news = tickerInfo.stockNews("market")
for i in range(len(news["title"])):
message = f"{message}\n\n[{news['title'][i]}]({news['link'][i]})"
update.message.reply_text(
text=message, parse_mode=telegram.ParseMode.MARKDOWN
)
else:
tickerData = tickerInfo.tickerQuote(tickers)
for ticker in tickers:
ticker = ticker.upper()
# Makes sure ticker exists
if tickerData[ticker] == 1:
name = tickerData[ticker + "Name"]
price = tickerData[ticker + "Price"]
change = tickerData[ticker + "Change"]
message = f"The current stock price of {name} is $**{price}**"
if change > 0:
message = f"{message}, the stock is currently **up {change}%**"
elif change < 0:
message = (
f"{message}, the stock is currently **down {change}%**"
)
else:
message = (
f"{message}, the stock hasn't shown any movement today."
)
news = tickerInfo.stockNews(ticker)
for i in range(len(news["title"])):
message = (
f"{message}\n\n[{news['title'][i]}]({news['link'][i]})"
)
update.message.reply_text(
text=message, parse_mode=telegram.ParseMode.MARKDOWN
)
else:
update.message.reply_text(ticker + " Does not exist.")
except:
pass
def stockInfo(bot, update):
message = update.message.text
chat_id = update.message.chat_id
try:
# 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)
if len(tickers) > 0:
tickerData = tickerInfo.tickerQuote(tickers)
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for ticker in tickers:
ticker = ticker.upper()
# Makes sure ticker exists
if tickerData[ticker] == 1:
name = tickerData[ticker + "Name"]
price = tickerData[ticker + "Price"]
change = tickerData[ticker + "Change"]
message = f"The current stock price of {name} is $**{price}**"
if change > 0:
message = f"{message}, the stock is currently **up {change}%**"
elif change < 0:
message = f"{message}, the stock is currently **down {change}%**"
else:
message = f"{message}, the stock hasn't shown any movement today."
update.message.reply_text(
text=message, parse_mode=telegram.ParseMode.MARKDOWN
)
else:
update.message.reply_text(ticker + " Does not exist.")
except:
pass
def dividend(bot, update):
message = update.message.text
chat_id = update.message.chat_id
try:
# 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 = tickerInfo.stockDividend(ticker)
update.message.reply_text(
text=message, parse_mode=telegram.ParseMode.MARKDOWN
)
except:
pass
def error(bot, update, error):
"""Log Errors caused by Updates."""
logger.warning('Update "%s" caused error "%s"', update, error)
def main():
"""Start the bot."""
# Create the EventHandler and pass it your bot's token.
updater = Updater(TOKEN)
# Get the dispatcher to register handlers
dp = updater.dispatcher
# 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))
# on noncommand i.e message - echo the message on Telegram
dp.add_handler(MessageHandler(Filters.text, stockInfo))
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == "__main__":
main()

View File

@ -1,105 +0,0 @@
import urllib.request
import json
from datetime import datetime
import time
def tickerQuote(tickers):
"""Gathers information from IEX api on stock"""
stockData = {}
IEXURL = (
"https://api.iextrading.com/1.0/stock/market/batch?symbols="
+ ",".join(tickers)
+ "&types=quote"
)
print("Gathering Quote from " + IEXURL)
with urllib.request.urlopen(IEXURL) as url:
IEXData = json.loads(url.read().decode())
for ticker in tickers:
ticker = ticker.upper()
# Makes sure ticker exists before populating a dictionary
if ticker in IEXData:
stockData[ticker] = 1
stockData[ticker + "Name"] = IEXData[ticker]["quote"]["companyName"]
stockData[ticker + "Price"] = IEXData[ticker]["quote"]["latestPrice"]
stockData[ticker + "Change"] = round(
(IEXData[ticker]["quote"]["changePercent"] * 100), 2
)
stockData[ticker + "Image"] = stockLogo(ticker)
print(ticker + " Quote Gathered")
else:
stockData[ticker] = 0
return stockData
def stockNews(ticker):
"""Makes a bunch of strings that are links to news websites for an input ticker"""
print("Gather News on " + ticker)
newsLink = f"https://api.iextrading.com/1.0/stock/{ticker}/news/last/5"
print(newsLink)
with urllib.request.urlopen(newsLink) as url:
data = json.loads(url.read().decode())
news = {"link": [], "title": []}
for i in range(len(data)):
news["link"].append(data[i]["url"])
news["title"].append(data[i]["headline"])
return news
def stockLogo(ticker):
"""returns a png of an input ticker"""
logoURL = f"https://g.foolcdn.com/art/companylogos/mark/{ticker}.png"
return logoURL
def stockInfo(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 stockDividend(ticker):
data = stockInfo(ticker)
print(data["divDate"])
if data["divDate"] == 0:
return "{} has no dividend.".format(data["companyName"])
line1 = "{} 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."
message = line1 + countdownMessage
return message