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

major refactor of functions

This commit is contained in:
Anson 2020-04-05 02:41:35 -07:00
parent 53d655fa48
commit aab4a4354e
2 changed files with 106 additions and 105 deletions

28
bot.py
View File

@ -3,11 +3,15 @@ import logging
import os import os
import telegram import telegram
from functions import * from functions import Symbol
from telegram.ext import CommandHandler, Filters, MessageHandler, Updater from telegram.ext import CommandHandler, Filters, MessageHandler, Updater
TELEGRAM_TOKEN = os.environ["TELEGRAM"] TELEGRAM_TOKEN = (
"724630968:AAHL_cpMgrw-B9zSbVlVe7iTYyo0XXL8fi4" # os.environ["TELEGRAM"]
)
IEX_TOKEN = "sk_9e8d93b7cac84cd4b800f34d15b72ad6" # os.environ["IEX"]
s = Symbol(IEX_TOKEN)
# Enable logging # Enable logging
logging.basicConfig( logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
@ -30,19 +34,19 @@ def help(bot, update):
update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN) update.message.reply_text(text=message, parse_mode=telegram.ParseMode.MARKDOWN)
def symbolDetect(bot, update): def symbol_detect(bot, update):
""" """
Runs on any message that doesn't have a command and searches for symbols, then returns the prices of any symbols found. Runs on any message that doesn't have a command and searches for symbols, then returns the prices of any symbols found.
""" """
message = update.message.text message = update.message.text
chat_id = update.message.chat_id chat_id = update.message.chat_id
symbols = getSymbols(message) symbols = s.find_symbols(message)
if symbols: if symbols:
# Let user know bot is working # Let user know bot is working
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING) bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in symbolDataReply(symbols).items(): for reply in s.price_reply(symbols).items():
update.message.reply_text( update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
@ -55,12 +59,12 @@ def dividend(bot, update):
""" """
message = update.message.text message = update.message.text
chat_id = update.message.chat_id chat_id = update.message.chat_id
symbols = getSymbols(message) symbols = s.find_symbols(message)
if symbols: if symbols:
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING) bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in symbolDividend(symbols).items(): for reply in s.symbol_name(symbols).items():
update.message.reply_text( update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
@ -73,12 +77,12 @@ def news(bot, update):
""" """
message = update.message.text message = update.message.text
chat_id = update.message.chat_id chat_id = update.message.chat_id
symbols = getSymbols(message) symbols = s.find_symbols(message)
if symbols: if symbols:
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING) bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in symbolNews(symbols).items(): for reply in s.news_reply(symbols).items():
update.message.reply_text( update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
@ -91,12 +95,12 @@ def info(bot, update):
""" """
message = update.message.text message = update.message.text
chat_id = update.message.chat_id chat_id = update.message.chat_id
symbols = getSymbols(message) symbols = s.find_symbols(message)
if symbols: if symbols:
bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING) bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in symbolInfo(symbols).items(): for reply in s.info_reply(symbols).items():
update.message.reply_text( update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
@ -125,7 +129,7 @@ def main():
dp.add_handler(CommandHandler("info", info)) dp.add_handler(CommandHandler("info", info))
# on noncommand i.e message - echo the message on Telegram # on noncommand i.e message - echo the message on Telegram
dp.add_handler(MessageHandler(Filters.text, symbolDetect)) dp.add_handler(MessageHandler(Filters.text, symbol_detect))
# log all errors # log all errors
dp.add_error_handler(error) dp.add_error_handler(error)

View File

@ -6,119 +6,116 @@ from datetime import datetime
import requests import requests
class Symbol:
def getSymbols(text: str):
"""
Takes a blob of text and returns a list of symbols without any repeats.
"""
SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})" SYMBOL_REGEX = "[$]([a-zA-Z]{1,4})"
return list(set(re.findall(SYMBOL_REGEX, text))) def __init__(self, IEX_TOKEN: str):
self.IEX_TOKEN = IEX_TOKEN
def find_symbols(self, text: str):
"""
Takes a blob of text and returns a list of symbols without any repeats.
"""
def symbolDataReply(symbols: list): return list(set(re.findall(self.SYMBOL_REGEX, text)))
"""
Takes a list of symbols and returns a dictionary of strings with information about the symbol.
"""
dataMessages = {}
for symbol in symbols:
IEXurl = (
f"https://cloud.iexapis.com/stable/stock/{symbol}/quote?token={IEX_TOKEN}"
)
response = requests.get(IEXurl) def price_reply(self, symbols: list):
if response.status_code == 200: """
IEXData = response.json() Takes a list of symbols and returns a dictionary of strings with information about the symbol.
message = f"The current stock price of {IEXData['companyName']} is $**{IEXData['latestPrice']}**" """
dataMessages = {}
for symbol in symbols:
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/quote?token={self.IEX_TOKEN}"
# Determine wording of change text response = requests.get(IEXurl)
change = round(IEXData["changePercent"] * 100, 2) if response.status_code == 200:
if change > 0: IEXData = response.json()
message += f", the stock is currently **up {change}%**" message = f"The current stock price of {IEXData['companyName']} is $**{IEXData['latestPrice']}**"
elif change < 0:
message += f", the stock is currently **down {change}%**" # 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: else:
message += ", the stock hasn't shown any movement today." message = f"The symbol: {symbol} was not found."
else:
message = f"The symbol: {symbol} was not found."
dataMessages[symbol] = message dataMessages[symbol] = message
return dataMessages return dataMessages
def symbol_name(self, symbols: list):
divMessages = {}
def symbolDividend(symbols: list): for symbol in symbols:
divMessages = {} IEXurl = f"https://cloud.iexapis.com/stable/data-points/{symbol}/NEXTDIVIDENDDATE?token={self.IEX_TOKEN}"
response = requests.get(IEXurl)
if response.status_code == 200:
for symbol in symbols: # extract date from json
IEXurl = f"https://cloud.iexapis.com/stable/data-points/{symbol}/NEXTDIVIDENDDATE?token={IEX_TOKEN}" date = response.json()
response = requests.get(IEXurl) # Pattern IEX uses for dividend date.
if response.status_code == 200: pattern = "%Y-%m-%d"
divDate = datetime.strptime(date, pattern)
# extract date from json daysDelta = (divDate - datetime.now()).days
date = response.json() datePretty = divDate.strftime("%A, %B %w")
# Pattern IEX uses for dividend date. if daysDelta < 0:
pattern = "%Y-%m-%d" divMessages[
divDate = datetime.strptime(date, pattern) 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."
daysDelta = (divDate - datetime.now()).days else:
datePretty = divDate.strftime("%A, %B %w")
if daysDelta < 0:
divMessages[ divMessages[
symbol symbol
] = f"{symbol.upper()} dividend was on {datePretty} and a new date hasn't been announced yet." ] = f"{symbol} either doesn't exist or pays no dividend."
elif daysDelta > 0:
divMessages[ return divMessages
symbol
] = f"{symbol.upper()} dividend is on {datePretty} which is in {daysDelta} Days." def news_reply(self, symbols: list):
newsMessages = {}
for symbol in symbols:
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/3?token={self.IEX_TOKEN}"
response = requests.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
else: else:
divMessages[symbol] = f"{symbol.upper()} is today." newsMessages[
symbol
] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
else: return newsMessages
divMessages[symbol] = f"{symbol} either doesn't exist or pays no dividend."
return divMessages 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 = requests.get(IEXurl)
def symbolNews(symbols: list): if response.status_code == 200:
newsMessages = {} 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"
for symbol in symbols: else:
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/3?token={IEX_TOKEN}" infoMessages[
response = requests.get(IEXurl) symbol
if response.status_code == 200: ] = f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
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
else:
newsMessages[
symbol
] = f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
return newsMessages return infoMessages
def symbolInfo(symbols: list):
infoMessages = {}
for symbol in symbols:
IEXurl = (
f"https://cloud.iexapis.com/stable/stock/{symbol}/company?token={IEX_TOKEN}"
)
response = requests.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