483 lines
23 KiB
Python
483 lines
23 KiB
Python
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'<td[^>]*nowrap[^>]*>([^<]+):</td>\s*<td[^>]*colspan=.10.[^>]*><b>([^<]*)</b></td>'
|
|
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'<td[^>]*>([^<]+):</td>\s*<td[^>]*><b>([^<]*)</b></td>'
|
|
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() |