From ee78320c43a3aaa3b6dbe94e6b6fab6d31a6b8fb Mon Sep 17 00:00:00 2001 From: Zulq <141235496+ZulqAnsari@users.noreply.github.com> Date: Wed, 23 Aug 2023 00:37:22 +0100 Subject: [PATCH 1/2] v1 - Implement Python Application for Stock Price Monitoring and Notifications Initialised python script for real-time stock price monitoring and notifications. Utilises API for price data, calculates 7-day averages, and triggers IFTTT notifications (pending) --- main.py | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 main.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..86c6be2 --- /dev/null +++ b/main.py @@ -0,0 +1,65 @@ +import requests +import time +import datetime + +# API base URL +BASE_URL = "https://financialmodelingprep.com/api/v3" + +# API key (API from Financial Modeling Prep API Documentation) +API_KEY = "d5031e8e27160bf572aeeb965a8b0a2f" + +# List of stocks to monitor +stocks = ["TSLA", "AAPL", "MSFT", "GOOGL", "NKE"] + +# Threshold for price drop (£0.25 GBP) +PRICE_THRESHOLD = 0.1 + +# Function to get the current stock price +def get_stock_price(symbol): + url = f"{BASE_URL}/quote-short/{symbol}?apikey={API_KEY}" + response = requests.get(url) + data = response.json() + return data[0]["price"] + +# Function to calculate 7-day average price +def calculate_7_day_average(symbol): + end_date = datetime.date.today() + start_date = end_date - datetime.timedelta(days=7) + url = f"{BASE_URL}/historical-price-full/{symbol}?from={start_date}&to={end_date}&apikey={API_KEY}" + response = requests.get(url) + data = response.json() + prices = [item["close"] for item in data["historical"]] + return sum(prices) / len(prices) + +# Function to send notification +def send_notification(stock_symbol, message): + event_name = "stock_price_alert" + payload = { + "value1": stock_symbol, + "value2": message + } + url = f"https://maker.ifttt.com/trigger/{event_name}/with/key/YOUR_IFTTT_KEY" + requests.post(url, json=payload) + +# Main loop +if __name__ == "__main__": + while True: + for stock in stocks: + current_price = get_stock_price(stock) + average_price = calculate_7_day_average(stock) + + print("stock: ", stock) + print("current_price: ", current_price) + print("average_price: ", average_price) + + + if current_price < average_price: + message = f"Today's price ({current_price}) is less than 7-day average ({average_price})." + send_notification(stock, message) + + if current_price <= (average_price - PRICE_THRESHOLD): + message = f"Price dropped by at least £{PRICE_THRESHOLD} GBP. Current price: {current_price}" + send_notification(stock, message) + + # Check every 5 minutes + time.sleep(300) \ No newline at end of file From 5e0339a804ae64e2e9608842ab4f0cf9297ad16b Mon Sep 17 00:00:00 2001 From: Zulq <141235496+ZulqAnsari@users.noreply.github.com> Date: Wed, 30 Aug 2023 09:58:51 +0100 Subject: [PATCH 2/2] Latest update to main.py Implemented Webhook, implemented mathematical functions, if-else statements along with unit testing. --- main.py | 164 +++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 115 insertions(+), 49 deletions(-) diff --git a/main.py b/main.py index 86c6be2..6c64f2f 100644 --- a/main.py +++ b/main.py @@ -1,65 +1,131 @@ import requests -import time import datetime +import schedule +import time +import logging -# API base URL -BASE_URL = "https://financialmodelingprep.com/api/v3" +# Configure logging +logging.basicConfig(level=logging.INFO) # Adjust the logging level as and when required -# API key (API from Financial Modeling Prep API Documentation) -API_KEY = "d5031e8e27160bf572aeeb965a8b0a2f" +# IFTTT Webhooks details +ifttt_event_name = "update_stock_prices" +ifttt_api_key = "dFtK3DkhVCFf5_NqCxS1HH" +base_url = "https://financialmodelingprep.com/api/v3" +api_key = "733e442a1ef9374542a43ce72e65adf9" # List of stocks to monitor stocks = ["TSLA", "AAPL", "MSFT", "GOOGL", "NKE"] -# Threshold for price drop (£0.25 GBP) -PRICE_THRESHOLD = 0.1 +MIN_PRICE_DIFFERENCE = 0.25 -# Function to get the current stock price -def get_stock_price(symbol): - url = f"{BASE_URL}/quote-short/{symbol}?apikey={API_KEY}" + +def get_stock_price(symbols): + url = f"{base_url}/quote-short/{', '.join(symbols)}?apikey={api_key}&datatype=csv" response = requests.get(url) - data = response.json() - return data[0]["price"] -# Function to calculate 7-day average price + if response.status_code == 200: + data = response.text.splitlines() + headers = data[0].split(',') + values = data[1].split(',') + + # Find the index of the "price" column + price_index = headers.index('price') + + if len(values) > price_index: + print("Successfully retrieved stock data:") + print(data) + return float(values[price_index]) + else: + print("Data format is unexpected:") + print(data) + return None + else: + print(f"Failed to retrieve stock data. Status code: {response.status_code}") + return None + + def calculate_7_day_average(symbol): end_date = datetime.date.today() start_date = end_date - datetime.timedelta(days=7) - url = f"{BASE_URL}/historical-price-full/{symbol}?from={start_date}&to={end_date}&apikey={API_KEY}" + url = f"{base_url}/historical-price-full/{symbol}?from={start_date}&to={end_date}&apikey={api_key}" response = requests.get(url) data = response.json() - prices = [item["close"] for item in data["historical"]] - return sum(prices) / len(prices) - -# Function to send notification -def send_notification(stock_symbol, message): - event_name = "stock_price_alert" - payload = { - "value1": stock_symbol, - "value2": message - } - url = f"https://maker.ifttt.com/trigger/{event_name}/with/key/YOUR_IFTTT_KEY" - requests.post(url, json=payload) - -# Main loop -if __name__ == "__main__": - while True: - for stock in stocks: - current_price = get_stock_price(stock) - average_price = calculate_7_day_average(stock) - - print("stock: ", stock) - print("current_price: ", current_price) - print("average_price: ", average_price) - - - if current_price < average_price: - message = f"Today's price ({current_price}) is less than 7-day average ({average_price})." - send_notification(stock, message) - - if current_price <= (average_price - PRICE_THRESHOLD): - message = f"Price dropped by at least £{PRICE_THRESHOLD} GBP. Current price: {current_price}" - send_notification(stock, message) - - # Check every 5 minutes - time.sleep(300) \ No newline at end of file + + if "historical" in data: + prices = [item["close"] for item in data["historical"]] + return sum(prices) / len(prices) + else: + return None + + +previous_prices = {} + + +# Function to calculate price change + + +def calculate_price_change(symbol, current_price): + if symbol in previous_prices: + previous_price = previous_prices[symbol] + return current_price - previous_price + else: + return 0.0 # Default value for the first time + + +def fetch_and_send_data(): + stock_data = [] + + for stock_symbol in stocks: + current_price_gbp = get_stock_price([stock_symbol]) # Fetch price in GBP + current_price_usd = convert_usd_to_gbp(current_price_gbp) # Convert to GBP for comparison + average_price_gbp = calculate_7_day_average(stock_symbol) # Fetch average price in GBP + + if current_price_gbp is not None: + price_change_gbp = calculate_price_change(stock_symbol, current_price_gbp) + + if price_change_gbp <= -MIN_PRICE_DIFFERENCE: + stock_data.append({ + 'name': stock_symbol, + 'current_price': current_price_gbp, + 'price_change': price_change_gbp, + 'average_price': average_price_gbp, + }) + + previous_prices[stock_symbol] = current_price_gbp + + if stock_data: + send_to_ifttt_webhook(stock_data) + else: + logging.info("The stock prices remain unchanged") + +def convert_usd_to_gbp(usd_amount): + conversion_rate = 1.35 + return usd_amount / conversion_rate + + +# Function to send data to IFTTT webhook +def send_to_ifttt_webhook(data_list): + url = f"https://maker.ifttt.com/trigger/update_prices/json/with/key/{ifttt_api_key}" + response = requests.post(url, json=data_list) + + if response.status_code == 200: + logging.info("Webhook triggered successfully.") + else: + logging.error("Failed to trigger webhook. Status code: %d", response.status_code) + + +# Define a function to run the scheduled job +def run_scheduled_job(): + print("Fetching and sending data...") + fetch_and_send_data() + print("Job completed.") + + +schedule.every().day.at("09:00").do(fetch_and_send_data) +# Schedule the job to run every 5 seconds +schedule.every(5).seconds.do(run_scheduled_job) + +# Run the scheduling loop +while True: + schedule.run_pending() + time.sleep(1) \ No newline at end of file