''' File: mt5_control.py Description: Логика управления запуском и остановкой терминалов MetaTrader 5 ''' __version__ = '0.5.1' import copy import os import sqlite3 import subprocess import time from pathlib import Path import psutil from config import instance_directory class MT5_Control: # Сколько секунд с последней записи в states считается "свежими" данными. # Сервис пишет раз в 2-3 секунды, порог даёт запас на редкие пропуски. FRESHNESS_THRESHOLD_SEC = 30 STATE_QUERY = ''' SELECT t.path AS terminal_path, t.company AS terminal_company, t.build AS terminal_build, s.updated_at, s.connected, s.ping_last, s.login, s.server, s.currency, s.leverage, s.trade_mode, s.trade_allowed, s.balance, s.equity, s.profit, s.margin, s.margin_free, s.margin_level, s.credit, s.positions, s.company AS account_company, s.holder_name, s.error_code, s.error_text FROM terminals t LEFT JOIN states s ON s.terminal_id = t.id WHERE t.folder = ? ''' def __init__(self, terminals: dict, mt5_folder: str, database: str, mt5_exe: str = 'terminal64.exe'): ''' Инициализирует экземпляр класса MT5_Control. Args: terminals (dict): Словарь с информацией о терминалах. mt5_folder (str): Путь к папке с экземплярами терминалов. database (str): Путь к общей базе SQLite с состоянием терминалов. mt5_exe (str): Имя исполняемого файла терминала (по умолчанию 'terminal64.exe'). ''' # Путь к папке с экземплярами терминалов self.mt5_folder = mt5_folder # Путь к общей базе SQLite self.database = database # Имя исполняемого файла терминала self.mt5_exe = mt5_exe self.configure(terminals, mt5_folder, database) @staticmethod def normalized_path(path: str) -> str: return os.path.normcase(os.path.abspath(path)) def configure(self, terminals: dict, mt5_folder: str, database: str) -> None: self.mt5_folder = mt5_folder self.database = database # Пароли остаются в ConfigStore; словари карточек содержат только # разрешённые для выдачи поля конфигурации и состояние процесса. self.terminal_config = { folder: {**{key: value for key, value in record.items() if key in ('name', 'login', 'server')}, 'path': str(instance_directory(mt5_folder, folder))} for folder, record in terminals.items() } # Список имён экземпляров self.instances_folders = list(terminals.keys()) # Словарь соответствия имя -> путь self.instances_paths = {folder: self.instance_path( folder) for folder in self.instances_folders} # Словарь соответствия путь -> имя self.paths_instances = {self.normalized_path(self.instance_path( folder)): folder for folder in self.instances_folders} # Глубокая копия: instances хранит pid/status, не должен делить # вложенные словари с конфигурацией self.instances = copy.deepcopy(self.terminal_config) def instance_path(self, folder: str) -> str: ''' Формирует полный путь по имени экземпляра терминала Args: folder (str): Имя экземпляра терминала. Returns: path (str): Полный путь к запускаемому файлу экземпляра терминала. ''' return str(instance_directory(self.mt5_folder, folder) / self.mt5_exe) def read_state(self, folder: str) -> sqlite3.Row | None: ''' Читает последнее известное состояние терминала из общей базы. Args: folder (str): Имя экземпляра терминала. Returns: row (sqlite3.Row | None): Строка состояния, или None, если терминал ещё не регистрировался в базе (сервис ни разу не запускался) либо база недоступна. ''' database = Path(self.database).resolve() try: # Создание файла и таблиц — задача MQL5-сервиса. Ошибка пути # менеджера не должна создавать вторую пустую базу. con = sqlite3.connect(database.as_uri() + '?mode=ro', uri=True, timeout=3) con.row_factory = sqlite3.Row try: return con.execute(self.STATE_QUERY, (folder,)).fetchone() finally: con.close() except sqlite3.OperationalError as e: # База ещё не создана сервисом, временно занята, либо у неё # неожиданно не совпадает схема со STATE_QUERY (например, таблица # уже существовала до того, как в неё добавили колонки, а # CREATE TABLE IF NOT EXISTS не выполняет миграцию) - в любом из # этих случаев продолжаем без данных, но печатаем причину, чтобы # рассинхронизацию схемы не пришлось второй раз откапывать вручную message = str(e) if 'no such table:' in message: hint = (' В этой базе ещё нет схемы сервиса. Проверьте его запуск, ' 'журнал и совпадение database с Common/Files/InpDatabase.') elif not database.is_file(): hint = (' Файл базы ещё не создан сервисом или неверно задан database. ' 'Менеджер не создаёт его при чтении.') else: hint = '' print(f'read_state({folder!r}): {message}; database={database}.{hint}') return None def instance_info(self, folder: str) -> dict: ''' Получает информацию о терминале по имени экземпляра терминала. Статус процесса определяется через psutil, данные счёта и терминала - из общей базы SQLite, которую пишет сервис ServiceStateWriter. Совмещение даёт три состояния вместо простого running/stopped: процесс жив и данные свежие - "running"; процесс жив, но данные устарели или их ещё нет (терминал грузится, сервис не запущен или завис) - "starting"; процесса нет - "stopped". Args: folder (str): Имя экземпляра терминала. Returns: info (dict): Информация о терминале или обнаруженных ошибках. ''' # Если терминала с таким именем нет в конфигурации if folder not in self.instances: return { 'last_error': {'code': -1, 'description': 'Manager: Terminal not found'}, 'status': 'not found', } info = self.instances[folder] # Если для терминала нет запущенного процесса if not info.get('pid'): info['last_error'] = {'code': 0, 'description': 'Stopped'} info['status'] = 'stopped' return info row = self.read_state(folder) if row is not None: expected = os.path.dirname(self.instance_path(folder)) if not row['terminal_path'] or self.normalized_path(row['terminal_path']) != self.normalized_path(expected): row = None fresh = ( row is not None and row['updated_at'] is not None and (time.time() - row['updated_at']) < self.FRESHNESS_THRESHOLD_SEC ) if row is not None: info['terminal'] = { 'path': row['terminal_path'], 'company': row['terminal_company'], 'build': row['terminal_build'], 'connected': bool(row['connected']), 'ping_last': row['ping_last'], } if not fresh: # Терминал грузится, сервис ещё не написал данные, либо завис info['last_error'] = {'code': 0, 'description': 'Starting'} info['status'] = 'starting' return info info['account'] = { 'login': row['login'], 'server': row['server'], 'currency': row['currency'], 'leverage': row['leverage'], 'trade_mode': row['trade_mode'], 'trade_allowed': bool(row['trade_allowed']), 'balance': row['balance'], 'equity': row['equity'], 'profit': row['profit'], 'margin': row['margin'], 'margin_free': row['margin_free'], 'margin_level': row['margin_level'], 'credit': row['credit'], 'positions': row['positions'], 'company': row['account_company'], 'holder_name': row['holder_name'], } # Секунды unix: сервис пишет TimeGMT(), миллисекундная точность не нужна info['last_update'] = row['updated_at'] info['last_error'] = {'code': row['error_code'], 'description': row['error_text']} info['status'] = 'running' 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 = self.normalized_path(proc.info['exe']) if proc.info['exe'] else '' # Если он есть среди путей экземпляров терминала if exe_path and exe_path in self.paths_instances: # Определяем имя экземпляра для данного пути folder = self.paths_instances[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() 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.instances_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'}}