Spaces:
Sleeping
Sleeping
File size: 11,172 Bytes
e97cf97 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 |
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()
|