A python client for MetaTrader 5 terminal (disclaimer: only for educational purpose)

MetaTrader.py

# Authors: MohammadReza Tayyebi, Microsoft Copilot, DeepSeek-V3
# Description: A Python script for algorithmic trading using MetaTrader 5 (MT5).
# Features: Symbol management, order management, real-time trading signals, manual command execution, and audible feedback.

import MetaTrader5 as mt5
import pandas as pd
import time
import threading
import winsound  # For Windows beep sounds
import os  # For non-Windows systems

# Function to play a beep sound
def play_beep():
    try:
        # Windows
        winsound.Beep(1000, 200)  # Frequency: 1000 Hz, Duration: 200 ms
    except:
        # Non-Windows systems (e.g., Linux, macOS)
        os.system('echo -n "\a"')  # Play a system beep

# Initialize MetaTrader 5
if not mt5.initialize():
    print("Failed to initialize MetaTrader 5.")
    quit()

# Check if the user is already logged in
if not mt5.account_info():
    print("No active trading session found. Please log in to MetaTrader 5 first.")
    mt5.shutdown()
    quit()

# Default trading parameters
symbol = "EURUSD"
lot = 0.1
default_risk = 10  # in pips
default_reward = 20  # in pips
timeframe = mt5.TIMEFRAME_M5

# Global variable to store user commands
user_command = None
current_group = "main"  # Tracks the current command group

# Function to listen for user commands
def listen_for_commands():
    global user_command, current_group
    while True:
        if current_group == "main":
            print("\nMain Menu:")
            print("1. Symbol Management (type 'symbol')")
            print("2. Order Management (type 'order')")
            print("3. Exit (type 'exit')")
            print("Shortcut Commands:")
            print("- Buy with default R/R (type 'b')")
            print("- Sell with default R/R (type 's')")
            print("- Close all positions (type 'c')")
            cmd = input("Enter command: ").strip().lower()
            if cmd == "symbol":
                current_group = "symbol"
            elif cmd == "order":
                current_group = "order"
            elif cmd == "exit":
                break
            elif cmd == "b":
                execute_trade("Buy", default_risk, default_reward)
            elif cmd == "s":
                execute_trade("Sell", default_risk, default_reward)
            elif cmd == "c":
                close_all_positions()
            else:
                print("Invalid command. Use 'symbol', 'order', 'exit', 'b', 's', or 'c'.")
                play_beep()  # Beep for invalid command

        elif current_group == "symbol":
            print("\nSymbol Management:")
            print("1. Change symbol (type 'change_symbol')")
            print("2. Change lot size (type 'change_lot')")
            print("3. Change default R/R (type 'change_rr')")
            print("4. Change timeframe (type 'change_tf')")
            print("5. Back to main menu (type 'back')")
            cmd = input("Enter command: ").strip().lower()
            if cmd == "change_symbol":
                new_symbol = input("Enter new symbol (e.g., EURUSD): ").strip()
                symbol = new_symbol
                print(f"Symbol changed to {symbol}.")
                play_beep()  # Beep for successful action
            elif cmd == "change_lot":
                new_lot = float(input("Enter new lot size (e.g., 0.1): "))
                lot = new_lot
                print(f"Lot size changed to {lot}.")
                play_beep()  # Beep for successful action
            elif cmd == "change_rr":
                new_risk = float(input("Enter new default risk (in pips): "))
                new_reward = float(input("Enter new default reward (in pips): "))
                default_risk = new_risk
                default_reward = new_reward
                print(f"Default R/R changed to {default_risk}/{default_reward}.")
                play_beep()  # Beep for successful action
            elif cmd == "change_tf":
                new_tf = int(input("Enter new timeframe in minutes (e.g., 5 for M5): "))
                timeframes = {1: mt5.TIMEFRAME_M1, 5: mt5.TIMEFRAME_M5, 15: mt5.TIMEFRAME_M15, 60: mt5.TIMEFRAME_H1}
                timeframe = timeframes.get(new_tf, mt5.TIMEFRAME_M5)
                print(f"Timeframe changed to {new_tf} minutes.")
                play_beep()  # Beep for successful action
            elif cmd == "back":
                current_group = "main"
            else:
                print("Invalid command.")
                play_beep()  # Beep for invalid command

        elif current_group == "order":
            print("\nOrder Management:")
            print("1. Close profitable positions (type 'close_profitable')")
            print("2. Close losing positions (type 'close_losing')")
            print("3. Close all positions (type 'close_all')")
            print("4. Buy with default R/R (type 'buy_default')")
            print("5. Buy with custom R/R (type 'buy_custom <risk> <reward>')")
            print("6. Sell with default R/R (type 'sell_default')")
            print("7. Sell with custom R/R (type 'sell_custom <risk> <reward>')")
            print("8. Back to main menu (type 'back')")
            cmd = input("Enter command: ").strip().lower()
            if cmd == "close_profitable":
                close_profitable_positions()
            elif cmd == "close_losing":
                close_losing_positions()
            elif cmd == "close_all":
                close_all_positions()
            elif cmd == "buy_default":
                execute_trade("Buy", default_risk, default_reward)
            elif cmd.startswith("buy_custom"):
                try:
                    _, risk, reward = cmd.split()
                    execute_trade("Buy", float(risk), float(reward))
                except ValueError:
                    print("Invalid command. Use 'buy_custom <risk> <reward>'.")
                    play_beep()  # Beep for invalid command
            elif cmd == "sell_default":
                execute_trade("Sell", default_risk, default_reward)
            elif cmd.startswith("sell_custom"):
                try:
                    _, risk, reward = cmd.split()
                    execute_trade("Sell", float(risk), float(reward))
                except ValueError:
                    print("Invalid command. Use 'sell_custom <risk> <reward>'.")
                    play_beep()  # Beep for invalid command
            elif cmd == "back":
                current_group = "main"
            else:
                print("Invalid command.")
                play_beep()  # Beep for invalid command

