Automating Stock Market Updates with Python

main.py

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)