Spaces:
Sleeping
Sleeping
| 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() | |