""" Advanced MT5 to Laravel Sync Bridge with retry logic, error recovery, and multi-account support """ import MetaTrader5 as mt5 import requests import json import time import logging from datetime import datetime, timedelta from typing import Dict, List, Optional from dataclasses import dataclass import os from dotenv import load_dotenv # Load environment variables load_dotenv() # Configuration LARAVEL_API_URL = os.getenv('LARAVEL_API_URL', 'https://your-laravel-app.com/api/v1/mt5/sync') API_KEY = os.getenv('API_KEY', 'your-secure-api-key-here') SYNC_INTERVAL_SECONDS = int(os.getenv('SYNC_INTERVAL_SECONDS', '300')) MAX_RETRIES = int(os.getenv('MAX_RETRIES', '3')) RETRY_DELAY_SECONDS = int(os.getenv('RETRY_DELAY_SECONDS', '30')) # Multi-account support MT5_ACCOUNTS = [ { 'login': int(os.getenv('MT5_LOGIN_1', '12345678')), 'password': os.getenv('MT5_PASSWORD_1', 'password1'), 'server': os.getenv('MT5_SERVER_1', 'server1.com') }, # Add more accounts as needed ] # Logging setup logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('mt5_sync_advanced.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) @dataclass class SyncResult: success: bool account_number: str trades_synced: int error_message: Optional[str] = None sync_time: Optional[datetime] = None class MT5Account: def __init__(self, login: int, password: str, server: str): self.login = login self.password = password self.server = server self.connected = False def connect(self) -> bool: """Connect to MT5 terminal for this account""" try: if not mt5.initialize(): logger.error(f"MT5 initialize failed for account {self.login}: {mt5.last_error()}") return False if not mt5.login(self.login, self.password, self.server): logger.error(f"MT5 login failed for account {self.login}: {mt5.last_error()}") mt5.shutdown() return False self.connected = True logger.info(f"Connected to MT5 account: {self.login}") return True except Exception as e: logger.error(f"Connection error for account {self.login}: {str(e)}") return False def disconnect(self): """Disconnect from MT5 terminal""" if self.connected: mt5.shutdown() self.connected = False logger.info(f"Disconnected from MT5 account: {self.login}") def get_account_info(self) -> Dict: """Fetch account information""" try: account_info = mt5.account_info() if account_info is None: logger.error(f"Failed to get account info for {self.login}") return None return { 'account_number': str(account_info.login), 'account_name': account_info.name, 'server': account_info.server, 'balance': float(account_info.balance), 'equity': float(account_info.equity), 'margin': float(account_info.margin), 'free_margin': float(account_info.margin_free), 'floating_profit': float(account_info.equity - account_info.balance), 'currency': account_info.currency, 'leverage': account_info.leverage } except Exception as e: logger.error(f"Error getting account info for {self.login}: {str(e)}") return None def get_open_trades(self) -> List[Dict]: """Fetch all open trades""" try: positions = mt5.positions_get() if positions is None: logger.error(f"Failed to get open trades for {self.login}") return [] trades = [] for position in positions: trades.append({ 'ticket': str(position.ticket), 'symbol': position.symbol, 'type': 'BUY' if position.type == 0 else 'SELL', 'volume': float(position.volume), 'open_price': float(position.price_open), 'current_price': float(position.price_current), 'sl': float(position.sl) if position.sl > 0 else None, 'tp': float(position.tp) if position.tp > 0 else None, 'profit': float(position.profit), 'swap': float(position.swap), 'commission': float(position.commission), 'open_time': datetime.fromtimestamp(position.time).isoformat(), 'close_time': None, 'status': 'OPEN', 'comment': position.comment }) logger.info(f"Account {self.login}: Fetched {len(trades)} open trades") return trades except Exception as e: logger.error(f"Error getting open trades for {self.login}: {str(e)}") return [] class MT5SyncBridge: def __init__(self): self.accounts = [] self.last_sync_time = None # Initialize accounts for account_config in MT5_ACCOUNTS: self.accounts.append(MT5Account( account_config['login'], account_config['password'], account_config['server'] )) def sync_account(self, account: MT5Account, retry_count: int = 0) -> SyncResult: """Sync a single MT5 account to Laravel API with retry logic""" try: if not account.connect(): return SyncResult( success=False, account_number=str(account.login), trades_synced=0, error_message="Failed to connect to MT5" ) # Get account info account_info = account.get_account_info() if not account_info: account.disconnect() return SyncResult( success=False, account_number=str(account.login), trades_synced=0, error_message="Failed to get account info" ) # Get open trades open_trades = account.get_open_trades() # Prepare sync payload payload = { 'api_key': API_KEY, 'timestamp': int(time.time()), 'account': account_info, 'trades': open_trades } # Send to Laravel API response = requests.post( LARAVEL_API_URL, json=payload, headers={'Content-Type': 'application/json'}, timeout=30 ) account.disconnect() if response.status_code == 202: logger.info(f"Account {account.login}: Sync queued successfully") return SyncResult( success=True, account_number=str(account.login), trades_synced=len(open_trades), sync_time=datetime.now() ) else: logger.error(f"Account {account.login}: Sync failed: {response.status_code} - {response.text}") # Retry logic if retry_count < MAX_RETRIES: logger.info(f"Retrying sync for account {account.login} (attempt {retry_count + 1}/{MAX_RETRIES})") time.sleep(RETRY_DELAY_SECONDS) return self.sync_account(account, retry_count + 1) return SyncResult( success=False, account_number=str(account.login), trades_synced=0, error_message=f"API error: {response.status_code}" ) except requests.exceptions.Timeout: account.disconnect() logger.error(f"Account {account.login}: Sync request timed out") if retry_count < MAX_RETRIES: logger.info(f"Retrying sync for account {account.login} (attempt {retry_count + 1}/{MAX_RETRIES})") time.sleep(RETRY_DELAY_SECONDS) return self.sync_account(account, retry_count + 1) return SyncResult( success=False, account_number=str(account.login), trades_synced=0, error_message="Request timeout" ) except Exception as e: account.disconnect() logger.error(f"Account {account.login}: Sync error: {str(e)}") if retry_count < MAX_RETRIES: logger.info(f"Retrying sync for account {account.login} (attempt {retry_count + 1}/{MAX_RETRIES})") time.sleep(RETRY_DELAY_SECONDS) return self.sync_account(account, retry_count + 1) return SyncResult( success=False, account_number=str(account.login), trades_synced=0, error_message=str(e) ) def sync_all_accounts(self) -> List[SyncResult]: """Sync all configured MT5 accounts""" results = [] for account in self.accounts: logger.info(f"Starting sync for account {account.login}") result = self.sync_account(account) results.append(result) # Small delay between account syncs time.sleep(2) return results def run_sync_loop(self): """Run continuous sync loop for all accounts""" logger.info("Starting MT5 sync loop for all accounts") try: while True: try: logger.info("=" * 50) logger.info("Starting sync cycle") results = self.sync_all_accounts() # Log summary successful = sum(1 for r in results if r.success) failed = len(results) - successful total_trades = sum(r.trades_synced for r in results) logger.info(f"Sync cycle completed: {successful}/{len(results)} accounts successful, {total_trades} trades synced") if failed > 0: for result in results: if not result.success: logger.error(f"Failed: Account {result.account_number} - {result.error_message}") self.last_sync_time = datetime.now() logger.info(f"Next sync in {SYNC_INTERVAL_SECONDS}s") logger.info("=" * 50) time.sleep(SYNC_INTERVAL_SECONDS) except KeyboardInterrupt: logger.info("Received interrupt signal. Stopping...") break except Exception as e: logger.error(f"Error in sync loop: {str(e)}") time.sleep(60) # Wait 1 minute before retry finally: logger.info("Shutting down MT5 sync bridge") def main(): """Main entry point""" bridge = MT5SyncBridge() bridge.run_sync_loop() if __name__ == "__main__": main()