249 Zeilen
9,7 KiB
Python
249 Zeilen
9,7 KiB
Python
'''
|
|
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'}}
|