Gr8/historical.py

import MetaTrader5 as mt5
import pytz
import json
import os
import smtplib
import pandas as pd
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
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.")

# -----------------------------------
# Fetch Weekly Data
# -----------------------------------
def fetch_weekly_data(symbol, timeframe=mt5.TIMEFRAME_H1):
    tz = pytz.timezone("America/New_York")

    # Define last week's start and end (NY time)
    now = datetime.now(tz)
    monday = now - timedelta(days=now.weekday())  # Start of this week
    last_monday = monday - timedelta(days=7)      # Start of previous week
    last_sunday = monday - timedelta(seconds=1)   # End of previous week

    logging.info(f"Fetching {symbol} data from {last_monday} to {last_sunday}")

    # Pull data from MT5
    rates = mt5.copy_rates_range(symbol, timeframe, last_monday, last_sunday)
    if rates is None or len(rates) == 0:
        logging.warning(f"No data found for {symbol}")
        return None

    # Convert to DataFrame
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    return df

# -----------------------------------
# Export and Email
# -----------------------------------
def export_and_email(data_dict):
    # Create email object
    msg = MIMEMultipart()
    msg['From'] = EMAIL_CONFIG["sender_email"]
    msg['To'] = EMAIL_CONFIG["recipient_email"]
    msg['Subject'] = "Weekly Historical Market Data"

    body_text = "Attached are last week's OHLC data for XAUUSD, USTEC, and US30."
    msg.attach(MIMEText(body_text, "plain"))

    # Attach CSV files
    for symbol, df in data_dict.items():
        if df is not None:
            filename = f"{symbol}_last_week.csv"
            df.to_csv(filename, index=False)
            logging.info(f"Exported {symbol} data → {filename}")

            with open(filename, "rb") as f:
                attachment = MIMEBase('application', 'octet-stream')
                attachment.set_payload(f.read())
                encoders.encode_base64(attachment)
                attachment.add_header('Content-Disposition', f'attachment; filename={filename}')
                msg.attach(attachment)

    # Send email
    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 with historical data sent → {EMAIL_CONFIG['recipient_email']}")
    except Exception as e:
        logging.error(f"Email failed: {e}")

# -----------------------------------
# Main
# -----------------------------------
def main():
    logging.info("===== Starting Weekly Data Extraction =====")
    try:
        initialize_mt5()

        # Use USTEC instead of US100
        symbols = ["XAUUSD", "USTEC", "US30"]
        all_data = {}

        for symbol in symbols:
            df = fetch_weekly_data(symbol, mt5.TIMEFRAME_H1)
            if df is not None:
                logging.info(f"{symbol} → {len(df)} rows fetched")
            else:
                logging.warning(f"No data fetched for {symbol}")
            all_data[symbol] = df

        export_and_email(all_data)

    except Exception as e:
        logging.error(f"Fatal error: {e}")
    finally:
        logging.info("Shutting down MT5...")
        mt5.shutdown()
        logging.info("MT5 shutdown complete.")

if __name__ == "__main__":
    main()