import os import csv import xml.etree.ElementTree as ET from datetime import datetime from typing import Dict, List, Optional from bs4 import BeautifulSoup import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class ResultParser: def __init__(self, results_dir: str = "results"): self.results_dir = results_dir def parse_xml_report(self, report_path: str) -> Optional[Dict]: if not os.path.exists(report_path): logger.warning(f"Report not found: {report_path}") return None try: tree = ET.parse(report_path) root = tree.getroot() result = { "report_file": os.path.basename(report_path), "parse_time": datetime.now().isoformat(), "test_info": {}, "metrics": {}, "trades": [] } tester = root.find("Tester") if tester is not None: result["test_info"] = { "expert": tester.findtext("Expert", ""), "symbol": tester.findtext("Symbol", ""), "period": tester.findtext("Period", ""), "model": tester.findtext("Model", ""), "from_date": tester.findtext("FromDate", ""), "to_date": tester.findtext("ToDate", ""), } equity = root.find("Equity") if equity is not None: result["metrics"]["initial_deposit"] = float(equity.findtext("Initial", "0")) result["metrics"]["final_balance"] = float(equity.findtext("Final", "0")) result["metrics"]["gross_profit"] = float(equity.findtext("GrossProfit", "0")) result["metrics"]["gross_loss"] = float(equity.findtext("GrossLoss", "0")) result["metrics"]["profit_factor"] = float(equity.findtext("ProfitFactor", "0")) result["metrics"]["expected_payoff"] = float(equity.findtext("ExpectedPayoff", "0")) trades_elem = root.find("Trades") if trades_elem is not None: result["metrics"]["total_trades"] = int(trades_elem.findtext("Total", "0")) result["metrics"]["short_positions"] = int(trades_elem.findtext("Short", "0")) result["metrics"]["long_positions"] = int(trades_elem.findtext("Long", "0")) result["metrics"]["winning_trades"] = int(trades_elem.findtext("ProfitTrades", "0")) result["metrics"]["losing_trades"] = int(trades_elem.findtext("LossTrades", "0")) if result["metrics"]["total_trades"] > 0: result["metrics"]["win_rate"] = ( result["metrics"]["winning_trades"] / result["metrics"]["total_trades"] * 100 ) else: result["metrics"]["win_rate"] = 0.0 return result except Exception as e: logger.error(f"Error parsing XML report {report_path}: {e}") return None def _detect_encoding(self, report_path: str) -> str: with open(report_path, 'rb') as f: bom = f.read(4) if bom[:2] == b'\xff\xfe': return 'utf-16-le' elif bom[:2] == b'\xfe\xff': return 'utf-16-be' return 'utf-8' def parse_html_report(self, report_path: str) -> Optional[Dict]: if not os.path.exists(report_path): logger.warning(f"Report not found: {report_path}") return None try: import re 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') result = { "report_file": os.path.basename(report_path), "parse_time": datetime.now().isoformat(), "test_info": {}, "metrics": {}, "trades": [] } plain_pattern = r']*nowrap[^>]*>([^<]+):\s*]*colspan=.10.[^>]*>([^<]*)' plain_matches = re.findall(plain_pattern, content) for key, value in plain_matches: key = key.strip() value = value.strip() if key in result["test_info"] and result["test_info"].get(key): continue self._parse_html_cell(key, value, result) pattern = r']*>([^<]+):\s*]*>([^<]*)' matches = re.findall(pattern, content) for key, value in matches: key = key.strip() value = value.strip() if key in result["test_info"] and result["test_info"].get(key): continue if key in result["metrics"] and result["metrics"].get(key): continue self._parse_html_cell(key, value, result) return result except Exception as e: logger.error(f"Error parsing HTML report {report_path}: {e}") return None try: with open(report_path, 'r', encoding='utf-16') as f: content = f.read() content_utf8 = content.encode('utf-8').decode('utf-8') soup = BeautifulSoup(content_utf8, 'html.parser') result = { "report_file": os.path.basename(report_path), "parse_time": datetime.now().isoformat(), "test_info": {}, "metrics": {}, "trades": [] } tables = soup.find_all('table') for table in tables: rows = table.find_all('tr') for row in rows: cells = row.find_all(['td', 'th']) if len(cells) >= 2: non_empty = [c for c in cells if c.get_text(strip=True)] if len(non_empty) >= 2: key = non_empty[0].get_text(strip=True) value = non_empty[1].get_text(strip=True) self._parse_html_cell(key, value, result) return result except Exception as e: logger.error(f"Error parsing HTML report {report_path}: {e}") return None def _extract_number(self, text: str) -> float: import re 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: import re numbers = re.findall(r'\d+\.?\d*%', text) if numbers: try: return float(numbers[0].replace('%', '')) except: return 0.0 return 0.0 def _parse_html_cell(self, key: str, value: str, result: Dict): import re key_lower = key.lower() if '专家' in key or 'Expert' in key: if not result["test_info"].get("expert"): result["test_info"]["expert"] = value elif '交易品种' in key or '交易品' in key or 'Symbol' in key or 'symbol' in key_lower: current = result["test_info"].get("symbol") if (not current or current in ('1', '0', '')) and value and value not in ('1', ''): result["test_info"]["symbol"] = value if not result["test_info"].get("symbol") or result["test_info"].get("symbol") == "0": fn = result["report_file"] m = re.search(r'_([A-Z]{5,6})_', fn) if m: result["test_info"]["symbol"] = m.group(1) elif '期间' in key or 'Period' in key: if not result["test_info"].get("period"): result["test_info"]["period"] = value elif '模型' in key or 'Model' in key: result["test_info"]["model"] = value elif '公司' in key or 'Company' in key: result["test_info"]["company"] = value elif '货币' in key or 'Currency' in key: result["test_info"]["currency"] = value elif '杠杆' in key or 'Leverage' in key: result["test_info"]["leverage"] = value elif '初始入金' in key or ('Initial' in key and 'Deposit' in key): result["metrics"]["initial_deposit"] = self._extract_number(value) elif '总净盈利' in key or 'Net Profit' in key: result["metrics"]["net_profit"] = self._extract_number(value) elif '毛利' in key or 'Gross Profit' in key: result["metrics"]["gross_profit"] = self._extract_number(value) elif '毛损' in key or 'Gross Loss' in key: result["metrics"]["gross_loss"] = self._extract_number(value) elif '盈利因子' in key or 'Profit Factor' in key: result["metrics"]["profit_factor"] = self._extract_number(value) elif '预期收益' in key or 'Expected Payoff' in key: result["metrics"]["expected_payoff"] = self._extract_number(value) elif ('总' in key and '交易' in key.lower()) or ('Total' in key and 'trades' in key_lower): result["metrics"]["total_trades"] = self._extract_number(value) elif '采收率' in key or 'Recovery Factor' in key: result["metrics"]["recovery_factor"] = self._extract_number(value) elif '夏普比率' in key or 'Sharpe Ratio' in key: result["metrics"]["sharpe_ratio"] = self._extract_number(value) elif 'AHPR' in key: result["metrics"]["ahpr"] = value elif 'GHPR' in key: result["metrics"]["ghpr"] = value elif 'LR 相关性' in key or 'LR Correlation' in key: result["metrics"]["lr_correlation"] = self._extract_number(value) elif 'LR 标准误差' in key or 'LR Standard Error' in key: result["metrics"]["lr_standard_error"] = self._extract_number(value) elif '最大结余亏损' in key or 'Maximal Drawdown' in key: result["metrics"]["max_drawdown"] = self._extract_number(value) elif '最大净值亏损' in key or 'Max Equity Drawdown' in key: result["metrics"]["max_equity_drawdown"] = self._extract_number(value) elif '绝对结余亏损' in key or 'Absolute Drawdown' in key: result["metrics"]["absolute_drawdown"] = self._extract_number(value) elif '预付款维持率' in key or 'Margin Level' in key: result["metrics"]["margin_level"] = self._extract_number(value) elif '卖出交易' in key or 'Short Positions' in key: result["metrics"]["short_trades"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif '买入交易' in key or 'Long Positions' in key: result["metrics"]["long_trades"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif '总成交' in key or 'Total Deals' in key: result["metrics"]["total_deals"] = self._extract_number(value) elif '盈利交易' in key or 'Profit Trades' in key: result["metrics"]["winning_trades"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) result["metrics"]["win_rate"] = self._extract_percentage(value) elif '亏损交易' in key or 'Loss Trades' in key: result["metrics"]["losing_trades"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif '最大 获利交易' in key or 'Max Profit Trade' in key: result["metrics"]["max_profit_trade"] = self._extract_number(value) elif '最大 亏损交易' in key or 'Max Loss Trade' in key: result["metrics"]["max_loss_trade"] = self._extract_number(value) elif '平均 获利交易' in key or 'Avg Profit Trade' in key: result["metrics"]["avg_profit_trade"] = self._extract_number(value) elif '平均 亏损交易' in key or 'Avg Loss Trade' in key: result["metrics"]["avg_loss_trade"] = self._extract_number(value) elif '最大值 连胜' in key or 'Longest Winning Streak' in key: result["metrics"]["longest_win_streak"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif '最大值 连败' in key or 'Longest Losing Streak' in key: result["metrics"]["longest_lose_streak"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif '平均 连胜' in key or 'Average Winning Streak' in key: result["metrics"]["avg_win_streak"] = self._extract_number(value) elif '平均 连败' in key or 'Average Losing Streak' in key: result["metrics"]["avg_lose_streak"] = self._extract_number(value) elif '最小持仓时间' in key or 'Min Hold Time' in key: result["metrics"]["min_hold_time"] = value elif '最大持仓时间' in key or 'Max Hold Time' in key: result["metrics"]["max_hold_time"] = value elif '平均持仓时间' in key or 'Avg Hold Time' in key: result["metrics"]["avg_hold_time"] = value elif '质量历史' in key or 'Quality' in key: result["metrics"]["quality"] = self._extract_number(value) elif '柱' in key or 'Bars' in key: result["metrics"]["bars"] = self._extract_number(value) elif '报价' in key or 'Quotes' in key: result["metrics"]["quotes"] = self._extract_number(value) elif '分值' in key or 'Score' in key: result["metrics"]["score"] = self._extract_number(value.split('(')[0]) if '(' in value else self._extract_number(value) elif 'OnTester结果' in key or 'OnTester' in key: result["metrics"]["on_tester"] = self._extract_number(value) def parse_all_reports(self, pattern: str = "*.xml") -> List[Dict]: import glob pattern_base = pattern.replace('*', '') if pattern_base == '.xml': report_files = glob.glob(os.path.join(self.results_dir, '*.xml')) report_files.extend(glob.glob(os.path.join(self.results_dir, '*.htm'))) report_files.extend(glob.glob(os.path.join(self.results_dir, '*.html'))) else: report_files = glob.glob(os.path.join(self.results_dir, pattern)) results = [] for report_file in report_files: if report_file.endswith('.xml'): parsed = self.parse_xml_report(report_file) elif report_file.endswith(('.html', '.htm')): parsed = self.parse_html_report(report_file) else: continue if parsed: parsed["file_path"] = os.path.abspath(report_file) parsed.setdefault("report_file", os.path.basename(report_file)) results.append(parsed) logger.info(f"Parsed: {os.path.basename(report_file)}") else: logger.warning(f"Failed to parse: {report_file}") logger.info(f"Total reports parsed: {len(results)}") return results def get_summary(self, results: List[Dict]) -> Dict: if not results: return {} summary = { "total_tests": len(results), "total_trades": 0, "avg_win_rate": 0, "avg_profit_factor": 0, "best_test": None, "worst_test": None } total_trades = sum(r["metrics"].get("total_trades", 0) for r in results) win_rates = [r["metrics"].get("win_rate", 0) for r in results if "win_rate" in r["metrics"]] profit_factors = [r["metrics"].get("profit_factor", 0) for r in results if "profit_factor" in r["metrics"]] summary["total_trades"] = total_trades if win_rates: summary["avg_win_rate"] = sum(win_rates) / len(win_rates) if profit_factors: summary["avg_profit_factor"] = sum(profit_factors) / len(profit_factors) completed = [r for r in results if r.get("test_info", {}).get("expert")] if completed: summary["best_test"] = max(completed, key=lambda x: x["metrics"].get("profit_factor", 0)) summary["worst_test"] = min(completed, key=lambda x: x["metrics"].get("profit_factor", 0)) return summary def export_to_csv(self, results: List[Dict], output_path: str, sort_by: str = "profit_factor", reverse: bool = True): if not results: return fieldnames = [ "expert", "symbol", "period", "company", "currency", "leverage", "initial_deposit", "net_profit", "gross_profit", "gross_loss", "profit_factor", "expected_payoff", "recovery_factor", "sharpe_ratio", "ahpr", "ghpr", "lr_correlation", "lr_standard_error", "max_drawdown", "max_equity_drawdown", "absolute_drawdown", "margin_level", "total_trades", "total_deals", "short_trades", "long_trades", "winning_trades", "losing_trades", "win_rate", "max_profit_trade", "max_loss_trade", "avg_profit_trade", "avg_loss_trade", "longest_win_streak", "longest_lose_streak", "avg_win_streak", "avg_lose_streak", "min_hold_time", "max_hold_time", "avg_hold_time", "quality", "bars", "quotes", "score", "on_tester" ] rows = [] for r in results: info = r.get("test_info", {}) metrics = r.get("metrics", {}) row = { "expert": info.get("expert", ""), "symbol": info.get("symbol", ""), "period": info.get("period", ""), "company": info.get("company", ""), "currency": info.get("currency", ""), "leverage": info.get("leverage", ""), "initial_deposit": metrics.get("initial_deposit", ""), "net_profit": metrics.get("net_profit", ""), "gross_profit": metrics.get("gross_profit", ""), "gross_loss": metrics.get("gross_loss", ""), "profit_factor": metrics.get("profit_factor", ""), "expected_payoff": metrics.get("expected_payoff", ""), "recovery_factor": metrics.get("recovery_factor", ""), "sharpe_ratio": metrics.get("sharpe_ratio", ""), "ahpr": metrics.get("ahpr", ""), "ghpr": metrics.get("ghpr", ""), "lr_correlation": metrics.get("lr_correlation", ""), "lr_standard_error": metrics.get("lr_standard_error", ""), "max_drawdown": metrics.get("max_drawdown", ""), "max_equity_drawdown": metrics.get("max_equity_drawdown", ""), "absolute_drawdown": metrics.get("absolute_drawdown", ""), "margin_level": metrics.get("margin_level", ""), "total_trades": metrics.get("total_trades", ""), "total_deals": metrics.get("total_deals", ""), "short_trades": metrics.get("short_trades", ""), "long_trades": metrics.get("long_trades", ""), "winning_trades": metrics.get("winning_trades", ""), "losing_trades": metrics.get("losing_trades", ""), "win_rate": metrics.get("win_rate", ""), "max_profit_trade": metrics.get("max_profit_trade", ""), "max_loss_trade": metrics.get("max_loss_trade", ""), "avg_profit_trade": metrics.get("avg_profit_trade", ""), "avg_loss_trade": metrics.get("avg_loss_trade", ""), "longest_win_streak": metrics.get("longest_win_streak", ""), "longest_lose_streak": metrics.get("longest_lose_streak", ""), "avg_win_streak": metrics.get("avg_win_streak", ""), "avg_lose_streak": metrics.get("avg_lose_streak", ""), "min_hold_time": metrics.get("min_hold_time", ""), "max_hold_time": metrics.get("max_hold_time", ""), "avg_hold_time": metrics.get("avg_hold_time", ""), "quality": metrics.get("quality", ""), "bars": metrics.get("bars", ""), "quotes": metrics.get("quotes", ""), "score": metrics.get("score", ""), "on_tester": metrics.get("on_tester", ""), } rows.append(row) rows.sort(key=lambda x: x.get(sort_by, ""), reverse=reverse) os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else '.', exist_ok=True) with open(output_path, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) logger.info(f"CSV exported to: {output_path}") def main(): import argparse parser = argparse.ArgumentParser(description="MT5 Report Parser") parser.add_argument("--results-dir", "-r", default="reports", help="Directory containing report files") parser.add_argument("--output", "-o", default="reports/parsed_results.json", help="Output file for parsed results") parser.add_argument("--csv", "-c", default=None, help="CSV output path (e.g. reports/results.csv)") parser.add_argument("--sort", "-s", default="profit_factor", choices=["expert", "symbol", "period", "net_profit", "gross_profit", "profit_factor", "total_trades", "win_rate", "max_drawdown"], help="Field to sort by") parser.add_argument("--asc", action="store_true", help="Sort in ascending order (default: descending)") args = parser.parse_args() parser = ResultParser(args.results_dir) results = parser.parse_all_reports() if results: summary = parser.get_summary(results) print("\n" + "="*60) print("PARSED RESULTS SUMMARY") print("="*60) print(f"Total tests parsed: {summary['total_tests']}") print(f"Total trades: {summary['total_trades']}") print(f"Average win rate: {summary['avg_win_rate']:.2f}%") print(f"Average profit factor: {summary['avg_profit_factor']:.2f}") if summary['best_test']: print(f"\nBest test: {summary['best_test']['test_info'].get('expert', 'N/A')}") print(f" Profit factor: {summary['best_test']['metrics'].get('profit_factor', 0):.2f}") print(f" Win rate: {summary['best_test']['metrics'].get('win_rate', 0):.2f}%") os.makedirs(os.path.dirname(args.output) if os.path.dirname(args.output) else '.', exist_ok=True) import json with open(args.output, 'w', encoding='utf-8') as f: json.dump({"results": results, "summary": summary}, f, indent=2, ensure_ascii=False) print(f"\nResults saved to: {args.output}") if args.csv: parser.export_to_csv(results, args.csv, sort_by=args.sort, reverse=not args.asc) else: print("No results to parse") if __name__ == "__main__": main()