358 lines
12 KiB
Python
358 lines
12 KiB
Python
# 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()
|