from __future__ import annotations import argparse import csv import json import math import os import re import shlex import subprocess import time import ast from dataclasses import dataclass, field, asdict from pathlib import Path from statistics import median from typing import Any Range = tuple[Any, Any, Any] MQL_ENUM_VALUES = { "PERIOD_M1": 1, "PERIOD_M2": 2, "PERIOD_M3": 3, "PERIOD_M4": 4, "PERIOD_M5": 5, "PERIOD_M6": 6, "PERIOD_M10": 10, "PERIOD_M12": 12, "PERIOD_M15": 15, "PERIOD_M20": 20, "PERIOD_M30": 30, "PERIOD_H1": 16385, "PERIOD_H2": 16386, "PERIOD_H3": 16387, "PERIOD_H4": 16388, "PERIOD_H6": 16390, "PERIOD_H8": 16392, "PERIOD_H12": 16396, "PERIOD_D1": 16408, "PERIOD_W1": 32769, "PERIOD_MN1": 49153, } @dataclass class Subgroup: tag: str title: str group: str = "Default" ranges: dict[str, Range] = field(default_factory=dict) fixed: dict[str, Any] = field(default_factory=dict) @dataclass class OptimizerConfig: terminal_exe: str data_path: str expert_name: str csv_prefix: str run_dir: str symbols: list[str] from_date: str to_date: str period: str model: int = 1 optimization_criterion: int = 6 deposit: float = 10000 currency: str = "USD" leverage: str = "1:100" genetic_threshold_passes: int = 500 max_boundary_expansions_per_subgroup: int = 1 timeout_seconds: int = 7200 close_terminal_before_run: bool = True selection_mode: str = "semi_auto" # auto | semi_auto | manual top_sets_to_show: int = 20 pause_after_each_subgroup: bool = True allow_user_override: bool = True base: dict[str, Any] = field(default_factory=dict) subgroups: list[dict[str, Any]] = field(default_factory=list) expand_limits: dict[str, tuple[float, float]] = field(default_factory=dict) INPUT_RE = re.compile( r"^\s*input\s+(?Pbool|int|long|double|float|string|ENUM_[A-Za-z0-9_]+|[A-Za-z0-9_]+)\s+" r"(?P[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?P[^;]+);" ) def parse_value(value: str) -> Any: v = str(value).strip() if len(v) >= 2 and ((v[0] == v[-1] == '"') or (v[0] == v[-1] == "'")): return v[1:-1] lv = v.lower() if lv == "true": return True if lv == "false": return False try: if "." not in v and "," not in v: return int(v) return float(v.replace(",", ".")) except ValueError: arithmetic = re.fullmatch(r"[0-9\s\+\-\*\/\(\)\.]+", v) if arithmetic: try: return eval_arithmetic(v) except Exception: pass return v def eval_arithmetic(expr: str) -> int | float: allowed = ( ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.USub, ast.UAdd, ) tree = ast.parse(expr, mode="eval") if not all(isinstance(node, allowed) for node in ast.walk(tree)): raise ValueError(expr) value = eval(compile(tree, "", "eval"), {"__builtins__": {}}, {}) if abs(value - round(value)) < 1e-9: return int(round(value)) return float(value) def mt5_value(value: Any) -> str: if isinstance(value, bool): return "true" if value else "false" if isinstance(value, float): return f"{value:.8g}" if isinstance(value, str) and value in MQL_ENUM_VALUES: return str(MQL_ENUM_VALUES[value]) return str(value) def as_float(value: Any, default: float = 0.0) -> float: try: return float(str(value).replace(",", ".")) except (TypeError, ValueError): return default def parse_kv_line(line: str) -> dict[str, Any]: clean = line.strip().replace("//", "", 1).strip() for prefix in ("@opt", "@fixed", "@system"): if clean.startswith(prefix): clean = clean[len(prefix):].strip() break lexer = shlex.shlex(clean, posix=True) lexer.whitespace_split = True lexer.commenters = "" result: dict[str, Any] = {} for token in lexer: if "=" not in token: continue k, v = token.split("=", 1) result[k.strip()] = parse_value(v.strip()) return result INCLUDE_RE = re.compile(r'^\s*#include\s+[<"](?P[^>"]+)[>"]') def collect_mq5_lines(path: Path, visited: set[Path] | None = None) -> list[tuple[Path, str]]: """ Reads the main .mq5 file and locally included #include files, when they exist. """ visited = visited or set() path = path.resolve() if path in visited: return [] visited.add(path) if not path.exists(): return [] try: raw_lines = path.read_text(encoding="utf-8", errors="ignore").splitlines() except Exception: raw_lines = path.read_text(encoding="latin1", errors="ignore").splitlines() result: list[tuple[Path, str]] = [] for raw in raw_lines: result.append((path, raw)) match = INCLUDE_RE.match(raw.strip()) if not match: continue include_name = match.group("file").replace("\\", "/") candidates = [ path.parent / include_name, path.parent / Path(include_name).name, path.parent.parent / include_name, path.parent.parent / Path(include_name).name, ] for candidate in candidates: if candidate.exists(): result.extend(collect_mq5_lines(candidate, visited)) break return result ENUM_DECL_RE = re.compile(r"\benum\s+[A-Za-z_][A-Za-z0-9_]*\s*\{") ENUM_ITEM_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*(?:=\s*([-+]?\d+))?\s*,?\s*(?://.*)?$") def update_enum_values_from_lines(lines: list[tuple[Path, str]]) -> None: """ Reads simple MQL5 enums and registers name -> value mappings for numeric .set output. """ in_enum = False next_value = 0 for _, raw in lines: line = raw.strip() if not in_enum: if ENUM_DECL_RE.search(line): in_enum = True next_value = 0 after_brace = line.split("{", 1)[1] if after_brace: line = after_brace else: continue else: continue if "}" in line: before = line.split("}", 1)[0].strip() if before: for part in before.split(","): registered = register_enum_item(part, next_value) if registered is not None: next_value = registered + 1 in_enum = False continue for part in line.split(","): registered = register_enum_item(part, next_value) if registered is not None: next_value = registered + 1 def register_enum_item(text: str, default_value: int) -> int | None: clean = text.strip() if not clean: return None match = ENUM_ITEM_RE.match(clean) if not match: return None name = match.group(1) explicit = match.group(2) value = int(explicit) if explicit is not None else default_value MQL_ENUM_VALUES[name] = value return value def parse_simple_opt_values(text: str) -> tuple[Any, Any, Any] | None: """ Reads compact formats: // @opt 10 2 40 // @opt false true For bool values with two entries, step is assumed as 1: false, 1, true """ clean = text.strip() clean = clean.replace("//", "", 1).strip() if not clean.startswith("@opt"): return None clean = clean[len("@opt"):].strip() if not clean: return None parts = clean.split() if len(parts) == 2: start_v = parse_value(parts[0]) stop_v = parse_value(parts[1]) return (start_v, 1, stop_v) if len(parts) >= 3: start_v = parse_value(parts[0]) step_v = parse_value(parts[1]) stop_v = parse_value(parts[2]) return (start_v, step_v, stop_v) return None def parse_mq5_annotations(mq5_path: Path) -> tuple[dict[str, Any], list[Subgroup]]: """ Flexible parser for two annotation styles. Style 1, explicit: // @opt group=Regime subgroup=G1A_EMAS label="EMAs" start=10 step=2 stop=40 input int fastEMAPeriod = 24; Style 2, compact: // @group G1_REGIME // @subgroup G1B_EMAS // @opt 10 2 40 input int fastEMAPeriod = 24; // @fixed input ulong magicBuy = 1003; Rules: - @group defines the current group. - @subgroup defines the current subgroup. - @opt with 3 values becomes start/step/stop. - @opt with 2 values becomes start/1/stop, useful for bool and simple enums. - @fixed immediately above an input adds that input to the current subgroup fixed values. - @opt in key=value format remains supported. """ if not mq5_path.exists(): raise FileNotFoundError(mq5_path) base: dict[str, Any] = {} subgroup_map: dict[str, Subgroup] = {} current_group = "Default" current_subgroup = "DEFAULT" current_title = "DEFAULT" pending_opt: dict[str, Any] | None = None pending_fixed = False lines = collect_mq5_lines(mq5_path) update_enum_values_from_lines(lines) for source_path, raw in lines: stripped = raw.strip() if not stripped: continue # ------------------------------------------------------------ # Compact format: // @group G1_REGIME # ------------------------------------------------------------ if stripped.startswith("// @group"): clean = stripped.replace("//", "", 1).strip() parts = clean.split(maxsplit=1) if len(parts) == 2: current_group = parts[1].strip() continue # ------------------------------------------------------------ # Compact format: // @subgroup G1B_EMAS # ------------------------------------------------------------ if stripped.startswith("// @subgroup"): clean = stripped.replace("//", "", 1).strip() parts = clean.split(maxsplit=1) if len(parts) == 2: current_subgroup = parts[1].strip() current_title = current_subgroup subgroup_map.setdefault( current_subgroup, Subgroup(tag=current_subgroup, title=current_title, group=current_group) ) continue # ------------------------------------------------------------ # Full @fixed with key=value: # // @fixed subgroup=G1A timeframeRegime=5 # ------------------------------------------------------------ if stripped.startswith("// @fixed") and "=" in stripped: fixed = parse_kv_line(stripped) subgroup_tag = str(fixed.pop("subgroup", current_subgroup)).strip() group = str(fixed.pop("group", current_group)).strip() title = str(fixed.pop("label", subgroup_tag)).strip() if subgroup_tag: sg = subgroup_map.setdefault( subgroup_tag, Subgroup(tag=subgroup_tag, title=title, group=group) ) if sg.group == "Default" and group != "Default": sg.group = group if not sg.title or sg.title == subgroup_tag: sg.title = title sg.fixed.update(fixed) pending_fixed = False continue # ------------------------------------------------------------ # Compact @fixed: # // @fixed # input Type parameter = value; # ------------------------------------------------------------ if stripped == "// @fixed" or stripped.startswith("// @fixed"): pending_fixed = True pending_opt = None continue # ------------------------------------------------------------ # @opt: accepts both key=value and compact formats # ------------------------------------------------------------ if stripped.startswith("// @opt"): if "=" in stripped: parsed = parse_kv_line(stripped) pending_opt = parsed else: simple = parse_simple_opt_values(stripped) if simple is not None: start_v, step_v, stop_v = simple pending_opt = { "group": current_group, "subgroup": current_subgroup, "label": current_subgroup, "start": start_v, "step": step_v, "stop": stop_v, } pending_fixed = False continue match = INPUT_RE.match(raw) if match: name = match.group("name") default_value = parse_value(match.group("value").strip()) base[name] = default_value # Optimizable input if pending_opt: subgroup_tag = str(pending_opt.get("subgroup", current_subgroup)).strip() group = str(pending_opt.get("group", current_group)).strip() title = str(pending_opt.get("label", subgroup_tag)).strip() start_v = pending_opt.get("start", default_value) step_v = pending_opt.get("step", 1) stop_v = pending_opt.get("stop", default_value) sg = subgroup_map.setdefault( subgroup_tag, Subgroup(tag=subgroup_tag, title=title, group=group) ) if sg.group == "Default" and group != "Default": sg.group = group if not sg.title or sg.title == subgroup_tag: sg.title = title sg.ranges[name] = (start_v, step_v, stop_v) pending_opt = None pending_fixed = False continue # Fixed input for the current subgroup if pending_fixed: sg = subgroup_map.setdefault( current_subgroup, Subgroup(tag=current_subgroup, title=current_title, group=current_group) ) sg.fixed[name] = default_value pending_fixed = False continue continue # If a pending @opt exists and another code line appears before the input, # discard it to avoid assigning the metadata to the wrong input. if pending_opt and not stripped.startswith("//"): pending_opt = None if pending_fixed and not stripped.startswith("//"): pending_fixed = False # Remove empty subgroups that have neither ranges nor fixed values. subgroups = [ sg for sg in subgroup_map.values() if sg.ranges or sg.fixed ] return base, subgroups def parse_set_file(set_path: Path) -> dict[str, Any]: defaults: dict[str, Any] = {} if not set_path or not set_path.exists(): return defaults for enc in ("utf-16", "utf-8", "ascii", "latin1"): try: lines = set_path.read_text(encoding=enc, errors="ignore").splitlines() local: dict[str, Any] = {} for line in lines: if not line or line.strip().startswith(";") or "=" not in line: continue key, rest = line.split("=", 1) value = rest.split("||", 1)[0] local[key.strip()] = parse_value(value.strip()) if local: defaults.update(local) break except Exception: continue return defaults def ask(prompt: str, default: Any | None = None) -> str: suffix = f" [{default}]" if default not in (None, "") else "" value = input(f"{prompt}{suffix}: ").strip() if not value and default is not None: return str(default) return value def ask_bool(prompt: str, default: bool = False) -> bool: d = "y" if default else "n" value = ask(prompt + " (y/n)", d).lower() return value in {"y", "yes", "1", "true"} def detect_csv_prefix_from_base(base: dict[str, Any], expert_name: str) -> str: """ Detects the CSV prefix from the csvExportPrefix input read from the .mq5 file. If the input does not exist or is empty, use the EA name. """ value = base.get("csvExportPrefix", "") if isinstance(value, str) and value.strip(): return value.strip() return expert_name def validate_required_optimizer_inputs(base: dict[str, Any]) -> list[str]: """ Checks whether the EA has the minimum inputs expected by the automation. """ required = ["optimizationTag", "saveOptimizationPasses", "dummyOptimizationPass", "csvExportPrefix"] return [name for name in required if name not in base] def detect_hardcoded_csv_prefix(mq5_path: Path) -> str | None: """ Detects the prefix in EAs that write CSV files with this literal pattern: "PREFIX_" + optimizationTag + "_frames.csv". """ if not mq5_path.exists(): return None text = mq5_path.read_text(encoding="utf-8", errors="ignore") match = re.search( r'"(?P[A-Za-z0-9_\-]+)_"\s*\+\s*optimizationTag\s*\+\s*"_frames\.csv"', text, ) if match: return match.group("prefix") return None def optimizer_dir_from_config_path(config_path: Path | None, cfg: OptimizerConfig) -> Path: if config_path: return config_path.resolve().parent return Path(cfg.data_path) / "MQL5" / "Experts" / "MT5_Global_Optimizer" def infer_data_path_from_mq5(mq5_path: Path) -> str | None: for parent in mq5_path.resolve().parents: if parent.name.upper() == "MQL5": return str(parent.parent) return None def infer_data_path_from_config(config_path: Path | None, expert_name: str) -> str | None: if not config_path: return None for parent in config_path.resolve().parents: if parent.name.upper() == "MQL5": data_path = parent.parent mq5_path = data_path / "MQL5" / "Experts" / f"{expert_name}.mq5" ex5_path = data_path / "MQL5" / "Experts" / f"{expert_name}.ex5" if mq5_path.exists() or ex5_path.exists(): return str(data_path) return None def default_terminal_exe() -> str: candidates = [ Path(r"C:\Program Files\MetaTrader 5\terminal64.exe"), Path(r"C:\Program Files\MetaTrader 5 Terminal\terminal64.exe"), ] for candidate in candidates: if candidate.exists(): return str(candidate) return str(candidates[0]) def default_data_path_hint() -> str: appdata = os.environ.get("APPDATA") if appdata: return str(Path(appdata) / "MetaQuotes" / "Terminal" / "YOUR_TERMINAL_ID") return str(Path.home() / "AppData" / "Roaming" / "MetaQuotes" / "Terminal" / "YOUR_TERMINAL_ID") def normalize_config(cfg: OptimizerConfig, config_path: Path | None = None) -> OptimizerConfig: """ Cleans manually created or copy-pasted configs and keeps the .set compatible only with real EA inputs. """ inferred_data_path = infer_data_path_from_config(config_path, cfg.expert_name) if inferred_data_path and Path(inferred_data_path).resolve() != Path(cfg.data_path).resolve(): print(f"Warning: config data_path points to another terminal. Using: {inferred_data_path}", flush=True) cfg.data_path = inferred_data_path data_path = Path(cfg.data_path) mq5_path = data_path / "MQL5" / "Experts" / f"{cfg.expert_name}.mq5" if mq5_path.exists(): parsed_base, parsed_subgroups = parse_mq5_annotations(mq5_path) valid_inputs = set(parsed_base) clean_base = { k: parse_value(v) if isinstance(v, str) else v for k, v in cfg.base.items() if k in valid_inputs } for key, value in parsed_base.items(): clean_base.setdefault(key, value) cfg.base = clean_base clean_subgroups: list[dict[str, Any]] = [] source_subgroups = cfg.subgroups or [asdict(sg) for sg in parsed_subgroups] for sg in source_subgroups: ranges = {k: v for k, v in sg.get("ranges", {}).items() if k in valid_inputs} fixed = {k: v for k, v in sg.get("fixed", {}).items() if k in valid_inputs} if ranges or fixed: item = dict(sg) item["ranges"] = ranges item["fixed"] = fixed clean_subgroups.append(item) if clean_subgroups: cfg.subgroups = clean_subgroups hardcoded_prefix = detect_hardcoded_csv_prefix(mq5_path) if hardcoded_prefix: cfg.csv_prefix = hardcoded_prefix else: cfg.csv_prefix = detect_csv_prefix_from_base(cfg.base, cfg.expert_name) run_dir = Path(cfg.run_dir) if not run_dir.is_absolute() or run_dir.suffix.lower() == ".csv": cfg.run_dir = str(optimizer_dir_from_config_path(config_path, cfg) / "runs" / f"{cfg.expert_name}_global") return cfg def wizard() -> None: print("\n=== MT5 Global Optimizer Wizard ===") print("Creates a config.json manually or by reading @opt/@fixed comments from the EA.\n") mq5_file = ask("Path to the .mq5 file for automatic @opt/@fixed reading, or press ENTER to skip", "") inferred_data_path = infer_data_path_from_mq5(Path(mq5_file)) if mq5_file else None terminal_exe = ask("Path to terminal64.exe", default_terminal_exe()) data_path = ask("MT5 Data Path", inferred_data_path or default_data_path_hint()) base: dict[str, Any] = {} subgroups: list[Subgroup] = [] if mq5_file: parsed_base, parsed_subgroups = parse_mq5_annotations(Path(mq5_file)) base.update(parsed_base) subgroups.extend(parsed_subgroups) print(f"OK: {len(base)} inputs read and {len(subgroups)} subgroups detected in the .mq5 file.") missing_required = validate_required_optimizer_inputs(base) if missing_required: print("Warning: required inputs were not found in the EA:", ", ".join(missing_required)) print("Add these inputs for full optimizer integration.") set_file = ask("Path to the base .set file to override defaults, or press ENTER to skip", "") if set_file: set_defaults = parse_set_file(Path(set_file)) base.update(set_defaults) print(f"OK: {len(set_defaults)} defaults imported from the .set file.") expert_name = ask("Expert Advisor name in the Strategy Tester", Path(mq5_file).stem if mq5_file else "MyEA") csv_prefix = (detect_hardcoded_csv_prefix(Path(mq5_file)) if mq5_file else None) or detect_csv_prefix_from_base(base, expert_name) run_dir = str(Path.cwd() / "runs" / f"{expert_name}_global") print(f"CSV prefix automatically detected: {csv_prefix}") print(f"Expected CSV file: {csv_prefix}__frames.csv") print(f"Run folder automatically set to: {run_dir}") symbols = [x.strip() for x in ask("Symbols separated by commas", "EURUSD,GBPUSD,USDJPY").split(",") if x.strip()] from_date = ask("Start date YYYY.MM.DD", "2022.01.01") to_date = ask("End date YYYY.MM.DD", "2025.12.31") period = ask("Test timeframe", "M5") model = int(ask("Tester model: 0=Every tick, 1=1min OHLC, 2=Open prices", "1")) currency = ask("Tester account currency", "USD").strip().upper() or "USD" timeout_seconds = int(ask("Timeout per run in seconds", "7200")) print("\nSet selection mode:") print(" auto = apply the recommended set automatically") print(" manual = always show TOP sets and ask") print(" semi_auto = ask only when the decision is sensitive") selection_mode = ask("Selection mode", "semi_auto").strip().lower() if selection_mode not in {"auto", "manual", "semi_auto"}: selection_mode = "semi_auto" top_sets_to_show = int(ask("How many TOP sets to show when selection is required", "20")) if not subgroups or ask_bool("Do you also want to add subgroups manually", False): n = int(ask("How many subgroups do you want to register manually", "0")) for i in range(n): print(f"\nManual subgroup {i+1}/{n}") group = ask("Main group", "Manual") tag = ask("Subgroup tag", f"G{i+1}") title = ask("Title", tag) ranges: dict[str, Range] = {} fixed: dict[str, Any] = {} pcount = int(ask("How many optimizable parameters in this subgroup", "1")) for _ in range(pcount): name = ask("Input name") start = parse_value(ask("Start")) step = parse_value(ask("Step")) stop = parse_value(ask("Stop")) ranges[name] = (start, step, stop) fixed_text = ask("Fixed values for this subgroup in the format a=1 b=true, or press ENTER", "") if fixed_text: lexer = shlex.shlex(fixed_text, posix=True) lexer.whitespace_split = True lexer.commenters = "" for token in lexer: if "=" in token: k, v = token.split("=", 1) fixed[k] = parse_value(v) subgroups.append(Subgroup(tag=tag, title=title, group=group, ranges=ranges, fixed=fixed)) base.setdefault("optimizationTag", "manual") base.setdefault("saveOptimizationPasses", True) base.setdefault("dummyOptimizationPass", 0) if "csvExportPrefix" in base: base.setdefault("csvExportPrefix", csv_prefix) config = OptimizerConfig( terminal_exe=terminal_exe, data_path=data_path, expert_name=expert_name, csv_prefix=csv_prefix, run_dir=run_dir, symbols=symbols, from_date=from_date, to_date=to_date, period=period, model=model, currency=currency, timeout_seconds=timeout_seconds, selection_mode=selection_mode, top_sets_to_show=top_sets_to_show, base=base, subgroups=[asdict(sg) for sg in subgroups], expand_limits={}, ) out = Path(ask("Output config.json name", f"config_{expert_name}_global.json")) config = normalize_config(config, out) out.write_text(json.dumps(asdict(config), indent=2, ensure_ascii=False), encoding="utf-8") print(f"\nOK: configuration saved to {out.resolve()}") def load_config(path: Path) -> OptimizerConfig: data = json.loads(path.read_text(encoding="utf-8-sig")) return normalize_config(OptimizerConfig(**data), path) def subgroup_from_dict(data: dict[str, Any]) -> Subgroup: ranges = {k: tuple(v) for k, v in data.get("ranges", {}).items()} return Subgroup( tag=data["tag"], title=data.get("title", data["tag"]), group=data.get("group", "Default"), ranges=ranges, fixed=data.get("fixed", {}), ) def estimate_passes(ranges: dict[str, Range]) -> int: total = 1 for start, step, stop in ranges.values(): if isinstance(start, bool) or isinstance(stop, bool): total *= 2 continue step_f = abs(as_float(step, 1.0)) or 1.0 total *= max(1, int(math.floor((as_float(stop) - as_float(start)) / step_f + 1.000001))) return total def paths(cfg: OptimizerConfig) -> dict[str, Path]: data_path = Path(cfg.data_path) return { "data_path": data_path, "tester_dir": data_path / "MQL5" / "Profiles" / "Tester", "files_dir": data_path / "MQL5" / "Files", "cache_dir": data_path / "Tester" / "cache", "run_dir": Path(cfg.run_dir), "terminal_exe": Path(cfg.terminal_exe), } def write_set(cfg: OptimizerConfig, tag: str, params: dict[str, Any], subgroup: Subgroup, baseline: bool = False) -> Path: p = paths(cfg) p["tester_dir"].mkdir(parents=True, exist_ok=True) merged = {**params, **subgroup.fixed, "optimizationTag": tag, "saveOptimizationPasses": True} set_path = p["tester_dir"] / f"{cfg.expert_name}_{tag}.set" lines: list[str] = [] for key, value in merged.items(): if key == "optimizationTag": lines.append(f"{key}={value}") elif baseline and key == "dummyOptimizationPass": lines.append(f"{key}=0||0||1||1||Y") elif not baseline and key in subgroup.ranges: start, step, stop = subgroup.ranges[key] lines.append(f"{key}={mt5_value(value)}||{mt5_value(start)}||{mt5_value(step)}||{mt5_value(stop)}||Y") else: lines.append(f"{key}={mt5_value(value)}||0||0||0||N") set_path.write_text("\n".join(lines) + "\n", encoding="ascii", errors="ignore") return set_path def write_ini(cfg: OptimizerConfig, symbol: str, tag: str, set_path: Path, genetic: bool) -> Path: p = paths(cfg) p["run_dir"].mkdir(parents=True, exist_ok=True) ini_path = p["run_dir"] / f"{tag}.ini" optimization = 2 if genetic else 1 report_path = p["run_dir"] / tag ini = f"""[Experts] AllowLiveTrading=0 AllowDllImport=0 Enabled=1 [Tester] Expert={cfg.expert_name} ExpertParameters={set_path.name} Symbol={symbol} Period={cfg.period} Login= Model={cfg.model} ExecutionMode=0 Optimization={optimization} OptimizationCriterion={cfg.optimization_criterion} FromDate={cfg.from_date} ToDate={cfg.to_date} ForwardMode=0 Report={report_path} ReplaceReport=1 ShutdownTerminal=1 Deposit={cfg.deposit} Currency={cfg.currency} Leverage={cfg.leverage} UseLocal=1 UseRemote=0 UseCloud=0 Visual=0 """ ini_path.write_text(ini, encoding="ascii", errors="ignore") return ini_path def clean_cache(cfg: OptimizerConfig, symbol: str) -> None: p = paths(cfg) pattern = f"{cfg.expert_name}.{symbol}.{cfg.period}.{cfg.from_date.replace('.', '')}.{cfg.to_date.replace('.', '')}*.opt" for file in p["cache_dir"].glob(pattern): try: file.unlink() except OSError: pass def run_mt5(cfg: OptimizerConfig, ini_path: Path) -> None: terminal = Path(cfg.terminal_exe) if not terminal.exists(): raise FileNotFoundError(f"terminal64.exe not found: {terminal}") if cfg.close_terminal_before_run: close_existing_terminal(terminal) proc = subprocess.Popen([str(terminal), f"/config:{ini_path}"], cwd=str(terminal.parent)) try: proc.wait(timeout=cfg.timeout_seconds) except subprocess.TimeoutExpired: proc.terminate() raise RuntimeError(f"MT5 timeout: {ini_path.name}") if proc.returncode not in (0, None): raise RuntimeError(f"MT5 exited with code {proc.returncode}: {ini_path.name}") def close_existing_terminal(terminal: Path) -> None: escaped = str(terminal).replace("'", "''") command = ( "$target = '" + escaped + "'; " "Get-Process -Name terminal64 -ErrorAction SilentlyContinue | " "Where-Object { $_.Path -eq $target } | " "Stop-Process -Force" ) subprocess.run( ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", command], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) time.sleep(1.0) def read_rows(path: Path) -> list[dict[str, Any]]: if not path.exists(): raise FileNotFoundError(path) rows: list[dict[str, Any]] = [] for enc in ("utf-8", "utf-8-sig", "latin1"): try: with path.open("r", encoding=enc, newline="") as fh: rows = list(csv.DictReader(fh, delimiter=";")) break except Exception: rows = [] for row in rows: if "optimizationTag" not in row and "tag" in row: row["optimizationTag"] = row.get("tag") for k, v in list(row.items()): if k in {"tag", "optimizationTag", "symbol", "worst_asset", "best_asset"}: continue try: row[k] = float(str(v).replace(",", ".")) except Exception: pass return rows def top_rows(rows: list[dict[str, Any]], min_trades: int = 30) -> list[dict[str, Any]]: filtered = [ r for r in rows if as_float(r.get("profit")) > 0 and as_float(r.get("pf")) >= 1.01 and as_float(r.get("trades")) >= min_trades and as_float(r.get("ddRel"), 999) <= 50 and as_float(r.get("recovery")) > 0 ] return sorted(filtered, key=lambda r: as_float(r.get("score")), reverse=True)[:20] def param_key(row: dict[str, Any], subgroup: Subgroup) -> tuple[Any, ...]: values = [] for param in subgroup.ranges: value = row.get(param, "") number = as_float(value) if abs(number - round(number)) < 1e-9: values.append(int(round(number))) else: values.append(round(number, 6)) return tuple(values) def score_global_metrics(cfg: OptimizerConfig, metrics: dict[str, Any]) -> float: n = max(1, len(cfg.symbols)) total_profit = as_float(metrics.get("total_profit")) avg_pf = as_float(metrics.get("avg_pf")) median_pf = as_float(metrics.get("median_pf")) positive_assets = as_float(metrics.get("positive_assets")) pf_gt_105 = as_float(metrics.get("pf_gt_105")) concentration = as_float(metrics.get("profit_concentration_best_asset"), 1.0) max_dd = as_float(metrics.get("max_ddRel"), 999) total_trades = as_float(metrics.get("total_trades")) tradable_assets = as_float(metrics.get("tradable_assets")) low_trade_assets = as_float(metrics.get("low_trade_assets")) avg_recovery = as_float(metrics.get("avg_recovery")) profit_score = max(total_profit, 0.0) breadth_score = (positive_assets / n) ** 2.0 pf_breadth_score = (pf_gt_105 / n) ** 1.4 median_pf_score = max(0.0, min(1.8, median_pf)) ** 1.8 avg_pf_score = max(0.0, min(1.8, avg_pf)) concentration_penalty = max(0.08, 1.0 - max(0.0, concentration - 0.40) * 1.7) dd_penalty = 1.0 / (1.0 + (max_dd / 10.0) ** 2) trade_score = min(1.0, total_trades / (n * 300.0)) trade_breadth_score = (tradable_assets / n) ** 1.5 low_trade_penalty = 0.65 ** low_trade_assets recovery_score = max(0.25, min(1.35, avg_recovery / 3.0)) return profit_score * breadth_score * pf_breadth_score * median_pf_score * avg_pf_score * concentration_penalty * dd_penalty * trade_score * trade_breadth_score * low_trade_penalty * recovery_score def aggregate_metric_rows(cfg: OptimizerConfig, by_symbol: dict[str, dict[str, Any]]) -> dict[str, Any]: symbols = cfg.symbols profits = [as_float(by_symbol.get(s, {}).get("profit")) for s in symbols] pfs = [as_float(by_symbol.get(s, {}).get("pf")) for s in symbols] trades = [as_float(by_symbol.get(s, {}).get("trades")) for s in symbols] dds = [as_float(by_symbol.get(s, {}).get("ddRel"), 0) for s in symbols] payoffs = [as_float(by_symbol.get(s, {}).get("payoff")) for s in symbols] recoveries = [as_float(by_symbol.get(s, {}).get("recovery")) for s in symbols] total_profit = sum(profits) positive_assets = sum(1 for p in profits if p > 0) pf_gt_105 = sum(1 for pf in pfs if pf > 1.05) median_pf = median(pfs) if pfs else 0 avg_pf = sum(pfs) / len(pfs) if pfs else 0 best_symbol = max(symbols, key=lambda s: as_float(by_symbol.get(s, {}).get("profit"))) worst_symbol = min(symbols, key=lambda s: as_float(by_symbol.get(s, {}).get("profit"))) positive_profit = sum(p for p in profits if p > 0) concentration = as_float(by_symbol.get(best_symbol, {}).get("profit")) / positive_profit if positive_profit > 0 else 1.0 metrics = { "total_profit": total_profit, "avg_pf": avg_pf, "median_pf": median_pf, "positive_assets": positive_assets, "pf_gt_105": pf_gt_105, "best_asset": best_symbol, "worst_asset": worst_symbol, "best_profit": as_float(by_symbol.get(best_symbol, {}).get("profit")), "worst_profit": as_float(by_symbol.get(worst_symbol, {}).get("profit")), "profit_concentration_best_asset": concentration, "max_ddRel": max(dds) if dds else 999, "total_trades": sum(trades), "tradable_assets": sum(1 for t in trades if t >= 30), "low_trade_assets": sum(1 for t in trades if t < 30), "avg_payoff": sum(payoffs) / len(payoffs) if payoffs else 0, "avg_recovery": sum(recoveries) / len(recoveries) if recoveries else 0, } metrics["global_score"] = score_global_metrics(cfg, metrics) return metrics def aggregate_global_rows(cfg: OptimizerConfig, symbol_rows: dict[str, list[dict[str, Any]]], subgroup: Subgroup) -> list[dict[str, Any]]: grouped: dict[tuple[Any, ...], dict[str, dict[str, Any]]] = {} for symbol, rows in symbol_rows.items(): for row in rows: grouped.setdefault(param_key(row, subgroup), {})[symbol] = row aggregates = [] for key, by_symbol in grouped.items(): metrics = aggregate_metric_rows(cfg, by_symbol) if as_float(metrics["total_trades"]) <= 0: continue row = dict(metrics) for i, param in enumerate(subgroup.ranges): row[param] = key[i] for symbol in cfg.symbols: src = by_symbol.get(symbol, {}) row[f"{symbol}_profit"] = as_float(src.get("profit")) row[f"{symbol}_pf"] = as_float(src.get("pf")) row[f"{symbol}_trades"] = as_float(src.get("trades")) aggregates.append(row) return sorted(aggregates, key=lambda r: as_float(r.get("global_score")), reverse=True) def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: if not rows: return path.parent.mkdir(parents=True, exist_ok=True) fields = list(rows[0].keys()) with path.open("w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fields) writer.writeheader() writer.writerows(rows) def collect_symbol_rows(cfg: OptimizerConfig, subgroup: Subgroup, params: dict[str, Any], baseline: bool = False) -> dict[str, list[dict[str, Any]]]: p = paths(cfg) p["files_dir"].mkdir(parents=True, exist_ok=True) p["run_dir"].mkdir(parents=True, exist_ok=True) passes = 1 if baseline else estimate_passes(subgroup.ranges) genetic = False if baseline else passes > cfg.genetic_threshold_passes rows_by_symbol: dict[str, list[dict[str, Any]]] = {} for symbol in cfg.symbols: prefix = "BASE" if baseline else "MSG" tag = f"{prefix}_{subgroup.tag}_{symbol}" cache_all = p["run_dir"] / f"{tag}_all.csv" if cache_all.exists(): rows_by_symbol[symbol] = read_rows(cache_all) continue csv_path = p["files_dir"] / f"{cfg.csv_prefix}_{tag}_frames.csv" if csv_path.exists(): csv_path.unlink() clean_cache(cfg, symbol) set_path = write_set(cfg, tag, params, subgroup, baseline=baseline) ini_path = write_ini(cfg, symbol, tag, set_path, genetic=genetic) mode = "baseline" if baseline else ("genetic" if genetic else "complete") print( f" {symbol}: {mode}, estimated passes={passes}, " f"threshold={cfg.genetic_threshold_passes}, ini={ini_path}", flush=True, ) run_mt5(cfg, ini_path) if not csv_path.exists(): raise FileNotFoundError( f"CSV not found: {csv_path}\n" f"Check whether the EA has csvExportPrefix='{cfg.csv_prefix}', " f"optimizationTag, saveOptimizationPasses and Frames-based export " f"with a file following the pattern csvExportPrefix + '_' + optimizationTag + '_frames.csv'." ) rows = read_rows(csv_path) write_csv(cache_all, rows) write_csv(p["run_dir"] / f"{tag}_top20.csv", top_rows(rows)) rows_by_symbol[symbol] = rows return rows_by_symbol def choose_cluster(cfg: OptimizerConfig, symbol_rows: dict[str, list[dict[str, Any]]], subgroup: Subgroup) -> tuple[dict[str, Any] | None, dict[str, Any], list[dict[str, Any]]]: p = paths(cfg) global_rows = aggregate_global_rows(cfg, symbol_rows, subgroup) global_top = global_rows[: max(1, cfg.top_sets_to_show)] write_csv(p["run_dir"] / f"{subgroup.tag}_global_all.csv", global_rows) write_csv(p["run_dir"] / f"{subgroup.tag}_global_top{cfg.top_sets_to_show}.csv", global_top) best = global_top[0] if global_top else None stats = { "tag": subgroup.tag, "title": subgroup.title, "group": subgroup.group, "passes_estimated": estimate_passes(subgroup.ranges), "global_top_count": len(global_top), "best_global": best or {}, "parameters": {}, "risk_flags": [], } if not best: return None, stats, global_top for param in subgroup.ranges: values = [as_float(row.get(param)) for row in global_top if row.get(param) not in ("", None)] if not values: continue counts: dict[str, int] = {} for v in values: key = str(int(round(v))) if abs(v - round(v)) < 1e-9 else str(round(v, 6)) counts[key] = counts.get(key, 0) + 1 mode, mode_count = sorted(counts.items(), key=lambda item: (-item[1], item[0]))[0] recommended: Any = parse_value(mode) if mode_count < 4: recommended = median(values) if abs(as_float(recommended) - round(as_float(recommended))) < 1e-9: recommended = int(round(as_float(recommended))) stats["parameters"][param] = { "recommended": recommended, "median": median(values), "mode": mode, "mode_count": mode_count, "min": min(values), "max": max(values), "unique_count": len(counts), "sensitive": mode_count < 4, } stats["risk_flags"] = build_risk_flags(cfg, stats, global_top) return best, stats, global_top def build_risk_flags(cfg: OptimizerConfig, stats: dict[str, Any], global_top: list[dict[str, Any]]) -> list[str]: if not global_top: return ["no_global_top"] best = global_top[0] flags: list[str] = [] if as_float(best.get("profit_concentration_best_asset"), 1.0) > 0.55: flags.append("high_concentration_in_best_asset") if as_float(best.get("median_pf")) < 1.05: flags.append("low_median_pf") if as_float(best.get("max_ddRel")) > 20: flags.append("high_drawdown") if as_float(best.get("positive_assets")) < len(cfg.symbols): flags.append("not_all_assets_positive") if as_float(best.get("total_trades")) < len(cfg.symbols) * 100: flags.append("few_trades") sensitive = [p for p, info in stats.get("parameters", {}).items() if info.get("sensitive")] if sensitive: flags.append("sensitive_parameters:" + ",".join(sensitive[:5])) if len(global_top) >= 2: s1 = as_float(global_top[0].get("global_score")) s2 = as_float(global_top[1].get("global_score")) if s1 > 0 and (s1 - s2) / s1 < 0.03: flags.append("top1_too_close_to_top2") return flags def accept_cluster(cfg: OptimizerConfig, baseline: dict[str, Any], best: dict[str, Any] | None) -> tuple[bool, str]: if not best: return False, "no valid global cluster" base_score = as_float(baseline.get("global_score")) opt_score = as_float(best.get("global_score")) base_positive = as_float(baseline.get("positive_assets")) opt_positive = as_float(best.get("positive_assets")) base_pf = as_float(baseline.get("median_pf")) opt_pf = as_float(best.get("median_pf")) base_dd = max(0.01, as_float(baseline.get("max_ddRel"))) opt_dd = as_float(best.get("max_ddRel")) opt_profit = as_float(best.get("total_profit")) if opt_profit <= 0: return False, "aggregate profit is negative or zero" if opt_pf < 1.0: return False, "median PF below 1.0" if opt_positive < max(1, math.ceil(len(cfg.symbols) * 0.60)): return False, "too few positive assets" delta = (opt_score - base_score) / abs(base_score) if abs(base_score) > 1e-9 else 1.0 if delta >= 0.05 and opt_positive >= base_positive and opt_pf >= base_pf - 0.03 and opt_dd <= base_dd * 1.30: return True, f"score improved {delta:.1%} without worsening stability" if opt_positive > base_positive: return True, "structural improvement: more positive assets" if opt_pf >= base_pf + 0.05 and opt_profit > 0: return True, "structural improvement: higher median PF" return False, f"did not beat baseline: score delta {delta:.1%}" def decision_is_sensitive(cfg: OptimizerConfig, baseline: dict[str, Any], best: dict[str, Any] | None, stats: dict[str, Any], accepted: bool) -> bool: if cfg.selection_mode == "manual": return True if cfg.selection_mode == "auto": return False if not best or not accepted: return True flags = stats.get("risk_flags", []) base_score = as_float(baseline.get("global_score")) opt_score = as_float(best.get("global_score")) delta = (opt_score - base_score) / abs(base_score) if abs(base_score) > 1e-9 else 1.0 if flags: return True if delta < 0.15: return True return False def format_float(v: Any, nd: int = 2) -> str: return f"{as_float(v):.{nd}f}" def print_top_sets(cfg: OptimizerConfig, subgroup: Subgroup, baseline: dict[str, Any], global_top: list[dict[str, Any]], stats: dict[str, Any]) -> None: print("\n" + "=" * 95) print(f"Subgroup: {subgroup.tag} - {subgroup.title}") print("Baseline:") print( f" Score={format_float(baseline.get('global_score'), 4)} | " f"Profit={format_float(baseline.get('total_profit'))} | " f"PFmed={format_float(baseline.get('median_pf'), 3)} | " f"Positive assets={int(as_float(baseline.get('positive_assets')))}/{len(cfg.symbols)} | " f"Trades={int(as_float(baseline.get('total_trades')))} | " f"DDmax={format_float(baseline.get('max_ddRel'), 2)}%" ) if stats.get("risk_flags"): print("Warnings:", ", ".join(stats["risk_flags"])) print("\nTOP sets:") print("# Score Profit PFmed Positive assets Trades DDmax Conc. Best/Worst") print("-" * 95) for i, row in enumerate(global_top[: cfg.top_sets_to_show], 1): print( f"{i:<2} {as_float(row.get('global_score')):<11.2f} " f"{as_float(row.get('total_profit')):<10.2f} " f"{as_float(row.get('median_pf')):<6.3f} " f"{int(as_float(row.get('positive_assets'))):>2}/{len(cfg.symbols):<2} " f"{int(as_float(row.get('total_trades'))):<7} " f"{as_float(row.get('max_ddRel')):<6.2f} " f"{as_float(row.get('profit_concentration_best_asset')):<6.2f} " f"{row.get('best_asset')}/{row.get('worst_asset')}" ) print("\nTOP 1 parameters:") if global_top: for param in subgroup.ranges: print(f" {param} = {global_top[0].get(param)}") print("=" * 95) def row_to_params(params: dict[str, Any], subgroup: Subgroup, selected: dict[str, Any]) -> dict[str, Any]: out = dict(params) out.update(subgroup.fixed) for param in subgroup.ranges: if param in selected: value = selected[param] if abs(as_float(value) - round(as_float(value))) < 1e-9: out[param] = int(round(as_float(value))) else: out[param] = as_float(value) return out def update_params_from_stats(params: dict[str, Any], subgroup: Subgroup, stats: dict[str, Any]) -> dict[str, Any]: out = dict(params) out.update(subgroup.fixed) for param, info in stats.get("parameters", {}).items(): out[param] = info["recommended"] return out def choose_set_interactively( cfg: OptimizerConfig, subgroup: Subgroup, params: dict[str, Any], baseline: dict[str, Any], global_top: list[dict[str, Any]], stats: dict[str, Any], accepted_auto: bool, auto_reason: str, ) -> tuple[str, dict[str, Any], str, int | None]: if not cfg.allow_user_override: if accepted_auto: return "ACCEPT", update_params_from_stats(params, subgroup, stats), auto_reason, 1 return "KEEP_BASELINE", params, auto_reason, None sensitive = decision_is_sensitive(cfg, baseline, global_top[0] if global_top else None, stats, accepted_auto) if not sensitive: if accepted_auto: return "ACCEPT_AUTO", update_params_from_stats(params, subgroup, stats), auto_reason, 1 return "KEEP_BASELINE_AUTO", params, auto_reason, None print_top_sets(cfg, subgroup, baseline, global_top, stats) print("Choose:") print(" A = accept recommended") print(" B = keep baseline") print(f" 1-{min(len(global_top), cfg.top_sets_to_show)} = apply selected set") print(" P = skip subgroup without changes") print(" S = save and exit") print(" R = mark for manual boundary reprocessing later") choice = ask("Your choice", "A" if accepted_auto else "B").strip().upper() if choice == "S": raise KeyboardInterrupt("User chose to save and exit") if choice in {"B", "P"}: return "KEEP_BASELINE", params, "user kept baseline", None if choice == "R": return "REPROCESS_BOUNDARIES", params, "user requested boundary reprocessing", None if choice.isdigit(): idx = int(choice) if 1 <= idx <= len(global_top): selected = global_top[idx - 1] return "USER_SELECTED", row_to_params(params, subgroup, selected), f"user selected TOP {idx}", idx print("Choice outside the valid range; applying recommended set.") if accepted_auto: return "ACCEPT", update_params_from_stats(params, subgroup, stats), "user accepted recommended: " + auto_reason, 1 return "KEEP_BASELINE", params, "user accepted the recommendation to keep baseline", None def save_final_set(cfg: OptimizerConfig, params: dict[str, Any]) -> Path: p = paths(cfg) p["tester_dir"].mkdir(parents=True, exist_ok=True) final = p["tester_dir"] / f"{cfg.expert_name}_GLOBAL_FINAL.set" lines = [] for key, value in params.items(): if key == "optimizationTag": lines.append(f"{key}=global_final") else: lines.append(f"{key}={mt5_value(value)}||0||0||0||N") final.write_text("\n".join(lines) + "\n", encoding="ascii", errors="ignore") return final def write_report(cfg: OptimizerConfig, manifest: list[dict[str, Any]], params: dict[str, Any]) -> None: p = paths(cfg) final_set = save_final_set(cfg, params) lines = [ f"# Report - {cfg.expert_name} Global Optimizer", "", f"- EA: {cfg.expert_name}", f"- Symbols: {', '.join(cfg.symbols)}", f"- Period: {cfg.from_date} to {cfg.to_date}", f"- Timeframe: {cfg.period}", f"- Selection mode: {cfg.selection_mode}", f"- Final set: {final_set}", "", "## Executed sequence", ] for item in manifest: rank = item.get("selected_rank") rank_text = f" | rank={rank}" if rank else "" lines.append( f"- {item['tag']} | {item['decision']}{rank_text} | " f"baseline={item.get('baseline_score', 0):.4g} | " f"optimized={item.get('optimized_score', 0):.4g} | " f"reason={item.get('reason', '')}" ) lines += ["", "## Final defaults"] for key, value in params.items(): lines.append(f"- {key} = {value}") (p["run_dir"] / "GLOBAL_OPTIMIZER_REPORT.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def build_rewind_plan(params: dict[str, Any], original: list[Subgroup]) -> list[Subgroup]: rewinds: list[Subgroup] = [] for sg in original: new_ranges: dict[str, Range] = {} for param, (start, step, stop) in sg.ranges.items(): center = params.get(param, start) if isinstance(start, bool) or isinstance(stop, bool): new_ranges[param] = (False, 1, True) continue step_f = abs(as_float(step, 1.0)) or 1.0 width = max(step_f * 2.0, abs(as_float(stop) - as_float(start)) * 0.20) c = as_float(center) new_start = round(c - width, 6) new_stop = round(c + width, 6) if isinstance(start, int) and isinstance(stop, int): new_ranges[param] = (int(round(new_start)), int(round(step_f)), int(round(new_stop))) else: new_ranges[param] = (new_start, step, new_stop) rewinds.append(Subgroup(tag=f"R_{sg.tag}", title=f"Fine rewind - {sg.title}", group=sg.group, ranges=new_ranges, fixed=sg.fixed)) return rewinds def run_stage(cfg: OptimizerConfig, stage: str = "main", start_at: str | None = None, stop_after: str | None = None) -> None: p = paths(cfg) p["run_dir"].mkdir(parents=True, exist_ok=True) params_path = p["run_dir"] / "current_defaults.json" manifest_path = p["run_dir"] / "manifest.json" params = dict(cfg.base) if params_path.exists(): params.update(json.loads(params_path.read_text(encoding="utf-8"))) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest_path.exists() else [] subgroups = [subgroup_from_dict(x) for x in cfg.subgroups] if stage == "rewind": subgroups = build_rewind_plan(params, subgroups) if start_at: while subgroups and subgroups[0].tag != start_at: subgroups.pop(0) try: for subgroup in subgroups: print(f"\n[{time.strftime('%H:%M:%S')}] {subgroup.tag} - {subgroup.title}", flush=True) if not subgroup.ranges: params.update(subgroup.fixed) manifest_item = { "tag": subgroup.tag, "title": subgroup.title, "group": subgroup.group, "decision": "FIXED_APPLIED", "reason": "subgroup without optimizable parameters; fixed values applied without calling MT5", "selected_rank": None, "baseline_score": 0.0, "optimized_score": 0.0, "risk_flags": [], "updated_params": list(subgroup.fixed.keys()), } manifest = [x for x in manifest if x.get("tag") != subgroup.tag] + [manifest_item] params_path.write_text(json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8") manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") write_report(cfg, manifest, params) print(" decision: FIXED_APPLIED | subgroup without optimizable ranges", flush=True) if stop_after and subgroup.tag == stop_after: break continue baseline_rows = collect_symbol_rows(cfg, subgroup, params, baseline=True) baseline_by_symbol = {} for sym, rows in baseline_rows.items(): if rows: baseline_by_symbol[sym] = sorted(rows, key=lambda r: as_float(r.get("score")), reverse=True)[0] baseline_metrics = aggregate_metric_rows(cfg, baseline_by_symbol) opt_rows = collect_symbol_rows(cfg, subgroup, params, baseline=False) best, stats, global_top = choose_cluster(cfg, opt_rows, subgroup) accepted_auto, auto_reason = accept_cluster(cfg, baseline_metrics, best) decision, new_params, reason, selected_rank = choose_set_interactively( cfg, subgroup, params, baseline_metrics, global_top, stats, accepted_auto, auto_reason ) if decision in {"ACCEPT", "ACCEPT_AUTO", "USER_SELECTED"}: params = new_params manifest_item = { "tag": subgroup.tag, "title": subgroup.title, "group": subgroup.group, "decision": decision, "reason": reason, "selected_rank": selected_rank, "baseline_score": as_float(baseline_metrics.get("global_score")), "optimized_score": as_float((best or {}).get("global_score")), "risk_flags": stats.get("risk_flags", []), "updated_params": list(stats.get("parameters", {}).keys()) if decision in {"ACCEPT", "ACCEPT_AUTO", "USER_SELECTED"} else [], } manifest = [x for x in manifest if x.get("tag") != subgroup.tag] + [manifest_item] params_path.write_text(json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8") manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") (p["run_dir"] / f"{subgroup.tag}_stats.json").write_text(json.dumps(stats, indent=2, ensure_ascii=False), encoding="utf-8") write_report(cfg, manifest, params) print(f" decision: {decision} | {reason}", flush=True) if stop_after and subgroup.tag == stop_after: break except KeyboardInterrupt: print("\nExecution interrupted by the user. Saving current state...") params_path.write_text(json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8") manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") write_report(cfg, manifest, params) final_set = save_final_set(cfg, params) print(f"\nOK: stage finished. Final set: {final_set}") def main() -> None: parser = argparse.ArgumentParser(description="MT5 Global Optimizer Wizard V3") parser.add_argument("--wizard", action="store_true", help="Create interactive config.json") parser.add_argument("--config", type=str, help="Path to config.json") parser.add_argument("--stage", choices=["main", "rewind", "all"], default="main") parser.add_argument("--start-at", type=str, default=None) parser.add_argument("--stop-after", type=str, default=None) parser.add_argument("--parse-mq5", type=str, help="Only read @opt/@fixed comments from an .mq5 file and print JSON") args = parser.parse_args() if args.parse_mq5: mq5_path = Path(args.parse_mq5) base, subgroups = parse_mq5_annotations(mq5_path) csv_prefix = detect_hardcoded_csv_prefix(mq5_path) or detect_csv_prefix_from_base(base, mq5_path.stem) print(json.dumps({ "detected_csv_prefix": csv_prefix, "expected_csv_file": f"{csv_prefix}__frames.csv", "inferred_data_path": infer_data_path_from_mq5(mq5_path), "missing_required_inputs": validate_required_optimizer_inputs(base), "base": base, "subgroups": [asdict(x) for x in subgroups] }, indent=2, ensure_ascii=False)) return if args.wizard: wizard() return if not args.config: raise SystemExit("Use --wizard or provide --config file.json") cfg = load_config(Path(args.config)) if args.stage == "all": print("\n=== STAGE 1/2: MAIN ===") run_stage(cfg, stage="main", start_at=args.start_at, stop_after=args.stop_after) # The rewind stage should start after the full main stage. # Therefore, by default, start_at/stop_after are not reused in rewind. # To run a partial rewind, use --stage rewind --start-at R_TAG. if args.stop_after: print("\nWarning: --stop-after was used in main. Automatic rewind will be skipped to avoid refining an incomplete sequence.") return print("\n=== STAGE 2/2: REWIND ===") run_stage(cfg, stage="rewind") else: run_stage(cfg, stage=args.stage, start_at=args.start_at, stop_after=args.stop_after) if __name__ == "__main__": main()