import json import os # import pyttsx3 import telepot # Import the Telepot library from datetime import datetime, timedelta, timezone from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger import time # Import the time module for sleep functionality # Path to the signals JSON file json_file_path = "/home/gjin/Documents/zscore/signals.json" # Telegram bot credentials BOT_TOKEN = '6787349532:AAGSwuiEJeykI2rQgsoB8C_iXZYKYuJyoOM' # Replace with your bot token CHAT_ID = '-4284060853' # Replace with your chat ID # Initialize the Telegram bot bot = telepot.Bot(BOT_TOKEN) # Initialize the pyttsx3 engine # engine = pyttsx3.init() # # Set a more natural speaking rate (around 150-160 words per minute) # engine.setProperty('rate', 155) # Adjust this as needed # engine.setProperty('volume', 1.0) # Max volume (range: 0.0 to 1.0) # Maintain a set of already processed alarms triggered_alarms = set() # Function to play alarm sound def play_alarm_sound(): try: os.system("echo -e '\a'") # Produces a beep sound on Linux/Unix. For Windows, use `winsound.Beep()`. except Exception as e: print(f"Error playing sound: {e}") # Function to speak the output (symbol, signal, zscore) # def speak_zscore_signal(symbol, signal, zscore, timeframe, realtime_ph_time): # message = f"The signal for {symbol} is {signal} with a zscore of {zscore:.2f} timeframe is {timeframe}\n realtime_ph_time:{realtime_ph_time}" # print(message) # Print the message to the console as well # engine.say(message) # engine.runAndWait() # Function to send a Telegram message def send_telegram_message(symbol, signal, zscore, timeframe, realtime_ph_time): try: message = f"🚨 Alarm Triggered! \nSymbol: {symbol}\nSignal: {signal}\nZ-score: {zscore:.2f} \nTimeframe: {timeframe} \n realtime_ph_time:{realtime_ph_time}" bot.sendMessage(CHAT_ID, message) print(f"Telegram message sent: {message}") except Exception as e: print(f"Error sending Telegram message: {e}") # def check_recent_zscore_and_alarm(json_file_path): # global triggered_alarms # Access the global set of triggered alarms # try: # # Load the JSON data # with open(json_file_path, "r") as file: # data = json.load(file) # # Get current UTC time and 1 hour ago in UTC # current_utc_time = datetime.now(timezone.utc) # one_hour_ago = current_utc_time - timedelta(hours=1) # btcdom_triggered = False # Flag to track if BTCDOM/USDT alarm has been triggered # btcdom_zscore = None # Variable to store the zscore for BTCDOM/USDT # # Get the zscore for BTCDOM/USDT from the data # for signal in data: # if signal["symbol"] == "BTCDOM/USDT": # btcdom_zscore = signal["zscore"] # break # for signal in data: # # Parse signal date and time as UTC # date_and_time = datetime.strptime(signal["date_and_time"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc) # zscore = signal["zscore"] # symbol = signal["symbol"] # time_frame = signal["time_frame"] # realtime_ph_time = signal["realtime_ph_time"] # # Create a unique identifier for this alarm # alarm_id = f"{symbol}_{date_and_time.isoformat()}_{zscore:.2f}" # # Check if the signal is within the last 1 hour in UTC # if one_hour_ago <= date_and_time <= current_utc_time: # # If BTCDOM/USDT zscore is < -2 and the altcoin zscore is > +2 # if btcdom_zscore < -2 and zscore > 2 and alarm_id not in triggered_alarms: # print(f"Strongly Sell Signal! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}, realtime_ph_time {realtime_ph_time}") # play_alarm_sound() # speak_zscore_signal(symbol, "Strongly Sell", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Strongly Sell", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # # If BTCDOM/USDT zscore is > +2 and the altcoin zscore is < -2 # elif btcdom_zscore > 2 and zscore < -2 and alarm_id not in triggered_alarms: # print(f"Strongly Buy Signal! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}") # play_alarm_sound() # speak_zscore_signal(symbol, "Strongly Buy", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Strongly Buy", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # # Check for other conditions for Sell, Buy, Prepare for Sell, or Prepare for Buy # elif zscore > 2 and alarm_id not in triggered_alarms: # Sell signal (zscore > +2) # print(f"Sell Signal! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}") # play_alarm_sound() # speak_zscore_signal(symbol, "Sell", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Sell", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # elif zscore < -2 and alarm_id not in triggered_alarms: # Buy signal (zscore < -2) # print(f"Buy Signal! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}") # play_alarm_sound() # speak_zscore_signal(symbol, "Buy", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Buy", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # elif 1.5 < zscore <= 2 and alarm_id not in triggered_alarms: # Prepare for Sell signal (0 < zscore <= 2) # print(f"Prepare for Sell! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}") # play_alarm_sound() # speak_zscore_signal(symbol, "Prepare for Sell/HOLD", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Prepare for Sell/HOLD", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # elif -2 < zscore < -1.5 and alarm_id not in triggered_alarms: # Prepare for Buy signal (-2 < zscore < 0) # print(f"Prepare for Buy! {symbol} at {signal['date_and_time']} with zscore {zscore} time frame is {time_frame}") # play_alarm_sound() # speak_zscore_signal(symbol, "Prepare for Buy/HOLD", zscore, time_frame, realtime_ph_time) # send_telegram_message(symbol, "Prepare for Buy/HOLD", zscore, time_frame, realtime_ph_time) # triggered_alarms.add(alarm_id) # # Wait for 1 minute to avoid spamming # time.sleep(60) # except Exception as e: # print(f"Error: {e}") # State tracking for each symbol zscore_states = {} def save_data(symbol, zscore, time_frame, realtime_ph_time): # Save the Z-score data to a file or database for review with open("zscore_data_log.txt", "a") as file: file.write(f"{datetime.now()}, {symbol}, {zscore}, {time_frame}, {realtime_ph_time}\n") def check_recent_zscore_and_alarm(json_file_path): global zscore_states # Access the global state tracking try: with open(json_file_path, "r") as file: data = json.load(file) current_utc_time = datetime.now(timezone.utc) one_hour_ago = current_utc_time - timedelta(hours=1) for signal in data: # Parse signal details symbol = signal["symbol"] zscore = signal["zscore"] date_and_time = datetime.strptime(signal["date_and_time"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc) time_frame = signal["time_frame"] realtime_ph_time = signal["realtime_ph_time"] # Skip processing signals outside the 1-hour window if not (one_hour_ago <= date_and_time <= current_utc_time): continue # Initialize state tracking for the symbol if symbol not in zscore_states: zscore_states[symbol] = {"was_low": False, "was_high": False} # Determine if an alarm should be triggered state = zscore_states[symbol] # Save data if Z-score hits extreme values (-2 or 2) if zscore == -2 or zscore == 2: save_data(symbol, zscore, time_frame, realtime_ph_time) print(f"Data saved for {symbol} at Z-score: {zscore}") # Alarm conditions if zscore < -2: # Z-score is very low (below -2, no alarm yet) state["was_low"] = True state["was_high"] = False elif zscore > 2: # Z-score is very high (above 2, no alarm yet) state["was_high"] = True state["was_low"] = False elif -1.85 <= zscore < -1: # Returning to mid-range from low extreme if state["was_low"]: # Trigger only if it was previously low print(f"Buy Alarm! {symbol} returning to range with Z-score {zscore}") play_alarm_sound() # speak_zscore_signal(symbol, "Buy Alarm", zscore, time_frame, realtime_ph_time) send_telegram_message(symbol, "Buy Alarm", zscore, time_frame, realtime_ph_time) state["was_low"] = False # Reset the state elif 1 <= zscore <= 1.85: # Returning to mid-range from high extreme if state["was_high"]: # Trigger only if it was previously high print(f"Sell Alarm! {symbol} returning to range with Z-score {zscore}") play_alarm_sound() # speak_zscore_signal(symbol, "Sell Alarm", zscore, time_frame, realtime_ph_time) send_telegram_message(symbol, "Sell Alarm", zscore, time_frame, realtime_ph_time) state["was_high"] = False # Reset the state # Wait for 1 minute to avoid spamming time.sleep(60) except Exception as e: print(f"Error: {e}") # Scheduler function to run every 3:55 to 4:10 and 7:55 to 8:10 def scheduled_task(): check_recent_zscore_and_alarm(json_file_path) # Set up the scheduler scheduler = BlockingScheduler() trigger1 = CronTrigger(minute="55-59") trigger2 = CronTrigger(minute="00-10") # Define triggers # trigger1 = CronTrigger(minute="55-59", hour="3,7,11,15,19,23") # trigger2 = CronTrigger(minute="00-10", hour="4,8,12,16,20,0") # Add the jobs to the scheduler scheduler.add_job(scheduled_task, trigger1) scheduler.add_job(scheduled_task, trigger2) # Start the scheduler scheduler.start()