It can be complex to keep track with all the developments in the stock market. Luckily, as a developer, we can help ourselves with automation. However, sometimes complexity can make this a daunting task. This does not need to be the case all the time. In this blog post, I explain how to write a script using Python to receive such an automated message about stocks in under 100 lines.
The plan
Firstly, we need a plan for our script. We want a program that sends us messages about the current developments from ETFs and stocks. Our program is consists of 3 different parts. The first module loads in a JSON file with all our stocks we want to watch. The second module makes the request to the API and the third and last module builds a message out of the responded data. This message is to be sent regularly with a timer unit or a cron job. We can use a Discord bot for receiving the push notification. This comes with the advantage, that we only need to request a webhook.
What API is used?
For this article, I use the market data API from the German startup lemon.markets. They deliver a fantastic developer experience and have extensive documentation. However, it is only a viable choice, if you are based in Germany. For US market, I would recommend taking a look at Alpaca. But the article stands more independent to the choice of the API. My focus lies on the requirements (10 request rate limit from the free plan in the case of lemon.markets) and how to use the given resources efficiently. So, let’s get going:
Part 1: Loading the data
We start with the first module of our script: Loading the JSON file into a dict. Our JSON file looks something like this:
{
"watchlist": [
{
"isin": "US4581401001",
"name": "Intel"
}
]
}This structure is not complex and that is great. We do not need more. At first, we create a dictionary of stocks. Why a dict and not a list? A dict will help us later by merging of different datasets. So, we open the file, load the JSON in and grab the watchlist array. Afterwards, we iterate over all the position.
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]The ISIN or International Securities Identification Number is an international identifier assigned to each stock. This identifier is unique and the perfect candidate for the key of our dictionaries. We use the ISIN and add the name as the value.
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_listNow we got a dict with our stocks. Let’s move on to the request.
Part 2: Get the stock data
We want to calculate a rough stock price development. For this, we only need to know how the stock performed in the morning and the evening. Here we can use the data points of a OHLC chart. But what is a OHLC chart?
What are OHLC charts?
An OHLC (Open-High-Low-Close) chart shows the price movement of a financial instrument. It consists of a candlestick or bar chart that shows the price range for a given time period. The data points for the candlestick consist of four parts: the opening price (Open), the highest price (High), the lowest price (Low) and the closing price (Close).
We are only interested in opening price and closing price of a day. For the query, we only need the ISIN of the stock and the API Key from lemon.markets. With d1, we select the day time range. We also make a placeholder for the ISIN so that we can replace them. Furthermore, we get the API Key from an environment variable. This ensures, that we do not accidentally leak API keys. Lastly, we convert the result into a JSON.
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()In the ideal case, we get the following back:
{
'status': 'ok',
'time': '2022-12-26T12:37:50.270+00:00',
'results':
[
{
'isin': 'US00507V1098',
'o': 70.88,
'h': 71.55,
'l': 70.82,
'c': 71.52,
'v': 269,
'pbv': 19210.93,
't': '2022-12-23T00:00:00.000+00:00',
'mic': 'XMUN'
}
],
'previous': None,
'next': None,
'total': 10,
'page': 1,
'pages': 1
}We only need to iterate on the results to obtain the necessary data.
open_price = data_day["results"][0]["o"]
close_price = data_day["results"][0]["c"]From here, it is easy to calculate the gains and losses of the day. We divide the opening price through the closing price and convert it to a percentage. Finally, we round the value to 4 digits to ensure that no long, crooked numbers appear in our message later on.
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)At the end, we only need to compose the string. Also, I include the latest stock price. But why do I take the Unicode numbers in the message instead of the real characters? Well, the Discord API can have problems with these characters, so I take this variant.
Together, it forms this function:
def get_message(isin: str, name: str) -> str:
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
open_price = data_day["results"][0]["o"]
close_price = data_day["results"][0]["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
message = f'{name} -> price: {close_price} \u20ac, 24h: {gain_loss_price_day} \u0025'
return messagePart 4: Sending the message
The last module is by far the easiest. We only need to make a POST request Discord. In the body of the request, we include our message. Of course, you need to create first a webhook on a given server. For this, you will get a special URL, which I also put into an environment variable.
def send_discord_message(discord_url: str, discord_message: str):
requests.post(discord_url, json={"content": f'{discord_message}'})At the end, we only need to iterate over the stock_config dict and build the messages.
All together results in this:
import json
import os
import requests
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_list
def get_message(isin: str, name: str) -> str:
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
open_price = data_day["results"][0]["o"]
close_price = data_day["results"][0]["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
message = f'{name} -> price: {close_price} \u20ac, 24h: {gain_loss_price_day} \u0025'
return message
def send_discord_message(discord_url: str, discord_message: str):
requests.post(discord_url, json={"content": f'{discord_message}'})
stock_config = read_config()
message_array = ""
for key, value in stock_config.items():
message_array += get_message(key, value)
send_discord_message(os.environ.get("DISCORDURL"), message_array)Part 5: Batching of the requests
All in all, we have a functioning script now. However, in the current version of the code, we dispatch a request to lemon.market for each individual share. This would result in only 10 shares per minute in the free plan. But we can do better!
We can request up to 10 shares per API call. So, let’s combine the different requests. This means that we need to restructure the second module. Now it needs to build different chunks out of the config data, process the data per chunk and merge all the data together. The result is that we replace the getMessage function with more specialized ones:
def get_data()
def merge_data()
def single_batch()
def build_message()Constructing a single batched request for stock data
The core is located in the get_data function. Here, we can borrow the most code from the old get_message. But how do we batch the requests? The lemon.markets API allows us to ask for multiple ISINs at once. We only need to chain all the ISINs together in something like this: isin=US00507V1098,US00724F1012,US0162551016,US02079K3059,US02079K1079,US0231351067,US0079031078,US0255371017,US0311621009,US0326541051.
This part is done right at the beginning. We simply take the keys from the dict join them together with a comma.
def get_data(stock_list_chunk: dict) -> (dict, dict):
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)Then, we make the call. Afterward, we just iterate over the result of the query and add the gains and closing prices to a dict. It is important that the dicts have the same key in the form of the ISIN. All in all, the code should look familiar to the get_message function.
def get_data(stock_list_chunk: dict) -> (dict, dict):
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin_string}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
gains_stock = {}
close_prices = {}
for stock in data_day["results"]:
open_price = stock["o"]
close_price = stock["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
gains_stock[stock["isin"]] = gain_loss_price_day
close_prices[stock["isin"]] = close_price
return gains_stock, close_pricesThe next step is to merge the individual dictionaries. For this, we iterate over the keys of the first dict and add the remaining elements into a tuple.
def merge_data(list_chunk: dict, gain_day: dict, end_price_day: dict) -> dict:
merged_dict = {}
for k in list_chunk.keys():
merged_dict[k] = (list_chunk[k], gain_day[k], end_price_day[k])
return merged_dictIn single_batch() we merge all the functions from above.
def single_batch(chunk: dict) -> dict:
gains, close_prices = get_data(chunk)
merged_data = merge_data(chunk, gains, close_prices)
return merged_dataLast but not least comes the build_message() method. This function gets a dict and iterates over it to create the larger message.
def build_message(data_dict: dict) -> str:
messages = ""
for item in data_dict.values():
messages += f'{item[0]} -> price: {item[2]} \u20ac, 24h: {item[1]} \u0025 \n'
return messagesAdditionally, we append our send_discord_message function with an error check.
def send_discord_message(discord_url: str, discord_message: str):
result = requests.post(discord_url, json={"content": f'{discord_message}'})
try:
result.raise_for_status()
except requests.exceptions.HTTPError as err:
print(err)
else:
print("Payload delivered successfully, code {}.".format(result.status_code))Finally, we also need to batch it properly. So we go over our stock_info list in the increment of 10 and create a chunk each time. Then this chunk is transferred into a larger data set.
if __name__ == "__main__":
stock_info = read_config()
all_data = {}
for count in range(0, len(stock_info), 10):
stock_chunk = {k: stock_info[k] for k in list(stock_info)[count:(count + 10)]}
batch = single_batch(stock_chunk)
all_data.update(batch)
message_array = build_message(all_data)
send_discord_message(os.environ.get("DISCORDURL"), message_array)Now, the code should look like this:
import json
import os
import requests
# function to read the config file and return a dictionary of ISIN codes and stock names
def read_config() -> dict:
stock_list = {}
with open(r"./config2.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_list
# function to get data for a list of ISIN codes
def get_data(stock_list_chunk: dict) -> (dict, dict):
# get the list of ISIN codes and convert the list of ISIN codes to a string separated by commas
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin_string}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
gains_stock = {}
close_prices = {}
# iterate through the results and calculate the gain/loss and the closing price for each stock
for stock in data_day["results"]:
open_price = stock["o"]
close_price = stock["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
gains_stock[stock["isin"]] = gain_loss_price_day
close_prices[stock["isin"]] = close_price
return gains_stock, close_prices
# function to merge the stock information, gain/loss, and closing price into a single dictionary
def merge_data(list_chunk: dict, gain_day: dict, end_price_day: dict) -> dict:
merged_dict = {}
# iterate through the list of ISIN codes and create a tuple of the stock information, gain/loss, and closing price
for k in list_chunk.keys():
merged_dict[k] = (list_chunk[k], gain_day[k], end_price_day[k])
return merged_dict
# function to get the data for a single chunk of ISIN codes
def single_batch(chunk: dict) -> dict:
gains, close_prices = get_data(chunk)
merged_data = merge_data(chunk, gains, close_prices)
return merged_data
# function to build a message string from the data dictionary
def build_message(data_dict: dict) -> str:
messages = ""
# iterate through the data and add it to the message string
for item in data_dict.values():
messages += f'{item[0]} -> price: {item[2]} \u20ac, 24h: {item[1]} \u0025 \n'
return messages
# function for sending the message to a discord webhook
def send_discord_message(discord_url: str, discord_message: str):
# post to discrod via webhook
result = requests.post(discord_url, json={"content": f'{discord_message}'})
# check potential errors
try:
result.raise_for_status()
except requests.exceptions.HTTPError as err:
print(err)
else:
print("Payload delivered successfully, code {}.".format(result.status_code))
if __name__ == "__main__":
stock_info = read_config()
all_data = {}
# create diffrent chunk from the stockInfo dict and call the API
for count in range(0, len(stock_info), 10):
stock_chunk = {k: stock_info[k] for k in list(stock_info)[count:(count + 10)]}
batch = single_batch(stock_chunk)
all_data.update(batch)
# build the complete message
message_array = build_message(all_data)
send_discord_message(os.environ.get("DISCORDURL"), message_array)
Automation
To ensure that the automated message is sent regularly, you need to make sure that the script is executed daily. To do this, you can create a systemd unit on the server that will run the script daily. Alternatively, a simple Cron job gets the task also done.
Conclusion
This is just the beginning. The script is quite modular, so it should be easy to extend. Sending a Discord message about stock updates is more or less a placeholder for some real applications. This is the advantage when thinks more modular. Each module can be replaced or rewritten to surf something different. Why loading in a static JSON. Let’s track a whole index by calling a different API. Alternatively, you can create a personal overview with the winners and losers of the day. Or you can write the query in the second module and add a RSI calculation to the third module. Now you can be informed when some share crosses a specific threshold. Or you create a small desktop with Rust and Tauri. The possibilities are endless.
The only thing I want to give on the way is that no matter how big and complex the application, still leave a clear and simple structure. Make the code clear and easy to understand. Needless complexity never solved a problem. Remember, this is an area that could impact your pocket.
Here is the code of this article:
https://gist.github.com/ngarske/25a6cd75e72c5008a79a3945a4331371
Schreibe einen Kommentar