From d5a8973c05c9ab1600aa6f41b6a038541a76c6a5 Mon Sep 17 00:00:00 2001 From: LEGSTECH Optimizer Date: Mon, 13 Apr 2026 19:55:16 +0000 Subject: [PATCH] =?UTF-8?q?feat(step1):=20Universal=20EA=20Schema=20Layer?= =?UTF-8?q?=20=E2=80=94=20replaces=20param=5Fmanifest.yaml?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NEW FILES: ea/__init__.py - EA module package ea/schema.py - ParameterDef + ParameterSchema (replaces YAML manifest) ea/set_parser.py - Parse ANY MT5 .set file → ParameterSchema ea/registry.py - EAProfile + EARegistry (ea_registry.yaml) ea_registry.yaml - LEGSTECH_EA_V2 pre-registered with automation overrides MODIFIED: mt5/ini_builder.py - Accepts ParameterSchema OR legacy manifest_path mt5/runner.py - Accepts optional EAProfile for EA-agnostic validation config.yaml - Removed hardcoded ea: block; added ea_registry path KEY DESIGN: - .set file IS the manifest: value|min|max|step auto-detects float/int/bool/enum/fixed - automation_overrides in EAProfile forces headless-safe values (InpShowPanel=0) - Phase A optimization INI: Optimization=2 (genetic), ranges from schema - Phase B single backtest: Optimization=0 (unchanged behavior) - LEGSTECH advanced mode: fully backwards compatible via legacy manifest fallback - runner.py still works for legacy callers (no EAProfile needed) TESTED: - 49/49 LEGSTECH params parsed correctly - InpShowPanel=0 automation override applied - INI output correct for both Phase A (Optimization=2) and Phase B (Optimization=0) - All imports pass --- config.yaml | 11 +- ea/__init__.py | 3 + ea/registry.py | 207 +++++++++++++++++++++++++++++++++ ea/schema.py | 231 +++++++++++++++++++++++++++++++++++++ ea/set_parser.py | 278 +++++++++++++++++++++++++++++++++++++++++++++ ea_registry.yaml | 35 ++++++ mt5/ini_builder.py | 134 +++++++++++++++------- mt5/runner.py | 53 +++++---- optimizer.db-shm | Bin 32768 -> 32768 bytes optimizer.db-wal | Bin 8272 -> 8272 bytes 10 files changed, 878 insertions(+), 74 deletions(-) create mode 100644 ea/__init__.py create mode 100644 ea/registry.py create mode 100644 ea/schema.py create mode 100644 ea/set_parser.py create mode 100644 ea_registry.yaml diff --git a/config.yaml b/config.yaml index dcdad8f..f41805f 100644 --- a/config.yaml +++ b/config.yaml @@ -1,14 +1,8 @@ # MT5 EA Strategy Optimizer — Master Configuration -# LEGSTECH_EA_V2 | XAUUSD | H1 +# EA identity (name, symbol, timeframe, .set file) is now managed +# in ea_registry.yaml via the EA Registry. See ea/registry.py. # ───────────────────────────────────────────── -ea: - name: "LEGSTECH_EA_V2" - file: "LEGSTECH_EA_V2" # .ex5 file name (no extension), in Experts/ - symbol: "XAUUSD" - timeframe: "H1" # Period code used in INI (H1 = 16385... see MT5 period codes) - mt5_period_code: 16385 # PERIOD_H1 numeric code for [Tester] Period - periods: train_start: "2022.01.01" train_end: "2023.12.31" @@ -101,3 +95,4 @@ paths: runs_dir: "runs" reports_dir: "reports" log_file: "optimizer.log" + ea_registry: "ea_registry.yaml" # EA profiles (symbol, .set path, mode) diff --git a/ea/__init__.py b/ea/__init__.py new file mode 100644 index 0000000..18a79ea --- /dev/null +++ b/ea/__init__.py @@ -0,0 +1,3 @@ +""" +ea/ — EA Profile management, .set file parsing, and parameter schema. +""" diff --git a/ea/registry.py b/ea/registry.py new file mode 100644 index 0000000..2d40d98 --- /dev/null +++ b/ea/registry.py @@ -0,0 +1,207 @@ +""" +ea/registry.py +EA Profile storage and retrieval. + +Profiles are stored in ea_registry.yaml (path from config.yaml paths.ea_registry). +The registry is the single source of truth for which EAs are registered +and how to find their .set files. +""" +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +import yaml +from loguru import logger + +from ea.schema import ParameterSchema +from ea.set_parser import SetParser + + +# ── EAProfile ───────────────────────────────────────────────────────────────── + +@dataclass +class EAProfile: + """Configuration for one EA registered in the optimizer.""" + name: str # Display name, e.g. "LEGSTECH_EA_V2" + ex5_file: str # MT5 Experts file name (without .ex5 extension) + set_template: str # Absolute path string to template .set file + symbol: str # e.g. "XAUUSD" + timeframe: str # e.g. "H1" + mode: str = "generic" # "generic" | "advanced" + registered_at: str = field( + default_factory=lambda: datetime.now(timezone.utc).isoformat() + ) + + # Params the user has chosen to optimize (param names → True/False). + # Empty dict means: use default_optimize logic in SetParser. + optimize_params: dict[str, bool] = field(default_factory=dict) + + # Automation overrides: param values that must be used during backtesting, + # regardless of what the .set template says. + # Example: {"InpShowPanel": 0, "InpTesterMode": 1} + automation_overrides: dict[str, object] = field(default_factory=dict) + + def __post_init__(self): + assert self.mode in ("generic", "advanced"), \ + f"EAProfile.mode must be 'generic' or 'advanced', got {self.mode!r}" + assert self.timeframe in ( + "M1","M5","M15","M30","H1","H4","D1","W1","MN" + ), f"Invalid timeframe: {self.timeframe}" + + @property + def set_template_path(self) -> Path: + return Path(self.set_template) + + def to_dict(self) -> dict: + return asdict(self) + + @classmethod + def from_dict(cls, d: dict) -> "EAProfile": + return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__}) + + +# ── EARegistry ──────────────────────────────────────────────────────────────── + +class EARegistry: + """ + Manages registered EA profiles, persisted in ea_registry.yaml. + + Usage: + reg = EARegistry("config.yaml") + profile = reg.get("LEGSTECH_EA_V2") + schema = reg.get_schema(profile) + """ + + def __init__(self, config_path: str | Path = "config.yaml"): + config_path = Path(config_path) + with open(config_path) as f: + cfg = yaml.safe_load(f) + + registry_rel = cfg.get("paths", {}).get("ea_registry", "ea_registry.yaml") + self._registry_path = config_path.parent / registry_rel + self._parser = SetParser() + self._profiles: dict[str, EAProfile] = {} + self._load() + + # ── Public API ──────────────────────────────────────────────────────────── + + def register(self, profile: EAProfile) -> None: + """Add or update an EA profile.""" + self._profiles[profile.name] = profile + self._save() + logger.info(f"EARegistry: registered {profile.name!r} (mode={profile.mode})") + + def get(self, name: str) -> EAProfile: + """Get a registered EA profile by name. Raises KeyError if not found.""" + if name not in self._profiles: + available = list(self._profiles.keys()) + raise KeyError( + f"EA {name!r} not registered. Available: {available}" + ) + return self._profiles[name] + + def list_all(self) -> list[EAProfile]: + """Return all registered profiles.""" + return list(self._profiles.values()) + + def remove(self, name: str) -> None: + """Unregister an EA.""" + self._profiles.pop(name, None) + self._save() + logger.info(f"EARegistry: removed {name!r}") + + def exists(self, name: str) -> bool: + return name in self._profiles + + def get_schema( + self, + profile: EAProfile, + apply_optimize_selection: bool = True, + ) -> ParameterSchema: + """ + Parse the EA's .set file and return a ParameterSchema. + Applies automation_overrides to ensure headless-safe defaults. + Applies optimize_params selection if present. + """ + if not profile.set_template_path.exists(): + raise FileNotFoundError( + f"Set template not found: {profile.set_template}\n" + f"Please update the path in EA Registry for {profile.name!r}." + ) + + schema = self._parser.parse( + path=profile.set_template_path, + ea_name=profile.name, + default_optimize=False, + ) + + # Apply automation overrides: force specific param values + # (e.g. InpShowPanel=0 so no GUI renders during headless backtests) + for pname, value in profile.automation_overrides.items(): + if pname in schema.parameters: + schema.parameters[pname].default = value + schema.parameters[pname].type = "fixed" + schema.parameters[pname].optimize = False + + if apply_optimize_selection and profile.optimize_params: + for pname, should_opt in profile.optimize_params.items(): + if pname in schema.parameters: + p = schema.parameters[pname] + if p.type != "fixed": + p.optimize = should_opt + elif not profile.optimize_params: + # No selection yet — mark all non-fixed as optimizable by default + for p in schema.parameters.values(): + if p.type != "fixed": + p.optimize = True + + return schema + + def update_optimize_params(self, name: str, optimize_params: dict[str, bool]) -> None: + """Update which parameters to optimize for a registered EA.""" + profile = self.get(name) + profile.optimize_params = optimize_params + self._save() + + # ── Internal ───────────────────────────────────────────────────────────── + + def _load(self) -> None: + """Load profiles from YAML. Creates the file if it doesn't exist.""" + if not self._registry_path.exists(): + logger.info(f"EARegistry: creating new registry at {self._registry_path}") + self._registry_path.write_text("profiles: []\n", encoding="utf-8") + return + + with open(self._registry_path, encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + + profiles_raw = data.get("profiles", []) + for item in profiles_raw: + try: + p = EAProfile.from_dict(item) + self._profiles[p.name] = p + except Exception as e: + logger.warning(f"EARegistry: skipped malformed profile {item}: {e}") + + logger.info(f"EARegistry: loaded {len(self._profiles)} profile(s) from {self._registry_path.name}") + + def _save(self) -> None: + """Persist all profiles to YAML.""" + data = {"profiles": [p.to_dict() for p in self._profiles.values()]} + with open(self._registry_path, "w", encoding="utf-8") as f: + yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False) + logger.debug(f"EARegistry: saved {len(self._profiles)} profile(s)") + + def verify_integrity(self) -> list[str]: + """ + Check that all registered EAs have accessible .set files. + Returns a list of error messages (empty = all OK). + """ + errors = [] + for name, profile in self._profiles.items(): + if not profile.set_template_path.exists(): + errors.append(f"{name}: .set file missing at {profile.set_template}") + return errors diff --git a/ea/schema.py b/ea/schema.py new file mode 100644 index 0000000..963b011 --- /dev/null +++ b/ea/schema.py @@ -0,0 +1,231 @@ +""" +ea/schema.py +ParameterDef and ParameterSchema — the universal parameter representation. +Replaces mutation/param_manifest.yaml entirely. +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + + +# ── Parameter types ─────────────────────────────────────────────────────────── + +PARAM_TYPES = {"float", "int", "bool", "enum", "fixed"} + + +@dataclass +class ParameterDef: + """ + One EA input parameter, as parsed from a .set file. + + type: + float — continuous, has decimal step + int — discrete integer range + bool — true/false toggle (min=0, max=1, step=1) + enum — small discrete set of integer values + fixed — never changed during optimization (min==max, or zeroed range) + """ + name: str + default: Any # Value at default (typed: float/int/bool) + type: str # float | int | bool | enum | fixed + min: Optional[float] = None + max: Optional[float] = None + step: Optional[float] = None + optimize: bool = False # User has selected this for optimization + enum_values: list = field(default_factory=list) # Populated for type==enum + + def __post_init__(self): + assert self.type in PARAM_TYPES, f"Unknown param type '{self.type}' for {self.name}" + if self.type == "enum" and not self.enum_values and self.min is not None: + # Auto-populate enum values from range + v = self.min + while v <= self.max + 1e-9: + self.enum_values.append(int(round(v))) + v += self.step + + @property + def range_label(self) -> str: + """Human-readable range string, e.g. '0.5 – 3.0 (step 0.5)'.""" + if self.type == "fixed": + return f"{self.default} (fixed)" + if self.type == "bool": + return "true / false" + if self.type == "enum": + return " | ".join(str(v) for v in self.enum_values) + return f"{self.min} – {self.max} (step {self.step})" + + def clamp(self, value: float) -> Any: + """Clamp a proposed value to valid range, return correctly typed value.""" + if self.type == "fixed": + return self.default + if self.type == "bool": + return bool(round(value)) + if self.min is not None: + value = max(float(self.min), min(float(self.max), float(value))) + if self.type == "int": + return int(round(value)) + if self.type == "enum": + # Snap to nearest enum value + return min(self.enum_values, key=lambda v: abs(v - value)) + # float — round to same decimal places as step + if self.step and self.step > 0: + decimals = len(str(self.step).rstrip("0").split(".")[-1]) if "." in str(self.step) else 0 + return round(value, decimals) + return value + + def step_up(self, current: Any) -> Optional[Any]: + """Return value one step above current, or None if already at max.""" + if self.type in ("fixed", "bool"): + return None + if self.type == "enum": + idx = self.enum_values.index(int(current)) if int(current) in self.enum_values else -1 + return self.enum_values[idx + 1] if idx < len(self.enum_values) - 1 else None + nxt = float(current) + float(self.step) + return self.clamp(nxt) if nxt <= self.max + 1e-9 else None + + def step_down(self, current: Any) -> Optional[Any]: + """Return value one step below current, or None if already at min.""" + if self.type in ("fixed", "bool"): + return None + if self.type == "enum": + idx = self.enum_values.index(int(current)) if int(current) in self.enum_values else -1 + return self.enum_values[idx - 1] if idx > 0 else None + nxt = float(current) - float(self.step) + return self.clamp(nxt) if nxt >= self.min - 1e-9 else None + + +# ── ParameterSchema ─────────────────────────────────────────────────────────── + +class ParameterSchema: + """ + Full parameter schema for one EA, derived from its .set file. + Replaces mutation/param_manifest.yaml. + """ + + def __init__(self, ea_name: str, source_set: Path, parameters: dict[str, ParameterDef]): + self.ea_name = ea_name + self.source_set = source_set + self.parameters = parameters # ordered dict: name → ParameterDef + + # ── Accessors ───────────────────────────────────────────────────────────── + + def optimizable(self) -> list[ParameterDef]: + """Parameters the user has selected to optimize.""" + return [p for p in self.parameters.values() if p.optimize] + + def fixed(self) -> list[ParameterDef]: + """Parameters that never change.""" + return [p for p in self.parameters.values() if not p.optimize] + + def all_params(self) -> list[ParameterDef]: + return list(self.parameters.values()) + + def get(self, name: str) -> Optional[ParameterDef]: + return self.parameters.get(name) + + def __len__(self) -> int: + return len(self.parameters) + + # ── Baseline / override helpers ─────────────────────────────────────────── + + def defaults(self) -> dict[str, Any]: + """All parameters at their default values.""" + return {name: p.default for name, p in self.parameters.items()} + + def with_overrides(self, overrides: dict[str, Any]) -> dict[str, Any]: + """ + Merge override values with defaults. + Overrides only apply to known parameters; unknown keys are dropped. + Values are clamped to valid range. + """ + result = self.defaults() + for name, value in overrides.items(): + if name in self.parameters: + result[name] = self.parameters[name].clamp(value) + return result + + # ── INI rendering ───────────────────────────────────────────────────────── + + def to_ini_inputs(self, params: dict[str, Any], optimize_mode: bool = False) -> str: + """ + Render [TesterInputs] block for an MT5 .ini file. + + optimize_mode=False → plain values (Phase B single backtest) + optimize_mode=True → value|min|max|step ranges (Phase A genetic search) + + Bool optimization ranges use 1/0 (not true/false) per MT5 spec. + """ + lines = [] + full = self.with_overrides(params) + + for name, p in self.parameters.items(): + value = full.get(name, p.default) + + if optimize_mode and p.optimize and p.type != "fixed": + # Write optimization range — booleans use 1/0 in range format + if p.type == "bool": + v = "1" if value else "0" + lines.append(f"{name}={v}|0|1|1") + else: + formatted = self._fmt(p, value) + lines.append( + f"{name}={formatted}|{self._fmt(p, p.min)}|{self._fmt(p, p.max)}|{self._fmt(p, p.step)}" + ) + else: + lines.append(f"{name}={self._fmt(p, value)}") + + return "\n".join(lines) + + def to_set_file(self, params: dict[str, Any], header_comment: str = "") -> str: + """ + Render a clean output .set file (no optimization ranges). + This is what gets downloaded by the user. + """ + lines = [] + if header_comment: + for line in header_comment.strip().splitlines(): + lines.append(f"; {line}") + lines.append("") + + full = self.with_overrides(params) + for name, p in self.parameters.items(): + value = full.get(name, p.default) + lines.append(f"{name}={self._fmt(p, value)}") + + return "\n".join(lines) + + # ── Internal formatting ─────────────────────────────────────────────────── + + @staticmethod + def _fmt(p: ParameterDef, value: Any) -> str: + """Format a value according to parameter type.""" + if value is None: + return str(p.default) + if p.type == "bool": + # Accept int (0/1) or bool + if isinstance(value, str): + return value.lower() + return "true" if value else "false" + if p.type in ("int", "enum"): + return str(int(round(float(value)))) + if p.type == "float": + step = p.step or 0.1 + decimals = 0 + if "." in str(step): + decimals = len(str(step).rstrip("0").split(".")[-1]) + return f"{float(value):.{decimals}f}" + # fixed or unknown + return str(value) + + # ── Summary ─────────────────────────────────────────────────────────────── + + def summary(self) -> str: + opt = self.optimizable() + return ( + f"ParameterSchema({self.ea_name}): " + f"{len(self.parameters)} params total, " + f"{len(opt)} optimizable" + ) diff --git a/ea/set_parser.py b/ea/set_parser.py new file mode 100644 index 0000000..d3a4b5a --- /dev/null +++ b/ea/set_parser.py @@ -0,0 +1,278 @@ +""" +ea/set_parser.py +Parse any MT5 .set file into a ParameterSchema. + +Handles both .set formats: + value|min|max|step (single pipe — most common) + value||min||max||step||Y/N (double pipe — some MT5 builds) + +Fixed detection: + min == max → type="fixed" (e.g. InpMagicNumber=202402|202402|202402|1) + min == 0 AND max == 0 → type="fixed" (zeroed range = "don't optimize") + No range at all → type="fixed" + +Type detection (non-fixed only): + min==0, max==1, step==1 → bool + "." in step string → float + max - min <= 8, step==1 → enum (small discrete integer set) + otherwise → int +""" +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Optional + +from loguru import logger + +from ea.schema import ParameterDef, ParameterSchema + + +# ── Tester section keys to skip (not EA inputs) ─────────────────────────────── + +_TESTER_KEYS = { + "expert", "symbol", "period", "optimization", "model", + "fromdate", "todate", "forwardmode", "report", "replacereport", + "shutdownterminal", "deposit", "currency", "leverage", + "optimizationmode", "forwarddate", "optimizationiterations", +} + +# EA params that should always be fixed even if they have a range +_FORCE_FIXED_PATTERNS = [ + "testermode", "testeri", "testerinit", "showpanel", + "magicnumber", "magic", +] + + +class SetParser: + """ + Parses a MT5 .set file into a ParameterSchema. + + Usage: + parser = SetParser() + schema = parser.parse( + path=Path("C:/MT5 Set files/LEGSTECH_EA_V2.set"), + ea_name="LEGSTECH_EA_V2", + default_optimize=False, # user chooses via UI + ) + """ + + def parse( + self, + path: Path, + ea_name: str, + default_optimize: bool = False, + force_optimize: Optional[set[str]] = None, + force_fixed: Optional[set[str]] = None, + ) -> ParameterSchema: + """ + Parse a .set file and return a ParameterSchema. + + Args: + path: Path to the .set file. + ea_name: Display name for the EA. + default_optimize: Whether to mark all optimizable params as optimize=True by default. + If False (default), the user selects via UI. + force_optimize: Set of param names that are always optimize=True regardless. + force_fixed: Set of param names that are always type="fixed". + """ + force_optimize = force_optimize or set() + force_fixed = force_fixed or set() + + path = Path(path) + if not path.exists(): + raise FileNotFoundError(f".set file not found: {path}") + + try: + text = path.read_text(encoding="utf-16") + except UnicodeError: + text = path.read_text(encoding="utf-8", errors="replace") + + parameters: dict[str, ParameterDef] = {} + current_section = "" + + for raw_line in text.splitlines(): + line = raw_line.strip() + if not line or line.startswith(";"): + continue + + # Section header + if line.startswith("[") and line.endswith("]"): + current_section = line[1:-1].lower() + continue + + # Skip lines without "=" + if "=" not in line: + continue + + name, _, rest = line.partition("=") + name = name.strip() + rest = rest.strip() + + # Skip tester-section metadata keys + if name.lower() in _TESTER_KEYS: + continue + + # Parse the value + optional range + param = self._parse_param(name, rest) + if param is None: + logger.debug(f"SetParser: skipped unrecognised line: {line!r}") + continue + + # Apply force-fixed overrides + if name in force_fixed or self._is_force_fixed(name): + param.type = "fixed" + param.optimize = False + elif name in force_optimize: + param.optimize = True + elif default_optimize and param.type != "fixed": + param.optimize = True + + parameters[name] = param + + if not parameters: + raise ValueError(f"No EA input parameters found in .set file: {path}") + + schema = ParameterSchema(ea_name=ea_name, source_set=path, parameters=parameters) + logger.info(f"SetParser: parsed {schema.summary()} from {path.name}") + return schema + + # ── Internal ───────────────────────────────────────────────────────────── + + def _parse_param(self, name: str, rest: str) -> Optional[ParameterDef]: + """ + Parse a single parameter line. + rest is everything after the first "=" on the line. + """ + # Normalise: double-pipe "||" → single "|" + rest = re.sub(r"\|\|", "|", rest) + + # Strip trailing Y/N optimize flag if present + yn_match = re.search(r"\|([YN])$", rest, re.IGNORECASE) + if yn_match: + rest = rest[:yn_match.start()] + + parts = [p.strip() for p in rest.split("|")] + + if len(parts) == 1: + # No range info → fixed + value = self._cast_value(parts[0]) + return ParameterDef( + name=name, default=value, type="fixed", + min=None, max=None, step=None, optimize=False, + ) + + if len(parts) < 4: + # Incomplete range — treat as fixed + value = self._cast_value(parts[0]) + return ParameterDef( + name=name, default=value, type="fixed", + min=None, max=None, step=None, optimize=False, + ) + + raw_val, raw_min, raw_max, raw_step = parts[0], parts[1], parts[2], parts[3] + + try: + default_f = float(raw_val) + min_f = float(raw_min) + max_f = float(raw_max) + step_f = float(raw_step) + except ValueError: + value = self._cast_value(raw_val) + return ParameterDef( + name=name, default=value, type="fixed", + min=None, max=None, step=None, optimize=False, + ) + + # Fixed detection + is_fixed = ( + abs(min_f - max_f) < 1e-9 # min == max + or (abs(min_f) < 1e-9 and abs(max_f) < 1e-9) # both zero (zeroed range) + ) + if is_fixed: + return ParameterDef( + name=name, + default=self._typed_default(raw_val, raw_step), + type="fixed", + min=min_f, max=max_f, step=step_f, + optimize=False, + ) + + # Type detection + ptype = self._detect_type(min_f, max_f, step_f, raw_step, raw_val) + default = self._typed_cast(ptype, default_f, raw_val) + + return ParameterDef( + name=name, + default=default, + type=ptype, + min=min_f, + max=max_f, + step=step_f, + optimize=False, # user sets this via UI; can be overridden by caller + ) + + @staticmethod + def _detect_type(min_f: float, max_f: float, step_f: float, + raw_step: str, raw_val: str) -> str: + """Infer parameter type from its range.""" + # Bool: exactly 0–1 with step 1 + if abs(min_f) < 1e-9 and abs(max_f - 1.0) < 1e-9 and abs(step_f - 1.0) < 1e-9: + return "bool" + + # Float: step has decimal component + if "." in raw_step and not raw_step.endswith(".0") and float(raw_step) % 1 != 0: + return "float" + # Also float if default value has meaningful decimal + if "." in raw_val and float(raw_val) % 1 != 0: + return "float" + + # Enum: small integer set (≤ 8 distinct values, step 1) + n_values = int(round((max_f - min_f) / step_f)) + 1 if step_f > 0 else 1 + if abs(step_f - 1.0) < 1e-9 and n_values <= 8: + return "enum" + + return "int" + + @staticmethod + def _typed_cast(ptype: str, value_f: float, raw: str) -> Any: + if ptype == "bool": + return value_f != 0 or raw.lower() in ("true", "1") + if ptype == "int": + return int(round(value_f)) + if ptype == "enum": + return int(round(value_f)) + return value_f # float + + @staticmethod + def _typed_default(raw: str, raw_step: str) -> Any: + """Cast a fixed-param value without range context.""" + lower = raw.lower() + if lower in ("true", "false"): + return lower == "true" + try: + f = float(raw) + # Return int if it's a whole number and step is integer-like + if "." not in raw_step or raw_step.endswith(".0"): + if f == int(f): + return int(f) + return f + except ValueError: + return raw + + @staticmethod + def _cast_value(raw: str) -> Any: + lower = raw.lower() + if lower in ("true", "false"): + return lower == "true" + try: + f = float(raw) + return int(f) if f == int(f) and "." not in raw else f + except ValueError: + return raw + + @staticmethod + def _is_force_fixed(name: str) -> bool: + """Return True for params that are always fixed regardless of their range.""" + lower = name.lower() + return any(pat in lower for pat in _FORCE_FIXED_PATTERNS) diff --git a/ea_registry.yaml b/ea_registry.yaml new file mode 100644 index 0000000..c4ac7b7 --- /dev/null +++ b/ea_registry.yaml @@ -0,0 +1,35 @@ +profiles: +- name: LEGSTECH_EA_V2 + ex5_file: LEGSTECH_EA_V2 + set_template: C:/Users/DELL/Desktop/MT5 Set files/LEGSTECH_EA_V2.set + symbol: XAUUSD + timeframe: H1 + mode: advanced + registered_at: '2026-04-13T19:52:30.824439+00:00' + optimize_params: + InpRiskPercent: true + InpMaxDailyLossPct: true + InpMaxTradesPerDay: true + InpRRRatio: true + InpUseTrailing: true + InpTrailStartPips: true + InpTrailStepPips: true + InpUseBreakeven: true + InpBEPips: true + InpBEBufferPips: true + InpUseSession: true + InpSessionStart: true + InpSessionEnd: true + InpMinScore: true + InpATRMultiplier: true + InpBotMode: false + InpRiskType: false + InpSLType: false + InpSLBuffer: false + InpFixedSLPips: false + InpUseSpreadGuard: true + InpMaxSpreadPips: true + automation_overrides: + InpShowPanel: 0 + InpTesterMode: 1 + InpTesterInitDeposit: 10000.0 diff --git a/mt5/ini_builder.py b/mt5/ini_builder.py index e3956cd..78cca25 100644 --- a/mt5/ini_builder.py +++ b/mt5/ini_builder.py @@ -1,15 +1,22 @@ """ mt5/ini_builder.py -Generates the MT5 strategy tester .ini file from a parameter dict + config. +Generates the MT5 strategy tester .ini file from a ParameterSchema + config. + +Accepts either: + - ParameterSchema object (new universal path) + - manifest_path YAML string (legacy path, still supported for transition) """ from __future__ import annotations import configparser from pathlib import Path -from typing import Any +from typing import Any, Optional, TYPE_CHECKING import yaml from loguru import logger +if TYPE_CHECKING: + from ea.schema import ParameterSchema + # MT5 period string names (used in [Tester] Period= field) TIMEFRAME_NAMES = { @@ -29,22 +36,37 @@ class IniBuilder: """ Builds MT5 strategy tester .ini files. - Usage: + New usage (universal, recommended): + builder = IniBuilder(config_path="config.yaml", schema=my_schema) + + Legacy usage (LEGSTECH transition): builder = IniBuilder(config_path="config.yaml", manifest_path="mutation/param_manifest.yaml") - ini_path = builder.build( - run_id="run_001", - params={"InpRiskPercent": 1.5, "InpUseTrailing": True, ...}, - period_start="2022.01.01", - period_end="2023.12.31", - output_dir=Path("runs/run_001"), - ) + + The EA identity (ex5_file, symbol, timeframe) comes from the EAProfile + passed to build(), or falls back to config["ea"] for backwards compat. """ - def __init__(self, config_path: str | Path, manifest_path: str | Path): + def __init__( + self, + config_path: str | Path, + manifest_path: Optional[str | Path] = None, + schema: Optional["ParameterSchema"] = None, + ): with open(config_path) as f: self.cfg = yaml.safe_load(f) - with open(manifest_path) as f: - self.manifest = yaml.safe_load(f)["parameters"] + + self._schema = schema + + # Legacy manifest support + self._manifest: dict = {} + if manifest_path is not None and schema is None: + with open(manifest_path) as f: + self._manifest = yaml.safe_load(f)["parameters"] + + def set_schema(self, schema: "ParameterSchema") -> None: + """Update the schema (useful when EA changes mid-session).""" + self._schema = schema + self._manifest = {} # ── Public ──────────────────────────────────────────────────────────────── @@ -56,10 +78,19 @@ class IniBuilder: period_end: str, output_dir: Path, phase: str = "explore", + ea_file: Optional[str] = None, + ea_symbol: Optional[str] = None, + ea_timeframe: Optional[str] = None, + optimize_mode: bool = False, ) -> Path: """ Write .ini to output_dir and return its path. - params: dict of EA input values (partial OK — missing params use manifest defaults) + + ea_file, ea_symbol, ea_timeframe: override config["ea"] values. + Pass these from EAProfile when using the universal path. + + optimize_mode=True: write Phase A optimization INI (Optimization=2). + optimize_mode=False: write Phase B single backtest INI (Optimization=0). """ output_dir.mkdir(parents=True, exist_ok=True) ini_path = output_dir / f"{run_id}.ini" @@ -67,8 +98,12 @@ class IniBuilder: report_dir.mkdir(parents=True, exist_ok=True) full_params = self._merge_with_defaults(params) - ini_text = self._render(run_id, full_params, period_start, period_end, - report_dir, phase) + ini_text = self._render( + run_id, full_params, period_start, period_end, + report_dir, phase, + ea_file=ea_file, ea_symbol=ea_symbol, ea_timeframe=ea_timeframe, + optimize_mode=optimize_mode, + ) ini_path.write_text(ini_text, encoding="utf-8") logger.debug(f"INI written: {ini_path}") @@ -76,34 +111,41 @@ class IniBuilder: def default_params(self) -> dict[str, Any]: """Return all EA parameters at their default values.""" + if self._schema is not None: + return self._schema.defaults() return self._merge_with_defaults({}) # ── Internal ────────────────────────────────────────────────────────────── def _merge_with_defaults(self, overrides: dict[str, Any]) -> dict[str, Any]: - """Merge caller-supplied overrides with manifest defaults.""" + """Merge caller-supplied overrides with defaults.""" + if self._schema is not None: + return self._schema.with_overrides(overrides) + # Legacy manifest path result: dict[str, Any] = {} - for name, spec in self.manifest.items(): - if name in overrides: - result[name] = overrides[name] - else: - result[name] = spec.get("default", 0) + manifest = self._manifest + for name, spec in manifest.items(): + result[name] = overrides.get(name, spec.get("default", 0)) return result def _format_value(self, name: str, value: Any) -> str: """Format a parameter value for the [TesterInputs] section.""" - spec = self.manifest.get(name, {}) + # Schema path + if self._schema is not None: + p = self._schema.get(name) + if p is not None: + return self._schema._fmt(p, value) + return str(value) + # Legacy manifest path + spec = self._manifest.get(name, {}) ptype = spec.get("type", "float") - if ptype == "bool": return "true" if value else "false" if ptype == "fixed": - # Fixed params: write exact default return str(value) - if ptype == "int" or ptype == "enum": + if ptype in ("int", "enum"): return str(int(value)) if ptype == "float": - # Determine decimal places from step step = spec.get("step", 0.1) decimals = len(str(step).split(".")[-1]) if "." in str(step) else 0 return f"{float(value):.{decimals}f}" @@ -117,30 +159,34 @@ class IniBuilder: period_end: str, report_dir: Path, phase: str, + ea_file: Optional[str] = None, + ea_symbol: Optional[str] = None, + ea_timeframe: Optional[str] = None, + optimize_mode: bool = False, ) -> str: """Render final INI content as a string.""" - ea_cfg = self.cfg["ea"] mt5_cfg = self.cfg["mt5"] broker_cfg = self.cfg["broker"] - # Period must be the string name (H1, M30 etc) — NOT the ENUM integer - tf_name = TIMEFRAME_NAMES.get(ea_cfg["timeframe"].upper(), "H1") + # EA identity: prefer explicit args, fall back to config["ea"] for legacy + ea_cfg = self.cfg.get("ea", {}) + _ea_file = ea_file or ea_cfg.get("file", "EA") + _symbol = ea_symbol or ea_cfg.get("symbol", "XAUUSD") + _tf_str = ea_timeframe or ea_cfg.get("timeframe", "H1") + tf_name = TIMEFRAME_NAMES.get(_tf_str.upper(), "H1") - # Model: 0=Every Tick (slow), 4=OHLC M1 (fast, reliable for ini-based launch) - model_code = mt5_cfg.get("tester_model", 4) - - # Report path must be RELATIVE to the MT5 terminal data folder - # MT5 appends its own base path. Use run_id as the report name. - report_name = f"Optimizer_{run_id}" + model_code = mt5_cfg.get("tester_model", 4) + report_name = f"Optimizer_{run_id}" + opt_value = 2 if optimize_mode else 0 # 2=genetic, 0=single backtest lines = [ f"; MT5 Optimizer INI — run_id={run_id} phase={phase}", f"", f"[Tester]", - f"Expert={ea_cfg['file']}", - f"Symbol={ea_cfg['symbol']}", + f"Expert={_ea_file}", + f"Symbol={_symbol}", f"Period={tf_name}", - f"Optimization=0", + f"Optimization={opt_value}", f"Model={model_code}", f"FromDate={period_start}", f"ToDate={period_end}", @@ -155,9 +201,13 @@ class IniBuilder: f"[TesterInputs]", ] - for name, value in params.items(): - formatted = self._format_value(name, value) - lines.append(f"{name}={formatted}") + # Use schema's INI renderer if available + if self._schema is not None: + lines.append(self._schema.to_ini_inputs(params, optimize_mode=optimize_mode)) + else: + for name, value in params.items(): + formatted = self._format_value(name, value) + lines.append(f"{name}={formatted}") lines.append("") # trailing newline return "\n".join(lines) diff --git a/mt5/runner.py b/mt5/runner.py index e802f58..782e4d8 100644 --- a/mt5/runner.py +++ b/mt5/runner.py @@ -15,13 +15,16 @@ import subprocess import time import psutil from pathlib import Path -from typing import Optional +from typing import Optional, TYPE_CHECKING import yaml from loguru import logger from data.models import RunResult +if TYPE_CHECKING: + from ea.registry import EAProfile + # ── Custom Exceptions ───────────────────────────────────────────────────────── @@ -68,15 +71,12 @@ class MT5Runner: ini_path: Path, report_dir: Path, log_csv_search_dir: Optional[Path] = None, + profile: Optional["EAProfile"] = None, ) -> RunResult: """ - Full execution pipeline with retry: - 1. Kill existing MT5 - 2. Validate environment - 3. Launch fresh MT5 - 4. Wait for data readiness - 5. Poll for report - 6. Auto-retry once on failure + Full execution pipeline with retry. + profile: EAProfile used for validation + TradeLog lookup. + Falls back to config['ea'] if not provided (legacy mode). """ report_dir.mkdir(parents=True, exist_ok=True) report_stem = f"Optimizer_{run_id}" @@ -94,7 +94,7 @@ class MT5Runner: self._clear_stale_reports(run_id) # ── Step 2: Pre-run validation ─────────────────────────── - self._validate(run_id) + self._validate(run_id, profile=profile) # ── Step 3: Launch fresh MT5 ───────────────────────────── proc = self._launch_mt5(run_id, ini_path) @@ -107,7 +107,9 @@ class MT5Runner: result = self._wait_for_report(run_id, proc, report_dir, report_stem) # ── Step 6: Find TradeLogger CSV ───────────────────────── - result.trade_log_csv = self._find_trade_log(run_id, log_csv_search_dir) + result.trade_log_csv = self._find_trade_log( + run_id, log_csv_search_dir, profile=profile + ) logger.success(f"[{run_id}] Run complete. Report: {result.report_xml}") return result @@ -151,35 +153,32 @@ class MT5Runner: # ── Step 2: Pre-run validation ──────────────────────────────────────────── - def _validate(self, run_id: str) -> None: + def _validate(self, run_id: str, profile: Optional["EAProfile"] = None) -> None: """Validate all required files and directories exist before launch.""" errors = [] - # Check terminal executable if not self.terminal_exe.exists(): errors.append(f"MT5 terminal not found: {self.terminal_exe}") - # Check EA compiled file - ea_name = self.cfg["ea"]["file"] + # EA file: use profile if provided, else fall back to config['ea'] + ea_name = profile.ex5_file if profile else self.cfg.get("ea", {}).get("file", "") + symbol = profile.symbol if profile else self.cfg.get("ea", {}).get("symbol", "") + ea_candidates = [ self.appdata_path / "MQL5" / "Experts" / f"{ea_name}.ex5", - self.appdata_path / "MQL5" / "Experts" / f"{ea_name}", + self.appdata_path / "MQL5" / "Experts" / ea_name, ] - ea_found = any(p.exists() for p in ea_candidates) - if not ea_found: + if not any(p.exists() for p in ea_candidates): errors.append( f"EA file not found: {ea_name}.ex5 — " f"ensure you compiled the EA in MetaEditor before running." ) - # Check appdata path if not self.appdata_path.exists(): errors.append(f"MT5 appdata folder not found: {self.appdata_path}") - # Symbol check (basic — just ensure it's set) - symbol = self.cfg["ea"].get("symbol", "") if not symbol: - errors.append("Symbol not configured in config.yaml ea.symbol") + errors.append("Symbol not configured (check EAProfile or config.yaml)") if errors: for e in errors: @@ -328,11 +327,17 @@ class MT5Runner: # ── TradeLogger CSV lookup ──────────────────────────────────────────────── def _find_trade_log( - self, run_id: str, search_dir: Optional[Path] + self, run_id: str, search_dir: Optional[Path], + profile: Optional["EAProfile"] = None, ) -> Optional[Path]: """Find the TradeLogger CSV written by the EA during the backtest.""" - ea_name = self.cfg["ea"]["file"] - symbol = self.cfg["ea"]["symbol"] + # Use profile if available, else fall back to config['ea'] + ea_cfg = self.cfg.get("ea", {}) + ea_name = profile.ex5_file if profile else ea_cfg.get("file", "") + symbol = profile.symbol if profile else ea_cfg.get("symbol", "") + + if not ea_name or not symbol: + return None candidates = [ self.mql5_files_dir / f"{ea_name}_{symbol}_TradeLog.csv", diff --git a/optimizer.db-shm b/optimizer.db-shm index 37127b657592c439c2125e208dcbd79be32a2262..7e24b22237724759d40e63c2415db89f52377d32 100644 GIT binary patch delta 95 zcmZo@U}|V!njj*h#K6Gt*-u7fN6PX~9+P!XHDxJZoIO`HTioNqGy>8S6Lc7rHcm7T Y-nc-Iar38;Qv#b$1Zy#FTp%S00KI@EZvX%Q delta 91 zcmZo@U}|V!njj)0%fP@;^vHa|uJ#A@yVgy<=KYRs=0B!Ce|9~;_mhD1!~`8inT->z UgE#L<)neTIDdd#E#sw}?0LrH<@&Et; diff --git a/optimizer.db-wal b/optimizer.db-wal index a1017bde6414b5b898bd59d0f51ecc781961f353..d2643b4deac03c9aeb0b493edda29b941d5337fa 100644 GIT binary patch delta 1083 zcmZ9KUuaWT9LE#aB(wCMlg>2B(#GDVb~ZD)-t*`FIhA#Nm{w7of()6CaC7guwxern zNLpk{XsFvB2HQBBG9Pp(=)+)A3gUx;2!ay^-5%!f#knyC+a8A79`-QUIUUik-`nT+ z`~J@F{J!U`O{`5coo2ihDgTg-M3NB|yG8k4>&@Y>FQm>K86EttB~yvce6>rQ+q9ay z+AWhvTaLh=!6xtz{1bcveuv%KL+v`4giW{#E`qrB4MaeP8SM@j10LL~org8}DSRK4 z;REoawhV4)!+%2#zJ=r9OYkLw~p>nMAf&$AAEjO_7$a<4WF+$kh zqQGW;NNmp^RzN5xj9|N z*UKxwB*56O1_P@PfEV3bkgrl`KVfQM{~wR91lKj0jKy|Dv@1P(Ha4E!7~8S3RE%X# z??pZNmqwPnzC@z&?yGHcMQ=9MWieu|uH-Ll$h4+SrD_$W>iE)R+x4faVs<>LAKc!P zA8H@9*nW-aI`R1?6Z^%^gW`U6H!_)3(hJ_g zu}Y2ma|^t7qA;doWDuFH@|n5C%GBap6@RWVTbBik5^>=9G+lnJ{M5F0o_y6O(-1%2 zn696w_>I~E-zA#^or>AMN|q4(pOZh`{^027U$-YdUYoykzk6IdxhqOVDZMSOBnU}A z4@v8Wy0B|7F{J0j&HhphIfl!mUnwE?7T3}+L*+f|b9HZ~GCw;bZ&)Ub8K$K)DN;6B P8M(UYjuCr|GUe01#E%)P delta 417 zcmccMaKWM8yq>LzLHCga1A_nq2<%!n`I`4TwwbG|HlF>HS-ux2#EvY!)c2QKpV}=Q zE})<^-y#P7=X`(pzwmwF=i;yD-^{m&-;iIN-=YyWMM9tVE(A6{z3aLUZ5Z=%N+(@ zBkp6IV(hC}?rhw6ibcHDTbf7KR<$uvl%2h_-QUDAHPtfNJY{l$gb0hFnW6FKU7`z_ zM5RD#nlyoG3LD*1l2Qy()65LvY9{ZJiJJUK;)$59B$8ReiAlz$#>o~YaCJblLK!FL z$tzAa6W5$PSz2LotGL$Wmq7X|khYgmn5-(HC9PB&TicW=fox-vk*Rrds*x$&B)E;< Y;z&07C#I#EnI~JOLDU(8T|b`*0M&YbmH+?%