DHBW/TelegramAktienBot
1
0
Fork 0
Code Issues 6 Pull Requests 26 Actions Packages Projects Releases Wiki Activity

Reformatting

Browse Source
This commit is contained in:
Administrator 2022-05-12 09:09:05 +02:00
parent 86264d38f5
commit 2a94c4c246
2 changed files with 56 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
telegram_bot/bot_updates.py
View File

@ -10,7 +10,7 @@ import os
import sys import sys
import time import time
from apscheduler.schedulers.background import BackgroundScheduler # scheduler for cron from apscheduler.schedulers.background import BackgroundScheduler # scheduler for cron
from dotenv import load_dotenv from dotenv import load_dotenv
import helper_functions as hf import helper_functions as hf
@ -64,27 +64,27 @@ def update_crontab(p_my_handler):
global user_crontab global user_crontab
global user_ids global user_ids
all_users = p_my_handler.get_all_users() # get all users so crontabs can update for everybody all_users = p_my_handler.get_all_users() # get all users so crontabs can update for everybody
user_ids = [] user_ids = []
user_crontab = [] user_crontab = []
for element in all_users: for element in all_users:
if element["cron"] != '' and element["telegram_user_id"] != '': # check if both values are existing so I have consistent data if element["cron"] != '' and element["telegram_user_id"] != '': # check if both values are existing so I have consistent data
try: try:
user_ids.append(int(element["telegram_user_id"])) user_ids.append(int(element["telegram_user_id"]))
try: try:
user_crontab.append(str(element["cron"])) user_crontab.append(str(element["cron"]))
except: except:
user_ids.pop() # if something goes wrong with cron I have to delete matching user id user_ids.pop() # if something goes wrong with cron I have to delete matching user id
except: continue except:
continue
print(user_ids) print(user_ids)
update_based_on_crontab(user_ids, user_crontab, p_my_handler) update_based_on_crontab(user_ids, user_crontab, p_my_handler)
update_crontab(p_my_handler) # restart the update after time sleep update_crontab(p_my_handler) # restart the update after time sleep
def update_based_on_crontab(p_user_ids, p_user_crontab, p_my_handler): def update_based_on_crontab(p_user_ids, p_user_crontab, p_my_handler):
@ -103,17 +103,18 @@ def update_based_on_crontab(p_user_ids, p_user_crontab, p_my_handler):
:rtype: none :rtype: none
""" """
my_scheduler = BackgroundScheduler() # schedule sends based on cron my_scheduler = BackgroundScheduler() # schedule sends based on cron
for i in range(len(p_user_ids)): for i in range(len(p_user_ids)):
cron_split = p_user_crontab[i].split(" ") # split it up to use in scheduler cron_split = p_user_crontab[i].split(" ") # split it up to use in scheduler
print(cron_split[4], cron_split[1], cron_split[0], cron_split[3], cron_split[2]) print(cron_split[4], cron_split[1], cron_split[0], cron_split[3], cron_split[2])
my_scheduler.add_job(update_for_user, 'cron', day_of_week=cron_split[4], hour=cron_split[1], minute=cron_split[0], month=cron_split[3], day=cron_split[2], args=(p_user_ids[i], p_my_handler)) my_scheduler.add_job(update_for_user, 'cron', day_of_week=cron_split[4], hour=cron_split[1], minute=cron_split[0], month=cron_split[3], day=cron_split[2], args=(p_user_ids[i], p_my_handler))
my_scheduler.start() my_scheduler.start()
time.sleep( 600 ) # scheduler runs in background and I wait 10mins time.sleep(600) # scheduler runs in background and I wait 10mins
my_scheduler.shutdown() # after this the new crontabs will be loaded my_scheduler.shutdown() # after this the new crontabs will be loaded
def update_for_user(p_user_id, p_my_handler): def update_for_user(p_user_id, p_my_handler):
""" Pull shares and send updates for specific user id """ Pull shares and send updates for specific user id
@ -130,7 +131,7 @@ def update_for_user(p_user_id, p_my_handler):
share_symbols = [] share_symbols = []
share_amounts = [] share_amounts = []
my_portfolio = p_my_handler.get_user_portfolio(p_user_id) # get all existing shares for user my_portfolio = p_my_handler.get_user_portfolio(p_user_id) # get all existing shares for user
for element in my_portfolio: for element in my_portfolio:
if element["count"] != '' and element["isin"] != '': if element["count"] != '' and element["isin"] != '':
@ -139,11 +140,11 @@ def update_for_user(p_user_id, p_my_handler):
share_amounts.append(element["count"]) share_amounts.append(element["count"])
my_user = p_my_handler.get_user(p_user_id) my_user = p_my_handler.get_user(p_user_id)
send_to_user("Hello %s this is your share update for today:"%str(my_user["username"]), pUser_id=p_user_id) send_to_user("Hello %s this is your share update for today:" % str(my_user["username"]), pUser_id=p_user_id)
shares = p_my_handler.get_user_shares(p_user_id) # all interest shares shares = p_my_handler.get_user_shares(p_user_id) # all interest shares
if len(share_symbols) != 0: # iterate through all shares if len(share_symbols) != 0: # iterate through all shares
for i in range(len(share_symbols)): for i in range(len(share_symbols)):
my_price = share_fetcher.get_share_price_no_currency(share_symbols[i]) my_price = share_fetcher.get_share_price_no_currency(share_symbols[i])
my_update_message = f'{share_fetcher.get_share_information_markdown(share_symbols[i])}\ncount: {hf.make_markdown_proof(share_amounts[i])}\nTotal: {hf.make_markdown_proof(round(float(my_price) * float(share_amounts[i]), 2))} EUR' my_update_message = f'{share_fetcher.get_share_information_markdown(share_symbols[i])}\ncount: {hf.make_markdown_proof(share_amounts[i])}\nTotal: {hf.make_markdown_proof(round(float(my_price) * float(share_amounts[i]), 2))} EUR'
@ -164,10 +165,10 @@ def update_for_user(p_user_id, p_my_handler):
news = news_fetcher.get_top_news_by_keyword(keyword)["articles"] news = news_fetcher.get_top_news_by_keyword(keyword)["articles"]
keyword = hf.make_markdown_proof(keyword) keyword = hf.make_markdown_proof(keyword)
if not news: # if empty news array if not news: # if empty news array
send_to_user(f"No news found for keyword _{keyword}_\.", pUser_id=p_user_id, md_mode=True) send_to_user(f"No news found for keyword _{keyword}_\.", pUser_id=p_user_id, md_mode=True)
elif news == None: # if news is none elif news == None: # if news is none
send_to_user(f"Server error for keyword _{keyword}_\.", pUser_id=p_user_id, md_mode=True) send_to_user(f"Server error for keyword _{keyword}_\.", pUser_id=p_user_id, md_mode=True)
else: else:
news_formatted = news_fetcher.format_article(news[0]) # format for message, only use the most popular article news_formatted = news_fetcher.format_article(news[0]) # format for message, only use the most popular article

