| import MetaTrader5 as mt5
|
| import json
|
| import os
|
| from datetime import datetime
|
| import pandas as pd
|
|
|
| class BacktestBridge:
|
| """Bridge to run MQL5 Expert Advisors and extract backtest results"""
|
|
|
| def __init__(self):
|
| self.mt5_path = None
|
|
|
| def init_mt5(self):
|
| """Initialize MT5 connection"""
|
| if not mt5.initialize():
|
| print(f"MT5 initialization failed: {mt5.last_error()}")
|
| return False
|
|
|
|
|
| terminal_info = mt5.terminal_info()
|
| self.mt5_path = terminal_info.path
|
| print(f"MT5 Path: {self.mt5_path}")
|
| return True
|
|
|
| def compile_ea(self, mq5_file_path):
|
| """
|
| Compile MQL5 Expert Advisor
|
| Note: This requires MetaEditor CLI or manual compilation
|
| """
|
|
|
| if not os.path.exists(mq5_file_path):
|
| return {'success': False, 'error': 'MQ5 file not found'}
|
|
|
|
|
|
|
| ex5_path = mq5_file_path.replace('.mq5', '.ex5')
|
|
|
| if not os.path.exists(ex5_path):
|
| return {
|
| 'success': False,
|
| 'error': 'EX5 file not found. Please compile in MetaEditor first.'
|
| }
|
|
|
| return {'success': True, 'ex5_path': ex5_path}
|
|
|
| def get_history_deals(self, from_date, to_date):
|
| """Get historical deals from MT5"""
|
| deals = mt5.history_deals_get(from_date, to_date)
|
|
|
| if deals is None:
|
| return []
|
|
|
| deals_list = []
|
| for deal in deals:
|
| deals_list.append({
|
| 'ticket': deal.ticket,
|
| 'order': deal.order,
|
| 'time': deal.time,
|
| 'type': 'BUY' if deal.type == 0 else 'SELL',
|
| 'entry': 'IN' if deal.entry == 0 else 'OUT',
|
| 'symbol': deal.symbol,
|
| 'volume': deal.volume,
|
| 'price': deal.price,
|
| 'profit': deal.profit,
|
| 'commission': deal.commission,
|
| 'swap': deal.swap,
|
| 'comment': deal.comment
|
| })
|
|
|
| return deals_list
|
|
|
| def get_history_orders(self, from_date, to_date):
|
| """Get historical orders from MT5"""
|
| orders = mt5.history_orders_get(from_date, to_date)
|
|
|
| if orders is None:
|
| return []
|
|
|
| orders_list = []
|
| for order in orders:
|
| orders_list.append({
|
| 'ticket': order.ticket,
|
| 'time_setup': order.time_setup,
|
| 'time_done': order.time_done,
|
| 'type': order.type,
|
| 'state': order.state,
|
| 'symbol': order.symbol,
|
| 'volume_initial': order.volume_initial,
|
| 'volume_current': order.volume_current,
|
| 'price_open': order.price_open,
|
| 'price_current': order.price_current,
|
| 'sl': order.sl,
|
| 'tp': order.tp,
|
| 'comment': order.comment
|
| })
|
|
|
| return orders_list
|
|
|
| def analyze_backtest_results(self, from_date, to_date):
|
| """Analyze backtest results and calculate statistics"""
|
| deals = self.get_history_deals(from_date, to_date)
|
|
|
| if not deals:
|
| return {'error': 'No deals found in the specified period'}
|
|
|
|
|
| df = pd.DataFrame(deals)
|
|
|
|
|
| total_trades = len(df[df['entry'] == 'OUT'])
|
| winning_trades = len(df[(df['entry'] == 'OUT') & (df['profit'] > 0)])
|
| losing_trades = len(df[(df['entry'] == 'OUT') & (df['profit'] < 0)])
|
|
|
| total_profit = df[df['entry'] == 'OUT']['profit'].sum()
|
| total_commission = df[df['entry'] == 'OUT']['commission'].sum()
|
| total_swap = df[df['entry'] == 'OUT']['swap'].sum()
|
|
|
| net_profit = total_profit + total_commission + total_swap
|
|
|
| win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0
|
|
|
|
|
| df_out = df[df['entry'] == 'OUT'].copy()
|
| df_out['cumulative_profit'] = df_out['profit'].cumsum()
|
| df_out['peak'] = df_out['cumulative_profit'].cummax()
|
| df_out['drawdown'] = df_out['peak'] - df_out['cumulative_profit']
|
| max_drawdown = df_out['drawdown'].max()
|
|
|
|
|
| gross_profit = df[(df['entry'] == 'OUT') & (df['profit'] > 0)]['profit'].sum()
|
| gross_loss = abs(df[(df['entry'] == 'OUT') & (df['profit'] < 0)]['profit'].sum())
|
| profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else 0
|
|
|
| return {
|
| 'total_trades': total_trades,
|
| 'winning_trades': winning_trades,
|
| 'losing_trades': losing_trades,
|
| 'win_rate': round(win_rate, 2),
|
| 'total_profit': round(total_profit, 2),
|
| 'total_commission': round(total_commission, 2),
|
| 'total_swap': round(total_swap, 2),
|
| 'net_profit': round(net_profit, 2),
|
| 'max_drawdown': round(max_drawdown, 2),
|
| 'profit_factor': round(profit_factor, 2),
|
| 'gross_profit': round(gross_profit, 2),
|
| 'gross_loss': round(gross_loss, 2),
|
| 'deals': deals
|
| }
|
|
|
| def export_backtest_report(self, from_date, to_date, output_file='backtest_report.json'):
|
| """Export backtest results to JSON file"""
|
| results = self.analyze_backtest_results(from_date, to_date)
|
|
|
| with open(output_file, 'w') as f:
|
| json.dump(results, f, indent=2, default=str)
|
|
|
| print(f"Backtest report exported to {output_file}")
|
| return results
|
|
|
|
|
| if __name__ == "__main__":
|
| bridge = BacktestBridge()
|
|
|
| if bridge.init_mt5():
|
|
|
| from_date = datetime(2024, 1, 1)
|
| to_date = datetime.now()
|
|
|
|
|
| results = bridge.analyze_backtest_results(from_date, to_date)
|
|
|
| print("\n=== BACKTEST RESULTS ===")
|
| print(f"Total Trades: {results.get('total_trades', 0)}")
|
| print(f"Win Rate: {results.get('win_rate', 0)}%")
|
| print(f"Net Profit: ${results.get('net_profit', 0)}")
|
| print(f"Max Drawdown: ${results.get('max_drawdown', 0)}")
|
| print(f"Profit Factor: {results.get('profit_factor', 0)}")
|
|
|
|
|
| bridge.export_backtest_report(from_date, to_date)
|
|
|
| mt5.shutdown() |