335 lines
13 KiB
Python
335 lines
13 KiB
Python
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'<td[^>]*>([^<]+):</td>\s*<td[^>]*><b>([^<]*)</b></td>'
|
|
matches = re.findall(pattern, content)
|
|
|
|
plain_pattern = r'<td[^>]*nowrap[^>]*>([^<]+):</td>\s*<td[^>]*colspan=.10.[^>]*><b>([^<]*)</b></td>'
|
|
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
|