Build A Real-Time Stock Alert Telegram Bot with Python
- Nikhil Adithyan
- 3 days ago
- 9 min read
A comprehensive guide to building an investment notifications system with APIs
In this fast-paced world, data is what drives an investment decision. Investing is not a random choice, and for an investor, timely and accurate information can mean the difference between profit and loss. Markets move quickly, and investors must react instantly to various events and price changes.
But what if you don’t want to spend all your time in front of a monitor trying to understand the excessive amount of information displayed? And even then, how can you effectively filter this data?
In this article, we are going to demonstrate how you can develop your custom-made alerting mechanism. We will use FinancialModelingPrep’s API endpoints to get the data, and Telegram to alert you immediately when needed.
Setting up Telegram
There are millions of tutorials on how to set up a Telegram bot and send messages using Python. Here, we will cover the basic steps that are enough to establish this setup, without diving into detailed explanations that can be quickly found through a simple search.
We will assume that you already have an account with Telegram. So first step is to go to the search in your chats @ BotFather (the space between @ and the name is because Medium understands it as a user name…). After you find it, you should send a message to this bot as /newbot . Telegram will ask you for its name, and you should give it. This is the name of a private bot, and you can name it whatever you want.
After naming your bot, Telegram will ask for your username. This should be unique among Telegram users and must end with the word’ bot’. You might need a few attempts to find a unique name. Once done, Telegram will provide you with a link to the bot, as well as an API token to communicate with it through HTTP.
Simply by clicking the link, you will open the bot’s page and should press start. Besides the token, you will also need a chat ID for your Python scripts. You can obtain this with a simple Python script. Keep the chat ID. You will need it. However, you won’t need that piece of code again.
import requests
# Your bot token from BotFather
TOKEN = "YOUR TELEGRAMs BOT TOKEN"
url = f"https://api.telegram.org/bot{TOKEN}/getUpdates"
response = requests.get(url)
data = response.json()
print(data)
Let’s Code
First of all, it will be a Python script rather than a Jupyter notebook. Therefore, all the code below should be included in a single script (.py). We will need to explain what this script does before diving into the code.
The script will do three things for the stocks we are monitoring: alert us to upcoming or announced earnings, send us news as they are published, and notify us of significant movements on the 1-hour price chart.
We will use the schedule library to schedule the alerts.
We will use a very simple log mechanism to record what we have processed so far, so that we do not send the same information to us repeatedly.
Now let’s explore the code, and things will become clearer. First, the imports and some parameters.
import json
import requests
from datetime import date, timedelta
import os
import schedule
import time
TELEGRAM_TOKEN = 'YOUR_TELEGRAM_TOKEN'
TELEGRAM_CHAT_ID = 'YOUR_TELEGRAM_CHATID'
FMP_API_KEY = 'YOUR_FMP_TOKEN'
STOCKS_TO_CHECK = ['AAPL','MSFT', 'GOOGL']
You will need to update the above with your tokens and the stocks you wish to check.
The first function we will define is one to send the actual telegram message. It should be very simple with just a few lines of code.
def send_telegram_message(message: str):
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown" # Optional: allows bold, italic, code style
}
response = requests.post(url, data=payload)
# Raise an error if Telegram returns a failure
response.raise_for_status()
print("Message sent:", response.json())
Now we should define a function that will get the log file of processed events.
def get_log(file_log_name):
if not os.path.exists(file_log_name):
with open(file_log_name, "w", encoding="utf-8") as f:
json.dump([], f, ensure_ascii=False, indent=2)
print(f"Created {file_log_name}")
return json.load(open(file_log_name, "r", encoding="utf-8"))
This will become clearer when we examine the actual actions of our script. For now, note that we pass a filename to this function, and:
It is trying to find the file,
If it is not there, it creates it as an empty list
If it exists, it returns the file’s content as a list of dictionaries.
This way, we will retain all the necessary dictionaries that we have already processed, so we will not process them again.
Earnings
The next part of the code concerns earnings. We will need two functions. One will retrieve earnings before and after a specific number of days using the FMP Earnings Calendar API endpoint.
For example, if today is the 22nd of October, it will retrieve earnings from October 15th to October 29th. This approach will give us both the announced earnings and the future ones. Ultimately, the function will filter only the stocks we are interested in.
def get_earnings_calendar(days_to_check=7):
today = date.today()
from_date = (today - timedelta(days=days_to_check)).strftime('%Y-%m-%d')
to_date = (today + timedelta(days=days_to_check)).strftime('%Y-%m-%d')
print(f"Will get the earnings calendar from {from_date} to {to_date}")
url = f'https://financialmodelingprep.com/stable/earnings-calendar'
querystring = {"apikey": FMP_API_KEY, "from": from_date, "to": to_date}
response = requests.get(url, querystring)
earnings_data = response.json()
# Filter for stocks in STOCKS_TO_CHECK
filtered_earnings = []
for earning in earnings_data:
if earning['symbol'] in STOCKS_TO_CHECK:
filtered_earnings.append(earning)
return filtered_earnings
The primary function responsible for processing the alerts will be the one outlined below:
def run_and_notify_earnings_calendar():
earnings = get_earnings_calendar(days_to_check=7)
file_log_name = 'earnings_calendar_log.txt'
earnings_log = get_log(file_log_name)
for earning in earnings:
try:
if earning not in earnings_log:
earnings_log.append(earning)
print(earning)
# Earnings are expected
if earning['epsActual'] is None:
if earning['epsEstimated'] is None:
message = f"Earnings for {earning['symbol']} on {earning['date']} and earnings estimates are unknown at this time."
else:
message = f"Earnings for {earning['symbol']} on {earning['date']} are expected to be {earning['epsEstimated']}."
# earnings are published
else:
if earning['epsEstimated'] is None:
message = f"Earnings for {earning['symbol']} on {earning['date']} are published at {earning['epsActual']}."
else:
message = f"Earnings for {earning['symbol']} on {earning['date']} are published at {earning['epsActual']} while estimates were {earning['epsEstimated']}."
send_telegram_message(message)
with open(file_log_name, "w", encoding="utf-8") as f:
json.dump(earnings_log, f, ensure_ascii=False, indent=2)
except Exception as e:
send_telegram_message(f"Error processing {earning} earnings: {e}")
return earnings
Let’s explain what the code does.
It configures the log to retain the already processed information.
Retrieves the earnings from the previous function.
For each earning, it checks if this is already in our log, which indicates that we have processed it.
If the epsActual exists, it indicates that earnings have been published, so it sends a Telegram message. The message is slightly modified depending on whether there was an estimation or not, allowing us to see if the stock performed better than expected quickly.
If the epsActiual does not exist, it means that the earnings have not yet been published, and we will send a message notifying that the earnings will be published on the specified date. In that case, we differentiate the message depending on whether there is an estimation or not.
Finally, we are logging the processed earnings so we don’t send the same message again during the next run.
Below are some messages I received while running my tests:
Stock News
Another important alert to monitor for your stocks is news. Fortunately, the FMP News API endpoint provides this functionality. You can specify a stock, and it will retrieve news from various sources about it.
We will set up two functions: one to fetch news for each stock and compile a consolidated list. By default, it retrieves three news stories per stock, but you can increase this number to get more information.
def get_stocks_news(days_to_check=10, limit=3):
# Calculate the current date and next 7 days
today = date.today()
from_date = (today - timedelta(days=days_to_check)).strftime('%Y-%m-%d')
to_date = today.strftime('%Y-%m-%d')
print(f"Will get the stock news from {from_date} to {to_date}")
news_data = []
for stock in STOCKS_TO_CHECK:
url = f"https://financialmodelingprep.com/stable/news/stock"
querystring = {"apikey": FMP_API_KEY, "symbols": stock, "from": from_date, "to": to_date, "limit": limit}
response = requests.get(url, querystring)
data = response.json()
print(f"Got {len(data)} news for {stock}")
news_data = news_data + data
return news_data
Then we will be setting the function to process and alert us:
def run_and_notify_stock_news():
file_log_name = 'stock_news_logs.txt'
news = get_stocks_news()
news_log = get_log(file_log_name)
for news_item in news:
try:
if news_item not in news_log:
news_log.append(news_item)
message = f"News for *{news_item['symbol']}* on {news_item['publishedDate']} is {news_item['title']}. You can read more [here]({news_item['url']})"
send_telegram_message(message)
with open(file_log_name, "w", encoding="utf-8") as f:
json.dump(news_log, f, ensure_ascii=False, indent=2)
except Exception as e:
send_telegram_message(f"Error processing {news_item} news: {e}")
The logic of the function remains exactly the same. It’s just that instead of earnings, it now sends news messages. Also, this time we made the messages a bit more attractive, and luckily, when sending a message to Telegram with a link, you get a nice preview of the article itself. Below is a nice screenshot of how the message can look like.
Price spikes
The final alert we’ll establish in this article is to monitor for price spikes, using the 1-hour interval. The first function will utilize the FMP Historical Chart API endpoint to retrieve the OHLC data for the last 1-hour timeframe.
def get_1_hour_stock_interval(symbol):
url = f"https://financialmodelingprep.com/stable/historical-chart/1hour"
querystring = {"apikey": FMP_API_KEY, "symbol": symbol}
response = requests.get(url, querystring)
data = response.json()
return data
Then, the second function will handle the actual processing. While the function will utilize the same logging mechanism to avoid processing the same events repeatedly, it is important to explain the remaining steps:
It will calculate the difference between the close and open prices.
It will check if the difference is greater than the threshold, which in our example is 0.5 (meaning we will get an alert if the price spikes up or down more than 0.5% in an hour).
Depending on whether the change exceeds the positive threshold (for an uptrend) and falls below the negative threshold (for a downtrend), we will send the respective message. Note that we have also added some emojis so that the messages are more informative at a glance, and — yes — you can add emojis to your Telegram messages.
def run_and_notify_stock_interval(threshold_pct=0.5):
file_log_name = 'stock_prices_log.txt'
prices_log = get_log(file_log_name)
for stock in STOCKS_TO_CHECK:
try:
data = get_1_hour_stock_interval(stock)
if len(data) > 0:
latest_price = data[0]
latest_price['symbol'] = stock
if latest_price not in prices_log:
prices_log.append(latest_price)
difference_pct = round((latest_price['close'] - latest_price['open']) / latest_price['open'] * 100,2)
print(f"Stock {stock} has a new price of {latest_price['close']} on {latest_price['date']} with a difference of {difference_pct}%")
if difference_pct > threshold_pct:
message = f"🔥Stock *{stock}* increased {difference_pct}% the last hour with a new price of {latest_price['close']} on {latest_price['date']}"
send_telegram_message(message)
elif difference_pct < -threshold_pct:
message = f"❌Stock *{stock}* decreased {difference_pct}% the last hour with a new price of {latest_price['close']} on {latest_price['date']}"
send_telegram_message(message)
with open(file_log_name, "w", encoding="utf-8") as f:
json.dump(prices_log, f, ensure_ascii=False, indent=2)
except Exception as e:
send_telegram_message(f"Error processing stock {stock} interval: {e}")
Below you can find a sample of a message that I received. 002891.SZ is a stock that we used during our testing that triggered the alert.
Schedule it
Lastly, we should create a main function that will be responsible for scheduling the execution of the alerts mentioned above at custom intervals. The scheduling library you should consider is similar to a cron job, but all you need to do is run the Python script and let it operate.
First, we send a message that the Bot started
Then we schedule the three alerting mechanisms we developed before
Next, we get into an infinite loop that runs the scheduled job if it is their time to be executed. We will put a second of delay between the loops
Also, we will include some basic error handling to receive a message if the bot crashes or stops due to our actions.
def main():
send_telegram_message("Bot Started!")
schedule.every().day.at("06:00").do(run_and_notify_earnings_calendar)
schedule.every().hour.at(":50").do(run_and_notify_stock_news)
schedule.every().hour.at(":05").do(run_and_notify_stock_interval)
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
send_telegram_message("Bot Stopped!")
except Exception as e:
send_telegram_message(f"Bot crashed with error: {e}")
Finally, at the end of the script, we should add the following code section, which will ensure that if you run the Python script from a BAT file or from the command line, it will execute the main function. (This will not be executed if the script is simply imported into another script.)
# Example usage
if __name__ == "__main__":
main()
Next steps and closure
Apparently, what an alerting mechanism should monitor is entirely personal, depending on your trading style or your strategy. Some additional ideas that can be easily implemented are:
Signals: You can receive notifications when technical indicators trigger them. The simplest example is an alert for when a slow and a fast-moving average cross each other.
Another one can be when the RSI enters the overbought or the oversold area
Alert when the stock is breaking the 52-week high or low.
A more advanced approach would be to calculate the rolling correlation of your portfolio and set alerts when it exceeds a threshold, so you know you must take some action to diversify.
Feed an AI model with live data and get some reports.
In conclusion, be cautious about the volume of alerts you receive. You can enhance the script with numerous alerts, but if you don’t read them, it becomes useless. Still, the ability to continuously fine-tune what alerts you want is a valuable benefit of this system.
I hope you enjoyed the article.