Anson/Simple-Telegram-Stock-Bot
1
0
Fork 0
mirror of https://gitlab.com/simple-stock-bots/simple-telegram-stock-bot.git synced 2025-06-16 06:56:46 +00:00
Code Releases Activity

start on #18

Browse Source
This commit is contained in:
MisterBiggs 2021-06-23 21:28:13 -07:00
parent baf45e09c8
commit acc0ebdd28
3 changed files with 79 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
IEX_Symbol.py
View File

@ -9,7 +9,7 @@ import requests as r
import schedule
from fuzzywuzzy import fuzz
import os
from logging import warning
from Symbol import Stock
@ -36,7 +36,7 @@ class IEX_Symbol:
self.IEX_TOKEN = os.environ["IEX"]
except KeyError:
self.IEX_TOKEN = ""
print(
warning(
"Starting without an IEX Token will not allow you to get market data!"
)

72
bot.py
View File

@ -7,6 +7,8 @@ import html
import json
import traceback
from logging import debug, info, warning, error, critical
import mplfinance as mpf
from uuid import uuid4
@ -36,7 +38,7 @@ try:
STRIPE_TOKEN = os.environ["STRIPE"]
except KeyError:
STRIPE_TOKEN = ""
print("Starting without a STRIPE Token will not allow you to accept Donations!")
warning("Starting without a STRIPE Token will not allow you to accept Donations!")
s = Router()
t = T_info()
@ -47,11 +49,12 @@ logging.basicConfig(
)
logger = logging.getLogger(__name__)
print("Bot Online")
info("Bot script started.")
def start(update: Update, context: CallbackContext):
"""Send a message when the command /start is issued."""
info(f"Start command ran by {update.message.chat.username}")
update.message.reply_text(
text=t.help_text,
parse_mode=telegram.ParseMode.MARKDOWN,
@ -61,6 +64,7 @@ def start(update: Update, context: CallbackContext):
def help(update: Update, context: CallbackContext):
"""Send link to docs when the command /help is issued."""
info(f"Help command ran by {update.message.chat.username}")
update.message.reply_text(
text=t.help_text,
parse_mode=telegram.ParseMode.MARKDOWN,
@ -70,6 +74,7 @@ def help(update: Update, context: CallbackContext):
def license(update: Update, context: CallbackContext):
"""Return bots license agreement"""
info(f"License command ran by {update.message.chat.username}")
update.message.reply_text(
text=t.license,
parse_mode=telegram.ParseMode.MARKDOWN,
@ -78,6 +83,7 @@ def license(update: Update, context: CallbackContext):
def status(update: Update, context: CallbackContext):
warning(f"Status command ran by {update.message.chat.username}")
bot_resp = datetime.datetime.now(update.message.date.tzinfo) - update.message.date
update.message.reply_text(
@ -90,6 +96,7 @@ def status(update: Update, context: CallbackContext):
def donate(update: Update, context: CallbackContext):
info(f"Donate command ran by {update.message.chat.username}")
chat_id = update.message.chat_id
if update.message.text.strip() == "/donate":
@ -107,7 +114,7 @@ def donate(update: Update, context: CallbackContext):
except ValueError:
update.message.reply_text(f"{amount} is not a valid donation amount or number.")
return
print(price)
info(f"Donation amount: {price}")
context.bot.send_invoice(
chat_id=chat_id,
@ -126,6 +133,7 @@ def donate(update: Update, context: CallbackContext):
def precheckout_callback(update: Update, context: CallbackContext):
info(f"precheckout_callback queried")
query = update.pre_checkout_query
query.answer(ok=True)
@ -138,6 +146,7 @@ def precheckout_callback(update: Update, context: CallbackContext):
def successful_payment_callback(update: Update, context: CallbackContext):
info(f"Successful payment!")
update.message.reply_text(
"Thank you for your donation! It goes a long way to keeping the bot free!"
)
@ -148,6 +157,7 @@ def symbol_detect(update: Update, context: CallbackContext):
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
chat_id = update.message.chat_id
symbols = s.find_symbols(message)
@ -155,7 +165,9 @@ def symbol_detect(update: Update, context: CallbackContext):
if symbols:
# Let user know bot is working
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
print(symbols)
info(f"User called for symbols: {update.message.chat.username}")
info(f"Symbols found: {symbols}")
for reply in s.price_reply(symbols):
update.message.reply_text(
text=reply,
@ -168,6 +180,7 @@ def dividend(update: Update, context: CallbackContext):
"""
waits for /dividend or /div command and then finds dividend info on that symbol.
"""
info(f"Dividend command ran by {update.message.chat.username}")
message = update.message.text
chat_id = update.message.chat_id
@ -193,6 +206,7 @@ def news(update: Update, context: CallbackContext):
"""
waits for /news command and then finds news info on that symbol.
"""
info(f"News command ran by {update.message.chat.username}")
message = update.message.text
chat_id = update.message.chat_id
@ -215,10 +229,11 @@ def news(update: Update, context: CallbackContext):
)
def info(update: Update, context: CallbackContext):
def information(update: Update, context: CallbackContext):
"""
waits for /info command and then finds info on that symbol.
"""
info(f"Information command ran by {update.message.chat.username}")
message = update.message.text
chat_id = update.message.chat_id
@ -242,6 +257,7 @@ def info(update: Update, context: CallbackContext):
def search(update: Update, context: CallbackContext):
info(f"Search command ran by {update.message.chat.username}")
message = update.message.text.replace("/search ", "")
chat_id = update.message.chat_id
@ -265,6 +281,7 @@ def search(update: Update, context: CallbackContext):
def intra(update: Update, context: CallbackContext):
info(f"Intra command ran by {update.message.chat.username}")
# TODO: Document usage of this command. https://iexcloud.io/docs/api/#historical-prices
message = update.message.text
@ -320,6 +337,7 @@ def intra(update: Update, context: CallbackContext):
def chart(update: Update, context: CallbackContext):
info(f"Chart command ran by {update.message.chat.username}")
# TODO: Document usage of this command. https://iexcloud.io/docs/api/#historical-prices
message = update.message.text
@ -350,7 +368,7 @@ def chart(update: Update, context: CallbackContext):
context.bot.send_chat_action(
chat_id=chat_id, action=telegram.ChatAction.UPLOAD_PHOTO
)
print(symbol)
buf = io.BytesIO()
mpf.plot(
df,
@ -375,6 +393,7 @@ def stat(update: Update, context: CallbackContext):
"""
https://iexcloud.io/docs/api/#key-stats
"""
info(f"Stat command ran by {update.message.chat.username}")
message = update.message.text
chat_id = update.message.chat_id
@ -401,6 +420,7 @@ def cap(update: Update, context: CallbackContext):
"""
Market Cap Information
"""
info(f"Cap command ran by {update.message.chat.username}")
message = update.message.text
chat_id = update.message.chat_id
@ -427,6 +447,7 @@ def trending(update: Update, context: CallbackContext):
"""
Trending Symbols
"""
info(f"Trending command ran by {update.message.chat.username}")
chat_id = update.message.chat_id
@ -444,14 +465,14 @@ def inline_query(update: Update, context: CallbackContext):
Handles inline query.
Does a fuzzy search on input and returns stocks that are close.
"""
info(f"Inline command ran by {update.message.chat.username}")
info(f"Query: {update.inline_query.query}")
matches = s.inline_search(update.inline_query.query)[:5]
symbols = " ".join([match[1].split(":")[0] for match in matches])
prices = s.batch_price_reply(s.find_symbols(symbols))
results = []
print(update.inline_query.query)
for match, price in zip(matches, prices):
try:
results.append(
@ -464,16 +485,20 @@ def inline_query(update: Update, context: CallbackContext):
)
)
except TypeError:
logging.warning(str(match))
warning(f"{match} caused error in inline query.")
pass
print(match[0], "\n\n\n")
if len(results) == 5:
update.inline_query.answer(results)
info("Inline Command was successful")
return
update.inline_query.answer(results)
def rand_pick(update: Update, context: CallbackContext):
info(
f"Someone is gambling! Random_pick command ran by {update.message.chat.username}"
)
update.message.reply_text(
text=s.random_pick(),
@ -484,22 +509,25 @@ def rand_pick(update: Update, context: CallbackContext):
def error(update: Update, context: CallbackContext):
"""Log Errors caused by Updates."""
logger.warning('Update "%s" caused error "%s"', update, error)
warning('Update "%s" caused error "%s"', update, error)
tb_list = traceback.format_exception(
None, context.error, context.error.__traceback__
)
tb_string = "".join(tb_list)
print(tb_string)
# if update:
# message = (
# f"An exception was raised while handling an update\n"
# f"<pre>update = {html.escape(json.dumps(update.to_dict(), indent=2, ensure_ascii=False))}"
# "</pre>\n\n"
# f"<pre>context.chat_data = {html.escape(str(context.chat_data))}</pre>\n\n"
# f"<pre>context.user_data = {html.escape(str(context.user_data))}</pre>\n\n"
# f"<pre>{html.escape(tb_string)}</pre>"
# )
if update:
message = (
f"An exception was raised while handling an update\n"
f"<pre>update = {html.escape(json.dumps(update.to_dict(), indent=2, ensure_ascii=False))}"
"</pre>\n\n"
f"<pre>context.chat_data = {html.escape(str(context.chat_data))}</pre>\n\n"
f"<pre>context.user_data = {html.escape(str(context.user_data))}</pre>\n\n"
f"<pre>{html.escape(tb_string)}</pre>"
)
warning(message)
else:
warning(tb_string)
update.message.reply_text(
text="An error has occured. Please inform @MisterBiggs if the error persists."
)
@ -524,7 +552,7 @@ def main():
dp.add_handler(CommandHandler("dividend", dividend))
dp.add_handler(CommandHandler("div", dividend))
dp.add_handler(CommandHandler("news", news))
dp.add_handler(CommandHandler("info", info))
dp.add_handler(CommandHandler("info", information))
dp.add_handler(CommandHandler("stat", stat))
dp.add_handler(CommandHandler("stats", stat))
dp.add_handler(CommandHandler("cap", cap))

50
symbol_router.py
View File

@ -7,7 +7,7 @@ import random
import datetime
from fuzzywuzzy import fuzz
from typing import List, Tuple
from logging import debug, info, warning, error, critical
from IEX_Symbol import IEX_Symbol
from cg_Crypto import cg_Crypto
@ -24,7 +24,7 @@ class Router:
self.stock = IEX_Symbol()
self.crypto = cg_Crypto()
def find_symbols(self, text: str) -> List[Symbol]:
def find_symbols(self, text: str) -> list[Symbol]:
"""Finds stock tickers starting with a dollar sign, and cryptocurrencies with two dollar signs
in a blob of text and returns them in a list.
Only returns each match once. Example: Whats the price of $tsla?
@ -36,7 +36,7 @@ class Router:
Returns
-------
List[str]
list[str]
List of stock symbols as strings without dollar sign.
"""
symbols = []
@ -45,15 +45,15 @@ class Router:
if stock.upper() in self.stock.symbol_list["symbol"].values:
symbols.append(Stock(stock))
else:
print(f"{stock} is not in list of stocks")
info(f"{stock} is not in list of stocks")
coins = set(re.findall(self.CRYPTO_REGEX, text))
for coin in coins:
if coin.lower() in self.crypto.symbol_list["symbol"].values:
symbols.append(Coin(coin.lower()))
else:
print(f"{coin} is not in list of coins")
print(symbols)
info(f"{coin} is not in list of coins")
info(symbols)
return symbols
def status(self, bot_resp) -> str:
@ -65,7 +65,7 @@ class Router:
Human readable text on status of the bot and relevant APIs
"""
return f"""
stats = f"""
Bot Status:
{bot_resp}
@ -76,7 +76,11 @@ class Router:
{self.crypto.status()}
"""
def search_symbols(self, search: str) -> List[Tuple[str, str]]:
warning(stats)
return stats
def search_symbols(self, search: str) -> list[tuple[str, str]]:
"""Performs a fuzzy search to find stock symbols closest to a search term.
Parameters
@ -86,7 +90,7 @@ class Router:
Returns
-------
List[tuple[str, str]]
list[tuple[str, str]]
A list tuples of every stock sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
"""
@ -113,7 +117,7 @@ class Router:
self.searched_symbols[search] = symbol_list
return symbol_list
def inline_search(self, search: str) -> List[Tuple[str, str]]:
def inline_search(self, search: str) -> list[tuple[str, str]]:
"""Searches based on the shortest symbol that contains the same string as the search.
Should be very fast compared to a fuzzy search.
@ -124,7 +128,7 @@ class Router:
Returns
-------
List[tuple[str, str]]
list[tuple[str, str]]
Each tuple contains: (Symbol, Issue Name).
"""
@ -141,7 +145,7 @@ class Router:
self.searched_symbols[search] = symbol_list
return symbol_list
def price_reply(self, symbols: list[Symbol]) -> List[str]:
def price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol.
Parameters
@ -158,17 +162,17 @@ class Router:
replies = []
for symbol in symbols:
print(symbol)
info(symbol)
if isinstance(symbol, Stock):
replies.append(self.stock.price_reply(symbol))
elif isinstance(symbol, Coin):
replies.append(self.crypto.price_reply(symbol))
else:
print(f"{symbol} is not a Stock or Coin")
info(f"{symbol} is not a Stock or Coin")
return replies
def dividend_reply(self, symbols: list) -> List[str]:
def dividend_reply(self, symbols: list) -> list[str]:
"""Returns the most recent, or next dividend date for a stock symbol.
Parameters
@ -192,7 +196,7 @@ class Router:
return replies
def news_reply(self, symbols: list) -> List[str]:
def news_reply(self, symbols: list) -> list[str]:
"""Gets recent english news on stock symbols.
Parameters
@ -220,12 +224,12 @@ class Router:
return replies
def info_reply(self, symbols: list) -> List[str]:
def info_reply(self, symbols: list) -> list[str]:
"""Gets information on stock symbols.
Parameters
----------
symbols : List[str]
symbols : list[str]
List of stock symbols.
Returns
@ -289,12 +293,12 @@ class Router:
print(f"{symbol} is not a Stock or Coin")
return pd.DataFrame()
def stat_reply(self, symbols: List[Symbol]) -> List[str]:
def stat_reply(self, symbols: list[Symbol]) -> list[str]:
"""Gets key statistics for each symbol in the list
Parameters
----------
symbols : List[str]
symbols : list[str]
List of stock symbols
Returns
@ -314,12 +318,12 @@ class Router:
return replies
def cap_reply(self, symbols: List[Symbol]) -> List[str]:
def cap_reply(self, symbols: list[Symbol]) -> list[str]:
"""Gets market cap for each symbol in the list
Parameters
----------
symbols : List[str]
symbols : list[str]
List of stock symbols
Returns
@ -375,7 +379,7 @@ class Router:
return f"{choice}\nBuy and hold until: {hold}"
def batch_price_reply(self, symbols: list[Symbol]) -> List[str]:
def batch_price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol.
Parameters