# import MetaTrader5 as mt5 # import pytz # import json # import os # import time # import smtplib # from datetime import datetime, timedelta # from email.mime.text import MIMEText # from email.mime.multipart import MIMEMultipart # import logging # # ----------------------------------- # # Logging Setup # # ----------------------------------- # logging.basicConfig( # format="%(asctime)s | %(levelname)s | %(message)s", # level=logging.INFO, # datefmt="%Y-%m-%d %H:%M:%S" # ) # # ----------------------------------- # # Load Configuration # # ----------------------------------- # def load_config(filepath="config.json"): # if not os.path.exists(filepath): # raise FileNotFoundError(f"Config file not found: {filepath}") # with open(filepath, "r") as f: # return json.load(f) # config = load_config() # EMAIL_CONFIG = config["email"] # # ----------------------------------- # # Initialize MT5 # # ----------------------------------- # def initialize_mt5(): # logging.info("Initializing MT5 terminal...") # if not mt5.initialize( # path=config["mt5Pathway"], # login=int(config["username"]), # password=config["password"], # server=config["server"] # ): # error = mt5.last_error() # logging.error(f"MT5 initialize failed → Code {error[0]}: {error[1]}") # raise RuntimeError(f"MT5 initialize failed: {error}") # logging.info("MT5 successfully connected.") # # ----------------------------------- # # Email Sender # # ----------------------------------- # def send_email(subject, body): # msg = MIMEMultipart() # msg['From'] = EMAIL_CONFIG["sender_email"] # msg['To'] = EMAIL_CONFIG["recipient_email"] # msg['Subject'] = subject # msg.attach(MIMEText(body, "plain")) # try: # with smtplib.SMTP(EMAIL_CONFIG["smtp_server"], EMAIL_CONFIG["smtp_port"]) as server: # server.starttls() # server.login(EMAIL_CONFIG["sender_email"], EMAIL_CONFIG["sender_password"]) # server.send_message(msg) # logging.info(f"Email sent → {EMAIL_CONFIG['recipient_email']}") # except Exception as e: # logging.error(f"Email failed: {e}") # # ----------------------------------- # # NY Reference Times # # ----------------------------------- # def get_ny_levels(): # tz = pytz.timezone("America/New_York") # now = datetime.now(tz) # today = now.date() # levels = { # "daily_open": tz.localize(datetime.combine(today, datetime.min.time())), # 00:00 NY # "news_time": tz.localize(datetime.combine(today, datetime.strptime("08:30", "%H:%M").time())), # "market_open": tz.localize(datetime.combine(today, datetime.strptime("09:30", "%H:%M").time())), # "session_close": tz.localize(datetime.combine(today, datetime.strptime("18:00", "%H:%M").time())) # } # return levels # # ----------------------------------- # # Get Price at a Specific Time # # ----------------------------------- # def get_price_at_time(symbol, target_time): # """Get the price closest to a specific timestamp.""" # utc_time = target_time.astimezone(pytz.utc) # rates = mt5.copy_rates_from(symbol, mt5.TIMEFRAME_M1, utc_time, 1) # if rates is None or len(rates) == 0: # return None # return rates[0]['close'] # # ----------------------------------- # # Monitor Prices & Email # # ----------------------------------- # def monitor_prices(symbols): # tz = pytz.timezone("America/New_York") # while True: # now = datetime.now(tz) # levels = get_ny_levels() # logging.info(f"Checking current prices... NY Time: {now.strftime('%Y-%m-%d %H:%M:%S')}") # email_body = f"Price Update - NY Time: {now.strftime('%Y-%m-%d %H:%M:%S')}\n\n" # for symbol in symbols: # tick = mt5.symbol_info_tick(symbol) # if tick is None: # logging.warning(f"No tick data for {symbol}") # email_body += f"{symbol}: No tick data available.\n\n" # continue # current_price = tick.last # logging.info(f"--- {symbol} ---") # logging.info(f"Current Price: {current_price:.2f}") # email_body += f"--- {symbol} ---\n" # email_body += f"Current Price: {current_price:.2f}\n" # # Calculate and log differences # for key, label in [ # ("daily_open", "00:00 Daily Open"), # ("news_time", "08:30 News Time"), # ("market_open", "09:30 Market Open"), # ("session_close", "18:00 Session Close") # ]: # ref_price = get_price_at_time(symbol, levels[key]) # if ref_price: # diff = current_price - ref_price # logging.info(f"{label}: {ref_price:.2f} | Δ {diff:.2f}") # email_body += f"{label}: {ref_price:.2f} | Δ {diff:.2f}\n" # email_body += "\n" # logging.info("-" * 50) # # Send email summary # send_email("NY Levels Price Update", email_body) # # Wait 5 minutes before next update # time.sleep(300) # # ----------------------------------- # # Main # # ----------------------------------- # def main(): # try: # initialize_mt5() # symbols = ["XAUUSD", "USTEC", "US30"] # Gold, Nasdaq, Dow # monitor_prices(symbols) # except Exception as e: # logging.error(f"Fatal error in main(): {e}") # finally: # logging.info("Shutting down MT5...") # mt5.shutdown() # if __name__ == "__main__": # main() # import MetaTrader5 as mt5 import pytz import logging import json import os import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime, timedelta # ====================================================== # Logging Configuration # ====================================================== logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", handlers=[logging.StreamHandler()] ) log = logging.getLogger() # ====================================================== # Load Config # ====================================================== def load_config(filepath="config.json"): if not os.path.exists(filepath): raise FileNotFoundError(f"Config file not found: {filepath}") with open(filepath, "r") as f: return json.load(f) # ====================================================== # MT5 Initialization # ====================================================== def initialize_mt5(config): log.info("Initializing MetaTrader 5...") if not mt5.initialize( path=config["mt5Pathway"], login=int(config["username"]), password=config["password"], server=config["server"] ): raise RuntimeError(f"MT5 initialize failed: {mt5.last_error()}") log.info("MT5 connected successfully") # ====================================================== # Symbol Management # ====================================================== def ensure_symbol_visible(symbol): info = mt5.symbol_info(symbol) if info is None: log.error(f"Symbol not found: {symbol}") return False if not info.visible: if mt5.symbol_select(symbol, True): log.info(f"Symbol {symbol} made visible") else: log.error(f"Failed to make symbol visible: {symbol}") return False return True # ====================================================== # Price Fetching # ====================================================== def get_current_price(symbol): if not ensure_symbol_visible(symbol): return None tick = mt5.symbol_info_tick(symbol) if tick: bid = getattr(tick, "bid", 0.0) or 0.0 ask = getattr(tick, "ask", 0.0) or 0.0 if bid > 0 and ask > 0: return (bid + ask) / 2.0 if getattr(tick, "last", 0.0) > 0: return float(tick.last) # fallback: get recent ticks utc_now = datetime.utcnow() ticks = mt5.copy_ticks_from(symbol, utc_now - timedelta(seconds=30), 50, mt5.COPY_TICKS_ALL) if ticks is not None and len(ticks) > 0: last_tick = ticks[-1] if "last" in last_tick.dtype.names and last_tick["last"] > 0: return float(last_tick["last"]) if "bid" in last_tick.dtype.names and "ask" in last_tick.dtype.names: b, a = float(last_tick["bid"]), float(last_tick["ask"]) if b > 0 and a > 0: return (b + a) / 2.0 return None def get_price_at_time(symbol, target_dt_ny): if not ensure_symbol_visible(symbol): return None # convert NY time to UTC utc_from = target_dt_ny.astimezone(pytz.utc) utc_to = utc_from + timedelta(minutes=1) rates = mt5.copy_rates_range(symbol, mt5.TIMEFRAME_M1, utc_from, utc_to) if rates is None or len(rates) == 0: log.warning(f"No M1 data for {symbol} at {target_dt_ny}") return None return float(rates[0]["open"]) # ====================================================== # Email Sending # ====================================================== def send_email(subject, body): smtp_config = { "smtp_server": "smtp.gmail.com", "smtp_port": 587, "sender_email": "ifairvaluegap@gmail.com", "sender_password": "pyxv pjnd jftb klnz", # Gmail App Password "recipient_email": "griffinskimutai@gmail.com" } msg = MIMEMultipart() msg['From'] = smtp_config["sender_email"] msg['To'] = smtp_config["recipient_email"] msg['Subject'] = subject msg.attach(MIMEText(body, "plain")) try: with smtplib.SMTP(smtp_config["smtp_server"], smtp_config["smtp_port"]) as server: server.starttls() server.login(smtp_config["sender_email"], smtp_config["sender_password"]) server.send_message(msg) log.info(f"Email sent to {smtp_config['recipient_email']}") except Exception as e: log.error(f"Email sending failed: {e}") # ====================================================== # Main # ====================================================== def main(): log.info("===== NY Levels Monitor Starting =====") # Load configuration config = load_config() # Initialize MT5 initialize_mt5(config) # Timezone tz = pytz.timezone("America/New_York") now_ny = datetime.now(tz) log.info(f"Current NY Time: {now_ny}") # Predefined levels levels = { "daily_open": tz.localize(datetime.combine(now_ny.date(), datetime.min.time())), "news_time": tz.localize(datetime.combine(now_ny.date(), datetime.strptime("08:30", "%H:%M").time())), "market_open": tz.localize(datetime.combine(now_ny.date(), datetime.strptime("09:30", "%H:%M").time())), "session_close": tz.localize(datetime.combine(now_ny.date(), datetime.strptime("18:00", "%H:%M").time())) } # Symbols symbols = ["XAUUSD", "USTEC", "US30"] email_body = [] for symbol in symbols: current_price = get_current_price(symbol) if current_price is None: log.error(f"Cannot fetch current price for {symbol}") continue log.info(f"{symbol} Current Price: {current_price}") email_body.append(f"{symbol} Current Price: {current_price}") for key, dt in levels.items(): ref_price = get_price_at_time(symbol, dt) if ref_price is not None: diff = current_price - ref_price log.info(f"{symbol} | {key} ({dt.strftime('%H:%M')} NY): {ref_price} | Δ={diff:.2f}") email_body.append(f"{symbol} {key}: {ref_price} | Δ={diff:.2f}") else: log.warning(f"{symbol} missing data for {key} at {dt.strftime('%H:%M')} NY") # Send Email send_email( subject="Daily Levels Update", body="\n".join(email_body) ) # Shutdown MT5 mt5.shutdown() log.info("===== Script Finished =====") # ====================================================== # Run # ====================================================== if __name__ == "__main__": try: main() except Exception as e: log.error(f"Fatal error in main(): {e}") mt5.shutdown()