21
telegram_bot/shares/share_fetcher.py
View File

@ -23,22 +23,22 @@ def get_share_price(str_search_for):
try: try:
search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1, countries=['germany']) search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1, countries=['germany'])
currency = str(search_result.retrieve_currency()) # retrieve currency from data currency = str(search_result.retrieve_currency()) # retrieve currency from data
# should always be Euro because of countries=['germany'] # should always be Euro because of countries=['germany']
recent_data = pandas.DataFrame(search_result.retrieve_recent_data()) # stock prices of last few days recent_data = pandas.DataFrame(search_result.retrieve_recent_data()) # stock prices of last few days
stock_price = recent_data.iloc[-1]["Close"] # retrieve latest stock price stock_price = recent_data.iloc[-1]["Close"] # retrieve latest stock price
stock_price = round(float(stock_price), 2) stock_price = round(float(stock_price), 2)
str_return =str(stock_price) + " " + str(currency) # return + currency str_return = str(stock_price) + " " + str(currency) # return + currency
return str_return return str_return
except RuntimeError: # if no shares are found for germany (e.g. isin: US.....) except RuntimeError: # if no shares are found for germany (e.g. isin: US.....)
try: try:
my_Converter = CurrencyConverter() # need a currency converter my_Converter = CurrencyConverter() # need a currency converter
search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1) search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1)
@ -48,7 +48,7 @@ def get_share_price(str_search_for):
stock_price = recent_data.iloc[-1]["Close"] stock_price = recent_data.iloc[-1]["Close"]
#convert stock price from currency to EUR # convert stock price from currency to EUR
stock_price = my_Converter.convert(float(stock_price), str(currency), 'EUR') stock_price = my_Converter.convert(float(stock_price), str(currency), 'EUR')
stock_price = round(float(stock_price), 2) stock_price = round(float(stock_price), 2)
@ -109,16 +109,15 @@ def get_share_information(str_search_for):
def get_share_information_markdown(str_search_for): def get_share_information_markdown(str_search_for):
try: try:
search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], search_result = investpy.search_quotes(text=str_search_for, products=['stocks'],
countries=['germany'], n_results=1) countries=['germany'], n_results=1)
except RuntimeError as e: except RuntimeError as e:
return hf.make_markdown_proof(f"no shares found for \"{str_search_for}\"") # if no shares are found, make error message markdown proof and return return hf.make_markdown_proof(f"no shares found for \"{str_search_for}\"") # if no shares are found, make error message markdown proof and return
except ConnectionError as e: except ConnectionError as e:
return hf.make_markdown_proof(f"connection not possible. Try again later.") # if no connection, make error message markdown proof and return return hf.make_markdown_proof(f"connection not possible. Try again later.") # if no connection, make error message markdown proof and return
str_return = f'*{hf.make_markdown_proof(search_result.name)}*\n_{hf.make_markdown_proof(search_result.symbol)}_\nworth: {hf.make_markdown_proof(get_share_price(str_search_for))}' str_return = f'*{hf.make_markdown_proof(search_result.name)}*\n_{hf.make_markdown_proof(search_result.symbol)}_\nworth: {hf.make_markdown_proof(get_share_price(str_search_for))}'
return str_return return str_return