# Function to execute a trade
def execute_trade(action, risk_pips, reward_pips):
    try:
        # Get the latest market price
        tick = mt5.symbol_info_tick(symbol)
        if tick is None:
            print(f"Failed to get tick data for {symbol}.")
            return

        if action == "Buy":
            order_type = mt5.ORDER_TYPE_BUY
            price = tick.ask
            sl = price - risk_pips * mt5.symbol_info(symbol).point
            tp = price + reward_pips * mt5.symbol_info(symbol).point
        elif action == "Sell":
            order_type = mt5.ORDER_TYPE_SELL
            price = tick.bid
            sl = price + risk_pips * mt5.symbol_info(symbol).point
            tp = price - reward_pips * mt5.symbol_info(symbol).point

        # Prepare the order request
        order_request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": lot,
            "type": order_type,
            "price": price,
            "sl": sl,
            "tp": tp,
            "deviation": 10,
        }

        # Send the order
        print("Order Request:", order_request)  # Debugging
        order_result = mt5.order_send(order_request)
        print("Order Result:", order_result)  # Debugging

        if order_result.retcode != mt5.TRADE_RETCODE_DONE:
            print(f"Failed to execute {action} order. Error: {order_result.comment}")
            play_beep()  # Beep for failed order
        else:
            print(f"{action} order executed successfully. Order ID: {order_result.order}")
            play_beep()  # Beep for successful order
    except Exception as e:
        print(f"An error occurred: {e}")
        play_beep()  # Beep for error

# Function to close profitable positions
def close_profitable_positions():
    positions = mt5.positions_get(symbol=symbol)
    if positions:
        for position in positions:
            if position.profit > 0:
                close_position(position)
    else:
        print("No open positions to close.")
        play_beep()  # Beep for no positions

# Function to close losing positions
def close_losing_positions():
    positions = mt5.positions_get(symbol=symbol)
    if positions:
        for position in positions:
            if position.profit < 0:
                close_position(position)
    else:
        print("No open positions to close.")
        play_beep()  # Beep for no positions

# Function to close all positions
def close_all_positions():
    positions = mt5.positions_get(symbol=symbol)
    if positions:
        for position in positions:
            close_position(position)
    else:
        print("No open positions to close.")
        play_beep()  # Beep for no positions

# Function to close a single position
def close_position(position):
    tick = mt5.symbol_info_tick(symbol)
    if tick is None:
        print(f"Failed to get tick data for {symbol}.")
        return

    close_type = mt5.ORDER_TYPE_SELL if position.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
    close_price = tick.bid if position.type == mt5.ORDER_TYPE_BUY else tick.ask
    close_request = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": symbol,
        "volume": position.volume,
        "type": close_type,
        "position": position.ticket,
        "price": close_price,
        "deviation": 10,
    }
    close_result = mt5.order_send(close_request)
    if close_result.retcode != mt5.TRADE_RETCODE_DONE:
        print(f"Failed to close position {position.ticket}. Error: {close_result.comment}")
        play_beep()  # Beep for failed close
    else:
        print(f"Closed position {position.ticket} successfully.")
        play_beep()  # Beep for successful close

# Start the command listener in a separate thread
command_thread = threading.Thread(target=listen_for_commands, daemon=True)
command_thread.start()

# Main loop
print("\nStarting live trading monitor. Press Ctrl+C to exit.\n")
try:
    while True:
        time.sleep(0.1)  # Small delay to avoid high CPU usage
except KeyboardInterrupt:
    print("\nScript terminated by user.")

# Shutdown MetaTrader 5
mt5.shutdown()