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:/VDS'

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


class MT5_Control:
    def __init__(self, terminals: dict, mt5_folder: str, mt5_exe: str = 'terminal64.exe'):
        self.mt5_folder = mt5_folder
        self.mt5_exe = mt5_exe

        # Список имён экземпляров (пока фиксированный)
        self.instances_folders = list(terminals.keys())

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

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

        self.instances = terminals.copy()

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

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

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

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

            if 'pid' in self.instances[folder] and self.instances[folder]['pid']:
                print(f'Start mt5.initialize[{folder}]')
                if mt5.initialize(path, timeout=200, 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())

                    code, description = mt5.last_error()
                    status = 'running'

                else:
                    print("initialize() failed, error code =", mt5.last_error())
                    if info['status'] == 'starting':
                        code, description = 0, 'Starting'
                        status = 'starting'
                    else:
                        code, description = mt5.last_error()
                        status = 'running'

                mt5.shutdown()
                print(f'End mt5.initialize[{folder}]')
            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 folder in self.instances_folders:
            self.instances[folder]['pid'] = 0
            self.instances[folder]['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:
                    # Определяем имя экземпляра для данного пути
                    folder = self.paths_instanсes[exe_path]

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

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

            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            # Берём идентификатор его процесса
            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 {folder: instance}

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

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

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

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

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

        # Если имя экземпляра есть среди имеющихся
        if folder 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 {folder: folder}

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