import os import re import json import glob import shutil from typing import List, Dict, Any, Optional from bs4 import BeautifulSoup import xml.etree.ElementTree as ET class ResultAnalyzer: def __init__(self, reports_dir: str = 'reports'): self.reports_dir = reports_dir def parse_mt5_html_report(self, report_path: str) -> Optional[Dict]: if not os.path.exists(report_path): return None try: with open(report_path, 'rb') as f: raw_bytes = f.read() if raw_bytes[:2] == b'\xff\xfe': content = raw_bytes[2:].decode('utf-16-le') else: content = raw_bytes.decode('utf-8', errors='ignore') content = content.encode('utf-8').decode('utf-8') soup = BeautifulSoup(content, 'html.parser') result = {} pattern = r']*>([^<]+):\s*]*>([^<]*)' matches = re.findall(pattern, content) plain_pattern = r']*nowrap[^>]*>([^<]+):\s*]*colspan=.10.[^>]*>([^<]*)' plain_matches = re.findall(plain_pattern, content) all_matches = dict(plain_matches) for k, v in matches: if k not in all_matches: all_matches[k] = v for key, value in all_matches.items(): key = key.strip() value = value.strip() if 'Net Profit' in key or '\u603b\u51c0\u76c8\u5229' in key: result['net_profit'] = self._extract_number(value) elif 'Gross Profit' in key or '\u6bdb\u5229' in key: result['gross_profit'] = self._extract_number(value) elif 'Gross Loss' in key or '\u6bdb\u635f' in key: result['gross_loss'] = self._extract_number(value) elif 'Profit Factor' in key or '\u76c8\u5229\u56e0\u5b50' in key: result['profit_factor'] = self._extract_number(value) elif 'Total Trades' in key or '\u4ea4\u6613\u603b\u8ba1' in key: result['total_trades'] = self._extract_number(value) elif 'Sharpe Ratio' in key or '\u590f\u666e\u6bd4\u7387' in key: result['sharpe_ratio'] = self._extract_number(value) elif 'Maximal Drawdown' in key or '\u6700\u5927\u7ed3\u4f59\u4e8f\u635f' in key: result['max_drawdown'] = self._extract_number(value) elif 'Win Rate' in key or '\u76c8\u5229\u4ea4\u6613' in key: result['win_rate'] = self._extract_percentage(value) return result if result else None except Exception as e: print('Error parsing ' + report_path + ': ' + str(e)) return None def _extract_number(self, text: str) -> float: numbers = re.findall(r'[-+]?\d*\.?\d+', text.replace(',', '').replace(' ', '')) if numbers: try: return float(numbers[0]) except: return 0.0 return 0.0 def _extract_percentage(self, text: str) -> float: numbers = re.findall(r'\d+\.?\d*%', text) if numbers: try: return float(numbers[0].replace('%', '')) except: return 0.0 return 0.0 def parse_mt5_xml_optimization(self, xml_path: str) -> List[Dict]: if not os.path.exists(xml_path): return [] results = [] try: tree = ET.parse(xml_path) root = tree.getroot() ns = {'ss': 'urn:schemas-microsoft-com:office:spreadsheet'} worksheet = root.find('.//ss:Worksheet', ns) if worksheet is None: return [] table = worksheet.find('ss:Table', ns) if table is None: return [] rows = table.findall('ss:Row', ns) if len(rows) < 2: return [] header_row = rows[0] headers = [] for cell in header_row.findall('ss:Cell', ns): data = cell.find('ss:Data', ns) if data is not None and data.text: headers.append(data.text.strip().lower()) for row_idx, row in enumerate(rows[1:], start=1): cells = row.findall('ss:Cell', ns) if not cells: continue result = {'pass': row_idx} for col_idx, cell in enumerate(cells): if col_idx >= len(headers): break header = headers[col_idx] data = cell.find('ss:Data', ns) value = data.text.strip() if data is not None and data.text else '' if 'pass' in header: result['pass'] = int(value) if value.isdigit() else row_idx elif 'result' in header: result['result'] = self._extract_number(value) elif 'profit' in header and 'expected' not in header and 'factor' not in header and 'drawdown' not in header: result['net_profit'] = self._extract_number(value) elif 'expected payoff' in header: result['expected_payoff'] = self._extract_number(value) elif 'profit factor' in header: if value: result['profit_factor'] = self._extract_number(value) elif 'recovery factor' in header: result['recovery_factor'] = self._extract_number(value) elif 'sharpe ratio' in header: result['sharpe_ratio'] = self._extract_number(value) elif 'custom' in header: result['custom'] = self._extract_number(value) elif 'equity dd' in header or 'drawdown' in header: result['max_drawdown'] = self._extract_number(value) elif 'trade' in header: result['total_trades'] = int(self._extract_number(value)) if result: results.append(result) except Exception as e: import traceback print('Error parsing MT5 XML ' + xml_path + ': ' + str(e)) traceback.print_exc() return results def copy_xml_results_to_reports(self, mt5_terminal_dir, ea_name: str, reports_dir: str) -> str: xml_patterns = [ f'{ea_name}_optimization.xml', f'{ea_name}_optimization[1].xml', f'{ea_name}_optimization[2].xml', ] os.makedirs(reports_dir, exist_ok=True) latest_xml = None latest_time = 0 search_dirs = mt5_terminal_dir if isinstance(mt5_terminal_dir, (list, tuple)) else [mt5_terminal_dir] for base_dir in search_dirs: if not base_dir or not os.path.isdir(base_dir): continue for pattern in xml_patterns: xml_path = os.path.join(base_dir, pattern) if os.path.exists(xml_path): mtime = os.path.getmtime(xml_path) if mtime > latest_time: latest_time = mtime latest_xml = xml_path if latest_xml: dest_path = os.path.join(reports_dir, os.path.basename(latest_xml)) shutil.copy2(latest_xml, dest_path) return dest_path return None def parse_all_reports(self, pattern: str = '*.htm*') -> List[Dict]: results = [] search_path = os.path.join(self.reports_dir, pattern) report_files = glob.glob(search_path) for report_file in report_files: parsed = self.parse_mt5_html_report(report_file) if parsed: parsed['report_file'] = os.path.basename(report_file) results.append(parsed) return results def merge_parameters_and_results(self, param_mapping: List[Dict], results: List[Dict]) -> List[Dict]: merged = [] result_map = {r.get('report_file', ''): r for r in results} for param_set in param_mapping: report_name = param_set.get('report_name', '') if report_name in result_map: combined = {**param_set, **result_map[report_name]} merged.append(combined) return merged def find_optimal_params(self, results: List[Dict], criterion: str = 'profit_factor', min_trades: int = 10, max_drawdown_pct: float = 50.0) -> List[Dict]: filtered = [] for r in results: trades = r.get('total_trades', 0) dd = r.get('max_drawdown', 0) pf = r.get('profit_factor', 0) if trades >= min_trades and dd <= max_drawdown_pct and pf > 0: filtered.append(r) filtered.sort(key=lambda x: x.get(criterion, 0), reverse=True) return filtered def generate_report(self, results: List[Dict], output_path: str = 'results/optimization_report.txt'): if not results: return os.makedirs(os.path.dirname(output_path), exist_ok=True) lines = [ '=' * 60, 'MT5 EA Optimization Report', '=' * 60, '', 'Total Results: ' + str(len(results)), '', ] if results: best = results[0] lines.extend([ 'Best Configuration:', ' Profit Factor: ' + str(best.get('profit_factor', 0)), ' Net Profit: ' + str(best.get('net_profit', 0)), ' Total Trades: ' + str(best.get('total_trades', 0)), ' Max Drawdown: ' + str(best.get('max_drawdown', 0)), ' Win Rate: ' + str(best.get('win_rate', 0)) + '%', '', ]) lines.append('Top 10 Configurations:') lines.append('-' * 60) for i, r in enumerate(results[:10], 1): lines.append( str(i) + '. PF=' + str(r.get('profit_factor', 0)) + ' Net=' + str(r.get('net_profit', 0)) + ' Trades=' + str(r.get('total_trades', 0)) ) with open(output_path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines)) return output_path def analyze_walk_forward(self, mt5_terminal_dir, ea_name: str, reports_dir: str) -> Dict: result = { 'ea_name': ea_name, 'in_sample': {}, 'out_of_sample': {}, 'overfitting_score': 0, 'verdict': 'Unknown' } xml_path = self.copy_xml_results_to_reports(mt5_terminal_dir, ea_name, reports_dir) if xml_path: is_results = self.parse_mt5_xml_optimization(xml_path) if is_results: best_is = sorted(is_results, key=lambda x: x.get('profit_factor', 0), reverse=True)[0] result['in_sample'] = best_is search_dirs = mt5_terminal_dir if isinstance(mt5_terminal_dir, (list, tuple)) else [mt5_terminal_dir] forward_xml = None for base_dir in search_dirs: if not base_dir or not os.path.isdir(base_dir): continue for candidate in (f'{ea_name}_optimization.forward.xml', f'{ea_name}_optimization[1].forward.xml'): p = os.path.join(base_dir, candidate) if os.path.exists(p): forward_xml = p break if forward_xml: break if forward_xml and os.path.exists(forward_xml): oos_results = self.parse_mt5_xml_optimization(forward_xml) if oos_results: best_oos = sorted(oos_results, key=lambda x: x.get('profit_factor', 0), reverse=True)[0] result['out_of_sample'] = best_oos is_pf = result.get('in_sample', {}).get('profit_factor', 0) oos_pf = result.get('out_of_sample', {}).get('profit_factor', 0) if is_pf > 0 and oos_pf > 0: decay = (is_pf - oos_pf) / is_pf result['pf_decay'] = round(decay * 100, 1) if oos_pf >= is_pf * 0.7 and oos_pf > 1.0: result['verdict'] = 'Robust (\u7a33\u5065)' elif oos_pf > 1.0: result['verdict'] = 'Mild Overfit (\u8f7b\u5ea6\u8fc7\u62df\u5408)' else: result['verdict'] = 'Overfit (\u4e25\u91cd\u8fc7\u62df\u5408)' elif oos_pf > 0: result['pf_decay'] = 0 result['verdict'] = 'Valid (\u6709\u6548\u4f46IS\u65e0\u7ed3\u679c)' is_dd = result.get('in_sample', {}).get('max_drawdown', 0) oos_dd = result.get('out_of_sample', {}).get('max_drawdown', 0) if is_dd > 0 and oos_dd > 0: result['dd_increase'] = round((oos_dd / is_dd - 1) * 100, 1) if is_dd > 0 else 0 return result