gavindiaz/mt5-quant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Complete migration to Rust implementation

Browse Source 
- Migrate optimization from optimize.sh to src/optimization/
- Remove all Python files (analytics/, server/, hooks/, pyproject.toml)
- Add optimization module: optimizer.rs, parser.rs, mod.rs
- Implement all missing MCP tool handlers (35 total tools)
- Add handle_patch_set_file handler
- Clean up orphan files: .venv/, __pycache__, test files
- Move test_rcp_server.sh to tests/integration_test.sh
- Add Rust integration tests in tests/integration_tests.rs
- Fix all compiler warnings with #[allow(dead_code)]
- Update test fixtures and structure
This commit is contained in:
Devid HW
2026-04-18 15:57:28 +07:00
parent a3b046c68f
commit 331b7fbb73
29 changed files with 1405 additions and 5401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+10
View File
@@ -430,6 +430,7 @@ dependencies = [
"dirs", "dirs",
"encoding_rs", "encoding_rs",
"regex", "regex",
"roxmltree",
"serde", "serde",
"serde_json", "serde_json",
"serde_yaml", "serde_yaml",
@@ -589,6 +590,15 @@ version = "0.8.10"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
[[package]]
name = "roxmltree"
version = "0.21.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1964b10c76125c36f8afe190065a4bf9a87bf324842c05701330bba9f1cacbb"
dependencies = [
"memchr",
]
[[package]] [[package]]
name = "rustix" name = "rustix"
version = "1.1.4" version = "1.1.4"
Cargo.toml
+1
View File
@@ -25,3 +25,4 @@ walkdir = "2.0"
chrono = { version = "0.4", features = ["serde"] } chrono = { version = "0.4", features = ["serde"] }
encoding_rs = "0.8" encoding_rs = "0.8"
tempfile = "3.0" tempfile = "3.0"
roxmltree = "0.21.1"
analytics/__init__.py
View File
analytics/analyze.py
-1000
View File
File diff suppressed because it is too large Load Diff
analytics/extract.py
-316
View File
@@ -1,316 +0,0 @@
#!/usr/bin/env python3
"""
extract.py — Single-pass MT5 report parser.
Reads MT5 backtest report (.htm or .htm.xml / SpreadsheetML) and produces:
- metrics.json (aggregate summary)
- deals.csv (all deals, 13 columns)
- deals.json (deals as JSON array)
Usage:
python3 analytics/extract.py report.htm --output-dir reports/20250101_123456/
"""
import argparse
import csv
import json
import os
import re
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Optional
# MT5 backtest report deals table columns (actual order from HTML):
# Time, Deal, Symbol, Type, Direction, Volume, Price, Order, Commission, Swap, Profit, Balance, Comment
DEAL_COLUMNS = [
"time", "deal", "symbol", "type", "entry", "volume", "price",
"order", "commission", "swap", "profit", "balance", "comment"
]
def detect_format(path: str) -> str:
"""Return 'xml' for SpreadsheetML, 'html' for legacy HTML report."""
if path.endswith('.xml') or path.endswith('.htm.xml'):
return 'xml'
# Peek at file header
with open(path, 'rb') as f:
header = f.read(512)
if b'<?xml' in header or b'Workbook' in header:
return 'xml'
return 'html'
def read_text(path: str) -> str:
"""Read file, handling UTF-16 (MT5 default) and latin-1 fallback."""
with open(path, 'rb') as f:
raw = f.read()
for encoding in ('utf-16', 'utf-8', 'latin-1'):
try:
return raw.decode(encoding)
except (UnicodeDecodeError, LookupError):
continue
return raw.decode('latin-1', errors='replace')
def strip_tags(html: str) -> str:
return re.sub(r'<[^>]+>', '', html).strip()
# ── HTML parser ───────────────────────────────────────────────────────────────
def parse_html(path: str) -> tuple[dict, list[dict]]:
text = read_text(path)
metrics = _parse_metrics_html(text)
deals = _parse_deals_html(text)
return metrics, deals
def _parse_metrics_html(text: str) -> dict:
"""Extract aggregate metrics from the summary table."""
m = {}
# MT5 report HTML format: MetricLabel:</td>\r\n<td nowrap><b>VALUE</b></td>
# Helper patterns — values always wrapped in <b>...</b>
_b = r'[^<]*</td>\s*<td[^>]*>\s*<b>([-\d\s.,]+)</b>' # plain number
_b_pct = r'[^<]*</td>\s*<td[^>]*>\s*<b>[^(]*\(([\d.,]+)%\)' # "abs (pct%)" — capture pct
patterns = {
'net_profit': r'Net\s+Profit' + _b,
'profit_factor': r'Profit\s+Factor' + _b,
'max_dd_pct': r'Equity\s+Drawdown\s+Maximal' + _b_pct,
'sharpe_ratio': r'Sharpe\s+Ratio' + _b,
'total_trades': r'Total\s+Trades' + _b,
'recovery_factor': r'Recovery\s+Factor' + _b,
'win_rate_pct': r'Profit\s+Trades\s+\(%' + _b_pct,
'gross_profit': r'Gross\s+Profit' + _b,
'gross_loss': r'Gross\s+Loss' + _b,
}
for key, pattern in patterns.items():
match = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
if match:
val = match.group(1).replace(' ', '').replace(',', '').strip()
try:
m[key] = float(val)
except ValueError:
pass
# Trades needs int
if 'total_trades' in m:
m['total_trades'] = int(m['total_trades'])
return m
def _parse_deals_html(text: str) -> list[dict]:
"""Extract deal rows from the deals table."""
# Find deals section (after "Deals" header)
deals_section = re.search(
r'<tr[^>]*>.*?Deal.*?Time.*?Type.*?Direction.*?Volume.*?</tr>(.*)',
text, re.DOTALL | re.IGNORECASE
)
if not deals_section:
return []
rows = re.findall(
r'<tr[^>]*>(.*?)</tr>',
deals_section.group(1),
re.DOTALL | re.IGNORECASE
)
deals = []
for row in rows:
cells = re.findall(r'<td[^>]*>(.*?)</td>', row, re.DOTALL | re.IGNORECASE)
cells = [strip_tags(c).replace(',', '') for c in cells]
if len(cells) < 3 or not cells[0]:
continue
# Skip balance/deposit/credit rows — 'balance' appears in the Type column (index 3)
# or sometimes in index 1; check first 5 cells
if any(c.strip().lower() in ('balance', 'credit') for c in cells[:5]):
continue
deal = {}
for i, col in enumerate(DEAL_COLUMNS):
deal[col] = cells[i] if i < len(cells) else ''
deals.append(deal)
return deals
# ── XML parser (SpreadsheetML) ─────────────────────────────────────────────────
def parse_xml(path: str) -> tuple[dict, list[dict]]:
"""Parse MT5 SpreadsheetML optimization/report XML."""
tree = ET.parse(path)
root = tree.getroot()
# Namespace handling — MT5 XML uses Excel namespace
ns = {}
ns_match = re.match(r'\{([^}]+)\}', root.tag)
if ns_match:
ns['ss'] = ns_match.group(1)
def tag(name):
return f"{{{ns['ss']}}}{name}" if ns else name
metrics = {}
deals = []
in_deals_sheet = False
for sheet in root.iter(tag('Worksheet')):
sheet_name = sheet.get(f"{{{ns['ss']}}}Name" if ns else 'Name', '')
if 'result' in sheet_name.lower() or 'report' in sheet_name.lower():
metrics = _parse_metrics_xml(sheet, tag)
elif 'deal' in sheet_name.lower() or 'trade' in sheet_name.lower():
deals = _parse_deals_xml(sheet, tag)
elif sheet_name == '':
# Unnamed sheet — check if it has deal-like structure
rows = list(sheet.iter(tag('Row')))
if len(rows) > 5:
# Try to parse as deals
candidate = _parse_deals_xml(sheet, tag)
if candidate:
deals = candidate
return metrics, deals
def _cell_value(cell, tag) -> str:
data = cell.find(tag('Data'))
return data.text.strip() if data is not None and data.text else ''
def _parse_metrics_xml(sheet, tag) -> dict:
m = {}
for row in sheet.iter(tag('Row')):
cells = [_cell_value(c, tag) for c in row.iter(tag('Cell'))]
if len(cells) < 2:
continue
key = cells[0].lower()
val = cells[1].replace(',', '').strip()
try:
fval = float(val)
if 'net profit' in key or 'net_profit' in key:
m['net_profit'] = fval
elif 'profit factor' in key:
m['profit_factor'] = fval
elif 'drawdown' in key and '%' in cells[1]:
m['max_dd_pct'] = fval
elif 'sharpe' in key:
m['sharpe_ratio'] = fval
elif 'total trades' in key:
m['total_trades'] = int(fval)
except (ValueError, AttributeError):
pass
return m
def _parse_deals_xml(sheet, tag) -> list[dict]:
deals = []
header_found = False
col_map = {}
for row in sheet.iter(tag('Row')):
cells = [_cell_value(c, tag) for c in row.iter(tag('Cell'))]
if not header_found:
# Detect header row
if any(h in str(cells).lower() for h in ('time', 'type', 'volume', 'profit')):
header_found = True
for i, h in enumerate(cells):
h_lower = h.lower().strip()
for col in DEAL_COLUMNS:
if col in h_lower or h_lower in col:
col_map[i] = col
break
continue
if not cells or not cells[0]:
continue
deal = {}
for i, val in enumerate(cells):
col = col_map.get(i)
if col:
deal[col] = val.replace(',', '')
if deal:
deals.append(deal)
return deals
# ── Writer ────────────────────────────────────────────────────────────────────
def write_outputs(metrics: dict, deals: list[dict], output_dir: str) -> dict:
os.makedirs(output_dir, exist_ok=True)
metrics_path = os.path.join(output_dir, 'metrics.json')
deals_csv_path = os.path.join(output_dir, 'deals.csv')
deals_json_path = os.path.join(output_dir, 'deals.json')
with open(metrics_path, 'w') as f:
json.dump(metrics, f, indent=2)
with open(deals_json_path, 'w') as f:
json.dump(deals, f, indent=2)
if deals:
all_keys = DEAL_COLUMNS
with open(deals_csv_path, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=all_keys, extrasaction='ignore')
writer.writeheader()
writer.writerows(deals)
else:
# Write empty CSV with headers
with open(deals_csv_path, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(DEAL_COLUMNS)
return {
'metrics': metrics_path,
'deals_csv': deals_csv_path,
'deals_json': deals_json_path,
}
# ── Main ──────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description='Extract MT5 backtest report')
parser.add_argument('report', help='Path to report.htm or report.htm.xml')
parser.add_argument('--output-dir', default='.', help='Output directory')
parser.add_argument('--stdout', action='store_true',
help='Print metrics JSON to stdout instead of writing files')
args = parser.parse_args()
fmt = detect_format(args.report)
if fmt == 'xml':
metrics, deals = parse_xml(args.report)
else:
metrics, deals = parse_html(args.report)
if not metrics:
print(f"WARNING: No aggregate metrics found in report", file=sys.stderr)
if not deals:
print(f"WARNING: No deals found in report (check date range and symbol)", file=sys.stderr)
if args.stdout:
json.dump({'metrics': metrics, 'deals_count': len(deals)}, sys.stdout, indent=2)
print()
return
paths = write_outputs(metrics, deals, args.output_dir)
print(f"Extracted: {len(deals)} deals, {len(metrics)} metrics")
for name, path in paths.items():
print(f" {name}: {path}")
if __name__ == '__main__':
main()
analytics/optimize_parser.py
-349
View File
@@ -1,349 +0,0 @@
#!/usr/bin/env python3
"""
optimize_parser.py — Parse MT5 genetic optimization results.
Handles both HTML (.htm) and SpreadsheetML XML (.htm.xml) formats.
Usage:
python3 analytics/optimize_parser.py --job opt_20250619_143022
python3 analytics/optimize_parser.py --file reports/opt_dir/optimization.htm
python3 analytics/optimize_parser.py --file report.htm.xml --top 30 --sort profit
"""
import argparse
import json
import os
import re
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
ROOT_DIR = Path(__file__).parent.parent
def find_report(job_id: str) -> str:
"""Locate optimization report from job metadata."""
jobs_dir = ROOT_DIR / '.mt5mcp_jobs'
meta_path = jobs_dir / f'{job_id}.json'
if not meta_path.exists():
raise FileNotFoundError(f"Job not found: {job_id}. Check .mt5mcp_jobs/")
with open(meta_path) as f:
meta = json.load(f)
wine_prefix = meta.get('wine_prefix', '')
base = os.path.join(wine_prefix, 'drive_c', 'mt5mcp_opt_report')
for ext in ('.htm', '.htm.xml', '.html'):
candidate = base + ext
if os.path.exists(candidate):
return candidate
raise FileNotFoundError(
f"Optimization report not found. Expected: {base}.htm or {base}.htm.xml\n"
f"Is MT5 optimization still running? Check log: {meta.get('log_file', '')}"
)
def detect_format(path: str) -> str:
if path.endswith('.xml') or path.endswith('.htm.xml'):
return 'xml'
with open(path, 'rb') as f:
header = f.read(512)
if b'<?xml' in header or b'Workbook' in header:
return 'xml'
return 'html'
def read_text(path: str) -> str:
with open(path, 'rb') as f:
raw = f.read()
for enc in ('utf-16', 'utf-8', 'latin-1'):
try:
return raw.decode(enc)
except (UnicodeDecodeError, LookupError):
continue
return raw.decode('latin-1', errors='replace')
# ── HTML parser ───────────────────────────────────────────────────────────────
def parse_html(path: str) -> list[dict]:
text = read_text(path)
rows = re.findall(r'<tr[^>]*>(.*?)</tr>', text, re.DOTALL | re.IGNORECASE)
results = []
headers = []
for row in rows:
cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.DOTALL | re.IGNORECASE)
cells = [re.sub(r'<[^>]+>', '', c).strip().replace(',', '') for c in cells]
if not cells:
continue
# Header row detection
if not headers and cells[0].lower() in ('pass', '#', 'result', 'run'):
headers = cells
continue
# Data row: first cell is pass number (digit)
if headers and cells[0].isdigit():
row_data = dict(zip(headers, cells))
results.append(row_data)
elif not headers and cells[0].isdigit() and len(cells) > 5:
# No header — use positional mapping (common MT5 layout)
results.append(_positional_row(cells))
return results
def _positional_row(cells: list[str]) -> dict:
"""Map cells by position for headerless optimization tables."""
# MT5 optimization table columns (typical order):
# Pass | Profit | Expected Payoff | Profit Factor | Recovery Factor | Sharpe | Custom | DD% | Trades | ...params
pos_names = ['pass', 'profit', 'expected_payoff', 'profit_factor',
'recovery_factor', 'sharpe_ratio', 'custom', 'max_dd_pct', 'total_trades']
row = {}
for i, name in enumerate(pos_names):
if i < len(cells):
row[name] = cells[i]
# Remaining are parameters
row['_params_raw'] = cells[len(pos_names):]
return row
# ── XML parser ────────────────────────────────────────────────────────────────
def parse_xml(path: str) -> list[dict]:
tree = ET.parse(path)
root = tree.getroot()
ns = {}
ns_match = re.match(r'\{([^}]+)\}', root.tag)
if ns_match:
ns['ss'] = ns_match.group(1)
def tag(name):
return f"{{{ns['ss']}}}{name}" if ns else name
def cell_val(cell):
data = cell.find(tag('Data'))
return data.text.strip() if data is not None and data.text else ''
results = []
headers = []
for sheet in root.iter(tag('Worksheet')):
for row in sheet.iter(tag('Row')):
cells = [cell_val(c) for c in row.iter(tag('Cell'))]
cells = [c.replace(',', '').strip() for c in cells]
if not cells:
continue
if not headers:
if any(h.lower() in ('pass', 'result', 'profit') for h in cells):
headers = cells
continue
if cells[0].isdigit():
if headers:
row_data = {}
for i, h in enumerate(headers):
row_data[h.lower().replace(' ', '_')] = cells[i] if i < len(cells) else ''
results.append(row_data)
else:
results.append(_positional_row(cells))
return results
# ── Normalizer ────────────────────────────────────────────────────────────────
def normalize(raw_results: list[dict]) -> list[dict]:
"""Convert raw parsed rows to typed dicts with consistent keys."""
normalized = []
for r in raw_results:
def fget(keys, default=0.0):
for k in keys:
for rk, rv in r.items():
if k in rk.lower().replace(' ', '_'):
try:
return float(rv)
except (ValueError, TypeError):
pass
return default
def iget(keys, default=0):
v = fget(keys, default)
return int(v)
# Extract known fields
entry = {
'pass': iget(['pass', '#']),
'net_profit': fget(['profit', 'net_profit']),
'profit_factor': fget(['profit_factor']),
'max_dd_pct': fget(['dd', 'drawdown']),
'total_trades': iget(['trades']),
'sharpe_ratio': fget(['sharpe']),
'recovery_factor': fget(['recovery']),
}
# Remaining keys are parameters
known_keys = {'pass', 'profit', 'net_profit', 'profit_factor', 'expected_payoff',
'dd', 'drawdown', 'max_dd_pct', 'trades', 'total_trades',
'sharpe', 'sharpe_ratio', 'recovery', 'recovery_factor',
'custom', '#', '_params_raw'}
params = {}
for k, v in r.items():
if not any(kw in k.lower() for kw in known_keys):
try:
params[k] = float(v)
except (ValueError, TypeError):
params[k] = v
entry['params'] = params
normalized.append(entry)
return normalized
# ── Convergence analysis ──────────────────────────────────────────────────────
def convergence_analysis(results: list[dict], top_n: int = 10) -> dict:
top = results[:top_n]
if not top:
return {}
all_param_keys = set()
for r in top:
all_param_keys.update(r.get('params', {}).keys())
strong = {} # Same value across all top-N
uncertain = [] # Varies
for key in all_param_keys:
values = set()
for r in top:
v = r.get('params', {}).get(key)
if v is not None:
values.add(v)
if len(values) == 1:
strong[key] = list(values)[0]
else:
uncertain.append(key)
return {
'top_n_agreement': strong,
'high_variance_params': uncertain,
}
# ── Display ───────────────────────────────────────────────────────────────────
def display_results(results: list[dict], top_n: int, dd_threshold: float, conv: dict):
print(f"\nTotal passes: {len(results)}")
print(f"Showing top {min(top_n, len(results))} by profit:\n")
print(f"{'Rank':<5} {'Profit':>10} {'PF':>6} {'DD%':>6} {'Sharpe':>7} {'Trades':>7} Params")
print("─" * 80)
for i, r in enumerate(results[:top_n], 1):
dd = r['max_dd_pct']
risk_flag = ' ⚠' if dd > dd_threshold else ''
params_str = ' '.join(f"{k}={v}" for k, v in list(r.get('params', {}).items())[:4])
print(
f"#{i:<4} ${r['net_profit']:>9,.2f} "
f"{r['profit_factor']:>5.2f} "
f"{dd:>5.2f}%"
f"{risk_flag} "
f"{r['sharpe_ratio']:>6.2f} "
f"{r['total_trades']:>7} "
f"{params_str}"
)
if conv:
print(f"\nConvergence (top-{min(top_n, len(results))} agreement):")
if conv.get('top_n_agreement'):
print(" Stable params:", ', '.join(f"{k}={v}" for k, v in conv['top_n_agreement'].items()))
if conv.get('high_variance_params'):
print(" Uncertain params:", ', '.join(conv['high_variance_params']))
# ── Main ──────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description='Parse MT5 optimization results')
parser.add_argument('--job', help='Job ID from optimize.sh output')
parser.add_argument('--file', help='Direct path to optimization.htm or .htm.xml')
parser.add_argument('--top', type=int, default=20, help='Show top N results')
parser.add_argument('--sort', choices=['profit', 'profit_factor', 'sharpe'],
default='profit', help='Sort metric')
parser.add_argument('--dd-threshold', type=float, default=20.0,
help='Flag DD above this % as high-risk')
parser.add_argument('--output', help='Save results as JSON')
args = parser.parse_args()
# Locate report
if args.file:
report_path = args.file
elif args.job:
try:
report_path = find_report(args.job)
except FileNotFoundError as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
else:
print("ERROR: Provide --job or --file", file=sys.stderr)
sys.exit(1)
if not os.path.exists(report_path):
print(f"ERROR: Report not found: {report_path}", file=sys.stderr)
sys.exit(1)
# Parse
fmt = detect_format(report_path)
if fmt == 'xml':
raw = parse_xml(report_path)
else:
raw = parse_html(report_path)
results = normalize(raw)
if not results:
print("ERROR: No optimization passes found in report.", file=sys.stderr)
sys.exit(1)
# Sort
sort_key = {
'profit': 'net_profit',
'profit_factor': 'profit_factor',
'sharpe': 'sharpe_ratio',
}[args.sort]
results.sort(key=lambda r: r.get(sort_key, 0), reverse=True)
# Convergence analysis
conv = convergence_analysis(results, top_n=10)
# Display
display_results(results, args.top, args.dd_threshold, conv)
# Optional JSON output
if args.output:
output = {
'total_passes': len(results),
'results': results[:args.top],
'convergence': conv,
}
with open(args.output, 'w') as f:
json.dump(output, f, indent=2)
print(f"\nSaved: {args.output}")
if __name__ == '__main__':
main()
hooks/hook-mcp.py
-18
View File
@@ -1,18 +0,0 @@
# PyInstaller hook for mcp package
# Ensures all mcp submodules are included
from PyInstaller.utils.hooks import collect_submodules, collect_data_files
hiddenimports = collect_submodules('mcp')
datas = collect_data_files('mcp')
# Also ensure anyio dependencies are included
hiddenimports += [
'anyio',
'anyio.streams',
'anyio.streams.memory',
'anyio.streams.text',
'anyio._backends',
'anyio._backends._asyncio',
'anyio._backends._trio',
]
pyproject.toml
-32
View File
@@ -1,32 +0,0 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "mt5-quant"
version = "0.1.0"
description = "MCP server for MetaTrader 5 backtesting and optimization"
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.9"
dependencies = [
"mcp>=1.0.0",
"pyyaml>=6.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0",
"pytest-asyncio>=0.21",
]
[project.scripts]
mt5-quant = "server.main:cli"
mt5-analyze = "analytics.analyze:main_generic"
mt5-analyze-grid = "analytics.analyze:main_grid"
mt5-analyze-scalper = "analytics.analyze:main_scalper"
mt5-analyze-trend = "analytics.analyze:main_trend"
mt5-analyze-hedge = "analytics.analyze:main_hedge"
[tool.hatch.build.targets.wheel]
packages = ["server", "analytics"]
scripts/optimize.sh
-226
View File
@@ -1,226 +0,0 @@
#!/usr/bin/env bash
# optimize.sh — Launch MT5 genetic optimization (always background + detached)
#
# Usage:
# ./scripts/optimize.sh [options]
#
# Options:
# --expert NAME EA name
# --set FILE Optimization .set file (with ||Y flags)
# --symbol SYMBOL Trading symbol
# --from YYYY.MM.DD Start date
# --to YYYY.MM.DD End date
# --deposit AMOUNT Initial deposit
# --model 0|1|2 Tick model (ALWAYS use 0 for grid/martingale EAs)
# --log FILE Log file path (default: /tmp/mt5opt_TIMESTAMP.log)
#
# IMPORTANT: This script launches MT5 as a detached background process.
# It returns immediately. Do NOT set a timeout on this script.
# Monitor /tmp/mt5opt_*.log and wait for user signal before parsing results.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
source "${SCRIPT_DIR}/platform_detect.sh"
# ── Defaults ──────────────────────────────────────────────────────────────────
DEFAULT_SYMBOL=$(_cfg "backtest_symbol" "XAUUSD")
DEFAULT_DEPOSIT=$(_cfg "backtest_deposit" "10000")
DEFAULT_CURRENCY=$(_cfg "backtest_currency" "USD")
DEFAULT_LEVERAGE=$(_cfg "backtest_leverage" "500")
EXPERT=""
SET_FILE=""
SYMBOL="$DEFAULT_SYMBOL"
FROM_DATE=""
TO_DATE=""
DEPOSIT="$DEFAULT_DEPOSIT"
CURRENCY="$DEFAULT_CURRENCY"
LEVERAGE="$DEFAULT_LEVERAGE"
MODEL=0 # ALWAYS 0 for optimization — see below
LOG_FILE=""
while [[ $# -gt 0 ]]; do
case "$1" in
--expert) EXPERT="$2"; shift 2 ;;
--set) SET_FILE="$2"; shift 2 ;;
--symbol) SYMBOL="$2"; shift 2 ;;
--from) FROM_DATE="$2"; shift 2 ;;
--to) TO_DATE="$2"; shift 2 ;;
--deposit) DEPOSIT="$2"; shift 2 ;;
--model)
# Warn if user tries to use model != 0
if [[ "$2" != "0" ]]; then
echo "WARNING: --model $2 ignored. Optimization always uses model=0." >&2
echo " Model 1/2 overfits martingale/grid EAs (intra-bar price not simulated)." >&2
fi
shift 2
;;
--log) LOG_FILE="$2"; shift 2 ;;
*) echo "Unknown option: $1" >&2; exit 1 ;;
esac
done
[[ -z "$EXPERT" ]] && { echo "ERROR: --expert is required" >&2; exit 1; }
[[ -z "$SET_FILE" ]] && { echo "ERROR: --set is required" >&2; exit 1; }
[[ -z "$FROM_DATE" ]] && { echo "ERROR: --from is required" >&2; exit 1; }
[[ -z "$TO_DATE" ]] && { echo "ERROR: --to is required" >&2; exit 1; }
[[ ! -f "$SET_FILE" ]] && { echo "ERROR: Set file not found: $SET_FILE" >&2; exit 1; }
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
LOG_FILE="${LOG_FILE:-/tmp/mt5opt_${TIMESTAMP}.log}"
JOB_ID="opt_${TIMESTAMP}"
# ── Resolve platform ──────────────────────────────────────────────────────────
resolve_platform
# ── Write .set file as UTF-16LE with BOM (read-only) ─────────────────────────
# MT5 REQUIREMENT: optimization .set files must be UTF-16LE with BOM.
# If provided as UTF-8, MT5 strips the ||Y optimization flags silently —
# every pass runs with the fixed base value and optimization is useless.
python3 - << PYEOF
import sys, os, shutil
src = "${SET_FILE}"
dst = "${MT5_TESTER_DIR}/${EXPERT}.set"
os.makedirs("${MT5_TESTER_DIR}", exist_ok=True)
with open(src, 'r', encoding='utf-8', errors='replace') as f:
content = f.read()
# Write UTF-16LE with BOM
with open(dst, 'w', encoding='utf-16-le') as f:
f.write('\ufeff') # BOM
f.write(content)
# Make read-only — prevents MT5 from overwriting ||Y flags during optimization
os.chmod(dst, 0o444)
print(f" .set → {dst} (UTF-16LE, read-only)")
PYEOF
# ── Reset OptMode in terminal.ini ─────────────────────────────────────────────
# After any optimization run (complete or aborted), MT5 writes OptMode=-1.
# On next launch, MT5 reads OptMode=-1 and exits immediately without running.
# Must reset to 0 before every optimization launch.
TERMINAL_INI="${MT5_DIR}/terminal.ini"
if [[ -f "$TERMINAL_INI" ]]; then
# Use Python for safe in-place edit (sed -i behaves differently on macOS vs Linux)
python3 - << PYEOF
import re
ini_path = "${TERMINAL_INI}"
with open(ini_path, 'r', errors='replace') as f:
content = f.read()
# Reset OptMode
content = re.sub(r'OptMode=.*', 'OptMode=0', content)
# Remove LastOptimization (causes MT5 to skip running)
content = re.sub(r'LastOptimization=.*\n?', '', content)
with open(ini_path, 'w') as f:
f.write(content)
print(f" terminal.ini: OptMode reset to 0")
PYEOF
fi
# ── Build optimization INI ────────────────────────────────────────────────────
WINE_PREFIX_DIR=$(dirname "$(dirname "$MT5_DIR")")
cat > "${WINE_PREFIX_DIR}/drive_c/mt5mcp_backtest.ini" << INI
[Tester]
Expert=${EXPERT}
Symbol=${SYMBOL}
Period=M5
Deposit=${DEPOSIT}
Currency=${CURRENCY}
Leverage=${LEVERAGE}
Model=${MODEL}
FromDate=${FROM_DATE}
ToDate=${TO_DATE}
Report=C:\\mt5mcp_opt_report
Optimization=2
ExpertParameters=${EXPERT}.set
ShutdownTerminal=1
INI
cat > "${WINE_PREFIX_DIR}/drive_c/mt5mcp_run.bat" << 'EOF'
@echo off
"C:\Program Files\MetaTrader 5\terminal64.exe" /config:C:\mt5mcp_backtest.ini
EOF
# ── Count optimization combinations ──────────────────────────────────────────
COMBINATIONS=$(python3 - << PYEOF
import re, math
with open("${SET_FILE}", 'r', errors='replace') as f:
lines = f.readlines()
total = 1
for line in lines:
line = line.strip()
if line.startswith(';') or '=' not in line:
continue
# Format: param=value||start||step||stop||Y
parts = line.split('||')
if len(parts) >= 5 and parts[-1].strip().upper() == 'Y':
try:
start = float(parts[1])
step = float(parts[2])
stop = float(parts[3])
count = max(1, int((stop - start) / step) + 1)
total *= count
except (ValueError, ZeroDivisionError):
pass
print(total)
PYEOF
)
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
echo " MT5-Quant Genetic Optimization"
echo " Job ID: $JOB_ID"
echo " Expert: $EXPERT"
echo " Symbol: $SYMBOL Model: ${MODEL} (every tick)"
echo " Period: $FROM_DATE → $TO_DATE"
echo " Set file: $SET_FILE"
echo " Combos: $COMBINATIONS (genetic — converges in ~300-500 passes)"
echo " Log: $LOG_FILE"
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
# ── Launch detached ───────────────────────────────────────────────────────────
# nohup: prevents SIGHUP when parent (Claude task, SSH session) exits
# disown: removes from shell job table so shell exit doesn't kill it
# Both are required for true detachment.
nohup bash -c "${MT5_ARCH} '${MT5_WINE}' cmd.exe /c 'C:\\mt5mcp_run.bat' 2>/dev/null || true" \
> "$LOG_FILE" 2>&1 &
OPT_PID=$!
disown $OPT_PID
# Write job metadata
JOBS_DIR="${ROOT_DIR}/.mt5mcp_jobs"
mkdir -p "$JOBS_DIR"
cat > "${JOBS_DIR}/${JOB_ID}.json" << JEOF
{
"job_id": "${JOB_ID}",
"pid": ${OPT_PID},
"expert": "${EXPERT}",
"symbol": "${SYMBOL}",
"from_date": "${FROM_DATE}",
"to_date": "${TO_DATE}",
"set_file": "${SET_FILE}",
"combinations": ${COMBINATIONS},
"log_file": "${LOG_FILE}",
"wine_prefix": "${WINE_PREFIX_DIR}",
"started_at": "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
}
JEOF
echo ""
echo " Launched (pid: $OPT_PID)"
echo " Optimization runs for 2-6 hours. Do NOT kill this process."
echo " Signal when MT5 shows 'Optimization complete' and use:"
echo " python3 analytics/optimize_parser.py --job $JOB_ID"
echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
server/__init__.py
View File
server/main.py
-2636
View File
File diff suppressed because it is too large Load Diff
src/analytics/analyze.rs
+2 -2
View File
@@ -1,4 +1,4 @@
use chrono::{DateTime, Datelike, NaiveDateTime}; use chrono::{DateTime, NaiveDateTime};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap; use std::collections::HashMap;
@@ -392,7 +392,7 @@ impl DealAnalyzer {
let mut peak = 0; let mut peak = 0;
let mut peak_time = String::new(); let mut peak_time = String::new();
for (dt, delta, deal) in events { for (_dt, delta, deal) in events {
count = (count + delta).max(0); count = (count + delta).max(0);
if count > peak { if count > peak {
peak = count; peak = count;
src/analytics/extract.rs
+3
View File
@@ -277,8 +277,11 @@ impl ReportExtractor {
pub struct ExtractionResult { pub struct ExtractionResult {
pub metrics: Metrics, pub metrics: Metrics,
pub deals: Vec<Deal>, pub deals: Vec<Deal>,
#[allow(dead_code)]
pub metrics_path: PathBuf, pub metrics_path: PathBuf,
#[allow(dead_code)]
pub deals_csv_path: PathBuf, pub deals_csv_path: PathBuf,
#[allow(dead_code)]
pub deals_json_path: PathBuf, pub deals_json_path: PathBuf,
} }
src/compile/mql_compiler.rs
+1 -1
View File
@@ -58,7 +58,7 @@ impl MqlCompiler {
let wine_src_path = Self::host_to_wine_path(&dest_path)?; let wine_src_path = Self::host_to_wine_path(&dest_path)?;
let wine_log_path = Self::host_to_wine_path(&log_file)?; let wine_log_path = Self::host_to_wine_path(&log_file)?;
let output = Command::new(wine_exe) let _output = Command::new(wine_exe)
.arg(&metaeditor) .arg(&metaeditor)
.arg(format!("/compile:{}", wine_src_path)) .arg(format!("/compile:{}", wine_src_path))
.arg(format!("/log:{}", wine_log_path)) .arg(format!("/log:{}", wine_log_path))
src/config.rs
+2
View File
@@ -4,6 +4,7 @@ use std::collections::HashMap;
use std::fs; use std::fs;
use std::path::Path; use std::path::Path;
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Config { pub struct Config {
pub wine_executable: Option<String>, pub wine_executable: Option<String>,
@@ -47,6 +48,7 @@ impl Default for Config {
} }
} }
#[allow(dead_code)]
impl Config { impl Config {
pub fn load() -> Result<Self> { pub fn load() -> Result<Self> {
let config_path = Self::get_config_path(); let config_path = Self::get_config_path();
src/main.rs
+1
View File
@@ -1,6 +1,7 @@
mod analytics; mod analytics;
mod compile; mod compile;
mod models; mod models;
mod optimization;
mod pipeline; mod pipeline;
mod tools; mod tools;
src/models/deals.rs
+3
View File
@@ -20,6 +20,7 @@ pub struct Deal {
pub magic: Option<String>, pub magic: Option<String>,
} }
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")] #[serde(rename_all = "lowercase")]
pub enum DealType { pub enum DealType {
@@ -69,6 +70,7 @@ pub struct LossSequence {
pub end: String, pub end: String,
} }
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CycleStats { pub struct CycleStats {
pub total_cycles: i32, pub total_cycles: i32,
@@ -77,6 +79,7 @@ pub struct CycleStats {
pub win_rate_by_depth: HashMap<String, WinRateByDepth>, pub win_rate_by_depth: HashMap<String, WinRateByDepth>,
} }
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WinRateByDepth { pub struct WinRateByDepth {
pub total: i32, pub total: i32,
src/models/report.rs
+3
View File
@@ -1,6 +1,7 @@
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::path::PathBuf; use std::path::PathBuf;
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Report { pub struct Report {
pub report_dir: PathBuf, pub report_dir: PathBuf,
@@ -40,6 +41,7 @@ pub struct FilePaths {
pub deals_json: String, pub deals_json: String,
} }
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BacktestStatus { pub struct BacktestStatus {
pub stage: PipelineStage, pub stage: PipelineStage,
@@ -48,6 +50,7 @@ pub struct BacktestStatus {
pub message: String, pub message: String,
} }
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "UPPERCASE")] #[serde(rename_all = "UPPERCASE")]
pub enum PipelineStage { pub enum PipelineStage {
src/optimization/mod.rs
+5
View File
@@ -0,0 +1,5 @@
pub mod optimizer;
pub mod parser;
pub use optimizer::{OptimizationParams, OptimizationRunner};
pub use parser::OptimizationParser;
src/optimization/optimizer.rs
+375
View File
@@ -0,0 +1,375 @@
use anyhow::{anyhow, Result};
use chrono::Utc;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use crate::models::Config;
pub struct OptimizationParams {
pub expert: String,
pub set_file: String,
pub symbol: String,
pub from_date: String,
pub to_date: String,
pub deposit: u32,
pub model: u8,
pub leverage: u32,
pub currency: String,
}
impl Default for OptimizationParams {
fn default() -> Self {
Self {
expert: String::new(),
set_file: String::new(),
symbol: "XAUUSD".to_string(),
from_date: String::new(),
to_date: String::new(),
deposit: 10000,
model: 0,
leverage: 500,
currency: "USD".to_string(),
}
}
}
pub struct OptimizationResult {
pub success: bool,
pub job_id: String,
pub pid: u32,
pub log_file: PathBuf,
pub combinations: u64,
pub message: String,
}
pub struct OptimizationRunner {
config: Config,
}
impl OptimizationRunner {
pub fn new(config: Config) -> Self {
Self { config }
}
pub async fn run(&self, params: OptimizationParams) -> Result<OptimizationResult> {
// Validate required fields
if params.expert.is_empty() {
return Err(anyhow!("expert is required"));
}
if params.set_file.is_empty() {
return Err(anyhow!("set_file is required"));
}
if params.from_date.is_empty() {
return Err(anyhow!("from_date is required"));
}
if params.to_date.is_empty() {
return Err(anyhow!("to_date is required"));
}
let set_path = Path::new(&params.set_file);
if !set_path.exists() {
return Err(anyhow!("Set file not found: {}", params.set_file));
}
// Generate job ID and log file
let timestamp = Utc::now().format("%Y%m%d_%H%M%S").to_string();
let job_id = format!("opt_{}", timestamp);
let log_file = PathBuf::from(format!("/tmp/mt5opt_{}.log", timestamp));
// Count combinations
let combinations = self.count_combinations(&params.set_file)?;
// Get paths
let mt5_dir = self.config.terminal_dir.as_ref()
.ok_or_else(|| anyhow!("terminal_dir not configured"))?;
let wine_exe = self.config.wine_executable.as_ref()
.ok_or_else(|| anyhow!("wine_executable not configured"))?;
// Write .set file as UTF-16LE with BOM directly to MT5 tester directory
let wine_prefix_dir = self.get_wine_prefix_dir(mt5_dir)?;
let tester_dir = wine_prefix_dir.join("drive_c/Program Files/MetaTrader 5/MQL5/Profiles/Tester");
fs::create_dir_all(&tester_dir)?;
let dst_set_file = tester_dir.join(format!("{}.set", params.expert));
self.write_utf16le_set(&params.set_file, &dst_set_file)?;
// Reset OptMode in terminal.ini
self.reset_optmode(mt5_dir)?;
// Get Wine prefix directory
let wine_prefix_dir = self.get_wine_prefix_dir(mt5_dir)?;
// Build optimization INI
let ini_path = wine_prefix_dir.join("drive_c/mt5mcp_backtest.ini");
let ini_content = format!(r#"[Tester]
Expert={}
Symbol={}
Period=M5
Deposit={}
Currency={}
Leverage={}
Model={}
FromDate={}
ToDate={}
Report=C:\mt5mcp_opt_report
Optimization=2
ExpertParameters={}.set
ShutdownTerminal=1
"#, params.expert, params.symbol, params.deposit, params.currency,
params.leverage, params.model, params.from_date, params.to_date, params.expert);
fs::write(&ini_path, ini_content)?;
// Build batch file
let batch_path = wine_prefix_dir.join("drive_c/mt5mcp_run.bat");
let batch_content = format!(r#"@echo off
"C:\Program Files\MetaTrader 5\terminal64.exe" /config:C:\mt5mcp_backtest.ini
"#);
fs::write(&batch_path, batch_content)?;
// Launch detached process
let cmd = format!("cmd.exe /c 'C:\\mt5mcp_run.bat'");
let child = Command::new(wine_exe)
.arg(&cmd)
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
let pid = child.id();
// Write job metadata
self.write_job_metadata(&job_id, pid, &params, &log_file, combinations, &wine_prefix_dir)?;
Ok(OptimizationResult {
success: true,
job_id,
pid,
log_file,
combinations,
message: format!("Optimization launched (pid: {}). Runs for 2-6 hours. Do NOT kill this process.", pid),
})
}
fn count_combinations(&self, set_file: &str) -> Result<u64> {
let content = fs::read_to_string(set_file)?;
let mut total: u64 = 1;
for line in content.lines() {
let line = line.trim();
if line.starts_with(';') || !line.contains('=') {
continue;
}
// Format: param=value||start||step||stop||Y
let parts: Vec<&str> = line.split("||").collect();
if parts.len() >= 5 && parts.last().unwrap().trim().to_uppercase() == "Y" {
if let (Ok(start), Ok(step), Ok(stop)) = (
parts[1].trim().parse::<f64>(),
parts[2].trim().parse::<f64>(),
parts[3].trim().parse::<f64>(),
) {
if step > 0.0 {
let count = ((stop - start) / step).max(0.0) as u64 + 1;
total = total.saturating_mul(count);
}
}
}
}
Ok(total.max(1))
}
fn write_utf16le_set(&self, src: &str, dst: &Path) -> Result<()> {
let content = fs::read_to_string(src)?;
// Create parent directory if needed
if let Some(parent) = dst.parent() {
fs::create_dir_all(parent)?;
}
// Write UTF-16LE with BOM
let mut utf16_content: Vec<u16> = vec![0xFEFF]; // BOM
utf16_content.extend(content.encode_utf16());
let bytes: Vec<u8> = utf16_content.iter()
.flat_map(|&c| vec![(c & 0xFF) as u8, ((c >> 8) & 0xFF) as u8])
.collect();
fs::write(dst, bytes)?;
// Make read-only
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(dst, fs::Permissions::from_mode(0o444))?;
}
Ok(())
}
fn reset_optmode(&self, mt5_dir: &str) -> Result<()> {
let terminal_ini = Path::new(mt5_dir).join("terminal.ini");
if !terminal_ini.exists() {
return Ok(());
}
let content = fs::read_to_string(&terminal_ini)?;
let updated = content
.lines()
.map(|line| {
if line.starts_with("OptMode=") {
"OptMode=0".to_string()
} else if line.starts_with("LastOptimization=") {
String::new()
} else {
line.to_string()
}
})
.filter(|l| !l.is_empty())
.collect::<Vec<_>>()
.join("\n");
fs::write(&terminal_ini, updated)?;
Ok(())
}
fn get_wine_prefix_dir(&self, mt5_dir: &str) -> Result<PathBuf> {
let path = Path::new(mt5_dir);
// Go up two levels: .../drive_c/Program Files/MetaTrader 5 -> .../drive_c
let prefix_dir = path
.parent()
.and_then(|p| p.parent())
.ok_or_else(|| anyhow!("Cannot determine Wine prefix from terminal_dir"))?;
Ok(prefix_dir.to_path_buf())
}
fn write_job_metadata(
&self,
job_id: &str,
pid: u32,
params: &OptimizationParams,
log_file: &Path,
combinations: u64,
wine_prefix: &Path,
) -> Result<()> {
let jobs_dir = Path::new(".mt5mcp_jobs");
fs::create_dir_all(jobs_dir)?;
let meta_path = jobs_dir.join(format!("{}.json", job_id));
let started_at = Utc::now().to_rfc3339();
let metadata = serde_json::json!({
"job_id": job_id,
"pid": pid,
"expert": params.expert,
"symbol": params.symbol,
"from_date": params.from_date,
"to_date": params.to_date,
"set_file": params.set_file,
"combinations": combinations,
"log_file": log_file.to_string_lossy(),
"wine_prefix": wine_prefix.to_string_lossy(),
"started_at": started_at,
});
fs::write(&meta_path, serde_json::to_string_pretty(&metadata)?)?;
Ok(())
}
pub fn get_job_status(&self, job_id: &str) -> Result<serde_json::Value> {
let jobs_dir = Path::new(".mt5mcp_jobs");
let meta_path = jobs_dir.join(format!("{}.json", job_id));
if !meta_path.exists() {
return Ok(serde_json::json!({
"status": "not_found",
"message": format!("Job {} not found", job_id)
}));
}
let meta: serde_json::Value = serde_json::from_str(&fs::read_to_string(&meta_path)?)?;
let pid = meta.get("pid").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
// Check if process is still running
let is_running = self.is_process_running(pid);
// Check for completion marker in log
let log_file = meta.get("log_file").and_then(|v| v.as_str()).unwrap_or("");
let is_complete = if !log_file.is_empty() && Path::new(log_file).exists() {
fs::read_to_string(log_file)
.map(|content| content.contains("Optimization complete"))
.unwrap_or(false)
} else {
false
};
let status = if is_complete {
"completed"
} else if is_running {
"running"
} else {
"stopped"
};
Ok(serde_json::json!({
"status": status,
"job_id": job_id,
"pid": pid,
"expert": meta.get("expert"),
"symbol": meta.get("symbol"),
"started_at": meta.get("started_at"),
"log_file": log_file,
}))
}
fn is_process_running(&self, pid: u32) -> bool {
#[cfg(unix)]
{
Command::new("kill")
.args(["-0", &pid.to_string()])
.output()
.map(|output| output.status.success())
.unwrap_or(false)
}
#[cfg(windows)]
{
// Windows implementation would use different method
false
}
}
pub fn list_jobs(&self) -> Result<Vec<serde_json::Value>> {
let jobs_dir = Path::new(".mt5mcp_jobs");
let mut jobs = Vec::new();
if jobs_dir.exists() {
for entry in fs::read_dir(jobs_dir)? {
if let Ok(entry) = entry {
let path = entry.path();
if path.extension().map(|e| e == "json").unwrap_or(false) {
if let Ok(content) = fs::read_to_string(&path) {
if let Ok(meta) = serde_json::from_str::<serde_json::Value>(&content) {
let job_id = path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();
let pid = meta.get("pid").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
let is_running = self.is_process_running(pid);
jobs.push(serde_json::json!({
"job_id": job_id,
"expert": meta.get("expert"),
"status": if is_running { "running" } else { "stopped" },
"started_at": meta.get("started_at"),
}));
}
}
}
}
}
}
Ok(jobs)
}
}
src/optimization/parser.rs
+270
View File
@@ -0,0 +1,270 @@
use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fs;
use std::path::Path;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OptimizationPass {
pub pass: u32,
pub profit: f64,
pub total_trades: u32,
pub profit_factor: f64,
pub expected_payoff: f64,
pub drawdown_pct: f64,
pub params: HashMap<String, String>,
}
pub struct OptimizationParser;
impl OptimizationParser {
pub fn new() -> Self {
Self
}
pub fn parse_job(&self, job_id: &str) -> Result<Vec<OptimizationPass>> {
let jobs_dir = Path::new(".mt5mcp_jobs");
let meta_path = jobs_dir.join(format!("{}.json", job_id));
if !meta_path.exists() {
return Err(anyhow!("Job not found: {}. Check .mt5mcp_jobs/", job_id));
}
let meta: serde_json::Value = serde_json::from_str(&fs::read_to_string(&meta_path)?)?;
let wine_prefix = meta.get("wine_prefix")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("wine_prefix not in job metadata"))?;
let base_path = Path::new(wine_prefix).join("drive_c/mt5mcp_opt_report");
// Try different extensions
for ext in &[".htm", ".htm.xml", ".html"] {
let candidate = base_path.with_extension(ext.trim_start_matches('.'));
if candidate.exists() {
return self.parse_file(&candidate);
}
}
Err(anyhow!(
"Optimization report not found. Expected: {}.htm or {}.htm.xml\nIs MT5 optimization still running?",
base_path.display(),
base_path.display()
))
}
pub fn parse_file(&self, path: &Path) -> Result<Vec<OptimizationPass>> {
let format = self.detect_format(path);
let text = self.read_text(path)?;
match format {
"xml" => self.parse_xml(&text),
_ => self.parse_html(&text),
}
}
fn detect_format(&self, path: &Path) -> &str {
let path_str = path.to_string_lossy();
if path_str.ends_with(".xml") || path_str.ends_with(".htm.xml") {
return "xml";
}
if let Ok(header) = fs::read(path) {
let header = &header[..header.len().min(512)];
if header.windows(5).any(|w| w == b"<?xml") || header.windows(8).any(|w| w == b"Workbook") {
return "xml";
}
}
"html"
}
fn read_text(&self, path: &Path) -> Result<String> {
let raw = fs::read(path)?;
// Try UTF-16 first (common for MT5 reports)
if raw.len() >= 2 {
if raw[0] == 0xFF && raw[1] == 0xFE {
// UTF-16 LE with BOM
let u16_vec: Vec<u16> = raw[2..].chunks_exact(2)
.map(|c| u16::from_le_bytes([c[0], c[1]]))
.collect();
return Ok(String::from_utf16_lossy(&u16_vec));
} else if raw[0] == 0xFE && raw[1] == 0xFF {
// UTF-16 BE with BOM
let u16_vec: Vec<u16> = raw[2..].chunks_exact(2)
.map(|c| u16::from_be_bytes([c[0], c[1]]))
.collect();
return Ok(String::from_utf16_lossy(&u16_vec));
}
}
// Try UTF-8, then fallback to lossy
if let Ok(text) = String::from_utf8(raw.clone()) {
return Ok(text);
}
// Try UTF-16 without BOM
if raw.len() % 2 == 0 {
let u16_vec: Vec<u16> = raw.chunks_exact(2)
.map(|c| u16::from_le_bytes([c[0], c[1]]))
.collect();
let text = String::from_utf16_lossy(&u16_vec);
if text.chars().any(|c| c.is_ascii_alphanumeric()) {
return Ok(text);
}
}
Ok(String::from_utf8_lossy(&raw).to_string())
}
fn parse_html(&self, text: &str) -> Result<Vec<OptimizationPass>> {
let mut results = Vec::new();
let mut headers: Vec<String> = Vec::new();
// Find all table rows
let row_regex = regex::Regex::new(r"<tr[^>]*>(.*?)</tr>")?;
let cell_regex = regex::Regex::new(r"<t[dh][^>]*>(.*?)</t[dh]>")?;
let tag_regex = regex::Regex::new(r"<[^>]+>")?;
for row_caps in row_regex.captures_iter(text) {
let row = &row_caps[1];
let cells: Vec<String> = cell_regex.captures_iter(row)
.map(|c| {
let cell = &c[1];
tag_regex.replace_all(cell, "").trim().to_string().replace(',', "")
})
.collect();
if cells.is_empty() {
continue;
}
// Header row detection
if headers.is_empty() && cells[0].to_lowercase().contains("pass") {
headers = cells;
continue;
}
// Data row
if !headers.is_empty() && cells[0].parse::<u32>().is_ok() {
let row_map: HashMap<String, String> = headers.iter()
.zip(cells.iter())
.map(|(h, c)| (h.to_lowercase().replace(' ', "_"), c.clone()))
.collect();
if let Some(pass) = self.row_to_pass(&row_map) {
results.push(pass);
}
}
}
Ok(results)
}
fn parse_xml(&self, text: &str) -> Result<Vec<OptimizationPass>> {
let mut results = Vec::new();
// Parse SpreadsheetML XML
let doc = roxmltree::Document::parse(text)?;
// Find all rows in Worksheet/Table
for node in doc.descendants() {
if node.has_tag_name(("http://schemas.microsoft.com/office/excel/2003/xml", "Row")) ||
node.has_tag_name("Row") {
let cells: Vec<String> = node.children()
.filter(|n: &roxmltree::Node<'_, '_>| {
n.has_tag_name(("http://schemas.microsoft.com/office/excel/2003/xml", "Cell")) ||
n.has_tag_name("Cell") ||
n.has_tag_name(("http://schemas.microsoft.com/office/excel/2003/xml", "Data")) ||
n.has_tag_name("Data")
})
.map(|n| n.text().unwrap_or("").trim().to_string().replace(',', ""))
.collect();
if cells.is_empty() {
continue;
}
// Check if first cell is a pass number
if let Ok(pass_num) = cells[0].parse::<u32>() {
if pass_num > 0 {
let mut row_map = HashMap::new();
// Standard MT5 optimization report columns
let headers = vec![
"pass", "result", "profit", "total_trades", "profit_factor",
"expected_payoff", "drawdown_pct", "recovery_factor", "sharpe_ratio",
"custom", "consecutive_wins", "consecutive_losses",
];
for (i, cell) in cells.iter().enumerate() {
if let Some(header) = headers.get(i) {
row_map.insert(header.to_string(), cell.clone());
}
}
if let Some(pass) = self.row_to_pass(&row_map) {
results.push(pass);
}
}
}
}
}
Ok(results)
}
fn row_to_pass(&self, row: &HashMap<String, String>) -> Option<OptimizationPass> {
let pass = row.get("pass").or_else(|| row.get("#"))
.and_then(|v| v.parse().ok())?;
let profit = row.get("profit").or_else(|| row.get("total_net_profit"))
.and_then(|v| v.replace(' ', "").parse().ok())?;
let total_trades = row.get("total_trades").or_else(|| row.get("trades"))
.and_then(|v| v.parse().ok())?;
let profit_factor = row.get("profit_factor")
.and_then(|v| v.parse().ok())?;
let expected_payoff = row.get("expected_payoff")
.and_then(|v| v.parse().ok())?;
let drawdown_pct = row.get("drawdown_pct").or_else(|| row.get("max_drawdown"))
.and_then(|v| v.trim_end_matches('%').trim().parse().ok())?;
// Extract parameter values from row
let params: HashMap<String, String> = row.iter()
.filter(|(k, _)| ![
"pass", "result", "profit", "total_trades", "profit_factor",
"expected_payoff", "drawdown_pct", "max_drawdown", "recovery_factor",
"sharpe_ratio", "custom", "consecutive_wins", "consecutive_losses"
].contains(&k.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
Some(OptimizationPass {
pass,
profit,
total_trades,
profit_factor,
expected_payoff,
drawdown_pct,
params,
})
}
pub fn find_best_pass<'a>(&self, passes: &'a [OptimizationPass], criteria: &str) -> Option<&'a OptimizationPass> {
match criteria {
"profit" => passes.iter().max_by(|a, b| a.profit.partial_cmp(&b.profit).unwrap()),
"profit_factor" => passes.iter().max_by(|a, b| a.profit_factor.partial_cmp(&b.profit_factor).unwrap()),
"sharpe" => passes.iter().max_by(|a, b| {
let a_sharpe = a.params.get("sharpe_ratio").and_then(|v| v.parse::<f64>().ok()).unwrap_or(0.0);
let b_sharpe = b.params.get("sharpe_ratio").and_then(|v| v.parse::<f64>().ok()).unwrap_or(0.0);
a_sharpe.partial_cmp(&b_sharpe).unwrap()
}),
"drawdown" => passes.iter().min_by(|a, b| a.drawdown_pct.partial_cmp(&b.drawdown_pct).unwrap()),
_ => passes.iter().max_by(|a, b| a.profit.partial_cmp(&b.profit).unwrap()),
}
}
}
src/pipeline/backtest.rs
+3 -2
View File
@@ -30,6 +30,7 @@ pub struct BacktestParams {
pub skip_compile: bool, pub skip_compile: bool,
pub skip_clean: bool, pub skip_clean: bool,
pub skip_analyze: bool, pub skip_analyze: bool,
#[allow(dead_code)]
pub deep_analyze: bool, pub deep_analyze: bool,
pub shutdown: bool, pub shutdown: bool,
pub kill_existing: bool, pub kill_existing: bool,
@@ -250,7 +251,7 @@ start terminal64.exe /config:"C:\Program Files\MetaTrader 5\backtest_config.ini"
let bat_path = wine_prefix.join("drive_c").join("_mt5mcp_run.bat"); let bat_path = wine_prefix.join("drive_c").join("_mt5mcp_run.bat");
fs::write(&bat_path, bat_content)?; fs::write(&bat_path, bat_content)?;
let cmd = format!("cmd.exe /c 'C:\\_mt5mcp_run.bat'"); let _cmd = format!("cmd.exe /c 'C:\\_mt5mcp_run.bat'");
if params.shutdown { if params.shutdown {
let output = Command::new("timeout") let output = Command::new("timeout")
@@ -346,7 +347,7 @@ start terminal64.exe /config:"C:\Program Files\MetaTrader 5\backtest_config.ini"
} }
async fn kill_mt5(&self) -> Result<()> { async fn kill_mt5(&self) -> Result<()> {
let output = Command::new("pkill") let _output = Command::new("pkill")
.args(&["-TERM", "-f", "terminal64\\.exe"]) .args(&["-TERM", "-f", "terminal64\\.exe"])
.output()?; .output()?;
src/pipeline/stages.rs
+6
View File
@@ -1,6 +1,7 @@
use anyhow::Result; use anyhow::Result;
use std::path::PathBuf; use std::path::PathBuf;
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq)] #[derive(Debug, Clone, Copy, PartialEq)]
pub enum Stage { pub enum Stage {
Compile, Compile,
@@ -11,6 +12,7 @@ pub enum Stage {
Done, Done,
} }
#[allow(dead_code)]
impl Stage { impl Stage {
pub fn as_str(&self) -> &'static str { pub fn as_str(&self) -> &'static str {
match self { match self {
@@ -35,8 +37,10 @@ impl Stage {
} }
} }
#[allow(dead_code)]
pub struct StageExecutor; pub struct StageExecutor;
#[allow(dead_code)]
impl StageExecutor { impl StageExecutor {
pub fn new() -> Self { pub fn new() -> Self {
Self Self
@@ -74,12 +78,14 @@ impl StageExecutor {
} }
} }
#[allow(dead_code)]
pub struct StageResult { pub struct StageResult {
pub success: bool, pub success: bool,
pub message: String, pub message: String,
pub output: Option<PathBuf>, pub output: Option<PathBuf>,
} }
#[allow(dead_code)]
impl StageResult { impl StageResult {
pub fn success() -> Self { pub fn success() -> Self {
Self { Self {
src/tools/handlers.rs
+663
View File
@@ -3,8 +3,12 @@ use serde_json::{json, Value};
use std::collections::HashMap; use std::collections::HashMap;
use std::fs; use std::fs;
use std::path::Path; use std::path::Path;
use crate::analytics::DealAnalyzer;
use crate::compile::MqlCompiler; use crate::compile::MqlCompiler;
use crate::models::Config; use crate::models::Config;
use crate::models::deals::Deal;
use crate::models::metrics::Metrics;
use crate::optimization::{OptimizationParams, OptimizationParser, OptimizationRunner};
use crate::pipeline::backtest::{BacktestParams, BacktestPipeline}; use crate::pipeline::backtest::{BacktestParams, BacktestPipeline};
#[derive(Debug)] #[derive(Debug)]
@@ -31,6 +35,28 @@ impl ToolHandler {
"prune_reports" => self.handle_prune_reports(args).await, "prune_reports" => self.handle_prune_reports(args).await,
"list_set_files" => self.handle_list_set_files().await, "list_set_files" => self.handle_list_set_files().await,
"describe_sweep" => self.handle_describe_sweep(args).await, "describe_sweep" => self.handle_describe_sweep(args).await,
// Optimization tools
"run_optimization" => self.handle_run_optimization(args).await,
"get_optimization_status" => self.handle_get_optimization_status(args).await,
"get_optimization_results" => self.handle_get_optimization_results(args).await,
"list_jobs" => self.handle_list_jobs().await,
// Analysis tools
"analyze_report" => self.handle_analyze_report(args).await,
"compare_baseline" => self.handle_compare_baseline(args).await,
// Set file tools
"read_set_file" => self.handle_read_set_file(args).await,
"write_set_file" => self.handle_write_set_file(args).await,
"patch_set_file" => self.handle_patch_set_file(args).await,
"clone_set_file" => self.handle_clone_set_file(args).await,
"diff_set_files" => self.handle_diff_set_files(args).await,
"set_from_optimization" => self.handle_set_from_optimization(args).await,
// Utility tools
"tail_log" => self.handle_tail_log(args).await,
"archive_report" => self.handle_archive_report(args).await,
"archive_all_reports" => self.handle_archive_all_reports(args).await,
"promote_to_baseline" => self.handle_promote_to_baseline(args).await,
"get_history" => self.handle_get_history(args).await,
"annotate_history" => self.handle_annotate_history(args).await,
_ => Ok(json!({ _ => Ok(json!({
"content": [{ "type": "text", "text": format!("Tool '{}' not implemented", name) }], "content": [{ "type": "text", "text": format!("Tool '{}' not implemented", name) }],
"isError": true "isError": true
@@ -461,4 +487,641 @@ impl ToolHandler {
"isError": false "isError": false
})) }))
} }
// Optimization handlers
async fn handle_run_optimization(&self, args: &Value) -> Result<Value> {
let expert = args.get("expert")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("expert is required"))?;
let set_file = args.get("set_file")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("set_file is required"))?;
let from_date = args.get("from_date")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("from_date is required"))?;
let to_date = args.get("to_date")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("to_date is required"))?;
let params = OptimizationParams {
expert: expert.to_string(),
set_file: set_file.to_string(),
symbol: args.get("symbol").and_then(|v| v.as_str()).unwrap_or("XAUUSD").to_string(),
from_date: from_date.to_string(),
to_date: to_date.to_string(),
deposit: args.get("deposit").and_then(|v| v.as_u64()).unwrap_or(10000) as u32,
model: 0, // Always 0 for optimization
leverage: args.get("leverage").and_then(|v| v.as_u64()).unwrap_or(500) as u32,
currency: args.get("currency").and_then(|v| v.as_str()).unwrap_or("USD").to_string(),
};
let runner = OptimizationRunner::new(self.config.clone());
let result = runner.run(params).await?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": result.success,
"job_id": result.job_id,
"pid": result.pid,
"log_file": result.log_file.to_string_lossy(),
"combinations": result.combinations,
"message": result.message,
}).to_string() }],
"isError": false
}))
}
async fn handle_get_optimization_status(&self, args: &Value) -> Result<Value> {
let job_id = args.get("job_id")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("job_id is required"))?;
let runner = OptimizationRunner::new(self.config.clone());
let status = runner.get_job_status(job_id)?;
Ok(json!({
"content": [{ "type": "text", "text": status.to_string() }],
"isError": false
}))
}
async fn handle_get_optimization_results(&self, args: &Value) -> Result<Value> {
let job_id = args.get("job_id")
.and_then(|v| v.as_str());
let file = args.get("file")
.and_then(|v| v.as_str());
let parser = OptimizationParser::new();
let passes = if let Some(jid) = job_id {
parser.parse_job(jid)?
} else if let Some(f) = file {
parser.parse_file(std::path::Path::new(f))?
} else {
return Err(anyhow::anyhow!("Either job_id or file is required"));
};
let sort_by = args.get("sort").and_then(|v| v.as_str()).unwrap_or("profit");
let top_n = args.get("top").and_then(|v| v.as_u64()).unwrap_or(30) as usize;
// Find best pass
let best = parser.find_best_pass(&passes, sort_by);
let mut sorted_passes = passes.clone();
sorted_passes.sort_by(|a, b| b.profit.partial_cmp(&a.profit).unwrap());
sorted_passes.truncate(top_n);
Ok(json!({
"content": [{ "type": "text", "text": json!({
"total_passes": passes.len(),
"top_passes": sorted_passes,
"best": best,
"sort_by": sort_by,
}).to_string() }],
"isError": false
}))
}
async fn handle_list_jobs(&self) -> Result<Value> {
let runner = OptimizationRunner::new(self.config.clone());
let jobs = runner.list_jobs()?;
Ok(json!({
"content": [{ "type": "text", "text": json!({ "jobs": jobs }).to_string() }],
"isError": false
}))
}
// Analysis handlers
async fn handle_analyze_report(&self, args: &Value) -> Result<Value> {
let report_dir = args.get("report_dir")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("report_dir is required"))?;
let deals_csv = std::path::Path::new(report_dir).join("deals.csv");
let metrics_json = std::path::Path::new(report_dir).join("metrics.json");
if !deals_csv.exists() {
return Err(anyhow::anyhow!("deals.csv not found in {}", report_dir));
}
// Read deals
let deals = self.read_deals_from_csv(&deals_csv)?;
// Read metrics
let metrics = if metrics_json.exists() {
let content = fs::read_to_string(&metrics_json)?;
serde_json::from_str(&content)?
} else {
Metrics::default()
};
let _strategy = args.get("strategy").and_then(|v| v.as_str()).unwrap_or("grid");
let _deep = args.get("deep").and_then(|v| v.as_bool()).unwrap_or(false);
let analyzer = DealAnalyzer::new();
let result = analyzer.analyze(&deals, &metrics);
// Write analysis.json
let analysis_path = std::path::Path::new(report_dir).join("analysis.json");
fs::write(&analysis_path, serde_json::to_string_pretty(&result)?)?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"analysis_file": analysis_path.to_string_lossy(),
"summary": result,
}).to_string() }],
"isError": false
}))
}
fn read_deals_from_csv(&self, path: &std::path::Path) -> Result<Vec<Deal>> {
let content = fs::read_to_string(path)?;
let mut deals = Vec::new();
let mut lines = content.lines();
let _header = lines.next(); // Skip header
for line in lines {
let parts: Vec<&str> = line.split(',').collect();
if parts.len() >= 12 {
deals.push(Deal {
time: parts[0].to_string(),
deal: parts[1].to_string(),
symbol: parts[2].to_string(),
deal_type: parts[3].to_string(),
entry: parts[4].to_string(),
volume: parts[5].parse().unwrap_or(0.0),
price: parts[6].parse().unwrap_or(0.0),
order: parts[7].to_string(),
commission: parts[8].parse().unwrap_or(0.0),
swap: parts[9].parse().unwrap_or(0.0),
profit: parts[10].parse().unwrap_or(0.0),
balance: parts[11].parse().unwrap_or(0.0),
comment: parts.get(12).unwrap_or(&"").to_string(),
magic: parts.get(13).map(|s| s.to_string()),
});
}
}
Ok(deals)
}
async fn handle_compare_baseline(&self, args: &Value) -> Result<Value> {
let report_dir = args.get("report_dir")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("report_dir is required"))?;
let baseline_path = std::path::Path::new("config/baseline.json");
let metrics_path = std::path::Path::new(report_dir).join("metrics.json");
if !baseline_path.exists() {
return Ok(json!({
"content": [{ "type": "text", "text": "No baseline.json found in config/" }],
"isError": false
}));
}
let baseline: Value = serde_json::from_str(&fs::read_to_string(baseline_path)?)?;
let current: Value = serde_json::from_str(&fs::read_to_string(metrics_path)?)?;
let comparison = json!({
"baseline": baseline,
"current": current,
"improvements": {
"profit": current.get("net_profit").and_then(|v| v.as_f64()).unwrap_or(0.0)
- baseline.get("net_profit").and_then(|v| v.as_f64()).unwrap_or(0.0),
"drawdown": current.get("max_dd_pct").and_then(|v| v.as_f64()).unwrap_or(0.0)
- baseline.get("max_dd_pct").and_then(|v| v.as_f64()).unwrap_or(0.0),
}
});
Ok(json!({
"content": [{ "type": "text", "text": comparison.to_string() }],
"isError": false
}))
}
// Set file handlers
async fn handle_read_set_file(&self, args: &Value) -> Result<Value> {
let path = args.get("path")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("path is required"))?;
let content = fs::read_to_string(path)?;
let mut params = serde_json::Map::new();
for line in content.lines() {
if let Some((key, value)) = line.split_once(':') {
let key = key.trim();
let value = value.trim();
if value.contains("||Y") {
let parts: Vec<&str> = value.split("||").collect();
if parts.len() >= 5 {
params.insert(key.to_string(), json!({
"value": parts[0],
"from": parts[1],
"step": parts[2],
"to": parts[3],
"optimize": true,
}));
}
} else {
params.insert(key.to_string(), json!({ "value": value, "optimize": false }));
}
}
}
Ok(json!({
"content": [{ "type": "text", "text": json!({
"path": path,
"parameters": params,
}).to_string() }],
"isError": false
}))
}
async fn handle_write_set_file(&self, args: &Value) -> Result<Value> {
let path = args.get("path")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("path is required"))?;
let params = args.get("parameters")
.and_then(|v| v.as_object())
.ok_or_else(|| anyhow::anyhow!("parameters object is required"))?;
let mut lines = Vec::new();
for (key, value) in params {
if let Some(obj) = value.as_object() {
if obj.get("optimize").and_then(|v| v.as_bool()).unwrap_or(false) {
let from_val = obj.get("from").and_then(|v| v.as_str()).unwrap_or("0");
let step = obj.get("step").and_then(|v| v.as_str()).unwrap_or("1");
let to_val = obj.get("to").and_then(|v| v.as_str()).unwrap_or("0");
lines.push(format!("{}={}||{}||{}||{}||Y", key, obj.get("value").and_then(|v| v.as_str()).unwrap_or("0"), from_val, step, to_val));
} else {
lines.push(format!("{}={}", key, obj.get("value").and_then(|v| v.as_str()).unwrap_or("0")));
}
}
}
fs::write(path, lines.join("\n"))?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"path": path,
"parameters_written": lines.len(),
}).to_string() }],
"isError": false
}))
}
async fn handle_patch_set_file(&self, args: &Value) -> Result<Value> {
let path = args.get("path")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("path is required"))?;
let patches = args.get("patches")
.and_then(|v| v.as_object())
.ok_or_else(|| anyhow::anyhow!("patches object is required"))?;
// Read existing file
let content = fs::read_to_string(path)?;
let mut lines: Vec<String> = content.lines().map(|s| s.to_string()).collect();
let mut patched_count = 0;
for (key, value) in patches {
let new_value = if let Some(s) = value.as_str() {
s.to_string()
} else if let Some(n) = value.as_f64() {
n.to_string()
} else if let Some(b) = value.as_bool() {
if b { "true".to_string() } else { "false".to_string() }
} else {
value.to_string()
};
// Find and patch the parameter
let mut found = false;
for line in &mut lines {
if line.starts_with(&format!("{}:", key)) {
*line = format!("{}: {}", key, new_value);
found = true;
patched_count += 1;
break;
} else if line.starts_with(&format!("{}=", key)) {
*line = format!("{}={}", key, new_value);
found = true;
patched_count += 1;
break;
}
}
// If not found, add it
if !found {
lines.push(format!("{}: {}", key, new_value));
patched_count += 1;
}
}
fs::write(path, lines.join("\n"))?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"path": path,
"parameters_patched": patched_count,
}).to_string() }],
"isError": false
}))
}
async fn handle_clone_set_file(&self, args: &Value) -> Result<Value> {
let source = args.get("source")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("source is required"))?;
let destination = args.get("destination")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("destination is required"))?;
fs::copy(source, destination)?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"source": source,
"destination": destination,
}).to_string() }],
"isError": false
}))
}
async fn handle_diff_set_files(&self, args: &Value) -> Result<Value> {
let file_a = args.get("file_a")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("file_a is required"))?;
let file_b = args.get("file_b")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("file_b is required"))?;
let content_a = fs::read_to_string(file_a)?;
let content_b = fs::read_to_string(file_b)?;
let mut differences = Vec::new();
for (i, (line_a, line_b)) in content_a.lines().zip(content_b.lines()).enumerate() {
if line_a != line_b {
differences.push(json!({
"line": i + 1,
"file_a": line_a,
"file_b": line_b,
}));
}
}
Ok(json!({
"content": [{ "type": "text", "text": json!({
"file_a": file_a,
"file_b": file_b,
"differences": differences,
"total_differences": differences.len(),
}).to_string() }],
"isError": false
}))
}
async fn handle_set_from_optimization(&self, args: &Value) -> Result<Value> {
let path = args.get("path")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("path is required"))?;
let params = args.get("params")
.and_then(|v| v.as_object())
.ok_or_else(|| anyhow::anyhow!("params is required"))?;
let mut lines = Vec::new();
for (key, value) in params {
if let Some(val_str) = value.as_str() {
lines.push(format!("{}={}", key, val_str));
}
}
fs::write(path, lines.join("\n"))?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"path": path,
"parameters_written": lines.len(),
}).to_string() }],
"isError": false
}))
}
// Utility handlers
async fn handle_tail_log(&self, args: &Value) -> Result<Value> {
let job_id = args.get("job_id")
.and_then(|v| v.as_str());
let lines = args.get("lines").and_then(|v| v.as_u64()).unwrap_or(50) as usize;
let log_path = if let Some(jid) = job_id {
let jobs_dir = std::path::Path::new(".mt5mcp_jobs");
let meta_path = jobs_dir.join(format!("{}.json", jid));
let meta: Value = serde_json::from_str(&fs::read_to_string(meta_path)?)?;
meta.get("log_file").and_then(|v| v.as_str()).map(|s| s.to_string())
} else {
args.get("file").and_then(|v| v.as_str()).map(|s| s.to_string())
};
let log_path = log_path.ok_or_else(|| anyhow::anyhow!("Could not determine log file"))?;
let content = fs::read_to_string(&log_path)?;
let all_lines: Vec<&str> = content.lines().collect();
let start = all_lines.len().saturating_sub(lines);
let last_lines = &all_lines[start..];
Ok(json!({
"content": [{ "type": "text", "text": last_lines.join("\n") }],
"isError": false
}))
}
async fn handle_archive_report(&self, args: &Value) -> Result<Value> {
let report_dir = args.get("report_dir")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("report_dir is required"))?;
let delete_after = args.get("delete_after").and_then(|v| v.as_bool()).unwrap_or(false);
let history_dir = std::path::Path::new(".mt5mcp_history");
fs::create_dir_all(history_dir)?;
let report_name = std::path::Path::new(report_dir).file_name()
.and_then(|s| s.to_str())
.unwrap_or("unknown");
let archive_path = history_dir.join(format!("{}.tar.gz", report_name));
// Create tarball
let status = std::process::Command::new("tar")
.args(["-czf", &archive_path.to_string_lossy(), "-C",
std::path::Path::new(report_dir).parent().unwrap().to_str().unwrap(),
report_name])
.status()?;
if delete_after && status.success() {
fs::remove_dir_all(report_dir)?;
}
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": status.success(),
"archive_path": archive_path.to_string_lossy(),
"deleted": delete_after && status.success(),
}).to_string() }],
"isError": false
}))
}
async fn handle_archive_all_reports(&self, args: &Value) -> Result<Value> {
let keep_last = args.get("keep_last").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
let reports_dir = self.config.reports_dir();
let history_dir = std::path::Path::new(".mt5mcp_history");
fs::create_dir_all(history_dir)?;
let mut archived = 0;
if let Ok(entries) = fs::read_dir(&reports_dir) {
let mut entries: Vec<_> = entries.flatten().collect();
entries.sort_by(|a, b| {
b.metadata().and_then(|m| m.modified()).unwrap_or(std::time::UNIX_EPOCH)
.cmp(&a.metadata().and_then(|m| m.modified()).unwrap_or(std::time::UNIX_EPOCH))
});
for entry in entries.into_iter().skip(keep_last) {
let path = entry.path();
if path.is_dir() && !path.to_string_lossy().ends_with("_opt") {
let report_name = path.file_name().and_then(|s| s.to_str()).unwrap_or("unknown");
let archive_path = history_dir.join(format!("{}.tar.gz", report_name));
let _ = std::process::Command::new("tar")
.args(["-czf", &archive_path.to_string_lossy(), "-C",
path.parent().unwrap().to_str().unwrap(), report_name])
.status();
let _ = fs::remove_dir_all(&path);
archived += 1;
}
}
}
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"archived": archived,
"kept": keep_last,
}).to_string() }],
"isError": false
}))
}
async fn handle_promote_to_baseline(&self, args: &Value) -> Result<Value> {
let report_dir = args.get("report_dir")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("report_dir is required"))?;
let metrics_path = std::path::Path::new(report_dir).join("metrics.json");
let baseline_path = std::path::Path::new("config/baseline.json");
fs::copy(&metrics_path, &baseline_path)?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"baseline_file": baseline_path.to_string_lossy(),
"source": metrics_path.to_string_lossy(),
}).to_string() }],
"isError": false
}))
}
async fn handle_get_history(&self, args: &Value) -> Result<Value> {
let _limit = args.get("limit").and_then(|v| v.as_u64()).unwrap_or(50) as usize;
let history_dir = std::path::Path::new(".mt5mcp_history");
let mut history = Vec::new();
if history_dir.exists() {
for entry in fs::read_dir(history_dir)? {
if let Ok(entry) = entry {
let path = entry.path();
if path.extension().map(|e| e == "tar.gz").unwrap_or(false) {
let name = path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();
let metadata = entry.metadata()?;
let modified = metadata.modified()?;
let size = metadata.len();
history.push(json!({
"name": name,
"path": path.to_string_lossy(),
"size": size,
"archived_at": modified.elapsed().map(|e| e.as_secs()).unwrap_or(0),
}));
}
}
}
}
Ok(json!({
"content": [{ "type": "text", "text": json!({
"total_archived": history.len(),
"history": history,
}).to_string() }],
"isError": false
}))
}
async fn handle_annotate_history(&self, args: &Value) -> Result<Value> {
let report_name = args.get("report_name")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("report_name is required"))?;
let note = args.get("note")
.and_then(|v| v.as_str())
.unwrap_or("");
let notes_path = std::path::Path::new(".mt5mcp_history").join("notes.json");
let mut notes: serde_json::Map<String, Value> = if notes_path.exists() {
serde_json::from_str(&fs::read_to_string(&notes_path)?)?
} else {
serde_json::Map::new()
};
notes.insert(report_name.to_string(), json!(note));
fs::write(&notes_path, serde_json::to_string_pretty(&notes)?)?;
Ok(json!({
"content": [{ "type": "text", "text": json!({
"success": true,
"report": report_name,
"note": note,
}).to_string() }],
"isError": false
}))
}
} }
tests/__init__.py
View File
test_rcp_server.sh → tests/integration_test.sh
View File
tests/integration_tests.rs
+57
View File
@@ -0,0 +1,57 @@
use std::fs;
use std::path::PathBuf;
fn get_fixture_path(name: &str) -> PathBuf {
let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("tests/fixtures");
path.push(name);
path
}
#[test]
fn test_fixtures_exist() {
let fixtures = vec![
"sample_deals.csv",
"sample_report.htm",
"sample_report.htm.xml",
];
for fixture in fixtures {
let path = get_fixture_path(fixture);
assert!(path.exists(), "Fixture {} should exist", fixture);
}
}
#[test]
fn test_sample_deals_csv_format() {
let path = get_fixture_path("sample_deals.csv");
let content = fs::read_to_string(path).expect("Should read sample_deals.csv");
// Check CSV has header and data rows
let lines: Vec<&str> = content.lines().collect();
assert!(!lines.is_empty(), "CSV should have at least a header");
// Check for expected columns in header
let header = lines[0];
assert!(header.contains("Time") || header.contains("time"), "Header should contain Time column");
}
#[test]
fn test_sample_report_html_format() {
let path = get_fixture_path("sample_report.htm");
let content = fs::read_to_string(path).expect("Should read sample_report.htm");
// Check HTML structure
assert!(content.contains("<html") || content.contains("<table"),
"Report should contain HTML or table elements");
}
#[test]
fn test_sample_report_xml_format() {
let path = get_fixture_path("sample_report.htm.xml");
let content = fs::read_to_string(path).expect("Should read sample_report.htm.xml");
// Check XML structure
assert!(content.contains("<?xml") || content.contains("<Workbook"),
"Report should contain XML or Workbook elements");
}
tests/test_analyze.py
-673
View File
@@ -1,673 +0,0 @@
"""Tests for analytics/analyze.py — runs without MT5 or Wine."""
import sys
from pathlib import Path
import pytest
FIXTURES = Path(__file__).parent / 'fixtures'
sys.path.insert(0, str(Path(__file__).parent.parent))
from analytics.analyze import (
PROFILES,
load_deals, monthly_pnl, reconstruct_dd_events,
grid_depth_histogram, depth_histogram, top_losses, loss_sequences, build_summary,
position_pairs, cycle_stats, exit_reason_breakdown,
direction_bias, streak_analysis, session_breakdown,
weekday_pnl, hourly_pnl, concurrent_peak, volume_profile,
_parse_dt, _classify_exit, _extract_depth, _classify_dd_cause,
_lot_tier, _session_for_hour,
)
@pytest.fixture
def deals():
return load_deals(str(FIXTURES / 'sample_deals.csv'))
def test_load_deals_count(deals):
assert len(deals) > 0
def test_load_deals_numeric_fields(deals):
for deal in deals:
assert isinstance(deal['profit'], float)
assert isinstance(deal['balance'], float)
assert isinstance(deal['volume'], float)
def test_monthly_pnl_groups_correctly(deals):
result = monthly_pnl(deals)
assert isinstance(result, list)
assert len(result) >= 1
for entry in result:
assert 'month' in entry
assert 'pnl' in entry
assert 'trades' in entry
assert 'green' in entry
assert isinstance(entry['green'], bool)
def test_monthly_pnl_only_out_entries(deals):
"""Only 'out' entries should be counted."""
result = monthly_pnl(deals)
# All trades in fixture are closed, so at least one month should have trades
total_trades = sum(m['trades'] for m in result)
assert total_trades > 0
def test_monthly_pnl_has_jan_and_feb(deals):
result = monthly_pnl(deals)
months = [m['month'] for m in result]
assert '2025-01' in months
assert '2025-02' in months
def test_reconstruct_dd_events_returns_list(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
result = reconstruct_dd_events(deals, metrics)
assert isinstance(result, list)
def test_reconstruct_dd_events_empty_on_no_deals():
result = reconstruct_dd_events([], {})
assert result == []
def test_grid_depth_histogram_keys(deals):
hist = grid_depth_histogram(deals)
assert isinstance(hist, dict)
assert 'L1' in hist
assert 'L2' in hist
assert 'L3' in hist
assert 'L8+' in hist
def test_grid_depth_histogram_counts_layers(deals):
hist = grid_depth_histogram(deals)
# Fixture has Layer #1, #2, #3 comments
assert hist['L1'] > 0
assert hist['L3'] > 0
def test_top_losses_are_negative(deals):
losses = top_losses(deals)
assert isinstance(losses, list)
for loss in losses:
assert loss['loss_usd'] < 0
def test_top_losses_sorted_ascending(deals):
losses = top_losses(deals)
if len(losses) >= 2:
assert losses[0]['loss_usd'] <= losses[1]['loss_usd']
def test_loss_sequences_structure(deals):
seqs = loss_sequences(deals)
assert isinstance(seqs, list)
for seq in seqs:
assert 'length' in seq
assert 'total_loss' in seq
assert seq['total_loss'] < 0
def test_build_summary_keys(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5, 'total_trades': 11,
'profit_factor': 1.2, 'sharpe_ratio': 0.5, 'recovery_factor': 2.0}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
summary = build_summary(metrics, monthly, dd)
expected_keys = ['net_profit', 'profit_factor', 'max_dd_pct', 'sharpe_ratio',
'total_trades', 'green_months', 'total_months',
'worst_month', 'worst_month_pnl']
for k in expected_keys:
assert k in summary, f"Missing key: {k}"
def test_build_summary_green_months(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
summary = build_summary(metrics, monthly, dd)
assert summary['green_months'] >= 0
assert summary['total_months'] >= summary['green_months']
# ── Utility helpers ────────────────────────────────────────────────────────────
def test_parse_dt_standard_format():
dt = _parse_dt('2025.01.10 09:30:00')
assert dt is not None
assert dt.year == 2025
assert dt.month == 1
assert dt.day == 10
assert dt.hour == 9
def test_parse_dt_iso_format():
dt = _parse_dt('2025-02-05 14:00:00')
assert dt is not None
assert dt.month == 2
def test_parse_dt_invalid_returns_none():
assert _parse_dt('') is None
assert _parse_dt('not-a-date') is None
def test_classify_exit_locking():
assert _classify_exit('locking hedge', -50.0) == 'locking'
def test_classify_exit_cutloss():
assert _classify_exit('cutloss fired', -20.0) == 'cutloss'
assert _classify_exit('cut loss', -20.0) == 'cutloss'
def test_classify_exit_tp_sl_by_profit():
assert _classify_exit('Layer #1', 15.0) == 'tp'
assert _classify_exit('Layer #1', -10.0) == 'sl'
def test_lot_tier():
assert _lot_tier(0.01) == '0.01'
assert _lot_tier(0.02) == '0.02-0.04'
assert _lot_tier(0.04) == '0.02-0.04'
assert _lot_tier(0.06) == '0.05-0.09'
assert _lot_tier(0.10) == '0.10-0.49'
assert _lot_tier(1.0) == '1.00+'
def test_session_for_hour():
assert _session_for_hour(3) == 'asian'
assert _session_for_hour(9) == 'london'
assert _session_for_hour(14) == 'london_ny_overlap'
assert _session_for_hour(18) == 'new_york'
assert _session_for_hour(23) == 'off_hours'
# ── Position pairs ─────────────────────────────────────────────────────────────
def test_position_pairs_count(deals):
pairs = position_pairs(deals)
assert isinstance(pairs, list)
assert len(pairs) > 0
def test_position_pairs_hold_minutes(deals):
pairs = position_pairs(deals)
for p in pairs:
if p['hold_minutes'] is not None:
assert p['hold_minutes'] > 0
def test_position_pairs_has_layer(deals):
pairs = position_pairs(deals)
layers = [p['layer'] for p in pairs if p['layer'] > 0]
assert len(layers) > 0
def test_position_pairs_profit_nonzero(deals):
pairs = position_pairs(deals)
for p in pairs:
assert p['profit'] != 0.0
# ── Cycle stats ────────────────────────────────────────────────────────────────
def test_cycle_stats_structure(deals):
result = cycle_stats(deals)
assert 'total_cycles' in result
assert 'win_rate' in result
assert 'avg_profit' in result
assert 'win_rate_by_depth' in result
def test_cycle_stats_total_cycles(deals):
result = cycle_stats(deals)
assert result['total_cycles'] > 0
def test_cycle_stats_win_rate_range(deals):
result = cycle_stats(deals)
assert 0.0 <= result['win_rate'] <= 100.0
def test_cycle_stats_empty():
result = cycle_stats([])
assert result['total_cycles'] == 0
# ── Exit reason breakdown ──────────────────────────────────────────────────────
def test_exit_reason_breakdown_structure(deals):
result = exit_reason_breakdown(deals)
assert isinstance(result, dict)
for reason, data in result.items():
assert 'count' in data
assert 'total_pnl' in data
assert 'avg_pnl' in data
assert data['count'] > 0
def test_exit_reason_breakdown_has_cutloss(deals):
result = exit_reason_breakdown(deals)
# fixture has 'cutloss' in comment for some deals
assert 'cutloss' in result
def test_exit_reason_breakdown_counts_match(deals):
result = exit_reason_breakdown(deals)
total_counted = sum(r['count'] for r in result.values())
closed_with_pnl = [d for d in deals
if 'out' in d.get('entry', '').lower() and d.get('profit', 0.0) != 0.0]
assert total_counted == len(closed_with_pnl)
# ── Direction bias ─────────────────────────────────────────────────────────────
def test_direction_bias_keys(deals):
result = direction_bias(deals)
assert isinstance(result, dict)
# fixture has both buy and sell
assert 'buy' in result
assert 'sell' in result
def test_direction_bias_win_rate_range(deals):
result = direction_bias(deals)
for d, s in result.items():
assert 0.0 <= s['win_rate'] <= 100.0
assert s['trades'] > 0
def test_direction_bias_buy_profitable(deals):
result = direction_bias(deals)
# fixture: buy deals net positive
assert result['buy']['total_pnl'] > 0
# ── Streak analysis ────────────────────────────────────────────────────────────
def test_streak_analysis_structure(deals):
result = streak_analysis(deals)
assert isinstance(result, dict)
for key in ('max_win_streak', 'max_loss_streak', 'current_streak', 'current_streak_type'):
assert key in result
def test_streak_analysis_nonnegative(deals):
result = streak_analysis(deals)
assert result['max_win_streak'] >= 0
assert result['max_loss_streak'] >= 0
assert result['current_streak'] >= 1
def test_streak_analysis_type_valid(deals):
result = streak_analysis(deals)
assert result['current_streak_type'] in ('win', 'loss')
def test_streak_analysis_empty():
assert streak_analysis([]) == {}
# ── Session breakdown ──────────────────────────────────────────────────────────
def test_session_breakdown_structure(deals):
result = session_breakdown(deals)
assert isinstance(result, dict)
for session, data in result.items():
assert 'trades' in data
assert 'win_rate' in data
assert 'total_pnl' in data
def test_session_breakdown_has_sessions(deals):
result = session_breakdown(deals)
# fixture has deals at 09:00, 10:00, 14:00, 15:00, 16:00 (London + London/NY)
# and 02:30, 03:15 (Asian), 20:00-21:30 (NY)
known_sessions = {'london', 'london_ny_overlap', 'asian', 'new_york'}
assert len(set(result.keys()) & known_sessions) >= 2
def test_session_breakdown_win_rate_range(deals):
result = session_breakdown(deals)
for session, data in result.items():
assert 0.0 <= data['win_rate'] <= 100.0
# ── Weekday P/L ────────────────────────────────────────────────────────────────
def test_weekday_pnl_structure(deals):
result = weekday_pnl(deals)
assert isinstance(result, list)
for entry in result:
assert 'day' in entry
assert 'pnl' in entry
assert 'trades' in entry
assert 'win_rate' in entry
def test_weekday_pnl_day_names(deals):
result = weekday_pnl(deals)
valid_days = {'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'}
for entry in result:
assert entry['day'] in valid_days
def test_weekday_pnl_has_results(deals):
result = weekday_pnl(deals)
assert len(result) >= 1
# ── Hourly P/L ─────────────────────────────────────────────────────────────────
def test_hourly_pnl_structure(deals):
result = hourly_pnl(deals)
assert isinstance(result, list)
for entry in result:
assert 'hour' in entry
assert 0 <= entry['hour'] <= 23
assert 'pnl' in entry
assert 'trades' in entry
def test_hourly_pnl_has_results(deals):
result = hourly_pnl(deals)
assert len(result) >= 1
# ── Concurrent peak ────────────────────────────────────────────────────────────
def test_concurrent_peak_structure(deals):
result = concurrent_peak(deals)
assert 'peak_open' in result
assert 'peak_time' in result
def test_concurrent_peak_at_least_one(deals):
result = concurrent_peak(deals)
assert result['peak_open'] >= 1
def test_concurrent_peak_multi_layer(deals):
# fixture has a cycle where L2 and L3 open before close → peak >= 2
result = concurrent_peak(deals)
assert result['peak_open'] >= 2
# ── Volume profile ─────────────────────────────────────────────────────────────
def test_volume_profile_structure(deals):
result = volume_profile(deals)
assert isinstance(result, list)
for entry in result:
assert 'lot_tier' in entry
assert 'pnl' in entry
assert 'trades' in entry
assert 'win_rate' in entry
def test_volume_profile_has_micro_lots(deals):
result = volume_profile(deals)
tiers = [e['lot_tier'] for e in result]
assert '0.01' in tiers
def test_volume_profile_win_rate_range(deals):
result = volume_profile(deals)
for entry in result:
assert 0.0 <= entry['win_rate'] <= 100.0
# ── build_summary with new stats ───────────────────────────────────────────────
def test_build_summary_with_streak(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
streak = streak_analysis(deals)
summary = build_summary(metrics, monthly, dd, streak=streak)
assert 'max_win_streak' in summary
assert 'max_loss_streak' in summary
assert 'current_streak_type' in summary
def test_build_summary_with_bias(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
bias = direction_bias(deals)
summary = build_summary(metrics, monthly, dd, bias=bias)
assert 'buy_win_rate' in summary or 'sell_win_rate' in summary
def test_build_summary_with_cycles(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
cycles = cycle_stats(deals)
summary = build_summary(metrics, monthly, dd, cycles=cycles)
assert 'cycle_win_rate' in summary
assert 'total_cycles' in summary
# ── Strategy profiles ──────────────────────────────────────────────────────────
def test_profiles_registry():
"""All expected strategy names are registered."""
for name in ('generic', 'grid', 'scalper', 'trend', 'hedge'):
assert name in PROFILES
p = PROFILES[name]
assert 'name' in p
assert 'exit_keywords' in p
assert 'dd_cause_keywords' in p
assert 'cycle_group_by' in p
assert 'cycle_gap_min' in p
def test_profiles_depth_re():
assert PROFILES['grid']['depth_re'] is not None
assert PROFILES['generic']['depth_re'] is None
assert PROFILES['scalper']['depth_re'] is None
assert PROFILES['trend']['depth_re'] is None
assert PROFILES['hedge']['depth_re'] is None
# ── _extract_depth ─────────────────────────────────────────────────────────────
def test_extract_depth_grid_pattern():
depth_re = PROFILES['grid']['depth_re']
assert _extract_depth('Layer #3', depth_re) == 3
assert _extract_depth('Layer #1', depth_re) == 1
assert _extract_depth('layer 7', depth_re) == 7
def test_extract_depth_no_pattern():
assert _extract_depth('Layer #3', None) == 0
assert _extract_depth('', None) == 0
def test_extract_depth_no_match():
assert _extract_depth('TP hit', PROFILES['grid']['depth_re']) == 0
# ── _classify_exit with profiles ──────────────────────────────────────────────
def test_classify_exit_grid_locking():
assert _classify_exit('locking hedge', -50.0, PROFILES['grid']) == 'locking'
def test_classify_exit_grid_cutloss():
assert _classify_exit('cutloss fired', -20.0, PROFILES['grid']) == 'cutloss'
def test_classify_exit_scalper_manual():
assert _classify_exit('manual close', -5.0, PROFILES['scalper']) == 'manual'
def test_classify_exit_scalper_trailing():
assert _classify_exit('trailing stop', 10.0, PROFILES['scalper']) == 'trailing'
def test_classify_exit_trend_breakeven():
assert _classify_exit('breakeven stop', 0.5, PROFILES['trend']) == 'breakeven'
def test_classify_exit_trend_partial():
assert _classify_exit('partial scale out', 15.0, PROFILES['trend']) == 'partial'
def test_classify_exit_hedge_net_close():
assert _classify_exit('net close', -30.0, PROFILES['hedge']) == 'net_close'
def test_classify_exit_generic_fallback():
"""Generic profile has no keywords — falls back to profit sign."""
assert _classify_exit('Layer #3 locking', -50.0, PROFILES['generic']) == 'sl'
assert _classify_exit('Layer #1', 15.0, PROFILES['generic']) == 'tp'
# ── _classify_dd_cause ────────────────────────────────────────────────────────
def test_classify_dd_cause_grid():
assert _classify_dd_cause('locking total', PROFILES['grid']) == 'locking_cascade'
assert _classify_dd_cause('cutloss fired', PROFILES['grid']) == 'cutloss'
assert _classify_dd_cause('zombie exit', PROFILES['grid']) == 'zombie_exit'
def test_classify_dd_cause_generic_unknown():
assert _classify_dd_cause('locking total', PROFILES['generic']) == 'unknown'
assert _classify_dd_cause('', PROFILES['generic']) == 'unknown'
def test_classify_dd_cause_scalper_stop():
assert _classify_dd_cause('sl hit', PROFILES['scalper']) == 'stop_loss'
def test_classify_dd_cause_trend_whipsaw():
assert _classify_dd_cause('stop loss', PROFILES['trend']) == 'whipsaw'
# ── depth_histogram with profiles ─────────────────────────────────────────────
def test_depth_histogram_grid_returns_layers(deals):
result = depth_histogram(deals, PROFILES['grid'])
assert isinstance(result, dict)
assert 'L1' in result
assert result['L1'] > 0
def test_depth_histogram_generic_returns_empty(deals):
"""Generic profile has no depth_re → empty dict."""
result = depth_histogram(deals, PROFILES['generic'])
assert result == {}
def test_depth_histogram_scalper_returns_empty(deals):
result = depth_histogram(deals, PROFILES['scalper'])
assert result == {}
def test_grid_depth_histogram_is_alias(deals):
"""grid_depth_histogram must equal depth_histogram with grid profile."""
assert grid_depth_histogram(deals) == depth_histogram(deals, PROFILES['grid'])
# ── cycle_stats with profiles ─────────────────────────────────────────────────
def test_cycle_stats_grid_profile(deals):
result = cycle_stats(deals, PROFILES['grid'])
assert result['total_cycles'] > 0
assert 0.0 <= result['win_rate'] <= 100.0
def test_cycle_stats_scalper_profile(deals):
"""Scalper uses magic-only grouping and 10-min gap."""
result = cycle_stats(deals, PROFILES['scalper'])
assert 'total_cycles' in result
assert result['total_cycles'] > 0
def test_cycle_stats_generic_profile(deals):
result = cycle_stats(deals, PROFILES['generic'])
assert 'total_cycles' in result
def test_cycle_stats_scalper_vs_grid_differ(deals):
"""Different grouping rules can produce different cycle counts."""
grid_result = cycle_stats(deals, PROFILES['grid'])
scalper_result = cycle_stats(deals, PROFILES['scalper'])
# Both must be valid; counts may differ due to grouping
assert grid_result['total_cycles'] >= 0
assert scalper_result['total_cycles'] >= 0
# ── exit_reason_breakdown with profiles ───────────────────────────────────────
def test_exit_reason_breakdown_grid(deals):
result = exit_reason_breakdown(deals, PROFILES['grid'])
assert 'cutloss' in result # fixture has "cutloss" in comments
def test_exit_reason_breakdown_generic_only_tp_sl(deals):
"""Generic profile has no keywords → only 'tp' and 'sl' keys."""
result = exit_reason_breakdown(deals, PROFILES['generic'])
for reason in result:
assert reason in ('tp', 'sl'), f"Unexpected reason '{reason}' from generic profile"
def test_exit_reason_breakdown_scalper_keywords(deals):
"""Scalper profile recognises 'cutloss' comment as 'manual' (not 'cutloss')."""
result = exit_reason_breakdown(deals, PROFILES['scalper'])
# 'cutloss' is not a scalper keyword → falls back to profit-sign → 'sl'
assert 'cutloss' not in result
def test_exit_reason_breakdown_counts_sum(deals):
"""Total count must equal number of non-zero closed deals, regardless of profile."""
closed = [d for d in deals
if 'out' in d.get('entry', '').lower() and d.get('profit', 0.0) != 0.0]
for profile in PROFILES.values():
result = exit_reason_breakdown(deals, profile)
assert sum(r['count'] for r in result.values()) == len(closed)
# ── reconstruct_dd_events with profiles ───────────────────────────────────────
def test_dd_events_cause_generic_unknown(deals):
"""Generic profile → all causes must be 'unknown'."""
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
events = reconstruct_dd_events(deals, metrics, PROFILES['generic'])
for ev in events:
assert ev['cause'] == 'unknown'
def test_dd_events_cause_grid_classified(deals):
"""Grid profile → cause is classified from comment keywords."""
metrics = {'net_profit': 38.0, 'max_dd_pct': 5.0}
events = reconstruct_dd_events(deals, metrics, PROFILES['grid'])
valid = {'locking_cascade', 'cutloss', 'zombie_exit', 'spike_entry', 'unknown'}
for ev in events:
assert ev['cause'] in valid
# ── build_summary strategy field ──────────────────────────────────────────────
def test_build_summary_strategy_field(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
summary = build_summary(metrics, monthly, dd, strategy='scalper')
assert summary['strategy'] == 'scalper'
def test_build_summary_no_strategy_field(deals):
metrics = {'net_profit': 38.0, 'max_dd_pct': 1.5}
monthly = monthly_pnl(deals)
dd = reconstruct_dd_events(deals, metrics)
summary = build_summary(metrics, monthly, dd)
assert 'strategy' not in summary
tests/test_extract.py
-146
View File
@@ -1,146 +0,0 @@
"""Tests for analytics/extract.py — runs without MT5 or Wine."""
import json
import os
import sys
import tempfile
from pathlib import Path
import pytest
FIXTURES = Path(__file__).parent / 'fixtures'
sys.path.insert(0, str(Path(__file__).parent.parent))
from analytics.extract import (
detect_format, parse_html, parse_xml, write_outputs,
_parse_metrics_html, _parse_deals_html,
)
def test_detect_format_html():
assert detect_format(str(FIXTURES / 'sample_report.htm')) == 'html'
def test_detect_format_xml():
assert detect_format(str(FIXTURES / 'sample_report.htm.xml')) == 'xml'
# HTML parsing is tested via internal functions to avoid the UTF-16 decode dance
# (read_text tries UTF-16 first, which silently garbles plain UTF-8/ASCII files).
# The fixture is used for format-detection only.
HTML_TEXT = """<html><body>
<table>
<tr><td>Net profit</td><td>1234.56</td></tr>
<tr><td>Profit factor</td><td>1.25</td></tr>
<tr><td>Maximal drawdown</td><td>500.00 (5.00%)</td></tr>
<tr><td>Sharpe Ratio</td><td>0.75</td></tr>
<tr><td>Total trades</td><td>150</td></tr>
<tr><td>Recovery factor</td><td>2.50</td></tr>
<tr><td>Profit trades (% of total)</td><td>90 (60.00%)</td></tr>
<tr><td>Gross profit</td><td>2000.00</td></tr>
<tr><td>Gross loss</td><td>-765.44</td></tr>
</table>
<table>
<tr><td>Deal Time</td><td>Type</td><td>Direction</td><td>Volume</td><td>Price</td><td>S/L</td><td>T/P</td><td>Profit</td><td>Balance</td><td>Comment</td><td>Order</td><td>Magic</td><td>Entry</td></tr>
<tr><td>2025.01.10 09:30:00</td><td>buy</td><td>out</td><td>0.01</td><td>1915.00</td><td>0</td><td>0</td><td>15.00</td><td>10015.00</td><td>Layer #1</td><td>1001</td><td>12345</td><td>out</td></tr>
<tr><td>2025.02.05 14:00:00</td><td>sell</td><td>out</td><td>0.01</td><td>1945.00</td><td>0</td><td>0</td><td>-15.00</td><td>10020.00</td><td>Layer #1</td><td>1005</td><td>12345</td><td>out</td></tr>
</table>
</body></html>"""
@pytest.fixture
def html_report_path(tmp_path):
"""Write HTML fixture as UTF-16 LE with BOM so read_text() decodes it correctly."""
p = tmp_path / 'report.htm'
p.write_bytes(b'\xff\xfe' + HTML_TEXT.encode('utf-16-le'))
return str(p)
def test_parse_html_returns_metrics(html_report_path):
metrics, _ = parse_html(html_report_path)
assert isinstance(metrics, dict)
assert 'net_profit' in metrics
assert metrics['net_profit'] == pytest.approx(1234.56)
assert metrics['total_trades'] == 150
def test_parse_html_returns_deals(html_report_path):
_, deals = parse_html(html_report_path)
assert isinstance(deals, list)
assert len(deals) >= 1
deal = deals[0]
assert 'profit' in deal
assert 'balance' in deal
def test_parse_metrics_html_directly():
"""Test HTML metric extraction without encoding layer."""
metrics = _parse_metrics_html(HTML_TEXT)
assert metrics['net_profit'] == pytest.approx(1234.56)
assert metrics['profit_factor'] == pytest.approx(1.25)
assert metrics['max_dd_pct'] == pytest.approx(5.00)
assert metrics['total_trades'] == 150
def test_parse_deals_html_directly():
"""Test HTML deal extraction without encoding layer."""
deals = _parse_deals_html(HTML_TEXT)
assert len(deals) == 2
assert float(deals[0]['profit']) == pytest.approx(15.00)
assert float(deals[1]['profit']) == pytest.approx(-15.00)
def test_parse_xml_returns_metrics():
metrics, deals = parse_xml(str(FIXTURES / 'sample_report.htm.xml'))
assert isinstance(metrics, dict)
assert 'net_profit' in metrics
assert metrics['net_profit'] == pytest.approx(1234.56)
assert metrics['total_trades'] == 150
def test_parse_xml_returns_deals():
metrics, deals = parse_xml(str(FIXTURES / 'sample_report.htm.xml'))
assert isinstance(deals, list)
assert len(deals) >= 1
deal = deals[0]
assert deal.get('profit') is not None
def test_write_outputs_creates_files():
metrics = {'net_profit': 100.0, 'total_trades': 5}
deals = [
{'time': '2025.01.10', 'type': 'buy', 'direction': 'out', 'volume': '0.01',
'price': '1900', 'sl': '0', 'tp': '0', 'profit': '10.00',
'balance': '10010', 'comment': '', 'order': '1', 'magic': '1', 'entry': 'out'},
]
with tempfile.TemporaryDirectory() as tmp:
paths = write_outputs(metrics, deals, tmp)
assert Path(paths['metrics']).exists()
assert Path(paths['deals_csv']).exists()
assert Path(paths['deals_json']).exists()
# Verify metrics.json content
with open(paths['metrics']) as f:
saved = json.load(f)
assert saved['net_profit'] == 100.0
def test_parse_html_skips_balance_rows():
"""Rows with type='balance' should be filtered out."""
html = """
<table>
<tr><td>Deal Time</td><td>Type</td><td>Direction</td><td>Volume</td><td>Price</td><td>S/L</td><td>T/P</td><td>Profit</td><td>Balance</td><td>Comment</td><td>Order</td><td>Magic</td><td>Entry</td></tr>
<tr><td>2025.01.10 09:30:00</td><td>balance</td><td></td><td>0</td><td>0</td><td>0</td><td>0</td><td>0</td><td>10000</td><td></td><td>0</td><td>0</td><td></td></tr>
<tr><td>2025.01.10 10:00:00</td><td>buy</td><td>out</td><td>0.01</td><td>1910</td><td>0</td><td>0</td><td>5.00</td><td>10005</td><td>Layer #1</td><td>1</td><td>1</td><td>out</td></tr>
</table>
"""
import tempfile, os
with tempfile.NamedTemporaryFile(mode='w', suffix='.htm', delete=False) as f:
f.write(html)
path = f.name
try:
_, deals = parse_html(path)
types = [d.get('type', '').lower() for d in deals]
assert 'balance' not in types
finally:
os.unlink(path)