mt5-manager/mt5_control.py

'''
File: mt5_control.py
Description: Логика управления запуском и остановкой терминалов MetaTrader 5
'''

__version__ = '0.1.0'

import subprocess
import psutil
from datetime import datetime
import MetaTrader5 as mt5


# Путь к папке с экземплярами терминалов
MT5_FOLDER = 'C:/MT5'

# Имя запускаемого файла терминала
MT5_EXE = 'terminal64.exe'


class MT5_Control:
    def __init__(self, terminals: dict):
        # Список имён экземпляров (пока фиксированный)
        self.instances_names = list(terminals.keys())

        # Словарь соответствия имя -> путь
        self.instanсes_paths = {name: self.instance_path(
            name) for name in self.instances_names}

        # Словарь соответствия путь -> имя
        self.paths_instanсes = {self.instance_path(
            name): name for name in self.instances_names}
        
        self.instances = terminals.copy()

    def instance_path(self, name: str) -> str:
        '''
        Формирует полный путь по имени экземпляра терминала

        Args:
            name (str): Имя экземпляра терминала.

        Returns:
            path (str): Полый путь к запускаемому файлу экземпляра терминала.
        '''
        return f'{MT5_FOLDER}/{name}/{MT5_EXE}'

    def instance_info(self, name: str) -> dict:
        if name in self.instances:
            info = self.instances[name]
            path = self.instance_path(name)

            if 'pid' in self.instances[name] and self.instances[name]['pid']:
                print(f'Start mt5.initialize[{name}]')
                if mt5.initialize(path, timeout=10000, portable=True):
                    # выведем информацию о настройках и состоянии терминала
                    terminal_info = mt5.terminal_info()
                    if terminal_info != None:
                        info['terminal'] = terminal_info._asdict()

                    account_info = mt5.account_info()
                    if account_info != None:
                        info['account'] = account_info._asdict()

                    info['last_update'] = str(datetime.now())
                    
                else:
                    print("initialize() failed, error code =",mt5.last_error())

                code, description = mt5.last_error()    
                status = 'running'
                
                mt5.shutdown()
                print(f'End mt5.initialize[{name}]')
            else:
                code, description = 0, 'Stopped'
                status = 'stopped'
        else:
            code, description = -1, 'Manager: Terminal not found'
            status = 'not found'

        info['last_error'] = {'code': code, 'description': description}
        info['status'] = status
        return info

    def find_instances(self) -> None:
        '''Получение информации о статусе экземпляров 
        терминала MetaTrader 5 из заданных папок.

        Returns:
            instances (dict): Информация об экземплярах терминала.
        '''

        # Формируем словарь для всех экземпляров по списку имён
        for name in self.instances_names:
            self.instances[name]['pid'] = 0
            self.instances[name]['status'] = 'stopped'

        # Перебираем все запущенные процессы
        for proc in psutil.process_iter(['pid', 'name', 'exe']):
            try:
                # Получаем путь к запускаемому файлу процесса
                exe_path = str(proc.info['exe']).replace('\\', '/')

                # Если он есть среди путей экземпляров терминала
                if exe_path and exe_path in self.paths_instanсes:
                    # Определяем имя экземпляра для данного пути
                    name = self.paths_instanсes[exe_path]

                    # Берём нужный экземпляр
                    instance = self.instances[name]

                    # Запоминаем идентификатор и статус процесса
                    instance['pid'] = proc.info['pid']
                    instance['status'] = 'running'
                
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue

    def load_instances(self) -> dict:
        self.find_instances()
        print(self.instances)
        return self.instances  

    def start_mt5(self, name: str) -> dict:
        '''
        Запуск экземпляра терминала

        Args:
            name (str): Имя экземпляра терминала.

        Returns:
            instance (dict): Информация о запущенном терминале.
        '''

        # Получаем информацию об экземплярах терминала
        instances = self.load_instances()

        # Если имя экземпляра есть среди имеющихся
        if name in instances:
            # Берём нужный экземпляр
            instance = instances[name]

            # Если для него нет запущенного процесса, то
            if not instance['pid']:
                # Запускаем новый процесс
                process = subprocess.Popen(
                    [self.instanсes_paths[name], '/portable'])

                # Идентификатор запущенного процесса
                pid = process.pid

                # Запоминаем идентификатор и статус процесса
                instance['pid'] = pid
                instance['status'] = 'running'

            # Возвращаем результат - успешный запуск
            return {name: instance}

        # Возвращаем результат - имя не найдено
        return {name: {'status': 'not found'}}

    def stop_mt5(self, name: str) -> dict:
        '''
        Остановка терминала

        Args:
            name (str): Имя экземпляра терминала.

        Returns:
            instance (dict): Информация о запущенном терминале.
        '''

        # Получаем информацию об экземплярах терминала
        instances = self.load_instances()

        # Если имя экземпляра есть среди имеющихся
        if name in instances:
            # Берём нужный экземпляр
            instance = instances[name]

            # Берём идентификатор его процесса
            pid = instance['pid']

            # Если ранее терминал был запущен
            if pid:
                try:
                    # Получаем объект процесса терминала
                    p = psutil.Process(pid)

                    # Останавливаем процесс
                    p.terminate()
                    p.wait(10)
                except psutil.NoSuchProcess:
                    pass

            # Удаляем информацию о запущенном ранее процессе
            instance['pid'] = 0
            instance['status'] = 'stopped'

            # Возвращаем результат - успешная остановка
            return {name: instance}

        # Возвращаем результат - процесс не найден
        return {name: {'status': 'not found'}}

    def create_mt5(self, name: str) -> dict:
        '''
        Запуск экземпляра терминала

        Args:
            name (str): Имя экземпляра терминала.

        Returns:
            instance (dict): Информация о запущенном терминале.
        '''

        # Получаем информацию об экземплярах терминала
        instances = self.load_instances()

        # Если имя экземпляра есть среди имеющихся
        if name not in instances:
            info = dict()
            if mt5.initialize():
                # выведем информацию о настройках и состоянии терминала
                terminal_info = mt5.terminal_info()
                if terminal_info != None:
                    info['terminal'] = terminal_info._asdict()

                account_info = mt5.account_info()
                if account_info != None:
                    info['account'] = account_info._asdict()

            mt5.shutdown()
            # Возвращаем результат - успешный запуск
            return {name: name}

        # Возвращаем результат - имя не найдено
        return {name: {'status': 'already exists'}}