231 lines
8.2 KiB
Python
231 lines
8.2 KiB
Python
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
|
|
# ======================================================
|
|
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 Helpers
|
|
# ======================================================
|
|
def get_current_price(symbol):
|
|
"""Fetch current midpoint price"""
|
|
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
|
|
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"])
|
|
return None
|
|
|
|
def get_price_at_time(symbol, target_dt_ny):
|
|
"""Get open price of candle at specific NY time"""
|
|
if not ensure_symbol_visible(symbol):
|
|
return None
|
|
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:
|
|
return None
|
|
return float(rates[0]["open"])
|
|
|
|
# ======================================================
|
|
# Weekly Open Calculation
|
|
# ======================================================
|
|
def get_weekly_open(symbol):
|
|
"""Monday 00:00 NY open price"""
|
|
tz = pytz.timezone("America/New_York")
|
|
now_ny = datetime.now(tz)
|
|
monday = now_ny - timedelta(days=now_ny.weekday()) # Monday of current week
|
|
monday_open = tz.localize(datetime.combine(monday.date(), datetime.min.time()))
|
|
return get_price_at_time(symbol, monday_open)
|
|
|
|
# ======================================================
|
|
# Previous Month High/Low
|
|
# ======================================================
|
|
def get_previous_month_high_low(symbol):
|
|
"""Fetch previous month's high and low using D1 candles"""
|
|
if not ensure_symbol_visible(symbol):
|
|
return None, None
|
|
|
|
tz = pytz.timezone("America/New_York")
|
|
now = datetime.now(tz)
|
|
first_day_this_month = tz.localize(datetime(now.year, now.month, 1))
|
|
last_day_prev_month = first_day_this_month - timedelta(days=1)
|
|
first_day_prev_month = tz.localize(datetime(last_day_prev_month.year, last_day_prev_month.month, 1))
|
|
|
|
rates = mt5.copy_rates_range(symbol, mt5.TIMEFRAME_D1, first_day_prev_month.astimezone(pytz.utc), first_day_this_month.astimezone(pytz.utc))
|
|
if rates is None or len(rates) == 0:
|
|
return None, None
|
|
|
|
highs = [r['high'] for r in rates]
|
|
lows = [r['low'] for r in rates]
|
|
return max(highs), min(lows)
|
|
|
|
# ======================================================
|
|
# Email Sending
|
|
# ======================================================
|
|
def send_email(config, subject, body):
|
|
smtp_config = config["email"]
|
|
|
|
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("===== Weekly & Monthly Levels Monitor =====")
|
|
|
|
# Load configuration
|
|
config = load_config()
|
|
|
|
# Initialize MT5
|
|
initialize_mt5(config)
|
|
|
|
# Symbols to monitor
|
|
symbols = ["XAUUSD", "USTEC", "US30"]
|
|
|
|
tz = pytz.timezone("America/New_York")
|
|
now_ny = datetime.now(tz)
|
|
|
|
email_body = [f"Report Generated at NY Time: {now_ny.strftime('%Y-%m-%d %H:%M:%S')}", ""]
|
|
|
|
for symbol in symbols:
|
|
log.info(f"Processing {symbol}...")
|
|
|
|
# Current price
|
|
current_price = get_current_price(symbol)
|
|
if current_price is None:
|
|
log.error(f"Cannot fetch current price for {symbol}")
|
|
continue
|
|
|
|
# Weekly open
|
|
weekly_open = get_weekly_open(symbol)
|
|
|
|
# Daily open and close
|
|
today_open_time = tz.localize(datetime.combine(now_ny.date(), datetime.min.time()))
|
|
daily_open = get_price_at_time(symbol, today_open_time)
|
|
|
|
# Use last completed D1 candle for daily close
|
|
daily_rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_D1, 1, 2)
|
|
daily_close = float(daily_rates[-2]['close']) if daily_rates is not None and len(daily_rates) >= 2 else None
|
|
|
|
# Previous month high/low
|
|
prev_month_high, prev_month_low = get_previous_month_high_low(symbol)
|
|
|
|
# Build report
|
|
report = [
|
|
f"Symbol: {symbol}",
|
|
f"Current Price: {current_price:.2f}",
|
|
f"Weekly Open (Mon 00:00 NY): {weekly_open if weekly_open else 'N/A'}",
|
|
f"Daily Open: {daily_open if daily_open else 'N/A'}",
|
|
f"Previous Daily Close: {daily_close if daily_close else 'N/A'}",
|
|
f"Prev Month High: {prev_month_high if prev_month_high else 'N/A'}",
|
|
f"Prev Month Low: {prev_month_low if prev_month_low else 'N/A'}",
|
|
]
|
|
|
|
# Relative calculations
|
|
if weekly_open and current_price:
|
|
report.append(f"Δ Current vs Weekly Open: {current_price - weekly_open:.2f}")
|
|
if daily_open and current_price:
|
|
report.append(f"Δ Current vs Daily Open: {current_price - daily_open:.2f}")
|
|
if daily_close and current_price:
|
|
report.append(f"Δ Current vs Daily Close: {current_price - daily_close:.2f}")
|
|
|
|
email_body.extend(report)
|
|
email_body.append("") # spacer
|
|
|
|
# Send email
|
|
send_email(config, subject="Weekly & Monthly Market Report", 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: {e}")
|
|
mt5.shutdown()
|